diff --git a/README.md b/README.md index c2e1227..e72305c 100644 --- a/README.md +++ b/README.md @@ -52,6 +52,8 @@ Vite 通常使用 http://localhost:5173。 - POST `/api/knowledge/{id}/review`、GET `/api/knowledge/{id}/history` - GET/POST `/api/sources`、GET/PATCH/DELETE `/api/sources/{id}`、GET `/api/sources/{id}/history` - 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}`：检索历史和原始结果 ### 阶段 2 接口约定 @@ -77,7 +79,26 @@ Vite 通常使用 http://localhost:5173。 } ``` -响应新增 `version/updated_at/is_deleted` 及关联、采集、审核字段；完整类型以 `/openapi.json` 和 `/docs` 为准。前端现有录入与数组列表可继续使用；来源、文档、审核及完整分页页面仍需按接口联调。 +响应新增 `version/updated_at/is_deleted` 及关联、采集、审核字段；完整类型以 `/openapi.json` 和 `/docs` 为准。前端已接入知识录入和来源登记/查询，文档、审核及完整分页页面仍需按接口联调。 + +### 关键词检索基线 + +先安装更新后的依赖并执行 `alembic upgrade head`，再启动后端。检索请求示例： +```json +{"query": "那拉提旅游资源", "mode": "keyword", "top_k": 5, "category": "旅游资源"} +``` + +`query` 为 1–500 字符；`top_k` 默认 5、最大 20；`category` 可省略。当前只接受 `keyword` 模式。纯标点、空白或仅含停用词的查询返回 422。 + +- 仅使用 `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` 和空结果，当前不会调用模型生成答案。 +- 每次成功检索（含零命中）保存不可变结果快照、实际检索耗时、语料指纹和算法参数。`elapsed_ms` 包含读取、分词、排序和片段生成，不包含记录写入和网络时间。历史记录用于复核，不能当作仍然有效的当前知识；依据变更后的新检索会重新应用审核条件。 +- `GET /api/retrieval/runs` 使用标准分页参数；`GET /api/retrieval/runs/{run_id}` 读取当时结果。当前知识修改、失效或删除不会改写历史结果。检索记录保存在数据库，运行日志和实际查询不要提交到 Git。 +- 本基线按请求计算语料词频，面向小规模资料集：筛选后的语料最多 5000 条、标题与正文合计最多 200 万字符。超过容量明确返回 503，不截断语料或伪造空结果；可缩小类别范围，大规模索引留到后续实现。 + +前端 `src/api.ts` 已提供带类型的检索及历史调用；检索页面交互仍由前端阶段接入。本阶段没有接入 Embedding、向量/混合检索或 LLM，不产生论文实验分数。 前端包括问答、知识库管理、来源与更新、实验评估四个页面。实验评估页面没有虚构数据；所有页面明确标记当前功能状态。 @@ -94,7 +115,7 @@ pytest -q 测试需要真实 PostgreSQL + pgvector：设置 `TEST_DATABASE_URL`（`postgresql+psycopg://...`），或另行 `pip install pgserver` 使用可选的本机临时实例。`pgserver` 仅用于测试，不是应用运行依赖；外部测试数据库账号需有建 Schema 权限且已启用 vector 扩展。每个测试使用独立 Schema，不会清空业务表。测试使用的占位正文和审核人不代表真实资料。 -迁移命令为 `alembic upgrade head`，可重复执行。需要撤回阶段 2 时，先备份，再执行 `alembic downgrade 0001_baseline`；它保留基础知识字段，但会删除新增来源、文档、审核字段与历史记录，软删除的基础条目也会重新可见。恢复完整数据应使用备份。不要在有业务数据的数据库执行 `downgrade base`。 +迁移命令为 `alembic upgrade head`，可重复执行。仅撤回关键词检索的数据库变更时，备份后执行 `alembic downgrade 0002_catalog`，只删除检索历史表。需要撤回阶段 2 时，先备份，再执行 `alembic downgrade 0001_baseline`；它保留基础知识字段，但会删除新增来源、文档、审核字段与所有历史记录，软删除的基础条目也会重新可见。降级须配合对应旧版后端，当前应用会拒绝旧 Schema。恢复完整数据应使用备份。不要在有业务数据的数据库执行 `downgrade base`。 ## 团队协作文档 diff --git a/backend/app/api/routes/retrieval.py b/backend/app/api/routes/retrieval.py new file mode 100644 index 0000000..8ddbc15 --- /dev/null +++ b/backend/app/api/routes/retrieval.py @@ -0,0 +1,47 @@ +from uuid import UUID + +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from app.db.session import get_db +from app.models import RetrievalRun +from app.schemas.catalog import Page +from app.schemas.retrieval import RetrievalRequest, RetrievalResponse +from app.services.retrieval import retrieve + +router = APIRouter(prefix="/retrieval", tags=["retrieval"]) + + +@router.post("", response_model=RetrievalResponse) +def keyword_retrieval(payload: RetrievalRequest, db: Session = Depends(get_db)): + """只检索已核验知识；保存原始检索结果供复核，不调用大模型。""" + return retrieve(db, payload) + + +@router.get("/runs", response_model=Page[RetrievalResponse]) +def list_runs( + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + db: Session = Depends(get_db), +): + rows = db.scalars( + select(RetrievalRun) + .order_by(RetrievalRun.created_at.desc(), RetrievalRun.id) + .offset((page - 1) * page_size) + .limit(page_size) + ) + return { + "items": [row.response for row in rows], + "total": db.scalar(select(func.count()).select_from(RetrievalRun)), + "page": page, + "page_size": page_size, + } + + +@router.get("/runs/{run_id}", response_model=RetrievalResponse) +def read_run(run_id: UUID, db: Session = Depends(get_db)): + row = db.get(RetrievalRun, str(run_id)) + if row is None: + raise HTTPException(404, "检索记录不存在") + return row.response diff --git a/backend/app/main.py b/backend/app/main.py index 7f1739c..8a7fed0 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -7,9 +7,10 @@ from sqlalchemy import text from sqlalchemy.exc import IntegrityError, OperationalError, ProgrammingError from sqlalchemy.orm.exc import StaleDataError -from app.api.routes import chat, documents, health, knowledge, sources +from app.api.routes import chat, documents, health, knowledge, retrieval, sources from app.core.config import settings from app.db.session import engine +from app.services.retrieval import tokenizer def create_app(db_engine=engine) -> FastAPI: @@ -18,7 +19,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 != "0002_catalog": + if revision != "0003_retrieval_runs": raise RuntimeError( "数据库版本不匹配，请先在 backend 目录执行 alembic upgrade head" ) @@ -26,12 +27,13 @@ def create_app(db_engine=engine) -> FastAPI: raise RuntimeError( "数据库尚未初始化，请先在 backend 目录执行 alembic upgrade head" ) from exc + tokenizer() try: yield finally: db_engine.dispose() - app = FastAPI(title=settings.app_name, version="0.2.0", lifespan=lifespan) + app = FastAPI(title=settings.app_name, version="0.3.0", lifespan=lifespan) app.add_middleware( CORSMiddleware, allow_origins=settings.cors_origin_list, @@ -39,7 +41,14 @@ def create_app(db_engine=engine) -> FastAPI: allow_methods=["*"], allow_headers=["*"], ) - for router in (health.router, chat.router, knowledge.router, sources.router, documents.router): + for router in ( + health.router, + chat.router, + knowledge.router, + sources.router, + documents.router, + retrieval.router, + ): app.include_router(router, prefix=settings.api_prefix) @app.exception_handler(StaleDataError) diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 771117b..89e1a73 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -1,4 +1,5 @@ from app.models.catalog import AuditEvent, Document, Source from app.models.knowledge import KnowledgeItem +from app.models.retrieval import RetrievalRun -__all__ = ["AuditEvent", "Document", "KnowledgeItem", "Source"] +__all__ = ["AuditEvent", "Document", "KnowledgeItem", "RetrievalRun", "Source"] diff --git a/backend/app/models/retrieval.py b/backend/app/models/retrieval.py new file mode 100644 index 0000000..8c41441 --- /dev/null +++ b/backend/app/models/retrieval.py @@ -0,0 +1,18 @@ +from datetime import datetime + +from sqlalchemy import JSON, DateTime, String, Text, func +from sqlalchemy.orm import Mapped, mapped_column + +from app.db.session import Base + + +class RetrievalRun(Base): + __tablename__ = "retrieval_runs" + id: Mapped[str] = mapped_column(String(36), primary_key=True) + query: Mapped[str] = mapped_column(Text, nullable=False) + mode: Mapped[str] = mapped_column(String(20), nullable=False) + corpus_fingerprint: Mapped[str] = mapped_column(String(64), nullable=False) + response: Mapped[dict] = mapped_column(JSON, nullable=False) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), index=True + ) diff --git a/backend/app/schemas/retrieval.py b/backend/app/schemas/retrieval.py new file mode 100644 index 0000000..10dcd2c --- /dev/null +++ b/backend/app/schemas/retrieval.py @@ -0,0 +1,59 @@ +from datetime import datetime +from typing import Literal +from uuid import UUID + +from pydantic import BaseModel, Field + +from app.schemas.catalog import Category, Input + + +class RetrievalRequest(Input): + query: str = Field(min_length=1, max_length=500) + mode: Literal["keyword"] = "keyword" + top_k: int = Field(default=5, ge=1, le=20) + category: Category | None = None + + +class RetrievalHit(BaseModel): + rank: int + 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 + snippet: str + snippet_start: int + snippet_end: int + score: float + matched_terms: list[str] + + +class RetrievalConfig(BaseModel): + tokenizer: str + scorer: str + k1: float + b: float + title_weight: int + stopwords_version: str + + +class RetrievalResponse(BaseModel): + run_id: UUID + query: str + mode: Literal["keyword"] + status: Literal["ok", "no_evidence"] + top_k: int + category: str | None + query_terms: list[str] + corpus_size: int + corpus_fingerprint: str + total_hits: int + results: list[RetrievalHit] + elapsed_ms: float + created_at: datetime + config: RetrievalConfig diff --git a/backend/app/services/retrieval.py b/backend/app/services/retrieval.py new file mode 100644 index 0000000..a769dae --- /dev/null +++ b/backend/app/services/retrieval.py @@ -0,0 +1,215 @@ +"""Deterministic lexical baseline over a bounded, reviewed knowledge corpus.""" + +import hashlib +import json +import logging +import math +import unicodedata +from collections import Counter +from functools import lru_cache +from time import perf_counter +from uuid import uuid4 + +import jieba +from fastapi import HTTPException +from sqlalchemy import func, or_, select +from sqlalchemy.orm import Session + +from app.models import Document, KnowledgeItem, RetrievalRun, Source +from app.models.catalog import utc_now +from app.schemas.retrieval import RetrievalConfig, RetrievalHit, RetrievalRequest, RetrievalResponse + +MAX_ITEMS = 5000 +MAX_CHARACTERS = 2_000_000 +SNIPPET_LENGTH = 240 +STOPWORDS = frozenset("的 了 和 与 是 在 有 哪些 什么 请问 如何 吗 呢 我 你 想 了解".split()) +CONFIG = RetrievalConfig( + tokenizer="jieba-0.42.1-search-hmm-off-nalati-v1", + scorer="bm25-weighted-text-v1", + k1=1.2, + b=0.75, + title_weight=2, + stopwords_version="zh-small-v1", +) + + +@lru_cache(maxsize=1) +def tokenizer(): + jieba.setLogLevel(logging.WARNING) + instance = jieba.Tokenizer() + instance.initialize() + instance.add_word("那拉提", freq=100000) + return instance + + +def normalized_with_offsets(text: str) -> tuple[str, list[int]]: + pieces, offsets = [], [] + for position, character in enumerate(text): + piece = unicodedata.normalize("NFKC", character).casefold() + pieces.append(piece) + offsets.extend([position] * len(piece)) + return "".join(pieces), offsets + + +def tokenize(text: str) -> list[str]: + normalized, _ = normalized_with_offsets(text) + return [ + term + for term in tokenizer().cut_for_search(normalized, HMM=False) + if term not in STOPWORDS and any(character.isalnum() for character in term) + ] + + +def bm25(counters: list[Counter], query_terms: list[str]) -> list[float]: + """Positive log-IDF BM25; repeated query terms count once.""" + count = len(counters) + lengths = [sum(counter.values()) for counter in counters] + average = sum(lengths) / count if count else 0 + if not average: + return [0.0] * count + scores = [0.0] * count + for term in dict.fromkeys(query_terms): + frequency = sum(counter[term] > 0 for counter in counters) + if not frequency: + continue + idf = math.log(1 + (count - frequency + 0.5) / (frequency + 0.5)) + for index, counter in enumerate(counters): + tf = counter[term] + if tf: + denominator = tf + CONFIG.k1 * (1 - CONFIG.b + CONFIG.b * lengths[index] / average) + scores[index] += idf * tf * (CONFIG.k1 + 1) / denominator + return scores + + +def excerpt(content: str, matched_terms: list[str]) -> tuple[str, int, int]: + normalized, offsets = normalized_with_offsets(content) + positions = [normalized.find(term) for term in matched_terms if term in normalized] + start = max(0, offsets[min(positions)] - 60) if positions else 0 + end = min(len(content), start + SNIPPET_LENGTH) + return content[start:end], start, end + + +def load_corpus(db: Session, category: str | None) -> list[dict]: + statement = ( + select( + KnowledgeItem, + Source.name.label("source_name"), + Source.version.label("source_version"), + Document.version.label("document_version"), + func.count().over().label("corpus_total"), + func.sum(func.length(KnowledgeItem.title) + func.length(KnowledgeItem.content)) + .over() + .label("corpus_characters"), + ) + .outerjoin(Source, KnowledgeItem.source_id == Source.id) + .outerjoin(Document, KnowledgeItem.document_id == Document.id) + .where( + KnowledgeItem.verification_status == "verified", + KnowledgeItem.is_deleted.is_(False), + KnowledgeItem.source_url.is_not(None), + func.length(func.trim(KnowledgeItem.source_url)) > 0, + KnowledgeItem.collected_at.is_not(None), + KnowledgeItem.reviewed_at.is_not(None), + KnowledgeItem.reviewer.is_not(None), + KnowledgeItem.verification_notes.is_not(None), + or_(KnowledgeItem.source_id.is_(None), Source.is_deleted.is_(False)), + or_( + KnowledgeItem.document_id.is_(None), + (Document.is_deleted.is_(False)) & (Document.source_id == KnowledgeItem.source_id), + ), + or_(KnowledgeItem.source_id.is_(None), KnowledgeItem.source_url == Source.source_url), + ) + .order_by(KnowledgeItem.id) + .limit(MAX_ITEMS + 1) + ) + if category is not None: + statement = statement.where(KnowledgeItem.category == category) + corpus = [] + # 一条 SQL 的快照同时给出语料与容量；流式读取避免先载入超限正文。 + with db.execute(statement.execution_options(stream_results=True, max_row_buffer=1)) as rows: + for row in rows: + if row.corpus_total > MAX_ITEMS or row.corpus_characters > MAX_CHARACTERS: + raise HTTPException( + 503, "已核验语料超过关键词基线容量，请缩小分类范围或建立检索索引" + ) + item = row.KnowledgeItem + corpus.append( + { + "knowledge_id": item.id, + "knowledge_version": item.version, + "document_id": item.document_id, + "document_version": row.document_version, + "source_id": item.source_id, + "source_version": row.source_version, + "source_name": row.source_name, + "source_url": item.source_url, + "title": item.title, + "category": item.category, + "content": item.content, + } + ) + return corpus + + +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() + counters = [ + Counter(tokenize(item["title"]) * CONFIG.title_weight + tokenize(item["content"])) + for item in corpus + ] + scores = bm25(counters, terms) + ranked = sorted( + (index for index, score in enumerate(scores) if score > 0), + key=lambda index: (-scores[index], corpus[index]["knowledge_id"]), + ) + results = [] + for rank, index in enumerate(ranked[: payload.top_k], start=1): + item = corpus[index] + matched = [term for term in terms if counters[index][term]] + snippet, start, end = excerpt(item["content"], matched) + results.append( + RetrievalHit( + **{key: value for key, value in item.items() if key != "content"}, + rank=rank, + score=round(scores[index], 8), + matched_terms=matched, + snippet=snippet, + snippet_start=start, + snippet_end=end, + ) + ) + response = RetrievalResponse( + run_id=uuid4(), + query=payload.query, + mode="keyword", + status="ok" if results else "no_evidence", + top_k=payload.top_k, + category=payload.category, + query_terms=terms, + corpus_size=len(corpus), + corpus_fingerprint=fingerprint, + total_hits=len(ranked), + results=results, + elapsed_ms=round((perf_counter() - started) * 1000, 3), + created_at=utc_now(), + config=CONFIG, + ) + db.add( + RetrievalRun( + id=str(response.run_id), + query=response.query, + mode=response.mode, + corpus_fingerprint=fingerprint, + response=response.model_dump(mode="json"), + created_at=response.created_at, + ) + ) + db.commit() + return response diff --git a/backend/migrations/versions/0003_retrieval_runs.py b/backend/migrations/versions/0003_retrieval_runs.py new file mode 100644 index 0000000..b393d5c --- /dev/null +++ b/backend/migrations/versions/0003_retrieval_runs.py @@ -0,0 +1,28 @@ +"""Persist keyword retrieval evidence and reproducible configuration.""" + +import sqlalchemy as sa +from alembic import op + +revision = "0003_retrieval_runs" +down_revision = "0002_catalog" +branch_labels = None +depends_on = None + + +def upgrade(): + op.create_table( + "retrieval_runs", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("query", sa.Text(), nullable=False), + sa.Column("mode", sa.String(20), nullable=False), + sa.Column("corpus_fingerprint", sa.String(64), nullable=False), + sa.Column("response", sa.JSON(), nullable=False), + sa.Column( + "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False + ), + ) + op.create_index("ix_retrieval_runs_created_at", "retrieval_runs", ["created_at"]) + + +def downgrade(): + op.drop_table("retrieval_runs") diff --git a/backend/requirements.txt b/backend/requirements.txt index 1be4a2c..73bf48b 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -6,3 +6,4 @@ SQLAlchemy>=2.0,<3.0 psycopg[binary]>=3.2,<4.0 pgvector>=0.3,<1.0 alembic>=1.13,<2.0 +jieba==0.42.1 diff --git a/backend/tests/test_migrations.py b/backend/tests/test_migrations.py index 0747b06..de0ec81 100644 --- a/backend/tests/test_migrations.py +++ b/backend/tests/test_migrations.py @@ -15,6 +15,13 @@ def test_startup_requires_migration(database): pass +def test_startup_rejects_previous_schema(database): + migrate(database, "0002_catalog") + with pytest.raises(RuntimeError, match="alembic upgrade head"): + with TestClient(create_app(database)): + pass + + def test_upgrade_preserves_original_rows_and_repeated_upgrade(database): migrate(database, "0001_baseline") with database.begin() as connection: @@ -33,7 +40,10 @@ def test_upgrade_preserves_original_rows_and_repeated_upgrade(database): assert row["collected_at"] is None assert row["verification_status"] == "pending" assert connection.scalar(text("SELECT count(*) FROM audit_events")) == 2 - assert connection.scalar(text("SELECT version_num FROM alembic_version")) == "0002_catalog" + assert ( + connection.scalar(text("SELECT version_num FROM alembic_version")) + == "0003_retrieval_runs" + ) assert compare_metadata(MigrationContext.configure(connection), Base.metadata) == [] migrate(database, "0001_baseline", downgrade=True) with database.connect() as connection: diff --git a/backend/tests/test_retrieval.py b/backend/tests/test_retrieval.py new file mode 100644 index 0000000..950b25e --- /dev/null +++ b/backend/tests/test_retrieval.py @@ -0,0 +1,212 @@ +import math +from collections import Counter +from uuid import uuid4 + +import pytest +from sqlalchemy import text + +from app.services import retrieval + + +def add_item(client, *, status="verified", **fields): + response = client.post( + "/api/knowledge", + json={ + "title": "检索占位资料", + "content": "那拉提旅游资源检索测试，不代表真实文旅事实。", + "source_url": "https://example.com/", + "collected_at": "2026-10-15T09:00:00+08:00", + **fields, + }, + ) + assert response.status_code == 201, response.text + item = response.json() + if status != "pending": + response = client.post( + f"/api/knowledge/{item['id']}/review", + json={ + "expected_version": item["version"], + "verification_status": status, + "reviewer": "test-reviewer", + "verification_notes": "仅为自动化审核流程测试", + }, + ) + assert response.status_code == 200, response.text + item = response.json() + return item + + +def search(client, query="那拉提", **fields): + response = client.post("/api/retrieval", json={"query": query, **fields}) + assert response.status_code == 200, response.text + return response.json() + + +def test_bm25_matches_hand_calculated_scores_and_query_deduplication(): + counters = [Counter({"x": 1}), Counter({"x": 1, "y": 1})] + scores = retrieval.bm25(counters, ["x"]) + idf = math.log(1 + 0.5 / 2.5) + assert scores[0] == pytest.approx(idf * 2.2 / (1 + 1.2 * (0.25 + 0.75 / 1.5))) + assert scores[1] == pytest.approx(idf * 2.2 / (1 + 1.2 * (0.25 + 0.75 * 2 / 1.5))) + assert scores[0] > scores[1] + assert retrieval.bm25(counters, ["x", "x"]) == scores + assert retrieval.bm25([], ["x"]) == [] + assert retrieval.bm25([Counter()], ["x"]) == [0.0] + + +def test_tokenization_and_exact_original_snippet_offsets(): + terms = retrieval.tokenize("请问那拉提 ＡＰＩ api！") + assert "那拉提" in terms and terms.count("api") == 2 + assert "请问" not in terms + original = "占位前文。" * 80 + "这里有ＡＰＩ检索说明。" + "占位后文。" * 80 + snippet, start, end = retrieval.excerpt(original, ["api"]) + assert start > 0 and "ＡＰＩ" in snippet + assert snippet == original[start:end] and len(snippet) <= 240 + + +def test_only_reviewed_active_knowledge_is_retrieved(client): + good = add_item(client) + add_item(client, status="pending") + add_item(client, status="rejected") + deleted = add_item(client) + assert ( + client.delete( + f"/api/knowledge/{deleted['id']}?expected_version={deleted['version']}" + ).status_code + == 204 + ) + result = search(client) + assert result["corpus_size"] == 1 and result["total_hits"] == 1 + assert result["results"][0]["knowledge_id"] == good["id"] + assert result["results"][0]["knowledge_version"] == good["version"] + assert result["results"][0]["source_url"] == "https://example.com/" + assert result["elapsed_ms"] >= 0 + + +def test_tied_results_category_top_k_and_reproducibility(client): + first = add_item(client, category="交通出行") + second = add_item(client, category="交通出行") + third = add_item(client, category="民宿服务") + result = search(client, top_k=2) + again = search(client, top_k=2) + assert result["results"] == again["results"] + assert result["corpus_fingerprint"] == again["corpus_fingerprint"] + assert result["config"] == again["config"] + assert result["run_id"] != again["run_id"] + assert result["total_hits"] == 3 + assert [hit["knowledge_id"] for hit in result["results"]] == [first["id"], second["id"]] + filtered = search(client, category="民宿服务") + assert filtered["corpus_size"] == 1 + assert filtered["results"][0]["knowledge_id"] == third["id"] + + +def test_title_terms_have_more_weight_than_body_terms(client): + title_match = add_item(client, title="旅游", content="交通") + body_match = add_item(client, title="交通", content="旅游") + result = search(client, query="旅游") + assert [hit["knowledge_id"] for hit in result["results"]] == [ + title_match["id"], + body_match["id"], + ] + assert result["results"][0]["score"] > result["results"][1]["score"] + + +def test_no_evidence_and_history_are_persisted(client): + empty = search(client) + assert empty["status"] == "no_evidence" and empty["results"] == [] + add_item(client) + unmatched = search(client, query="unmatcheduniquelexeme") + assert unmatched["status"] == "no_evidence" and unmatched["corpus_size"] == 1 + assert client.get(f"/api/retrieval/runs/{empty['run_id']}").json() == empty + history = client.get("/api/retrieval/runs?page_size=1").json() + assert history["total"] == 2 and len(history["items"]) == 1 + assert history["items"][0] == unmatched + assert client.get(f"/api/retrieval/runs/{uuid4()}").status_code == 404 + assert client.get("/api/retrieval/runs/invalid").status_code == 422 + assert client.get("/api/retrieval/runs?page_size=101").status_code == 422 + + +def test_source_snapshot_and_original_run_survive_invalidation(client): + src = client.post( + "/api/sources", + json={ + "name": "测试来源", + "source_url": "https://example.com/", + }, + ).json() + content = "占位前言。" * 80 + "那拉提检索关键词。" + "占位后文。" * 80 + doc = client.post( + "/api/documents", + json={ + "source_id": src["id"], + "title": "测试文档", + "content": content, + "collected_at": "2026-10-15T09:00:00+08:00", + }, + ).json() + item = add_item(client, document_id=doc["id"], content=content) + original = search(client) + hit = original["results"][0] + assert hit["document_id"] == doc["id"] and hit["document_version"] == 1 + assert hit["source_id"] == src["id"] and hit["source_version"] == 1 + assert hit["source_name"] == "测试来源" + assert "那拉提" in hit["snippet"] and hit["snippet_start"] > 0 + assert hit["snippet"] == content[hit["snippet_start"] : hit["snippet_end"]] + assert ( + client.patch( + f"/api/sources/{src['id']}", + json={ + "expected_version": src["version"], + "name": "修订来源名称", + }, + ).status_code + == 200 + ) + revised = search(client) + assert revised["corpus_size"] == 0 and revised["results"] == [] + assert revised["corpus_fingerprint"] != original["corpus_fingerprint"] + stored = client.get(f"/api/retrieval/runs/{original['run_id']}").json() + assert stored == original and stored["results"][0]["knowledge_id"] == item["id"] + + +def test_inactive_parent_is_excluded_even_if_status_was_not_reset(client, migrated_database): + src = client.post( + "/api/sources", json={"name": "测试", "source_url": "https://example.com/"} + ).json() + add_item(client, source_id=src["id"]) + assert search(client)["corpus_size"] == 1 + with migrated_database.begin() as connection: + connection.execute( + text("UPDATE sources SET is_deleted = true WHERE id = :id"), {"id": src["id"]} + ) + assert search(client)["corpus_size"] == 0 + + +@pytest.mark.parametrize( + "payload", + [ + {"query": " "}, + {"query": "!!!"}, + {"query": "的了"}, + {"query": "那拉提", "top_k": 0}, + {"query": "那拉提", "top_k": 21}, + {"query": "那拉提", "mode": "vector"}, + {"query": "那拉提", "unexpected": True}, + {"query": "a" * 501}, + ], +) +def test_invalid_query_does_not_create_run(client, payload): + assert client.post("/api/retrieval", json=payload).status_code == 422 + assert client.get("/api/retrieval/runs").json()["total"] == 0 + + +@pytest.mark.parametrize("setting,value", [("MAX_ITEMS", 1), ("MAX_CHARACTERS", 1)]) +def test_corpus_budget_fails_explicitly_without_silent_truncation( + client, monkeypatch, setting, value +): + add_item(client) + add_item(client) + monkeypatch.setattr(retrieval, setting, value) + response = client.post("/api/retrieval", json={"query": "那拉提"}) + assert response.status_code == 503 and "容量" in response.json()["detail"] + assert client.get("/api/retrieval/runs").json()["total"] == 0 diff --git a/backend/tests/test_runtime.py b/backend/tests/test_runtime.py index eb3536f..63f20cc 100644 --- a/backend/tests/test_runtime.py +++ b/backend/tests/test_runtime.py @@ -76,3 +76,34 @@ def test_default_startup_and_persistence_across_restart(migrated_database): with server(migrated_database) as client: assert client.get(f"/api/knowledge/{item['id']}").json()["content"] == "仅为自动化测试数据" assert client.get("/api/knowledge").json()[0]["id"] == item["id"] + + +def test_retrieval_run_persists_across_restart(migrated_database): + with server(migrated_database) as client: + item = client.post( + "/api/knowledge", + json={ + "title": "HTTP检索测试", + "content": "仅用于自动化测试的那拉提关键词。", + "source_url": "https://example.com/", + "collected_at": "2026-10-15T09:00:00+08:00", + }, + ).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 + ) + response = client.post("/api/retrieval", json={"query": "那拉提"}) + assert response.status_code == 200 + run = response.json() + assert run["results"][0]["knowledge_id"] == item["id"] + with server(migrated_database) as client: + assert client.get(f"/api/retrieval/runs/{run['run_id']}").json() == run diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index d698576..32e196c 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -14,7 +14,7 @@ FastAPI + Pydantic |-- 来源获取 -> 人工核验 -> 文本切分 -> 生成向量 -> 发布版本 ``` -当前后端实现健康/数据库就绪检查、来源与文档管理、知识增删改查、分页筛选、人工审核及历史快照；问答仍为占位接口。数据库由 Alembic 显式迁移并启用 pgvector，尚无向量字段、向量索引和 RAG。前端来源管理与实验页仍为静态原型，审核与分页页面需继续联调。 +当前后端实现健康/数据库就绪检查、来源与文档管理、知识增删改查、分页筛选、人工审核、历史快照及独立关键词检索；问答仍为占位接口。数据库由 Alembic 显式迁移并启用 pgvector，尚无向量字段、向量索引和 RAG。前端来源登记/查询已接入，实验页仍为静态原型，文档、审核、完整分页与检索页面需继续联调。 ## 阶段 2 数据边界 @@ -28,3 +28,11 @@ FastAPI + Pydantic ## 安全边界 当前无认证与权限控制，仅供本机开发；正式部署前须补充登录鉴权、接口限流、数据库迁移、密钥管理、日志脱敏、CORS 收敛和 HTTPS。 + +## 关键词检索基线（T06） + +`POST /api/retrieval` 在同一 SQL 语料快照中筛选有效的已审核知识，读取出处和记录版本，再进行固定中文分词、标题加权 BM25 排序和原文片段截取。语料指纹覆盖本次实际参与排序的内容与出处版本；算法标识和参数独立记录。管理页的字面子串筛选继续用于管理，正式关键词基线使用专用检索接口。 + +新增 `retrieval_runs` 保存查询、模式、语料指纹和完整结果快照，支持分页读取及按 UUID 复核。零命中如实记录；当前依据修改后失效不会改写历史，但后续检索必须重新检查审核与父记录状态。响应耗时是本次真实读取、分词、排序及片段生成耗时，不含写日志或模型调用。 + +当前按知识条目检索，显示片段最长 240 字符；尚无持久化 chunks、倒排索引、Embedding、向量检索、RRF 或 RAG。全语料计算明确限制条目数及总字符数，超限报错，适用于首批小规模资料与后续对照实验准备，不能当作生产检索性能结论。 diff --git a/docs/DECISIONS.md b/docs/DECISIONS.md index 5e955fd..adb116b 100644 --- a/docs/DECISIONS.md +++ b/docs/DECISIONS.md @@ -17,8 +17,14 @@ ## ADR-006 阶段 2 知识管理 -状态：分支实现，待合并请求审阅。采用来源、文档、知识条目和追加式事件日志；记录版本用于并发控制，快照用于追溯。所有删除均为软删除，被有效记录引用的父记录禁止删除。依据或正文修改即回待审核；核验通过必须有来源链接、采集时间、核验人和说明。保留旧版知识创建请求及列表数组结构，新页面使用独立分页接口。当前为本机人工审核流程，认证身份、向量切分和索引发布仍属后续阶段。 +状态：已通过 PR #4 合入 `dev`。采用来源、文档、知识条目和追加式事件日志；记录版本用于并发控制，快照用于追溯。所有删除均为软删除，被有效记录引用的父记录禁止删除。依据或正文修改即回待审核；核验通过必须有来源链接、采集时间、核验人和说明。保留旧版知识创建请求及列表数组结构，新页面使用独立分页接口。当前为本机人工审核流程，认证身份、向量切分和索引发布仍属后续阶段。 ## ADR-007 显式迁移与旧数据保留 -状态：分支实现，待合并请求审阅。以 Alembic 取代应用启动时的 `create_all`。基础迁移接管旧知识表，新增字段和表通过增量迁移完成；不推测旧数据采集时间。旧审核结论缺少完整记录时保留原快照并转待审核。迁移回滚需先备份，新增历史与审核元数据无法通过降级完整保留。 +状态：已通过 PR #4 合入 `dev`。以 Alembic 取代应用启动时的 `create_all`。基础迁移接管旧知识表，新增字段和表通过增量迁移完成；不推测旧数据采集时间。旧审核结论缺少完整记录时保留原快照并转待审核。迁移回滚需先备份，新增历史与审核元数据无法通过降级完整保留。 + +## ADR-008 可复核的中文关键词基线 + +状态：分支实现，待审阅。采用固定 Jieba 搜索分词及关闭 HMM 的词典模式，加入项目名称词，并使用固定停用词表。选择标题重复权重 2、正文权重 1 的 BM25 正值平滑 IDF 变体，不称为 BM25F；查询词去重，同分按知识 ID 排序。参数、分词标识、实际语料指纹及出处版本随每次结果保存，方便冻结后续实验配置。分词接口参考 [Jieba 官方说明](https://github.com/fxsjy/jieba)，BM25 基础参考 [Stanford IR 教材](https://nlp.stanford.edu/IR-book/html/htmledition/okapi-bm25-a-non-binary-model-1.html)；具体平滑公式与权重以本仓库实现和手算测试为准。 + +当前检索单元为已审核知识条目，在设置的容量内按请求计算，不提前实现向量切分/索引更新；超过容量明确失败。`retrieval_runs` 仅记录当时结果，不代表该条目之后仍然获得审核批准。网络搜索、模型选择和真实论文实验保持原阶段边界。 diff --git a/docs/TASKS.md b/docs/TASKS.md index 263f347..b7fd832 100644 --- a/docs/TASKS.md +++ b/docs/TASKS.md @@ -44,3 +44,11 @@ - 覆盖：新库迁移、已有基础表接管、重复升级、阶段 2 降级后基础正文保留、ORM 与迁移结构一致、审核前置条件、正文/来源/文档变更后重新审核、历史快照、软删除引用保护、筛选与分页、并发写入冲突、跨域预检、真实 HTTP 与服务重启持久化。 所有测试正文、来源和审核人均为占位数据，临时数据库不入库。当前问答仍是占位接口，未开展论文对照实验。Issue、合并请求和最终批准结论须补入 Gitea 后按团队规则归档。 + +## 2026-10-15 合并后续开发：关键词检索基线 + +- 阶段 2 后端功能已通过普通功能分支的 [PR #4](https://gitea.kronecker.cc/saint/border-red-post-nalati/pulls/4) 合入 `dev`（合并提交 `4d437ca`），包含测试客户端依赖修正。旧 AGit PR #2 仍打开，不作为该批功能是否已合入的唯一判据。 +- 从最新 `dev` 创建 `feat/backend-keyword-retrieval`，推进 T06 的独立关键词接口与检索记录。T05 的持久化切分/索引版本、T07 的模型评估和向量检索仍待后续实现。 +- 当前主机 Docker 服务仍未运行；本轮继续使用隔离 PostgreSQL 做接口与迁移测试，不把容器或浏览器端联调标记为完成。独立任务 Issue 仍须补齐。 +- 新增检索接口、中文分词和 BM25 基线、有效审核筛选、原文片段、来源与版本、语料指纹及检索历史；前端调用层同步类型与接口，页面接入由前端继续联调。 +- 分支验证：`ruff check .`、`ruff format --check .`、`pytest -q`；本轮 42 项 PostgreSQL/单元/真实 HTTP 测试通过，覆盖手算分数、标题权重、排序复现、审核边界、来源失效、容量拒绝、迁移兼容和服务重启后的检索历史。前端 `npm run build` 通过，包含新增检索调用类型的检查。全部使用占位资料，不是论文实验结果。 diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 6116f01..7b189eb 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -9,3 +9,9 @@ async function catalogRequest(path:string,init?:RequestInit):Promise{const export function getSources(q=''){return catalogRequest<{items:SourceItem[];total:number}>(`/api/sources?q=${encodeURIComponent(q)}`)} export function createSource(payload:{name:string;source_url?:string;notes:string}){return catalogRequest('/api/sources',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(payload)})} export function reviewKnowledge(id:number,payload:{expected_version:number;verification_status:VerificationStatus;reviewer:string;verification_notes:string}){return catalogRequest(`/api/knowledge/${id}/review`,{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(payload)})} + +export type RetrievalHit={rank:number;knowledge_id:number;knowledge_version:number;document_id:number|null;document_version:number|null;source_id:number|null;source_version:number|null;source_name:string|null;source_url:string;title:string;category:string;snippet:string;snippet_start:number;snippet_end:number;score:number;matched_terms:string[]} +export type RetrievalResult={run_id:string;query:string;mode:'keyword';status:'ok'|'no_evidence';top_k:number;category:string|null;query_terms:string[];corpus_size:number;corpus_fingerprint:string;total_hits:number;results:RetrievalHit[];elapsed_ms:number;created_at:string;config:{tokenizer:string;scorer:string;k1:number;b:number;title_weight:number;stopwords_version:string}} +export function retrieveKnowledge(payload:{query:string;mode?:'keyword';top_k?:number;category?:string}){return catalogRequest('/api/retrieval',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(payload)})} +export function getRetrievalRun(runId:string){return catalogRequest(`/api/retrieval/runs/${encodeURIComponent(runId)}`)} +export function getRetrievalRuns(page=1,pageSize=20){return catalogRequest<{items:RetrievalResult[];total:number;page:number;page_size:number}>(`/api/retrieval/runs?page=${page}&page_size=${pageSize}`)}