From 2722a59335a0cc8dae31fc2a0f7058c19f9e0665 Mon Sep 17 00:00:00 2001 From: Backend Contributors Date: Fri, 16 Oct 2026 08:44:30 +0800 Subject: [PATCH] feat: persist reviewed chunks and index publication versions --- README.md | 7 +- backend/app/api/routes/documents.py | 7 +- backend/app/api/routes/indexes.py | 91 +++++ backend/app/api/routes/knowledge.py | 7 +- backend/app/api/routes/sources.py | 7 +- backend/app/main.py | 7 +- backend/app/models/__init__.py | 7 +- backend/app/models/indexing.py | 74 ++++ backend/app/schemas/indexing.py | 89 +++++ backend/app/services/indexing.py | 147 ++++++++ backend/app/services/retrieval.py | 10 +- .../migrations/versions/0005_chunk_indexes.py | 81 +++++ backend/tests/test_indexing.py | 320 ++++++++++++++++++ backend/tests/test_migrations.py | 29 +- backend/tests/test_runtime.py | 36 ++ docs/ARCHITECTURE.md | 11 +- docs/DECISIONS.md | 7 + docs/INDEX_API.md | 81 +++++ docs/TASKS.md | 9 + frontend/src/services/indexes.ts | 16 + 20 files changed, 1027 insertions(+), 16 deletions(-) create mode 100644 backend/app/api/routes/indexes.py create mode 100644 backend/app/models/indexing.py create mode 100644 backend/app/schemas/indexing.py create mode 100644 backend/app/services/indexing.py create mode 100644 backend/migrations/versions/0005_chunk_indexes.py create mode 100644 backend/tests/test_indexing.py create mode 100644 docs/INDEX_API.md create mode 100644 frontend/src/services/indexes.ts diff --git a/README.md b/README.md index 93fc0cc..3f0aedb 100644 --- a/README.md +++ b/README.md @@ -34,7 +34,7 @@ uvicorn app.main:app --reload --host 127.0.0.1 配置始终从 `backend/.env` 读取，环境变量优先。必须先迁移再启动；应用启动只检查数据库版本，不再执行 `create_all`。旧版 `knowledge_items` 会被非破坏性接管；首次迁移前仍须备份。迁移会保留原条目及导入快照，不推测缺失的采集时间。旧版 `verified` 条目因缺少完整审核记录，会转为 `pending`，原状态保存在导入日志中。 -认证迁移为 `0004_auth_sessions`。知识管理、来源、文档及检索历史现要求管理员登录；迁移不创建默认账号。运行 `python -m app.manage create-admin --username local_admin`，通过隐藏输入设置独立密码。默认 Cookie 要求 HTTPS；本机 HTTP 联调须在未入库的 `.env` 显式设置 `AUTH_COOKIE_SECURE=false`（仅 development/test 允许）。完整来源、CSRF、字段和权限说明见 [认证接口契约](docs/AUTH_API.md)。 +认证迁移为 `0004_auth_sessions`。知识管理、来源、文档及检索历史现要求管理员登录；迁移不创建默认账号。运行 `python -m app.manage create-admin --username local_admin`，通过隐藏输入设置独立密码。默认 Cookie 要求 HTTPS；本机 HTTP 联调须在未入库的 `.env` 显式设置 `AUTH_COOKIE_SECURE=false`（仅 development/test 允许）。完整来源、CSRF、字段和权限说明见 [认证接口契约](docs/AUTH_API.md)。当前数据库最新迁移为 `0005_chunk_indexes`，索引构建/发布说明见 [切分与索引版本接口](docs/INDEX_API.md)。 ## 启动前端（另开终端） ```powershell @@ -57,6 +57,7 @@ Vite 通常使用 http://localhost:5173。 - GET/POST `/api/documents`、GET/PATCH/DELETE `/api/documents/{id}`、GET `/api/documents/{id}/history` - POST `/api/retrieval`：已核验知识的关键词检索 - GET `/api/retrieval/runs`、GET `/api/retrieval/runs/{run_id}`：检索历史和原始结果 +- `/api/indexes`：管理员构建/查询切分快照、发布/查询当前版本、查询 chunks 和发布历史，详见 [接口契约](docs/INDEX_API.md) ### 阶段 2 接口约定 @@ -96,7 +97,7 @@ Vite 通常使用 http://localhost:5173。 - 仅使用 `verified` 且未删除、采集/审核字段齐全的知识。关联来源/文档须仍有效，来源网址须与登记来源一致；`pending/rejected` 不参与分词、IDF 统计或结果排名。 - 使用固定 `jieba==0.42.1` 搜索分词，关闭 HMM，加入项目词“那拉提”，采用字符级 NFKC 和大小写折叠、小型固定停用词表。标题词频权重 2，正文权重 1；BM25 参数 `k1=1.2/b=0.75`，IDF 为 `ln(1+(N-df+0.5)/(df+0.5))`。重复查询词计一次，同分按知识 ID 升序。 - 返回 `run_id/status/query_terms/corpus_size/corpus_fingerprint/total_hits/results/elapsed_ms/config` 等字段。命中包含知识及关联文档/来源的 ID 与版本、标题、来源网址、分数、命中词和最长 240 字符的原文片段；`snippet_start/end` 是原文字符偏移，结束位置不包含在片段内。分数仅用于排序，不是事实正确率或置信度。独立网址的旧条目可没有文档 ID 或来源名称。 -- 检索单元是已审核知识条目；显示片段不等同于已经建立 `chunks` 或向量索引。没有匹配证据时返回 `status=no_evidence` 和空结果，当前不会调用模型生成答案。 +- 关键词基线的检索单元仍是已审核知识条目；该接口的显示片段独立于新增的持久化 chunks，尚无向量索引。没有匹配证据时返回 `status=no_evidence` 和空结果，当前不会调用模型生成答案。 - 每次成功检索（含零命中）保存不可变结果快照、实际检索耗时、语料指纹和算法参数。`elapsed_ms` 包含读取、分词、排序和片段生成，不包含记录写入和网络时间。历史记录用于复核，不能当作仍然有效的当前知识；依据变更后的新检索会重新应用审核条件。 - `GET /api/retrieval/runs` 使用标准分页参数；`GET /api/retrieval/runs/{run_id}` 读取当时结果。当前知识修改、失效或删除不会改写历史结果。检索记录保存在数据库，运行日志和实际查询不要提交到 Git。 - 本基线按请求计算语料词频，面向小规模资料集：筛选后的语料最多 5000 条、标题与正文合计最多 200 万字符。超过容量明确返回 503，不截断语料或伪造空结果；可缩小类别范围，大规模索引留到后续实现。 @@ -118,7 +119,7 @@ pytest -q 测试需要真实 PostgreSQL + pgvector：设置 `TEST_DATABASE_URL`（`postgresql+psycopg://...`），或另行 `pip install pgserver` 使用可选的本机临时实例。`pgserver` 仅用于测试，不是应用运行依赖；外部测试数据库账号需有建 Schema 权限且已启用 vector 扩展。每个测试使用独立 Schema，不会清空业务表。测试使用的占位正文和审核人不代表真实资料。 -迁移命令为 `alembic upgrade head`，可重复执行。仅撤回认证时，备份后执行 `alembic downgrade 0003_retrieval_runs`，删除用户/会话/预算表，保留知识和检索历史。继续执行 `alembic downgrade 0002_catalog` 还会删除检索历史。需要撤回阶段 2 时，先备份，再执行 `alembic downgrade 0001_baseline`；它保留基础知识字段，但会删除新增来源、文档、审核字段与所有历史记录，软删除的基础条目也会重新可见。降级须配合对应旧版后端，当前应用会拒绝旧 Schema。恢复完整数据应使用备份。不要在有业务数据的数据库执行 `downgrade base`。 +迁移命令为 `alembic upgrade head`，可重复执行。仅撤回切分/索引版本时，备份后执行 `alembic downgrade 0004_auth_sessions`，保留账号、会话、知识和检索记录。继续撤回认证时，备份后执行 `alembic downgrade 0003_retrieval_runs`，删除用户/会话/预算表，保留知识和检索历史。继续执行 `alembic downgrade 0002_catalog` 还会删除检索历史。需要撤回阶段 2 时，先备份，再执行 `alembic downgrade 0001_baseline`；它保留基础知识字段，但会删除新增来源、文档、审核字段与所有历史记录，软删除的基础条目也会重新可见。降级须配合对应旧版后端，当前应用会拒绝旧 Schema。恢复完整数据应使用备份。不要在有业务数据的数据库执行 `downgrade base`。 ## 团队协作文档 diff --git a/backend/app/api/routes/documents.py b/backend/app/api/routes/documents.py index 66c5b95..2be7898 100644 --- a/backend/app/api/routes/documents.py +++ b/backend/app/api/routes/documents.py @@ -15,8 +15,13 @@ from app.services.catalog import ( record_event, update_values, ) +from app.services.indexing import catalog_write_guard -router = APIRouter(prefix="/documents", tags=["documents"], dependencies=[Depends(require_admin)]) +router = APIRouter( + prefix="/documents", + tags=["documents"], + dependencies=[Depends(require_admin), Depends(catalog_write_guard)], +) @router.get("", response_model=Page[DocumentRead]) diff --git a/backend/app/api/routes/indexes.py b/backend/app/api/routes/indexes.py new file mode 100644 index 0000000..c1c08d9 --- /dev/null +++ b/backend/app/api/routes/indexes.py @@ -0,0 +1,91 @@ +from uuid import UUID + +from fastapi import APIRouter, Depends, Query +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.db.session import get_db +from app.models.indexing import Chunk, IndexPublication, IndexVersion +from app.schemas.catalog import Page +from app.schemas.indexing import ( + ActiveIndex, + BuildIndex, + ChunkRead, + IndexRead, + PublicationRead, + PublishIndex, +) +from app.services.auth import CurrentSession, require_admin +from app.services.catalog import page_result +from app.services.indexing import active_index, build_index, publish_index, version_or_404 + +router = APIRouter(prefix="/indexes", tags=["indexes"], dependencies=[Depends(require_admin)]) + + +@router.post("", response_model=IndexRead, status_code=201) +def build( + payload: BuildIndex, + current: CurrentSession = Depends(require_admin), + db: Session = Depends(get_db), +): + return build_index(db, payload, current.user.id) + + +@router.get("", response_model=Page[IndexRead]) +def list_versions( + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + db: Session = Depends(get_db), +): + return page_result( + db, + select(IndexVersion).order_by(IndexVersion.created_at.desc(), IndexVersion.id), + page, + page_size, + ) + + +@router.get("/active", response_model=ActiveIndex) +def active(db: Session = Depends(get_db)): + return active_index(db) + + +@router.get("/publications", response_model=Page[PublicationRead]) +def publications( + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + db: Session = Depends(get_db), +): + return page_result( + db, select(IndexPublication).order_by(IndexPublication.generation.desc()), page, page_size + ) + + +@router.get("/{version_id}", response_model=IndexRead) +def read_version(version_id: UUID, db: Session = Depends(get_db)): + return version_or_404(db, str(version_id)) + + +@router.post("/{version_id}/publish", response_model=ActiveIndex) +def publish( + version_id: UUID, + payload: PublishIndex, + current: CurrentSession = Depends(require_admin), + db: Session = Depends(get_db), +): + return publish_index(db, version_id, payload.expected_generation, current.user.id) + + +@router.get("/{version_id}/chunks", response_model=Page[ChunkRead]) +def chunks( + version_id: UUID, + knowledge_id: int | None = Query(None, gt=0), + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + db: Session = Depends(get_db), +): + version_or_404(db, str(version_id)) + statement = select(Chunk).where(Chunk.index_version_id == str(version_id)) + if knowledge_id is not None: + statement = statement.where(Chunk.knowledge_id == knowledge_id) + return page_result(db, statement.order_by(Chunk.knowledge_id, Chunk.ordinal), page, page_size) diff --git a/backend/app/api/routes/knowledge.py b/backend/app/api/routes/knowledge.py index ea453df..8245687 100644 --- a/backend/app/api/routes/knowledge.py +++ b/backend/app/api/routes/knowledge.py @@ -18,8 +18,13 @@ from app.services.catalog import ( resolve_knowledge_links, update_values, ) +from app.services.indexing import catalog_write_guard -router = APIRouter(prefix="/knowledge", tags=["knowledge"], dependencies=[Depends(require_admin)]) +router = APIRouter( + prefix="/knowledge", + tags=["knowledge"], + dependencies=[Depends(require_admin), Depends(catalog_write_guard)], +) def filtered_query(q="", category=None, verification_status=None, source_id=None, document_id=None): diff --git a/backend/app/api/routes/sources.py b/backend/app/api/routes/sources.py index d7198f5..8dbe254 100644 --- a/backend/app/api/routes/sources.py +++ b/backend/app/api/routes/sources.py @@ -15,8 +15,13 @@ from app.services.catalog import ( record_event, update_values, ) +from app.services.indexing import catalog_write_guard -router = APIRouter(prefix="/sources", tags=["sources"], dependencies=[Depends(require_admin)]) +router = APIRouter( + prefix="/sources", + tags=["sources"], + dependencies=[Depends(require_admin), Depends(catalog_write_guard)], +) @router.get("", response_model=Page[SourceRead]) diff --git a/backend/app/main.py b/backend/app/main.py index 15d4f6a..7a5f2b7 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -8,7 +8,7 @@ from sqlalchemy import text from sqlalchemy.exc import IntegrityError, OperationalError, ProgrammingError from sqlalchemy.orm.exc import StaleDataError -from app.api.routes import auth, chat, documents, health, knowledge, retrieval, sources +from app.api.routes import auth, chat, documents, health, indexes, knowledge, retrieval, sources from app.core.config import settings from app.db.session import engine from app.services.auth import dummy_hash @@ -21,7 +21,7 @@ def create_app(db_engine=engine) -> FastAPI: try: with db_engine.connect() as connection: revision = connection.scalar(text("SELECT version_num FROM alembic_version")) - if revision != "0004_auth_sessions": + if revision != "0005_chunk_indexes": raise RuntimeError( "数据库版本不匹配，请先在 backend 目录执行 alembic upgrade head" ) @@ -36,7 +36,7 @@ def create_app(db_engine=engine) -> FastAPI: finally: db_engine.dispose() - app = FastAPI(title=settings.app_name, version="0.4.0", lifespan=lifespan) + app = FastAPI(title=settings.app_name, version="0.5.0", lifespan=lifespan) app.add_middleware( CORSMiddleware, allow_origins=settings.cors_origin_list, @@ -47,6 +47,7 @@ def create_app(db_engine=engine) -> FastAPI: for router in ( health.router, auth.router, + indexes.router, chat.router, knowledge.router, sources.router, diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index e8ed958..b215e2e 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -1,4 +1,6 @@ +from app.models.auth import AuthRateLimit, AuthSession, User from app.models.catalog import AuditEvent, Document, Source +from app.models.indexing import Chunk, IndexPublication, IndexState, IndexVersion from app.models.knowledge import KnowledgeItem from app.models.retrieval import RetrievalRun @@ -6,10 +8,13 @@ __all__ = [ "AuditEvent", "AuthRateLimit", "AuthSession", + "Chunk", "Document", + "IndexPublication", + "IndexState", + "IndexVersion", "KnowledgeItem", "RetrievalRun", "Source", "User", ] -from app.models.auth import AuthRateLimit, AuthSession, User diff --git a/backend/app/models/indexing.py b/backend/app/models/indexing.py new file mode 100644 index 0000000..fa012e0 --- /dev/null +++ b/backend/app/models/indexing.py @@ -0,0 +1,74 @@ +"""Immutable chunk snapshots and atomic publication metadata (no vector index yet).""" + +from datetime import datetime + +from sqlalchemy import ( + JSON, + CheckConstraint, + DateTime, + ForeignKey, + Integer, + String, + Text, + UniqueConstraint, + func, +) +from sqlalchemy.orm import Mapped, mapped_column + +from app.db.session import Base + + +class IndexVersion(Base): + __tablename__ = "index_versions" + __table_args__ = ( + CheckConstraint( + "status IN ('ready', 'published', 'superseded')", name="ck_index_version_status" + ), + ) + id: Mapped[str] = mapped_column(String(36), primary_key=True) + status: Mapped[str] = mapped_column(String(20), nullable=False, default="ready") + config: Mapped[dict] = mapped_column(JSON, nullable=False) + corpus_fingerprint: Mapped[str] = mapped_column(String(64), nullable=False) + catalog_revision: Mapped[int] = mapped_column(Integer, nullable=False) + item_count: Mapped[int] = mapped_column(Integer, nullable=False) + chunk_count: Mapped[int] = mapped_column(Integer, nullable=False) + created_by: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + published_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + + +class IndexState(Base): + __tablename__ = "index_state" + __table_args__ = (CheckConstraint("id = 1", name="ck_index_state_singleton"),) + id: Mapped[int] = mapped_column(primary_key=True) + active_version_id: Mapped[str | None] = mapped_column(ForeignKey("index_versions.id")) + generation: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + catalog_revision: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + + +class Chunk(Base): + __tablename__ = "chunks" + __table_args__ = ( + UniqueConstraint("index_version_id", "knowledge_id", "ordinal", name="uq_chunk_position"), + CheckConstraint("start_offset >= 0 AND end_offset > start_offset", name="ck_chunk_offsets"), + ) + id: Mapped[str] = mapped_column(String(36), primary_key=True) + index_version_id: Mapped[str] = mapped_column(ForeignKey("index_versions.id"), index=True) + knowledge_id: Mapped[int] = mapped_column(ForeignKey("knowledge_items.id"), nullable=False) + ordinal: Mapped[int] = mapped_column(Integer, nullable=False) + start_offset: Mapped[int] = mapped_column(Integer, nullable=False) + end_offset: Mapped[int] = mapped_column(Integer, nullable=False) + text: Mapped[str] = mapped_column(Text, nullable=False) + text_sha256: Mapped[str] = mapped_column(String(64), nullable=False) + provenance: Mapped[dict] = mapped_column(JSON, nullable=False) + + +class IndexPublication(Base): + __tablename__ = "index_publications" + generation: Mapped[int] = mapped_column(primary_key=True) + previous_version_id: Mapped[str | None] = mapped_column(ForeignKey("index_versions.id")) + version_id: Mapped[str] = mapped_column(ForeignKey("index_versions.id"), nullable=False) + actor_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False) + published_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) diff --git a/backend/app/schemas/indexing.py b/backend/app/schemas/indexing.py new file mode 100644 index 0000000..37edf5a --- /dev/null +++ b/backend/app/schemas/indexing.py @@ -0,0 +1,89 @@ +from datetime import datetime, timezone +from typing import Literal +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator + +from app.schemas.catalog import Input + + +class BuildIndex(Input): + chunk_size: int = Field(500, ge=100, le=2000) + overlap: int = Field(80, ge=0, le=1000) + + @model_validator(mode="after") + def bounded_overlap(self): + if self.overlap > self.chunk_size // 2: + raise ValueError("重叠不得超过切分长度的一半") + return self + + +class PublishIndex(Input): + expected_generation: int = Field(ge=0) + + +class IndexConfig(BuildIndex): + splitter: Literal["unicode-window-v1"] + + +class ChunkProvenance(BaseModel): + knowledge_id: int + knowledge_version: int + document_id: int | None + document_version: int | None + source_id: int | None + source_version: int | None + source_name: str | None + source_url: str + title: str + category: str + + +class TimestampRead(BaseModel): + model_config = ConfigDict(from_attributes=True) + + @field_validator("created_at", "published_at", check_fields=False) + @classmethod + def utc_timestamp(cls, value): + return value.astimezone(timezone.utc) if value is not None else None + + +class IndexRead(TimestampRead): + id: UUID + status: Literal["ready", "published", "superseded"] + config: IndexConfig + corpus_fingerprint: str + catalog_revision: int + item_count: int + chunk_count: int + created_by: int + created_at: datetime + published_at: datetime | None + + +class ActiveIndex(BaseModel): + generation: int + catalog_revision: int + is_current: bool + active_version: IndexRead | None + + +class ChunkRead(BaseModel): + model_config = ConfigDict(from_attributes=True) + id: UUID + index_version_id: UUID + knowledge_id: int + ordinal: int + start_offset: int + end_offset: int + text: str + text_sha256: str + provenance: ChunkProvenance + + +class PublicationRead(TimestampRead): + generation: int + previous_version_id: UUID | None + version_id: UUID + actor_id: int + published_at: datetime diff --git a/backend/app/services/indexing.py b/backend/app/services/indexing.py new file mode 100644 index 0000000..34d74ec --- /dev/null +++ b/backend/app/services/indexing.py @@ -0,0 +1,147 @@ +"""Bounded chunk snapshots and serialized, optimistic publication.""" + +import hashlib +from uuid import UUID, uuid4, uuid5 + +from fastapi import Depends, HTTPException, Request +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.db.session import get_db +from app.models.catalog import utc_now +from app.models.indexing import Chunk, IndexPublication, IndexState, IndexVersion +from app.schemas.indexing import ActiveIndex, BuildIndex, IndexRead +from app.services.retrieval import corpus_fingerprint, load_corpus + +MAX_CHUNKS = 20_000 + + +def locked_state(db: Session) -> IndexState: + state = db.scalar( + select(IndexState) + .where(IndexState.id == 1) + .with_for_update() + .execution_options(populate_existing=True) + ) + if state is None: + raise HTTPException(503, "索引状态尚未初始化，请检查数据库迁移") + return state + + +def catalog_write_guard(request: Request, db: Session = Depends(get_db)): + if request.method in {"POST", "PATCH", "PUT", "DELETE"}: + # Always before catalog/parent row locks; failures roll back this revision as well. + state = locked_state(db) + state.catalog_revision += 1 + + +def version_or_404(db: Session, version_id: str) -> IndexVersion: + version = db.get(IndexVersion, version_id) + if version is None: + raise HTTPException(404, "索引版本不存在") + return version + + +def split_text(content: str, size: int, overlap: int): + """Offsets are original Python Unicode character offsets, end-exclusive.""" + for start in range(0, len(content), size - overlap): + end = min(len(content), start + size) + yield start, end, content[start:end] + if end == len(content): + break + + +def build_index(db: Session, payload: BuildIndex, actor_id: int) -> IndexRead: + state = locked_state(db) + corpus = load_corpus(db, None) + version_id = uuid4() + chunks = [] + for item in corpus: + provenance = {key: value for key, value in item.items() if key != "content"} + for ordinal, (start, end, content) in enumerate( + split_text(item["content"], payload.chunk_size, payload.overlap) + ): + if len(chunks) >= MAX_CHUNKS: + raise HTTPException(503, "切分数量超过本阶段容量，请增加切分长度或减少重叠") + chunks.append( + Chunk( + id=str(uuid5(version_id, f"{item['knowledge_id']}:{ordinal}")), + index_version_id=str(version_id), + knowledge_id=item["knowledge_id"], + ordinal=ordinal, + start_offset=start, + end_offset=end, + text=content, + text_sha256=hashlib.sha256(content.encode()).hexdigest(), + provenance=provenance, + ) + ) + version = IndexVersion( + id=str(version_id), + status="ready", + config={"splitter": "unicode-window-v1", **payload.model_dump()}, + corpus_fingerprint=corpus_fingerprint(corpus), + catalog_revision=state.catalog_revision, + item_count=len(corpus), + chunk_count=len(chunks), + created_by=actor_id, + ) + db.add(version) + db.flush() + db.add_all(chunks) + result = IndexRead.model_validate(version) + db.commit() + return result + + +def active_index(db: Session, state: IndexState | None = None) -> ActiveIndex: + state = state or locked_state(db) + version = db.get(IndexVersion, state.active_version_id) if state.active_version_id else None + current = bool(version and version.catalog_revision == state.catalog_revision) + if current: + current = version.corpus_fingerprint == corpus_fingerprint(load_corpus(db, None)) + return ActiveIndex( + generation=state.generation, + catalog_revision=state.catalog_revision, + is_current=current, + active_version=IndexRead.model_validate(version) if version else None, + ) + + +def publish_index(db: Session, version_id: UUID, expected_generation: int, actor_id: int): + state = locked_state(db) + if state.generation != expected_generation: + raise HTTPException(409, "已发布版本发生变化，请刷新后重试") + version = version_or_404(db, str(version_id)) + if ( + version.catalog_revision != state.catalog_revision + or version.corpus_fingerprint != corpus_fingerprint(load_corpus(db, None)) + ): + raise HTTPException(409, "资料或审核依据已变化，请重新构建后发布") + if state.active_version_id == version.id: + return active_index(db, state) + previous = state.active_version_id + if previous: + db.get(IndexVersion, previous).status = "superseded" + state.active_version_id = version.id + state.generation += 1 + version.status = "published" + version.published_at = utc_now() + db.add( + IndexPublication( + generation=state.generation, + previous_version_id=previous, + version_id=version.id, + actor_id=actor_id, + published_at=version.published_at, + ) + ) + # Freeze the response before commit expires ORM objects and releases the lock. + result = ActiveIndex( + generation=expected_generation + 1, + catalog_revision=version.catalog_revision, + is_current=True, + active_version=IndexRead.model_validate(version), + ) + db.commit() + return result diff --git a/backend/app/services/retrieval.py b/backend/app/services/retrieval.py index a769dae..f7eb0f7 100644 --- a/backend/app/services/retrieval.py +++ b/backend/app/services/retrieval.py @@ -151,15 +151,19 @@ def load_corpus(db: Session, category: str | None) -> list[dict]: return corpus +def corpus_fingerprint(corpus: list[dict]) -> str: + return hashlib.sha256( + json.dumps(corpus, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + + def retrieve(db: Session, payload: RetrievalRequest) -> RetrievalResponse: started = perf_counter() terms = list(dict.fromkeys(tokenize(payload.query))) if not terms: raise HTTPException(422, "问题中没有可检索词，请补充具体关键词") corpus = load_corpus(db, payload.category) - fingerprint = hashlib.sha256( - json.dumps(corpus, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode() - ).hexdigest() + fingerprint = corpus_fingerprint(corpus) counters = [ Counter(tokenize(item["title"]) * CONFIG.title_weight + tokenize(item["content"])) for item in corpus diff --git a/backend/migrations/versions/0005_chunk_indexes.py b/backend/migrations/versions/0005_chunk_indexes.py new file mode 100644 index 0000000..0081340 --- /dev/null +++ b/backend/migrations/versions/0005_chunk_indexes.py @@ -0,0 +1,81 @@ +"""Persist reviewed chunks, version snapshots and atomic publication history.""" + +import sqlalchemy as sa +from alembic import op + +revision = "0005_chunk_indexes" +down_revision = "0004_auth_sessions" +branch_labels = None +depends_on = None + + +def upgrade(): + op.create_table( + "index_versions", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("status", sa.String(20), nullable=False), + sa.Column("config", sa.JSON(), nullable=False), + sa.Column("corpus_fingerprint", sa.String(64), nullable=False), + sa.Column("catalog_revision", sa.Integer(), nullable=False), + sa.Column("item_count", sa.Integer(), nullable=False), + sa.Column("chunk_count", sa.Integer(), nullable=False), + sa.Column("created_by", sa.Integer(), sa.ForeignKey("users.id"), nullable=False), + sa.Column( + "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False + ), + sa.Column("published_at", sa.DateTime(timezone=True)), + sa.CheckConstraint( + "status IN ('ready', 'published', 'superseded')", name="ck_index_version_status" + ), + ) + op.create_table( + "index_state", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("active_version_id", sa.String(36), sa.ForeignKey("index_versions.id")), + sa.Column("generation", sa.Integer(), nullable=False), + sa.Column("catalog_revision", sa.Integer(), nullable=False), + sa.CheckConstraint("id = 1", name="ck_index_state_singleton"), + ) + op.execute( + sa.text("INSERT INTO index_state (id, generation, catalog_revision) VALUES (1, 0, 0)") + ) + op.create_table( + "chunks", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column( + "index_version_id", sa.String(36), sa.ForeignKey("index_versions.id"), nullable=False + ), + sa.Column( + "knowledge_id", sa.Integer(), sa.ForeignKey("knowledge_items.id"), nullable=False + ), + sa.Column("ordinal", sa.Integer(), nullable=False), + sa.Column("start_offset", sa.Integer(), nullable=False), + sa.Column("end_offset", sa.Integer(), nullable=False), + sa.Column("text", sa.Text(), nullable=False), + sa.Column("text_sha256", sa.String(64), nullable=False), + sa.Column("provenance", sa.JSON(), nullable=False), + sa.UniqueConstraint( + "index_version_id", "knowledge_id", "ordinal", name="uq_chunk_position" + ), + sa.CheckConstraint( + "start_offset >= 0 AND end_offset > start_offset", name="ck_chunk_offsets" + ), + ) + op.create_index("ix_chunks_index_version_id", "chunks", ["index_version_id"]) + op.create_table( + "index_publications", + sa.Column("generation", sa.Integer(), primary_key=True), + sa.Column("previous_version_id", sa.String(36), sa.ForeignKey("index_versions.id")), + sa.Column("version_id", sa.String(36), sa.ForeignKey("index_versions.id"), nullable=False), + sa.Column("actor_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=False), + sa.Column( + "published_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False + ), + ) + + +def downgrade(): + op.drop_table("index_publications") + op.drop_table("chunks") + op.drop_table("index_state") + op.drop_table("index_versions") diff --git a/backend/tests/test_indexing.py b/backend/tests/test_indexing.py new file mode 100644 index 0000000..1484f12 --- /dev/null +++ b/backend/tests/test_indexing.py @@ -0,0 +1,320 @@ +"""Versioned chunks remain evidence snapshots; publication uses real PostgreSQL locks.""" + +import hashlib +from uuid import UUID, uuid4 + +import pytest +from fastapi import HTTPException +from sqlalchemy import func, select, text +from sqlalchemy.exc import OperationalError +from sqlalchemy.orm import Session +from test_retrieval import add_item + +from app.models.indexing import Chunk, IndexPublication, IndexState, IndexVersion +from app.services import indexing +from app.services.indexing import locked_state + + +def build(client, **payload): + response = client.post("/api/indexes", json=payload) + assert response.status_code == 201, response.text + return response.json() + + +def publish(client, version, generation=0): + return client.post( + f"/api/indexes/{version['id']}/publish", json={"expected_generation": generation} + ) + + +def chunks(client, version): + return client.get(f"/api/indexes/{version['id']}/chunks?page_size=100").json()["items"] + + +def test_build_preserves_original_offsets_content_hash_and_provenance(client): + content = ("Ａ字符\n那拉提🙂占位资料。" * 37).strip() + item = add_item(client, content=content) + first = build(client, chunk_size=100, overlap=20) + second = build(client, chunk_size=100, overlap=20) + assert first["status"] == "ready" and first["item_count"] == 1 + assert first["corpus_fingerprint"] == second["corpus_fingerprint"] + assert first["config"] == second["config"] and first["id"] != second["id"] + actual = chunks(client, first) + assert len(actual) == first["chunk_count"] + assert [c["ordinal"] for c in actual] == list(range(len(actual))) + assert actual[0]["start_offset"] == 0 and actual[-1]["end_offset"] == len(content) + restored = [""] * len(content) + for chunk in actual: + start, end = chunk["start_offset"], chunk["end_offset"] + assert chunk["text"] == content[start:end] + assert chunk["text_sha256"] == hashlib.sha256(chunk["text"].encode()).hexdigest() + assert chunk["provenance"]["knowledge_version"] == item["version"] + assert chunk["provenance"]["source_url"] == "https://example.com/" + restored[start:end] = list(chunk["text"]) + assert "".join(restored) == content + assert [(c["text"], c["start_offset"], c["end_offset"]) for c in actual] == [ + (c["text"], c["start_offset"], c["end_offset"]) for c in chunks(client, second) + ] + state = client.get("/api/indexes/active").json() + assert state["generation"] == 0 and state["active_version"] is None and not state["is_current"] + + +def test_only_valid_reviewed_evidence_is_chunked(client): + good = add_item(client) + add_item(client, status="pending") + add_item(client, status="rejected") + removed = add_item(client) + assert ( + client.delete( + f"/api/knowledge/{removed['id']}?expected_version={removed['version']}" + ).status_code + == 204 + ) + version = build(client) + assert version["item_count"] == 1 + assert {c["knowledge_id"] for c in chunks(client, version)} == {good["id"]} + + +def test_publication_is_atomic_optimistic_and_audited(client): + add_item(client) + first, second = build(client), build(client) + result = publish(client, first) + assert result.status_code == 200 and result.json()["generation"] == 1 + assert result.json()["is_current"] + assert publish(client, second).status_code == 409 + assert publish(client, second, 1).json()["generation"] == 2 + assert client.get(f"/api/indexes/{first['id']}").json()["status"] == "superseded" + assert client.get(f"/api/indexes/{second['id']}").json()["status"] == "published" + # Same active version with the current generation is an idempotent no-op. + assert publish(client, second, 2).json()["generation"] == 2 + history = client.get("/api/indexes/publications").json() + assert history["total"] == 2 + assert history["items"][0]["previous_version_id"] == first["id"] + assert history["items"][0]["version_id"] == second["id"] + assert client.get("/api/indexes/active").json()["active_version"]["id"] == second["id"] + + +def test_catalog_mutation_marks_active_stale_and_rejects_old_build(client): + item = add_item(client) + version = build(client) + assert publish(client, version).status_code == 200 + old = chunks(client, version) + updated = client.patch( + f"/api/knowledge/{item['id']}", + json={"expected_version": item["version"], "content": "修订的占位正文"}, + ) + assert updated.status_code == 200 + active = client.get("/api/indexes/active").json() + assert not active["is_current"] and active["generation"] == 1 + assert active["active_version"]["id"] == version["id"] + assert chunks(client, version) == old + assert publish(client, version, 1).status_code == 409 + replacement = build(client) + assert replacement["item_count"] == 0 # revised body is pending, not approved evidence + assert publish(client, replacement, 1).json()["is_current"] + assert chunks(client, replacement) == [] + + +def test_failed_catalog_write_does_not_invalidate_current_version(client): + item = add_item(client) + version = build(client) + publish(client, version) + before = client.get("/api/indexes/active").json() + assert ( + client.patch( + f"/api/knowledge/{item['id']}", json={"expected_version": 999, "content": "占位"} + ).status_code + == 409 + ) + assert client.post("/api/knowledge", json={"title": "", "content": "占位"}).status_code == 422 + assert client.get("/api/indexes/active").json() == before + + +def test_source_and_document_mutations_preserve_old_provenance(client): + src = client.post( + "/api/sources", json={"name": "原始占位来源", "source_url": "https://example.com/"} + ).json() + doc = client.post( + "/api/documents", + json={ + "source_id": src["id"], + "title": "占位文档", + "content": "占位正文", + "collected_at": "2026-10-16T00:00:00Z", + }, + ).json() + add_item(client, document_id=doc["id"]) + version = build(client) + publish(client, version) + snapshot = chunks(client, version) + assert snapshot[0]["provenance"]["document_version"] == doc["version"] + assert snapshot[0]["provenance"]["source_name"] == src["name"] + assert ( + client.patch( + f"/api/documents/{doc['id']}", json={"expected_version": 1, "content": "文档新正文"} + ).status_code + == 200 + ) + assert not client.get("/api/indexes/active").json()["is_current"] + empty = build(client) + publish(client, empty, 1) + assert ( + client.patch( + f"/api/sources/{src['id']}", json={"expected_version": 1, "name": "新来源名"} + ).status_code + == 200 + ) + assert not client.get("/api/indexes/active").json()["is_current"] + assert chunks(client, version) == snapshot + assert publish(client, version, 2).status_code == 409 + + +def test_fingerprint_recheck_rejects_out_of_band_catalog_changes(client, migrated_database): + item = add_item(client) + version = build(client) + publish(client, version) + with migrated_database.begin() as c: + c.execute( + text("UPDATE knowledge_items SET content=:body WHERE id=:id"), + {"body": "直接修改的占位正文", "id": item["id"]}, + ) + assert not client.get("/api/indexes/active").json()["is_current"] + assert publish(client, version, 1).status_code == 409 + + +@pytest.mark.parametrize("failure", ["chunk_budget", "corpus_budget", "database_write"]) +def test_failed_build_keeps_previous_publication_and_leaves_no_partial_rows( + client, migrated_database, monkeypatch, failure +): + add_item(client, content="占位正文" * 100) + previous = build(client) + publish(client, previous) + if failure == "chunk_budget": + monkeypatch.setattr(indexing, "MAX_CHUNKS", 1) + response = client.post("/api/indexes", json={"chunk_size": 100, "overlap": 0}) + assert response.status_code == 503 + elif failure == "corpus_budget": + from app.services import retrieval + + monkeypatch.setattr(retrieval, "MAX_ITEMS", 0) + assert client.post("/api/indexes", json={}).status_code == 503 + monkeypatch.undo() + else: + # Real PostgreSQL constraint failure during chunk insert; the new version rolls back too. + with migrated_database.begin() as c: + c.execute( + text("ALTER TABLE chunks ADD CONSTRAINT test_reject_chunk CHECK (false) NOT VALID") + ) + assert client.post("/api/indexes", json={}).status_code == 409 + with Session(migrated_database) as db: + assert db.scalar(select(func.count()).select_from(IndexVersion)) == 1 + assert db.scalar(select(func.count()).select_from(Chunk)) == previous["chunk_count"] + assert db.get(IndexState, 1).active_version_id == previous["id"] + assert db.scalar(select(func.count()).select_from(IndexPublication)) == 1 + assert client.get("/api/indexes/active").json()["active_version"]["id"] == previous["id"] + + +@pytest.mark.parametrize( + "payload", + [ + {"chunk_size": 99}, + {"chunk_size": 2001}, + {"overlap": -1}, + {"chunk_size": 100, "overlap": 51}, + {"unexpected": 1}, + ], +) +def test_build_validation_has_no_side_effects(client, payload): + assert client.post("/api/indexes", json=payload).status_code == 422 + assert client.get("/api/indexes").json()["total"] == 0 + + +def test_index_management_permissions_csrf_and_pagination(client, anonymous_client): + # Same TestClient fixture is authenticated as admin here; first verify authorized validation. + version = build(client) + assert client.post("/api/indexes", json={}, headers={"X-CSRF-Token": ""}).status_code == 403 + assert client.get("/api/indexes?page_size=101").status_code == 422 + assert client.get(f"/api/indexes/{uuid4()}").status_code == 404 + assert client.get(f"/api/indexes/{uuid4()}/chunks").status_code == 404 + assert client.get("/api/indexes/invalid").status_code == 422 + assert client.get(f"/api/indexes/{version['id']}/chunks?page=2").json()["items"] == [] + client.cookies.clear() + for method, path in [ + ("GET", "/api/indexes"), + ("GET", "/api/indexes/active"), + ("GET", "/api/indexes/publications"), + ("POST", "/api/indexes"), + ("GET", f"/api/indexes/{version['id']}/chunks"), + ("POST", f"/api/indexes/{version['id']}/publish"), + ]: + assert client.request(method, path, json={}).status_code == 401 + assert ( + client.post( + "/api/auth/register", json={"username": "normal_user", "password": "test-password-123"} + ).status_code + == 201 + ) + auth = client.post( + "/api/auth/login", json={"username": "normal_user", "password": "test-password-123"} + ).json() + client.headers["X-CSRF-Token"] = auth["csrf_token"] + assert client.post("/api/indexes", json={}).status_code == 403 + assert client.get("/api/indexes/active").status_code == 403 + + +def test_publication_lock_serializes_competing_publishers(client, migrated_database): + add_item(client) + first, second = build(client), build(client) + actor = first["created_by"] + with Session(migrated_database) as owner, Session(migrated_database) as competitor: + locked_state(owner) + competitor.execute(text("SET LOCAL lock_timeout = '100ms'")) + with pytest.raises(OperationalError) as blocked: + indexing.publish_index(competitor, UUID(second["id"]), 0, actor) + assert blocked.value.orig.sqlstate == "55P03" + competitor.rollback() + assert indexing.publish_index(owner, UUID(first["id"]), 0, actor).generation == 1 + with pytest.raises(HTTPException) as stale: + indexing.publish_index(competitor, UUID(second["id"]), 0, actor) + assert stale.value.status_code == 409 + competitor.rollback() + assert client.get("/api/indexes/publications").json()["total"] == 1 + assert client.get("/api/indexes/active").json()["active_version"]["id"] == first["id"] + + +def test_catalog_writes_use_the_publication_lock(client, migrated_database): + # The guard must run before a catalog mutation, even for creation with no parent row. + with Session(migrated_database) as owner: + locked_state(owner) + from app.db.session import get_db + + factory = client.app.dependency_overrides[get_db] + + def short_timeout(): + for db in factory(): + db.execute(text("SET LOCAL lock_timeout = '100ms'")) + yield db + + client.app.dependency_overrides[get_db] = short_timeout + response = client.post("/api/knowledge", json={"title": "占位", "content": "占位"}) + assert response.status_code == 503 + client.app.dependency_overrides[get_db] = factory + owner.rollback() + assert client.get("/api/knowledge").json() == [] + + +def test_publication_failure_rolls_back_old_status_pointer_and_audit(client, migrated_database): + add_item(client) + first, second = build(client), build(client) + publish(client, first) + with migrated_database.begin() as c: + c.execute( + text( + "ALTER TABLE index_publications ADD CONSTRAINT test_reject_publication CHECK (false) NOT VALID" + ) + ) + assert publish(client, second, 1).status_code == 409 + assert client.get("/api/indexes/active").json()["active_version"]["id"] == first["id"] + assert client.get(f"/api/indexes/{first['id']}").json()["status"] == "published" + assert client.get(f"/api/indexes/{second['id']}").json()["status"] == "ready" + assert client.get("/api/indexes/publications").json()["total"] == 1 diff --git a/backend/tests/test_migrations.py b/backend/tests/test_migrations.py index 53e856a..82eede6 100644 --- a/backend/tests/test_migrations.py +++ b/backend/tests/test_migrations.py @@ -16,7 +16,7 @@ def test_startup_requires_migration(database): def test_startup_rejects_previous_schema(database): - migrate(database, "0003_retrieval_runs") + migrate(database, "0004_auth_sessions") with pytest.raises(RuntimeError, match="alembic upgrade head"): with TestClient(create_app(database)): pass @@ -42,7 +42,7 @@ def test_upgrade_preserves_original_rows_and_repeated_upgrade(database): assert connection.scalar(text("SELECT count(*) FROM audit_events")) == 2 assert ( connection.scalar(text("SELECT version_num FROM alembic_version")) - == "0004_auth_sessions" + == "0005_chunk_indexes" ) assert compare_metadata(MigrationContext.configure(connection), Base.metadata) == [] migrate(database, "0001_baseline", downgrade=True) @@ -73,3 +73,28 @@ def test_auth_upgrade_and_downgrade_preserve_knowledge_and_retrieval(database): assert c.scalar(text("SELECT content FROM knowledge_items")) == "保留的占位正文" assert c.scalar(text("SELECT count(*) FROM retrieval_runs")) == 1 migrate(database) + + +def test_chunk_index_downgrade_preserves_users_sessions_knowledge_and_retrieval( + client, migrated_database +): + from test_indexing import build, publish + from test_retrieval import add_item, search + + item = add_item(client) + run = search(client) + version = build(client) + assert publish(client, version).status_code == 200 + migrate(migrated_database, "0004_auth_sessions", downgrade=True) + with migrated_database.connect() as c: + assert c.scalar(text("SELECT count(*) FROM users")) == 1 + assert c.scalar(text("SELECT count(*) FROM auth_sessions")) == 1 + assert ( + c.scalar(text("SELECT content FROM knowledge_items WHERE id=:id"), {"id": item["id"]}) + == item["content"] + ) + assert c.scalar(text("SELECT id FROM retrieval_runs")) == run["run_id"] + migrate(migrated_database) + assert client.get("/api/auth/me").status_code == 200 + assert client.get("/api/indexes").json()["total"] == 0 + assert client.get("/api/indexes/active").json()["generation"] == 0 diff --git a/backend/tests/test_runtime.py b/backend/tests/test_runtime.py index af14b8a..6f51efe 100644 --- a/backend/tests/test_runtime.py +++ b/backend/tests/test_runtime.py @@ -154,3 +154,39 @@ def test_session_and_logout_survive_http_restart(migrated_database): with server(migrated_database, authenticate=False) as client: client.cookies.set(COOKIE_NAME, cookie) assert client.get("/api/auth/me").status_code == 401 + + +def test_published_index_and_chunks_survive_http_restart(migrated_database): + with server(migrated_database) as client: + item = client.post( + "/api/knowledge", + json={ + "title": "切分持久化占位", + "content": "那拉提测试正文。" * 50, + "source_url": "https://example.com/", + "collected_at": "2026-10-16T00:00:00Z", + }, + ).json() + assert ( + client.post( + f"/api/knowledge/{item['id']}/review", + json={ + "expected_version": item["version"], + "verification_status": "verified", + "reviewer": "test-reviewer", + "verification_notes": "仅为切分持久化测试", + }, + ).status_code + == 200 + ) + built = client.post("/api/indexes", json={"chunk_size": 100, "overlap": 20}).json() + response = client.post( + f"/api/indexes/{built['id']}/publish", json={"expected_generation": 0} + ) + assert response.status_code == 200 + active = response.json() + chunks = client.get(f"/api/indexes/{built['id']}/chunks").json() + with server(migrated_database) as client: + assert client.get("/api/indexes/active").json() == active + assert client.get(f"/api/indexes/{built['id']}/chunks").json() == chunks + assert client.get("/api/indexes/publications").json()["total"] == 1 diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index da47960..0b8b04d 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -37,4 +37,13 @@ FastAPI + Pydantic 新增 `retrieval_runs` 保存查询、模式、语料指纹和完整结果快照，支持分页读取及按 UUID 复核。零命中如实记录；当前依据修改后失效不会改写历史，但后续检索必须重新检查审核与父记录状态。响应耗时是本次真实读取、分词、排序及片段生成耗时，不含写日志或模型调用。 -当前按知识条目检索，显示片段最长 240 字符；尚无持久化 chunks、倒排索引、Embedding、向量检索、RRF 或 RAG。全语料计算明确限制条目数及总字符数，超限报错，适用于首批小规模资料与后续对照实验准备，不能当作生产检索性能结论。 +关键词基线仍按知识条目检索，显示片段最长 240 字符；独立的持久化 chunks 和文本索引版本见下方。当前尚无倒排索引、Embedding、向量检索、RRF 或 RAG。全语料计算明确限制条目数及总字符数，超限报错，适用于首批小规模资料与后续对照实验准备，不能当作生产检索性能结论。 + + +## 切分与索引版本（T05） + +`index_versions/chunks` 保存已审核语料的原文切分、出处与版本、参数、内容哈希及语料指纹。构建先保存 ready 快照，不切换当前发布；`index_state` 管理发布指针、代数与目录修订号，发布同一事务更新前后状态并追加 `index_publications`。 + +目录写入、构建与发布共用状态行锁，目录写入先于父记录锁获取该锁并递增修订号。发布核对代数、修订号和当前语料指纹；失败回滚，保留旧版本。目录变化后旧指针仍可追溯，但 is_current=false，不可据此绕过正式审核。历史快照不会被改写。接口均限管理员并沿用 Cookie/CSRF，完整边界见 [INDEX_API.md](INDEX_API.md)。 + +这只是后续索引的文本版本基础，关键词基线行为不变；没有生成向量或宣称已实现 ANN/混合检索。同步构建和全局写锁面向当前有限语料，后续异步/增量发布须重新设计并验证并发协议。 diff --git a/docs/DECISIONS.md b/docs/DECISIONS.md index cdece9b..fdc0bdd 100644 --- a/docs/DECISIONS.md +++ b/docs/DECISIONS.md @@ -35,3 +35,10 @@ 状态：Issue #5 分支实现，待审阅。使用 Argon2id 密码哈希、数据库保存摘要的随机 Cookie 会话，以及独立 CSRF 请求头与显式 Origin 白名单；不把角色或会话凭据放入浏览器持久存储。普通注册只能创建 user，管理员仅由本机隐藏密码输入命令创建。知识/来源/文档管理及检索历史限 admin，公开检索继续只使用已核验依据。完整契约与实现依据见 [AUTH_API.md](AUTH_API.md)。 为改密和会话撤销使用用户行锁与凭据版本，认证预算使用数据库原子计数；不依赖单进程内存限流。现有前端写请求同步 Cookie/CSRF 封装，页面和路由联调仍待前端完成。本分支叠加 PR #6，后续审阅需关注迁移顺序及匿名管理调用变为 401 的行为变化。 + + +## ADR-010 切分快照与显式发布 + +状态：T05 分支实现，待审阅。先对当前有效审核语料构建带原文偏移/哈希/出处的 chunks，再以独立的发布请求切换指针。固定 Unicode 字符窗口配置留存在版本中，关键词接口继续按条目运行，防止未经评估就改变实验口径。 + +为当前小规模语料选择统一数据库行锁：应用目录写入、构建、发布串行，发布同时校验代数、目录修订和语料指纹；失败回滚全部更新，旧版本保留为历史。任何成功目录写入保守地使旧索引过期，历史不能视作当前有效依据。实现、容量与后续并发扩展边界见 [INDEX_API.md](INDEX_API.md)。 diff --git a/docs/INDEX_API.md b/docs/INDEX_API.md new file mode 100644 index 0000000..716617d --- /dev/null +++ b/docs/INDEX_API.md @@ -0,0 +1,81 @@ +# 文本切分与索引版本接口（T05） + +本阶段保存可复核的已审核语料切分及发布指针，为后续检索索引提供版本边界。这里的“索引版本”是文本语料快照，不包含 Embedding、pgvector 向量字段/ANN 索引或 RRF。现有 `POST /api/retrieval` 仍运行原关键词基线，不读取这些 chunks；不能把本阶段称为向量检索已完成。 + +## 依赖、迁移与权限 + +分支 `feat/backend-chunk-index-versions` 叠加待审 PR #7（认证），并继承 PR #6（关键词）。三项须分别跟踪审阅；本分支不自行合并其他请求。升级时停止旧后端实例，安装依赖、备份并执行 `alembic upgrade head`，再启动新实例。所有实例必须使用当前版本，才能让目录写入遵守同一发布锁。 + +新增迁移 `0005_chunk_indexes`，依赖 `0004_auth_sessions`，创建： + +- `index_versions`：切分配置、语料指纹、目录修订号、条目/chunk 数、构建人、构建时间及发布状态。 +- `chunks`：所属版本、知识 ID、从 0 开始的片序、原文起止偏移、正文、UTF-8 内容 SHA-256 与出处快照。 +- `index_state`：单例发布指针、发布代数 `generation`、目录修订号 `catalog_revision`。 +- `index_publications`：每次发布的前后版本、代数、操作人和时间；重复发布当前版本不会新增记录。 + +迁移不改写既有知识、审核、用户、会话或检索记录，不创建初始发布版本。所有 `/api/indexes` 接口均要求 `admin`；POST 还必须携带有效 Cookie、明确 Origin 与 `X-CSRF-Token`，与 [认证契约](AUTH_API.md) 相同。 + +## 构建与原文切分 + +`POST /api/indexes` 创建一个 `ready` 版本，返回 201，不切换当前发布指针。请求可用默认值 `{}` 或： + +```json +{"chunk_size": 500, "overlap": 80} +``` + +`chunk_size` 为 100–2000 个 Python Unicode 字符，默认 500；`overlap` 默认 80，非负且不得超过切分长度一半。自定义较短长度时也须指定兼容的重叠，例如 `{"chunk_size":100,"overlap":20}`。未知字段拒绝，不支持按类别单独发布。 + +固定算法 `unicode-window-v1` 从知识条目原正文位置 0 开始（不是关联文档或原网页的偏移），以 `chunk_size-overlap` 前进，末片到正文结束即停止。保留空白、全半角、组合字符和 Emoji，不标准化或截断原文。`start_offset` 包含起点，`end_offset` 不包含终点；每片 `text == 原文[start_offset:end_offset]`。标题放在出处信息中，不拼进正文偏移。客户端若使用 JavaScript 字符串偏移，须注意其 UTF-16 索引与 Python Unicode 字符偏移的区别。 + +- 语料条件与关键词基线共用：只读取未删除、审核及采集信息完整的 `verified` 知识，关联来源/文档须有效、归属及来源网址一致；pending/rejected 不进入构建。 +- 语料指纹与关键词基线使用同一规范 JSON/算法，涵盖实际正文、标题、类别、出处及知识/父记录版本；切分参数独立保存在 `config`。 +- 出处快照包含 `knowledge_id/knowledge_version`、可空 `document_id/document_version`、可空 `source_id/source_version/source_name`、`source_url/title/category`。以后原资料修改或失效，不改写旧 chunk 内容及出处。 +- chunk ID 在一个版本内由版本 UUID、知识 ID 与顺序确定；新构建使用新版本 ID。相同资料及参数可重现偏移/正文/内容哈希，但新旧版本的 chunk ID 不相同。 +- 本阶段同步构建，沿用最多 5000 条及标题/正文总计 200 万字符的语料限制，每版本另限 20000 个 chunks。超限返回 503；事务写入失败也完整回滚，不保留半个版本，不改变旧发布指针。 +- 可以构建空语料版本，以便所有依据撤回后发布空快照；这不表示存在可用知识。 + +## 发布、过期与并发 + +先通过 `GET /api/indexes/active` 读取当前代数，再提交： + +```text +POST /api/indexes/{version_id}/publish +``` + +```json +{"expected_generation": 0} +``` + +发布时锁定 `index_state`，检查代数、目录修订号及当前有效语料指纹；资料或审核已变化、代数过期均返回 409，要求重新构建或刷新。成功后同一事务把旧版本设为 `superseded`、新版本设为 `published`、切换指针、增加代数并追加发布记录。校验失败或写入失败保留此前全部发布状态。持有当前代数重复发布当前版本是无变化的 200；仍有效的历史版本可用最新代数重新发布，发布记录保留每次转换。 + +知识、来源、文档的应用写入在修改父记录前持有同一状态锁，事务内将 `catalog_revision` 增加一次；失败写入一起回滚修订号。构建、发布和这些写入串行，避免校验到提交之间发生目录修改。为首批小规模资料选择此保守策略，长构建会暂时阻塞管理写入；尚无后台任务或大规模增量索引。 + +任一成功目录写入都会使旧快照过期，包括尚未核验的新条目；此规则比只比较有效语料更保守。`GET /active` 同时核对修订号与实际语料指纹，返回： + +```json +{"generation": 1, "catalog_revision": 2, "is_current": true, "active_version": {"id": "<版本 UUID>", "status": "published", "config": {"splitter":"unicode-window-v1","chunk_size":500,"overlap":80}, "...":"其余版本字段见 OpenAPI"}} +``` + +初始无发布时 `active_version=null/is_current=false`。资料变化后指针及历史保留，但 `is_current=false`；`published` 只表示仍是发布指针，不表示资料仍有效。历史 chunks 是当时的依据快照，不能绕过 `is_current` 直接用于后续正式检索。任何结果只描述该事务快照，之后的目录写入仍可让它过期。直接改库不属于支持的维护流程；额外指纹检查可发现绕过修订号的正文/出处变化，不能替代应用写锁。 + +## 查询与交接 + +| 接口 | 响应 | +|---|---| +| GET `/api/indexes` | 版本分页，构建时间降序，同时间按 ID | +| GET `/api/indexes/active` | 发布代数、目录修订号、当前性与当前版本 | +| GET `/api/indexes/publications` | 发布记录分页，代数降序 | +| GET `/api/indexes/{version_id}` | 版本元数据 | +| GET `/api/indexes/{version_id}/chunks` | chunk 分页；可按 `knowledge_id` 筛选，按知识 ID/片序排列 | + +分页沿用 `{items,total,page,page_size}`，默认 20、最大 100。版本元数据完整字段为 `id/status/config/corpus_fingerprint/catalog_revision/item_count/chunk_count/created_by/created_at/published_at`；`published_at` 是最近一次发布的时间，逐次时间保存在发布记录。版本及发布记录时间统一输出 UTC ISO 8601，避免数据库时区让重读结果出现不同表示。chunk 字段为 `id/index_version_id/knowledge_id/ordinal/start_offset/end_offset/text/text_sha256/provenance`。 + +前端 `src/services/indexes.ts` 提供类型和调用封装，写入共用 Cookie/CSRF；未增加页面或假数据。接口新增需由前端按该契约接入、审阅并验收。独立 T05 Issue 因当前缺少 Gitea 写入认证仍待补建，本 PR 和任务板先保留实现与验证记录。 + +## 验证及回滚 + +验证包含真实 PostgreSQL 的原文重建与哈希、审核筛选、重复构建、过期拒绝、来源/文档更新、失败写入、发布历史、数据库锁竞争、失败构建/发布的真实约束回滚、迁移兼容及真实 HTTP 重启后指针/chunks 持久化。这里只使用占位资料，不是论文实验结果。Docker 和浏览器端验收保持待办。 + +备份后配合旧后端执行 `alembic downgrade 0004_auth_sessions`，只撤回这四张表，保留用户、会话、知识及检索记录；索引历史与 chunks 恢复依赖备份。旧应用不可使用新版本数据库混跑。 + +并发实现参考 [PostgreSQL 16 行锁说明](https://www.postgresql.org/docs/16/explicit-locking.html) 与 [SQLAlchemy FOR UPDATE](https://docs.sqlalchemy.org/en/20/core/selectable.html#sqlalchemy.sql.expression.Select.with_for_update)。 diff --git a/docs/TASKS.md b/docs/TASKS.md index b6b2e63..1c0ac89 100644 --- a/docs/TASKS.md +++ b/docs/TASKS.md @@ -61,3 +61,12 @@ - 默认 Secure/HttpOnly/SameSite Cookie；写操作校验明确 Origin 及当前会话 CSRF。改密使所有旧会话失效，退出持久化撤销；参数错误不回显密码。字段、来源、CSRF 和前端交接见 [认证接口契约](AUTH_API.md)。 - 实际验证：`ruff check .`、`ruff format --check .` 通过；`pytest -q` **70 passed**，包含认证、授权、CSRF、输入越权、过期/轮换/撤销、限额、迁移及既有知识/检索回归；真实 HTTP 服务重启验证会话恢复和退出后不再有效。前端 `npm run build` 通过。 - 当前仍使用隔离 PostgreSQL 做自动化验证；Docker 容器、前端登录/角色页面及真实浏览器端流程仍待验收。远程 main 的注册原型最低密码提示需要按本契约同步到 15 字符，并接入带 Cookie/CSRF 的调用层。没有宣称已完成前端集成或生产部署。 + +## 2026-10-16 T05：持久化 chunks 与索引版本 + +- 分支 `feat/backend-chunk-index-versions` 叠加待审 PR #7（`f9827fe`），继承 #6；不等待合并继续后端阶段，同时保留各 PR 的审阅记录。 +- 新增 `0005_chunk_indexes`：版本、原文切分及出处、发布状态单例、逐次发布记录。构建 ready 快照与发布分开，保留字符偏移、内容哈希、知识/来源/文档版本和语料指纹。 +- 知识、来源、文档写入与构建/发布共用数据库行锁；目录修订与实际语料指纹一起决定当前性，旧构建/旧代数拒绝发布。构建或发布失败完整回滚，旧发布指针及历史不丢失。 +- 管理接口统一要求 admin 与写入 CSRF；前端只同步类型和调用层，页面仍待联调。现有关键词基线行为不变，向量、ANN、RRF 与 RAG 未在本阶段实现。 +- 实际验证：`ruff check .`、`ruff format --check .` 通过；`pytest -q` **91 passed**（隔离 PostgreSQL 16 + pgvector），包含真实锁竞争、失败构建/发布的数据库约束回滚、迁移保留账号/知识/检索记录、原文重建与来源快照、真实 HTTP 重启后的发布指针和 chunks 持久化。修正了发布时间在 UTC 与数据库时区之间表示不一致的问题，索引接口统一 UTC 输出。前端 `npm run build` 通过。 +- 完整请求/响应、切分/容量、失效策略及回滚说明见 [INDEX_API.md](INDEX_API.md)。本阶段仍是有限语料的同步文本快照；Docker/浏览器验收、独立 T05 Issue 补建及后续 Embedding 方案选择仍有待完成。 diff --git a/frontend/src/services/indexes.ts b/frontend/src/services/indexes.ts new file mode 100644 index 0000000..9d2b6c9 --- /dev/null +++ b/frontend/src/services/indexes.ts @@ -0,0 +1,16 @@ +import { apiRequest } from './session' +import type { RetrievalHit } from '../api' + +type Page = { items: T[]; total: number; page: number; page_size: number } +export type IndexConfig = { splitter: 'unicode-window-v1'; chunk_size: number; overlap: number } +export type IndexVersion = { id: string; status: 'ready' | 'published' | 'superseded'; config: IndexConfig; corpus_fingerprint: string; catalog_revision: number; item_count: number; chunk_count: number; created_by: number; created_at: string; published_at: string | null } +export type ActiveIndex = { generation: number; catalog_revision: number; is_current: boolean; active_version: IndexVersion | null } +export type Chunk = { id: string; index_version_id: string; knowledge_id: number; ordinal: number; start_offset: number; end_offset: number; text: string; text_sha256: string; provenance: Pick } +export type IndexPublication = { generation: number; previous_version_id: string | null; version_id: string; actor_id: number; published_at: string } +export const buildIndex = (config: Partial> = {}) => apiRequest('/api/indexes', { method: 'POST', body: JSON.stringify(config) }, true) +export const getActiveIndex = () => apiRequest('/api/indexes/active') +export const getIndexVersions = (page = 1, pageSize = 20) => apiRequest>(`/api/indexes?page=${page}&page_size=${pageSize}`) +export const getIndexVersion = (id: string) => apiRequest(`/api/indexes/${encodeURIComponent(id)}`) +export const publishIndex = (id: string, expectedGeneration: number) => apiRequest(`/api/indexes/${encodeURIComponent(id)}/publish`, { method: 'POST', body: JSON.stringify({ expected_generation: expectedGeneration }) }, true) +export const getChunks = (id: string, page = 1, pageSize = 20, knowledgeId?: number) => apiRequest>(`/api/indexes/${encodeURIComponent(id)}/chunks?page=${page}&page_size=${pageSize}${knowledgeId ? `&knowledge_id=${knowledgeId}` : ''}`) +export const getIndexPublications = (page = 1, pageSize = 20) => apiRequest>(`/api/indexes/publications?page=${page}&page_size=${pageSize}`)