Merge pull request 'feat: add chunk BM25 and reproducible RRF retrieval' (#14) from kronecker/feat/backend-rrf-retrieval into dev

README.md

@@ -1,6 +1,6 @@
 # 那拉提文旅智能问答
 
-Vue 3 + TypeScript + Vite 前端；FastAPI + Pydantic + SQLAlchemy 后端；PostgreSQL + pgvector。当前后端提供知识管理、来源、文档、人工审核、关键词检索、历史记录及认证/角色权限，问答仍为占位接口。
+Vue 3 + TypeScript + Vite 前端；FastAPI + Pydantic + SQLAlchemy 后端；PostgreSQL + pgvector。当前后端提供知识管理、来源、文档、人工审核、关键词/向量/RRF 检索、文本与向量发布、历史记录及认证/角色权限，问答仍为占位接口。
 
 ## 启动数据库
 需要可用的 Docker Engine（Windows 可使用 Docker Desktop），在项目根目录运行：
@@ -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)。当前数据库最新迁移为 `0005_chunk_indexes`，索引构建/发布说明见 [切分与索引版本接口](docs/INDEX_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)。当前数据库最新迁移为 `0006_vectors`；文本构建/发布见 [INDEX_API.md](docs/INDEX_API.md)，本地模型及向量发布见 [VECTOR_API.md](docs/VECTOR_API.md)。
 
 ## 启动前端（另开终端）
 ```powershell
@@ -102,7 +102,7 @@ Vite 通常使用 http://localhost:5173。
 - `GET /api/retrieval/runs` 使用标准分页参数；`GET /api/retrieval/runs/{run_id}` 读取当时结果。当前知识修改、失效或删除不会改写历史结果。检索记录保存在数据库，运行日志和实际查询不要提交到 Git。
 - 本基线按请求计算语料词频，面向小规模资料集：筛选后的语料最多 5000 条、标题与正文合计最多 200 万字符。超过容量明确返回 503，不截断语料或伪造空结果；可缩小类别范围，大规模索引留到后续实现。
 
-前端 `src/api.ts` 已提供带类型的检索及历史调用；检索页面交互仍由前端阶段接入。本阶段没有接入 Embedding、向量/混合检索或 LLM，不产生论文实验分数。
+前端 `src/api.ts` 已提供带类型的检索及历史调用；检索页面交互仍由前端阶段接入。chunk 关键词、精确向量与 RRF 的独立调用及比较口径见 [混合检索契约](docs/HYBRID_API.md)。LLM 问答与真实领域评测仍待后续阶段。
 
 前端包括问答、知识库管理、来源与更新、实验评估四个页面。实验评估页面没有虚构数据；所有页面明确标记当前功能状态。
 
@@ -119,7 +119,7 @@ pytest -q
 
 测试需要真实 PostgreSQL + pgvector：设置 `TEST_DATABASE_URL`（`postgresql+psycopg://...`），或另行 `pip install pgserver` 使用可选的本机临时实例。`pgserver` 仅用于测试，不是应用运行依赖；外部测试数据库账号需有建 Schema 权限且已启用 vector 扩展。每个测试使用独立 Schema，不会清空业务表。测试使用的占位正文和审核人不代表真实资料。
 
-迁移命令为 `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`。
+迁移命令为 `alembic upgrade head`，可重复执行。仅撤回向量快照时，备份后配合旧后端执行 `alembic downgrade 0005_chunk_indexes`，保留文本、账号、知识及检索历史；新模式历史的旧代码兼容边界见 [HYBRID_API.md](docs/HYBRID_API.md)。继续撤回切分/索引版本时，备份后执行 `alembic downgrade 0004_auth_sessions`，保留账号、会话、知识和检索记录。继续撤回认证时，备份后执行 `alembic downgrade 0003_retrieval_runs`，删除用户/会话/预算表，保留知识和检索历史。继续执行 `alembic downgrade 0002_catalog` 还会删除检索历史。需要撤回阶段 2 时，先备份，再执行 `alembic downgrade 0001_baseline`；它保留基础知识字段，但会删除新增来源、文档、审核字段与所有历史记录，软删除的基础条目也会重新可见。降级须配合对应旧版后端，当前应用会拒绝旧 Schema。恢复完整数据应使用备份。不要在有业务数据的数据库执行 `downgrade base`。
 
 ## 团队协作文档
 

backend/app/api/routes/retrieval.py

@@ -7,13 +7,23 @@ 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.hybrid import (
+    ChunkKeywordRequest,
+    ChunkKeywordResponse,
+    HybridRequest,
+    HybridResponse,
+)
 from app.schemas.retrieval import RetrievalRequest, RetrievalResponse
 from app.schemas.vectors import VectorRetrievalResponse
 from app.services.auth import require_admin
+from app.services.embedding_runtime import get_embedding_provider
+from app.services.hybrid import retrieve_chunks, retrieve_hybrid
 from app.services.retrieval import retrieve
 
 router = APIRouter(prefix="/retrieval", tags=["retrieval"])
 
+RunResponse = RetrievalResponse | VectorRetrievalResponse | ChunkKeywordResponse | HybridResponse
+
 
 @router.post("", response_model=RetrievalResponse)
 def keyword_retrieval(payload: RetrievalRequest, db: Session = Depends(get_db)):
@@ -21,9 +31,23 @@ def keyword_retrieval(payload: RetrievalRequest, db: Session = Depends(get_db)):
     return retrieve(db, payload)
 
 
+@router.post("/chunks", response_model=ChunkKeywordResponse)
+def chunk_keyword_retrieval(payload: ChunkKeywordRequest, db: Session = Depends(get_db)):
+    return retrieve_chunks(db, payload)
+
+
+@router.post("/hybrid", response_model=HybridResponse)
+def hybrid_retrieval(
+    payload: HybridRequest,
+    db: Session = Depends(get_db),
+    provider=Depends(get_embedding_provider),
+):
+    return retrieve_hybrid(db, payload, provider)
+
+
 @router.get(
     "/runs",
-    response_model=Page[RetrievalResponse | VectorRetrievalResponse],
+    response_model=Page[RunResponse],
     dependencies=[Depends(require_admin)],
 )
 def list_runs(
@@ -47,7 +71,7 @@ def list_runs(
 
 @router.get(
     "/runs/{run_id}",
-    response_model=RetrievalResponse | VectorRetrievalResponse,
+    response_model=RunResponse,
     dependencies=[Depends(require_admin)],
 )
 def read_run(run_id: UUID, db: Session = Depends(get_db)):

backend/app/schemas/hybrid.py

@@ -0,0 +1,95 @@
+"""Chunk-level lexical and hybrid contracts; the item-level baseline stays distinct."""
+
+from datetime import datetime
+from typing import Literal
+from uuid import UUID
+
+from pydantic import BaseModel, Field, model_validator
+
+from app.schemas.catalog import Category, Input
+from app.schemas.indexing import IndexConfig
+from app.schemas.retrieval import RetrievalConfig
+from app.schemas.vectors import SpaceRead, VectorHit
+
+
+class ChunkKeywordRequest(Input):
+    query: str = Field(min_length=1, max_length=500)
+    mode: Literal["keyword_chunks"] = "keyword_chunks"
+    top_k: int = Field(5, ge=1, le=20)
+    category: Category | None = None
+
+
+class HybridRequest(ChunkKeywordRequest):
+    mode: Literal["hybrid"] = "hybrid"
+    candidate_k: int = Field(50, ge=1, le=200)
+
+    @model_validator(mode="after")
+    def sufficient_candidates(self):
+        if self.candidate_k < self.top_k:
+            raise ValueError("candidate_k 不得小于 top_k")
+        return self
+
+
+class ChunkRetrievalConfig(BaseModel):
+    ranking_unit: Literal["chunk"] = "chunk"
+    lexical: RetrievalConfig
+    chunking: IndexConfig
+    tie_break: Literal["knowledge_id,chunk_ordinal,chunk_id"] = (
+        "knowledge_id,chunk_ordinal,chunk_id"
+    )
+
+
+class HybridConfig(ChunkRetrievalConfig):
+    scorer: Literal["rrf-v1"] = "rrf-v1"
+    rrf_k: Literal[60] = 60
+    weights: tuple[Literal[1], Literal[1]] = (1, 1)
+    candidate_k: int
+    vector_scorer: Literal["pgvector-cosine-exact-v1"] = "pgvector-cosine-exact-v1"
+    space: SpaceRead
+
+
+class RankedCandidate(BaseModel):
+    chunk_id: UUID
+    rank: int
+    score: float
+
+
+class HybridHit(VectorHit):
+    lexical_rank: int | None
+    lexical_score: float | None
+    vector_rank: int | None
+    vector_score: float | None
+    lexical_contribution: float
+    vector_contribution: float
+
+
+class ChunkKeywordResponse(BaseModel):
+    run_id: UUID
+    query: str
+    mode: Literal["keyword_chunks"] = "keyword_chunks"
+    status: Literal["ok", "no_evidence"]
+    top_k: int
+    category: str | None
+    query_terms: list[str]
+    text_index_version_id: UUID
+    text_generation: int
+    catalog_revision: int
+    corpus_fingerprint: str
+    corpus_size: int
+    total_hits: int
+    results: list[VectorHit]
+    elapsed_ms: float
+    created_at: datetime
+    config: ChunkRetrievalConfig
+
+
+class HybridResponse(ChunkKeywordResponse):
+    mode: Literal["hybrid"] = "hybrid"
+    vector_version_id: UUID
+    generation: int
+    lexical_total_hits: int
+    vector_total_hits: int
+    lexical_candidates: list[RankedCandidate]
+    vector_candidates: list[RankedCandidate]
+    results: list[HybridHit]
+    config: HybridConfig

backend/app/services/hybrid.py

@@ -0,0 +1,205 @@
+"""Bounded chunk BM25 and equal-weight reciprocal rank fusion on one publication."""
+
+from collections import Counter
+from time import perf_counter
+from uuid import uuid4
+
+from fastapi import HTTPException
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.models.catalog import utc_now
+from app.models.indexing import Chunk
+from app.models.retrieval import RetrievalRun
+from app.schemas.hybrid import (
+    ChunkKeywordRequest,
+    ChunkKeywordResponse,
+    ChunkRetrievalConfig,
+    HybridConfig,
+    HybridHit,
+    HybridRequest,
+    HybridResponse,
+    RankedCandidate,
+)
+from app.schemas.vectors import SpaceRead, VectorHit
+from app.services.indexing import locked_state
+from app.services.retrieval import CONFIG, bm25, tokenize
+from app.services.vectors import (
+    checked_chunks,
+    current_text,
+    encode,
+    rank_vectors,
+    retrieval_scope,
+)
+
+RRF_K = 60
+CHUNK_LEXICAL_CONFIG = CONFIG.model_copy(update={"scorer": "bm25-chunk-body-v1", "title_weight": 0})
+
+
+def chunk_key(chunk):
+    return chunk.knowledge_id, chunk.ordinal, chunk.id
+
+
+def chunks_in_scope(db, text_version, category):
+    statement = select(Chunk).where(Chunk.index_version_id == text_version.id)
+    if category is not None:
+        statement = statement.where(Chunk.provenance["category"].as_string() == category)
+    return list(db.scalars(statement.order_by(Chunk.knowledge_id, Chunk.ordinal, Chunk.id)))
+
+
+def lexical_ranking(chunks, terms):
+    counters = {chunk.id: Counter(tokenize(chunk.text)) for chunk in chunks}
+    scores = bm25(list(counters.values()), terms)
+    ranked = sorted(
+        ((chunk, score) for chunk, score in zip(chunks, scores) if score > 0),
+        key=lambda pair: (-pair[1], chunk_key(pair[0])),
+    )
+    return ranked, counters
+
+
+def candidate_list(rows):
+    return [
+        RankedCandidate(chunk_id=chunk.id, rank=rank, score=score)
+        for rank, (chunk, score) in enumerate(rows, 1)
+    ]
+
+
+def fuse(lexical, vector):
+    """Ranks start at one; an absent candidate contributes zero, never a padded rank."""
+    scores = {}
+    for leg in (lexical, vector):
+        for candidate in leg:
+            key = str(candidate.chunk_id)
+            scores[key] = scores.get(key, 0.0) + 1.0 / (RRF_K + candidate.rank)
+    return scores
+
+
+def hit_fields(chunk, terms, counters):
+    return dict(
+        **chunk.provenance,
+        chunk_id=chunk.id,
+        chunk_ordinal=chunk.ordinal,
+        snippet=chunk.text,
+        snippet_start=chunk.start_offset,
+        snippet_end=chunk.end_offset,
+        matched_terms=[term for term in terms if counters[chunk.id][term]],
+    )
+
+
+def save_run(db, response):
+    db.add(
+        RetrievalRun(
+            id=str(response.run_id),
+            query=response.query,
+            mode=response.mode,
+            corpus_fingerprint=response.corpus_fingerprint,
+            response=response.model_dump(mode="json"),
+            created_at=response.created_at,
+        )
+    )
+    db.commit()
+    return response
+
+
+def response_fields(payload, state, text_version, chunks, terms, total, started):
+    return dict(
+        run_id=uuid4(),
+        query=payload.query,
+        status="ok" if total else "no_evidence",
+        top_k=payload.top_k,
+        category=payload.category,
+        query_terms=terms,
+        text_index_version_id=text_version.id,
+        text_generation=state.generation,
+        catalog_revision=state.catalog_revision,
+        corpus_fingerprint=text_version.corpus_fingerprint,
+        corpus_size=len(chunks),
+        total_hits=total,
+        elapsed_ms=round((perf_counter() - started) * 1000, 3),
+        created_at=utc_now(),
+    )
+
+
+def retrieve_chunks(db: Session, payload: ChunkKeywordRequest):
+    started = perf_counter()
+    terms = list(dict.fromkeys(tokenize(payload.query)))
+    if not terms:
+        raise HTTPException(422, "问题中没有可检索词，请补充具体关键词")
+    state = locked_state(db)
+    text_version, corpus = current_text(db, state)
+    checked_chunks(db, text_version, corpus)
+    chunks = chunks_in_scope(db, text_version, payload.category)
+    ranked, counters = lexical_ranking(chunks, terms)
+    hits = [
+        VectorHit(**hit_fields(chunk, terms, counters), rank=rank, score=score)
+        for rank, (chunk, score) in enumerate(ranked[: payload.top_k], 1)
+    ]
+    return save_run(
+        db,
+        ChunkKeywordResponse(
+            **response_fields(payload, state, text_version, chunks, terms, len(ranked), started),
+            results=hits,
+            config=ChunkRetrievalConfig(lexical=CHUNK_LEXICAL_CONFIG, chunking=text_version.config),
+        ),
+    )
+
+
+def retrieve_hybrid(db: Session, payload: HybridRequest, provider):
+    started = perf_counter()
+    terms = list(dict.fromkeys(tokenize(payload.query)))
+    *_, token = retrieval_scope(db, provider)
+    db.commit()
+    query_vector = encode(provider, [payload.query], query=True)[0]
+    state, vector_state, text_version, version, space, current = retrieval_scope(db, provider)
+    if token != current:
+        raise HTTPException(409, "查询编码期间发布版本或审核依据已变化，请重试")
+    # Hold the shared catalog/publication locks until both rankings and the run are saved.
+    chunks = chunks_in_scope(db, text_version, payload.category)
+    ranked, counters = lexical_ranking(chunks, terms)
+    lexical = candidate_list(ranked[: payload.candidate_k])
+    vector_total, vector_rows = rank_vectors(
+        db, version, space, payload.category, payload.candidate_k, query_vector
+    )
+    vector = candidate_list(
+        [(chunk, max(-1.0, min(1.0, 1.0 - float(distance)))) for chunk, distance in vector_rows]
+    )
+    scores = fuse(lexical, vector)
+    by_id = {chunk.id: chunk for chunk in chunks}
+    ordered = sorted(scores, key=lambda key: (-scores[key], chunk_key(by_id[key])))
+    lexical_by_id = {str(c.chunk_id): c for c in lexical}
+    vector_by_id = {str(c.chunk_id): c for c in vector}
+    hits = []
+    for rank, key in enumerate(ordered[: payload.top_k], 1):
+        left, right = lexical_by_id.get(key), vector_by_id.get(key)
+        hits.append(
+            HybridHit(
+                **hit_fields(by_id[key], terms, counters),
+                rank=rank,
+                score=scores[key],
+                lexical_rank=left.rank if left else None,
+                lexical_score=left.score if left else None,
+                vector_rank=right.rank if right else None,
+                vector_score=right.score if right else None,
+                lexical_contribution=1.0 / (RRF_K + left.rank) if left else 0.0,
+                vector_contribution=1.0 / (RRF_K + right.rank) if right else 0.0,
+            )
+        )
+    return save_run(
+        db,
+        HybridResponse(
+            **response_fields(payload, state, text_version, chunks, terms, len(scores), started),
+            vector_version_id=version.id,
+            generation=vector_state.generation,
+            lexical_total_hits=len(ranked),
+            vector_total_hits=vector_total,
+            lexical_candidates=lexical,
+            vector_candidates=vector,
+            results=hits,
+            config=HybridConfig(
+                lexical=CHUNK_LEXICAL_CONFIG,
+                chunking=text_version.config,
+                candidate_k=payload.candidate_k,
+                space=SpaceRead.model_validate(space),
+            ),
+        ),
+    )

backend/app/services/vectors.py

@@ -287,26 +287,7 @@ def retrieve_vectors(db: Session, payload: VectorRequest, provider: EmbeddingPro
     state, vector, text_version, version, space, current = retrieval_scope(db, provider)
     if token != current:
         raise HTTPException(409, "查询编码期间发布版本或审核依据已变化，请重试")
-    # Extension-qualified operator works even with an isolated business schema search_path.
-    distance = ChunkVector.embedding.op("OPERATOR(public.<=>)", return_type=Float)(query_vector)
-    statement = (
-        select(Chunk, distance.label("distance"))
-        .join(ChunkVector, ChunkVector.chunk_id == Chunk.id)
-        .where(
-            ChunkVector.version_id == version.id,
-            ChunkVector.space_id == space.id,
-            ChunkVector.text_index_version_id == text_version.id,
-            ChunkVector.dimension == space.dimension,
-        )
-    )
-    if payload.category is not None:
-        statement = statement.where(Chunk.provenance["category"].as_string() == payload.category)
-    total = db.scalar(select(func.count()).select_from(statement.subquery()))
-    rows = db.execute(
-        statement.order_by(distance, Chunk.knowledge_id, Chunk.ordinal, Chunk.id).limit(
-            payload.top_k
-        )
-    )
+    total, rows = rank_vectors(db, version, space, payload.category, payload.top_k, query_vector)
     hits = [
         VectorHit(
             **chunk.provenance,
@@ -351,3 +332,25 @@ def retrieve_vectors(db: Session, payload: VectorRequest, provider: EmbeddingPro
     )
     db.commit()
     return result
+
+
+def rank_vectors(db, version, space, category, limit, query_vector):
+    # Extension-qualified operator works even with an isolated business schema search_path.
+    distance = ChunkVector.embedding.op("OPERATOR(public.<=>)", return_type=Float)(query_vector)
+    statement = (
+        select(Chunk, distance.label("distance"))
+        .join(ChunkVector, ChunkVector.chunk_id == Chunk.id)
+        .where(
+            ChunkVector.version_id == version.id,
+            ChunkVector.space_id == space.id,
+            ChunkVector.text_index_version_id == version.text_index_version_id,
+            ChunkVector.dimension == space.dimension,
+        )
+    )
+    if category is not None:
+        statement = statement.where(Chunk.provenance["category"].as_string() == category)
+    total = db.scalar(select(func.count()).select_from(statement.subquery()))
+    rows = db.execute(
+        statement.order_by(distance, Chunk.knowledge_id, Chunk.ordinal, Chunk.id).limit(limit)
+    )
+    return total, list(rows)

backend/tests/test_hybrid.py

@@ -0,0 +1,306 @@
+"""Synthetic coordinates/evidence test fusion semantics, not domain/model quality."""
+
+import math
+from dataclasses import replace
+from uuid import UUID, uuid4
+
+import pytest
+from sqlalchemy import func, select, text
+from sqlalchemy.orm import Session
+from test_indexing import build
+from test_retrieval import add_item
+from test_vectors import ToyProvider, text_version, vector_build, vector_publish
+
+from app.models.retrieval import RetrievalRun
+from app.schemas.hybrid import RankedCandidate
+from app.services import vectors
+from app.services.embedding import EmbeddingError
+from app.services.embedding_runtime import get_embedding_provider
+from app.services.hybrid import fuse
+from app.services.indexing import publish_index
+
+
+@pytest.fixture
+def provider(client):
+    value = ToyProvider()
+    client.app.dependency_overrides[get_embedding_provider] = lambda: value
+    return value
+
+
+def search(client, path="hybrid", **kwargs):
+    return client.post(f"/api/retrieval/{path}", json={"query": "query", **kwargs})
+
+
+def publish_vectors(client):
+    text_ = text_version(client)
+    version = vector_build(client, text_)
+    assert vector_publish(client, version).status_code == 200
+    return text_, version
+
+
+def test_rrf_hand_calculation_absent_leg_and_raw_score_independence():
+    a, b, c = uuid4(), uuid4(), uuid4()
+    lexical = [
+        RankedCandidate(chunk_id=a, rank=1, score=1000),
+        RankedCandidate(chunk_id=b, rank=2, score=1),
+    ]
+    vector = [
+        RankedCandidate(chunk_id=b, rank=1, score=-0.9),
+        RankedCandidate(chunk_id=c, rank=2, score=-1),
+    ]
+    scores = fuse(lexical, vector)
+    assert scores == {str(a): 1 / 61, str(b): 1 / 62 + 1 / 61, str(c): 1 / 62}
+    for candidate in lexical + vector:
+        candidate.score *= 10000
+    assert fuse(lexical, vector) == scores
+    assert fuse([], []) == {}
+
+
+def test_same_chunk_scope_standalone_rankings_formula_and_history(client, provider):
+    items = [
+        add_item(client, content="alpha query extra words words"),
+        add_item(client, content="query"),
+        add_item(client, content="beta"),
+    ]
+    add_item(client, status="pending", content="query alpha")
+    text_, version = publish_vectors(client)
+    response = search(client, top_k=3, candidate_k=3)
+    assert response.status_code == 200, response.text
+    result = response.json()
+    lexical = search(client, "chunks", top_k=3).json()
+    vector = search(client, "vector", top_k=3).json()
+    assert result["corpus_size"] == lexical["corpus_size"] == vector["corpus_size"] == 3
+    assert (
+        result["text_index_version_id"]
+        == lexical["text_index_version_id"]
+        == vector["text_index_version_id"]
+        == text_["id"]
+    )
+    assert result["vector_version_id"] == version["id"]
+    assert result["config"]["space"]["id"] == provider.spec.space_id
+    assert result["config"]["lexical"]["title_weight"] == 0
+    assert result["config"]["lexical"]["k1"] == 1.2
+    assert result["config"]["lexical"]["b"] == 0.75
+    assert result["config"]["chunking"] == text_["config"]
+    assert result["lexical_total_hits"] == 2 and result["vector_total_hits"] == 3
+    assert [c["chunk_id"] for c in result["lexical_candidates"]] == [
+        c["chunk_id"] for c in lexical["results"]
+    ]
+    assert [c["chunk_id"] for c in result["vector_candidates"]] == [
+        c["chunk_id"] for c in vector["results"]
+    ]
+    # Independently calculate positive-log-IDF BM25 on bodies of lengths 5, 1, 1.
+    expected = math.log(1 + 1.5 / 2.5) * 2.2 / (1 + 1.2 * (0.25 + 0.75 / (7 / 3)))
+    assert lexical["results"][0]["knowledge_id"] == items[1]["id"]
+    assert lexical["results"][0]["score"] == pytest.approx(expected)
+    for hit in result["results"]:
+        left = next(
+            (c for c in result["lexical_candidates"] if c["chunk_id"] == hit["chunk_id"]), None
+        )
+        right = next(
+            (c for c in result["vector_candidates"] if c["chunk_id"] == hit["chunk_id"]), None
+        )
+        assert hit["score"] == pytest.approx(
+            (1 / (60 + left["rank"]) if left else 0) + (1 / (60 + right["rank"]) if right else 0)
+        )
+        assert hit["score"] == hit["lexical_contribution"] + hit["vector_contribution"]
+        assert hit["source_url"] == "https://example.com/"
+    assert search(client, candidate_k=3, top_k=3).json()["results"] == result["results"]
+    for value in (result, lexical, vector):
+        assert client.get(f"/api/retrieval/runs/{value['run_id']}").json() == value
+    old = client.post("/api/retrieval", json={"query": "query"}).json()
+    assert old["mode"] == "keyword" and "chunk_id" not in old["results"][0]
+    history = client.get("/api/retrieval/runs?page_size=100").json()
+    assert {r["mode"] for r in history["items"]} == {
+        "keyword",
+        "vector",
+        "keyword_chunks",
+        "hybrid",
+    }
+    client.cookies.clear()
+    assert search(client).status_code == search(client, "chunks").status_code == 200
+    assert client.get("/api/retrieval/runs").status_code == 401
+
+
+def test_candidate_cutoffs_disjoint_legs_and_deterministic_tie(client, provider):
+    first = add_item(client, content="alpha")
+    second = add_item(client, content="query")
+    publish_vectors(client)
+    small = search(client, candidate_k=1, top_k=1).json()
+    assert small["total_hits"] == 2 and len(small["results"]) == 1
+    assert small["results"][0]["knowledge_id"] == first["id"]
+    assert small["results"][0]["lexical_rank"] is None
+    assert small["results"][0]["lexical_score"] is None
+    assert small["results"][0]["lexical_contribution"] == 0
+    assert small["results"][0]["score"] == 1 / 61
+    full = search(client, candidate_k=2, top_k=2).json()
+    assert full["results"][0]["knowledge_id"] == second["id"]
+    assert full["results"][0]["lexical_rank"] == 1 and full["results"][0]["vector_rank"] == 2
+
+
+def test_chunk_bodies_offsets_overlap_and_category_apply_before_both_rankings(client, provider):
+    item = add_item(client, title="uniquetitle", content="alpha query 文本。" * 30, category="交通")
+    add_item(client, content="query", category="其他")
+    publish_vectors(client)
+    result = search(client, category="交通", top_k=20, candidate_k=20).json()
+    assert result["corpus_size"] > 1 and result["total_hits"] == result["corpus_size"]
+    assert len({h["chunk_id"] for h in result["results"]}) == len(result["results"])
+    for hit in result["results"]:
+        assert hit["knowledge_id"] == item["id"]
+        assert hit["snippet"] == item["content"][hit["snippet_start"] : hit["snippet_end"]]
+    assert search(client, "chunks", query="uniquetitle").json()["status"] == "no_evidence"
+    empty = search(client, category="不存在分类").json()
+    assert empty["status"] == "no_evidence" and empty["results"] == []
+    assert empty["lexical_candidates"] == empty["vector_candidates"] == []
+
+
+def test_empty_lexical_leg_and_empty_corpus_are_explicit(client, provider):
+    add_item(client, content="alpha")
+    publish_vectors(client)
+    for query in ["unmatchedword", "！！"]:
+        result = search(client, query=query).json()
+        assert result["lexical_candidates"] == [] and result["status"] == "ok"
+        assert result["results"][0]["score"] == 1 / 61
+    assert search(client, "chunks", query="！！").status_code == 422
+
+
+def test_empty_published_corpus(client, provider):
+    publish_vectors(client)
+    for path in ["chunks", "hybrid"]:
+        result = search(client, path).json()
+        assert result["status"] == "no_evidence" and result["total_hits"] == 0
+
+
+def test_keyword_chunks_need_no_runtime_or_vector_publication(client):
+    add_item(client, content="query")
+    assert search(client, "chunks").status_code == 409
+    text_version(client)
+    assert search(client, "chunks").status_code == 200
+    assert search(client).status_code == 503
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [
+        {"candidate_k": 1, "top_k": 2},
+        {"candidate_k": 201},
+        {"candidate_k": 0},
+        {"top_k": 21},
+        {"top_k": 0},
+        {"mode": "vector"},
+        {"rrf_k": 10},
+        {"query": " "},
+        {"query": "x" * 501},
+    ],
+)
+def test_request_budget_and_fixed_config_validation(client, provider, payload):
+    assert search(client, **payload).status_code == 422
+
+
+@pytest.mark.parametrize("change", ["catalog", "text", "vector"])
+def test_query_encoding_releases_locks_and_rechecks_publication(
+    client, provider, migrated_database, change
+):
+    item = add_item(client, content="alpha query")
+    text_, first = publish_vectors(client)
+    replacement_text = build(client)
+    replacement_vector = vector_build(client, text_)
+
+    def mutate(query):
+        assert query
+        with Session(migrated_database) as other:
+            other.execute(text("SET LOCAL lock_timeout='1s'"))
+            if change == "catalog":
+                state = vectors.locked_state(other)
+                state.catalog_revision += 1
+                other.execute(
+                    text(
+                        "UPDATE knowledge_items SET content='changed synthetic evidence' WHERE id=:id"
+                    ),
+                    {"id": item["id"]},
+                )
+                other.commit()
+            elif change == "text":
+                publish_index(other, UUID(replacement_text["id"]), 1, first["created_by"])
+            else:
+                vectors.publish_vectors(
+                    other, UUID(replacement_vector["id"]), 1, first["created_by"]
+                )
+
+    provider.callback = mutate
+    assert search(client).status_code == 409
+    with Session(migrated_database) as db:
+        assert db.scalar(select(func.count()).select_from(RetrievalRun)) == 0
+
+
+@pytest.mark.parametrize("change", ["body", "chunks", "vectors", "space"])
+def test_corrupt_or_incompatible_snapshot_fails_closed(client, provider, migrated_database, change):
+    add_item(client, content="alpha query")
+    publish_vectors(client)
+    if change == "space":
+        provider.spec = replace(provider.spec, revision="b" * 40)
+    else:
+        with migrated_database.begin() as connection:
+            connection.execute(
+                text(
+                    {
+                        "body": "UPDATE knowledge_items SET content='out of band change'",
+                        "chunks": "UPDATE chunks SET text='tampered'",
+                        "vectors": "DELETE FROM chunk_vectors",
+                    }[change]
+                )
+            )
+    assert search(client).status_code == 409
+    with Session(migrated_database) as db:
+        assert db.scalar(select(func.count()).select_from(RetrievalRun)) == 0
+
+
+@pytest.mark.parametrize("failure,status", [("token", 422), ("runtime", 503)])
+def test_model_failure_has_no_fallback_or_partial_history(
+    client, provider, migrated_database, failure, status
+):
+    add_item(client, content="alpha query")
+    publish_vectors(client)
+
+    def fail(*args, **kwargs):
+        if failure == "token":
+            raise EmbeddingError("token budget")
+        raise OSError("private-runtime-path")
+
+    provider.encode = fail
+    response = search(client)
+    assert response.status_code == status and "private-runtime-path" not in response.text
+    with Session(migrated_database) as db:
+        assert db.scalar(select(func.count()).select_from(RetrievalRun)) == 0
+
+
+def test_failed_history_insert_rolls_back_both_leg_result(client, provider, migrated_database):
+    add_item(client, content="alpha query")
+    publish_vectors(client)
+    with migrated_database.begin() as connection:
+        connection.execute(
+            text("ALTER TABLE retrieval_runs ADD CONSTRAINT test_no_runs CHECK (false) NOT VALID")
+        )
+    assert search(client).status_code == 409
+    with Session(migrated_database) as db:
+        assert db.scalar(select(func.count()).select_from(RetrievalRun)) == 0
+
+
+def test_stale_history_survives_evidence_change_and_capacity_is_explicit(
+    client, provider, migrated_database, monkeypatch
+):
+    item = add_item(client, content="alpha query")
+    publish_vectors(client)
+    result = search(client).json()
+    monkeypatch.setattr(vectors, "MAX_VECTOR_CHUNKS", 0)
+    assert search(client).status_code == search(client, "chunks").status_code == 503
+    monkeypatch.undo()
+    assert (
+        client.patch(
+            f"/api/knowledge/{item['id']}",
+            json={"expected_version": item["version"], "content": "updated synthetic body"},
+        ).status_code
+        == 200
+    )
+    assert search(client).status_code == search(client, "chunks").status_code == 409
+    assert client.get(f"/api/retrieval/runs/{result['run_id']}").json() == result

backend/tests/test_vector_runtime.py

@@ -23,7 +23,7 @@ def test_real_local_model_vectors_retrieval_and_http_restart(migrated_database):
         "HF_HUB_DISABLE_IMPLICIT_TOKEN": "1",
         "TOKENIZERS_PARALLELISM": "false",
     }
-    query = {"query": "合成测试中交通安排应查阅哪份资料？"}
+    query = {"query": "合成测试中交通安排应查阅哪份资料？", "top_k": 3}
     with server(migrated_database, environment_overrides=options) as client:
         ids = []
         for topic, body in [
@@ -75,6 +75,18 @@ def test_real_local_model_vectors_retrieval_and_http_restart(migrated_database):
         assert {h["knowledge_id"] for h in first["results"]} == set(ids)
         assert all(-1 <= h["score"] <= 1 for h in first["results"])
         assert first["config"]["space"]["spec"]["revision"] == SMALL_ZH.revision
+        response = client.post(
+            "/api/retrieval/hybrid", json={**query, "candidate_k": 3}, timeout=60
+        )
+        assert response.status_code == 200, response.text
+        hybrid = response.json()
+        assert hybrid["vector_version_id"] == version["id"]
+        assert hybrid["text_index_version_id"] == text_["id"]
+        assert hybrid["config"]["space"]["spec"]["revision"] == SMALL_ZH.revision
+        assert {h["knowledge_id"] for h in hybrid["results"]} == set(ids)
+        assert hybrid["vector_candidates"]
+        keyword = client.post("/api/retrieval/chunks", json=query).json()
+        assert keyword["mode"] == "keyword_chunks" and keyword["results"]
     with migrated_database.connect() as c:
         assert (
             c.scalar(
@@ -89,3 +101,15 @@ def test_real_local_model_vectors_retrieval_and_http_restart(migrated_database):
         response = client.post("/api/retrieval/vector", json=query, timeout=60)
         assert response.status_code == 200, response.text
         assert response.json()["results"] == first["results"]
+        for result, endpoint in [(hybrid, "hybrid"), (keyword, "chunks")]:
+            assert client.get(f"/api/retrieval/runs/{result['run_id']}").json() == result
+            response = client.post(
+                f"/api/retrieval/{endpoint}",
+                json={**query, **({"candidate_k": 3} if endpoint == "hybrid" else {})},
+                timeout=60,
+            )
+            assert response.status_code == 200, response.text
+            assert response.json()["results"] == result["results"]
+            if endpoint == "hybrid":
+                assert response.json()["lexical_candidates"] == result["lexical_candidates"]
+                assert response.json()["vector_candidates"] == result["vector_candidates"]

docs/ARCHITECTURE.md

@@ -14,7 +14,7 @@ FastAPI + Pydantic
           |-- 来源获取 -> 人工核验 -> 文本切分 -> 生成向量 -> 发布版本
 ```
 
-当前后端实现健康/数据库就绪检查、来源与文档管理、知识增删改查、分页筛选、人工审核、历史快照、认证会话/角色权限及独立关键词检索；问答仍为占位接口。数据库由 Alembic 显式迁移并启用 pgvector，尚无向量字段、向量索引和 RAG。前端来源登记/查询已接入，实验页仍为静态原型，文档、审核、完整分页与检索页面需继续联调。
+当前后端实现健康/数据库就绪检查、来源与文档管理、知识增删改查、分页筛选、人工审核、历史快照、认证会话/角色权限、文本/向量快照及独立关键词/向量/RRF 检索；问答仍为占位接口。数据库由 Alembic 显式迁移并启用 pgvector，RAG 待后续实现。前端来源登记/查询已接入，实验页仍为静态原型，文档、审核、完整分页与检索页面需继续联调。
 
 ## 阶段 2 数据边界
 
@@ -37,7 +37,7 @@ FastAPI + Pydantic
 
 新增 `retrieval_runs` 保存查询、模式、语料指纹和完整结果快照，支持分页读取及按 UUID 复核。零命中如实记录；当前依据修改后失效不会改写历史，但后续检索必须重新检查审核与父记录状态。响应耗时是本次真实读取、分词、排序及片段生成耗时，不含写日志或模型调用。
 
-关键词基线仍按知识条目检索，显示片段最长 240 字符；独立的持久化 chunks 和文本索引版本见下方。当前尚无倒排索引、Embedding、向量检索、RRF 或 RAG。全语料计算明确限制条目数及总字符数，超限报错，适用于首批小规模资料与后续对照实验准备，不能当作生产检索性能结论。
+关键词基线仍按知识条目检索，显示片段最长 240 字符；独立的持久化 chunks 和文本索引版本见下方。该接口不读取 chunks；chunk 与向量/RRF 的独立接口见后文。全语料计算明确限制条目数及总字符数，超限报错，适用于首批小规模资料与后续对照实验准备，不能当作生产检索性能结论。
 
 
 ## 切分与索引版本（T05）
@@ -47,3 +47,10 @@ FastAPI + Pydantic
 目录写入、构建与发布共用状态行锁，目录写入先于父记录锁获取该锁并递增修订号。发布核对代数、修订号和当前语料指纹；失败回滚，保留旧版本。目录变化后旧指针仍可追溯，但 is_current=false，不可据此绕过正式审核。历史快照不会被改写。接口均限管理员并沿用 Cookie/CSRF，完整边界见 [INDEX_API.md](INDEX_API.md)。
 
 这只是后续索引的文本版本基础，关键词基线行为不变；没有生成向量或宣称已实现 ANN/混合检索。同步构建和全局写锁面向当前有限语料，后续异步/增量发布须重新设计并验证并发协议。
+
+
+## 向量与 RRF 检索（T07）
+
+文本发布之后，固定 Embedding 空间的完整向量快照通过独立代数发布，精确余弦检索只使用当前审核与发布版本。编码在发布锁外进行，完成后重新核对语料、空间及发布状态，失败保留旧版本。运行与空间身份见 [VECTOR_API.md](VECTOR_API.md)。
+
+新增 chunk 正文 BM25 与等权 RRF；两路限定同一文本/向量快照和分类，固定 k=60，先各取 candidate_k 再融合。保存完整候选排名、原始分数、贡献、切分/分词/BM25/模型配置及出处。旧条目关键词模式保留，评测不能混同两种证据单元。两路排序至检索记录写入共用当前性锁，面向最多 2000 chunks 的同步精确计算。字段、失败及回滚边界见 [HYBRID_API.md](HYBRID_API.md)。RAG、ANN 与真实领域质量验收仍待后续阶段。

docs/DECISIONS.md

@@ -42,3 +42,10 @@
 状态：T05 分支实现，待审阅。先对当前有效审核语料构建带原文偏移/哈希/出处的 chunks，再以独立的发布请求切换指针。固定 Unicode 字符窗口配置留存在版本中，关键词接口继续按条目运行，防止未经评估就改变实验口径。
 
 为当前小规模语料选择统一数据库行锁：应用目录写入、构建、发布串行，发布同时校验代数、目录修订和语料指纹；失败回滚全部更新，旧版本保留为历史。任何成功目录写入保守地使旧索引过期，历史不能视作当前有效依据。实现、容量与后续并发扩展边界见 [INDEX_API.md](INDEX_API.md)。
+
+
+## ADR-011 同粒度 chunk 排序与等权 RRF
+
+状态：后端分支实现，待审阅。基于已合入的向量快照，新增正文 BM25 的 chunk 基线，与向量模式使用同一文本发布及分类范围；保留条目关键词接口作为独立口径。两路候选各限 candidate_k，使用从 1 起的排名、固定 k=60、等权倒数排名融合，缺席一路贡献为零。保存候选全排名、原始分数、两路贡献和配置，确保实验准备可复核；不把 BM25 与余弦原始分数直接相加。
+
+公式与常数参考 [RRF 原论文](https://cormack.uwaterloo.ca/cormacksigir09-rrf.pdf)，本项目尚未依据核验领域问题集调优。同步精确计算限制 2000 chunks，固定配置、版本一致性、历史兼容与回滚见 [HYBRID_API.md](HYBRID_API.md)。

docs/HYBRID_API.md

@@ -0,0 +1,67 @@
+# Chunk 关键词与 RRF 混合检索
+
+本阶段基于已合入 `dev` 的文本/向量发布，增加可独立调用的 chunk BM25 和 RRF。既有条目关键词接口继续保留。三个 chunk 模式使用相同的当前文本版本、原文 chunk 与分类范围；比较时应固定发布版本、分类和 `top_k`，同时记录混合检索的 `candidate_k`。
+
+| 接口 | 请求 mode | 排序/证据单元 | 前置条件 |
+|---|---|---|---|
+| `POST /api/retrieval` | `keyword` | 知识条目，标题权重 2、正文权重 1 | 当前有效审核语料 |
+| `POST /api/retrieval/chunks` | `keyword_chunks` | chunk 正文 BM25，标题不参与打分 | 当前有效文本发布；无需模型或向量发布 |
+| `POST /api/retrieval/vector` | `vector` | chunk 精确余弦 | 当前有效文本/向量发布，运行模型空间匹配 |
+| `POST /api/retrieval/hybrid` | `hybrid` | 同一批 chunk 的两路 RRF | 与向量模式相同 |
+
+`mode` 可省略，由接口固定；传入其他模式或未知字段返回 422。公开检索无需登录；历史分页及详情继续限管理员。问答接口仍为占位，当前响应不生成 LLM 答案。
+
+## 请求与排名
+
+```json
+{"query":"合成资料中的交通查询入口","mode":"hybrid","top_k":5,"candidate_k":50,"category":"交通"}
+```
+
+- `query` 去除首尾空白后 1–500 字符，`top_k` 默认 5、范围 1–20，`category` 可省略。chunk 关键词请求不接收 `candidate_k`。
+- 混合检索每路最多取 `candidate_k` 条候选，默认 50、范围 1–200，并须不小于 `top_k`。这是固定排名后的候选截断，不会改变文本或输入 token。最终从候选并集返回最多 `top_k` 条。
+- chunk BM25 只对正文分词。沿用固定 Jieba 0.42.1 搜索分词、NFKC/casefold、HMM 关闭、项目词与 `zh-small-v1` 停用词表；查询词去重，`k1=1.2`、`b=0.75`，正值平滑 IDF。语料长度和词频在分类过滤后的全部 chunks 上计算，只将正分候选纳入关键词一路。标题仅作为出处字段。
+- 向量一路复用精确余弦排序，同样先按分类筛选，不设置相似度阈值。负分和零分仍可能进入候选，分数不是事实置信度，`status=ok` 只表示存在排序结果。
+- 两路先按各自原始分数排序；同分按 `knowledge_id, chunk_ordinal, chunk_id` 升序，排名从 1 开始。RRF 使用等权、固定 `k=60`：`score(c) = sum(1 / (60 + rank_leg(c)))`。不在某路候选列表中的 chunk，该路贡献为 0。原始 BM25 和余弦分数不直接相加；融合分数同分沿用相同 ID 次序。排序与日志使用未预先四舍五入的分数。
+- 融合以 chunk ID 去重。同一知识的不同片段（含重叠片段）保留各自排名和原文偏移，不做条目折叠或多样性重排。条目基线与 chunk 模式的分数、命中数、语料数及评测单元不同，不能直接混算指标。
+
+RRF 方法与常数参考 [Cormack、Clarke、Büttcher，SIGIR 2009 原论文](https://cormack.uwaterloo.ca/cormacksigir09-rrf.pdf)。这里把 `k=60` 作为首版可复现配置，并未通过本项目领域问题集调优。
+
+## 响应与复核
+
+全部模式保留原文、字符偏移（左闭右开）、知识/文档/来源 ID 及版本、URL、排名与耗时。chunk 关键词和混合响应另含：
+
+| 字段 | 含义 |
+|---|---|
+| `mode` | `keyword_chunks` 或 `hybrid`，可用于历史类型判别 |
+| `text_index_version_id`, `text_generation`, `catalog_revision` | 本次文本发布和目录修订 |
+| `corpus_fingerprint` | 整份已审核语料的指纹，分类筛选不改变它 |
+| `corpus_size` | 分类范围内参与两路打分的 chunk 数 |
+| `total_hits` | 关键词模式为全部正分 chunk 数；混合模式为两路截断后的候选并集大小，可能大于返回条数 |
+| `query_terms`, `config.lexical`, `config.chunking`, `config.tie_break` | 实际分词、BM25 参数、切分配置、稳定排序规则 |
+| `elapsed_ms` | 语料检查、模型编码（混合模式）、排序及响应组装耗时；不含记录写入及网络时间 |
+
+混合模式还保存：
+
+- `vector_version_id`、`generation`：向量发布版本与代数；`config.space` 包含完整模型 revision、维度、token 预算、指令、实现/池化/归一化与权重格式身份；与向量契约一致。
+- `config.rrf_k/weights/candidate_k/vector_scorer`：融合与候选预算，不允许请求覆盖固定算法参数。
+- `lexical_candidates/vector_candidates`：两路完整的截断候选列表，每条为 `chunk_id/rank/score`，即使未进最终前 `top_k` 也保留。
+- `lexical_total_hits/vector_total_hits`：截断前正分关键词候选数、分类范围内向量候选总数。
+- 命中的 `lexical_rank/lexical_score/vector_rank/vector_score` 和两路 `*_contribution`：缺席的排名/原始分数为 `null`、贡献为 0。`score` 等于两路贡献之和。
+
+结果完整写入现有 `retrieval_runs`，历史接口联合类型支持四种 mode，不需要新增迁移，当前 Schema 仍为 `0006_vectors`。修改依据后历史保持原样，只代表当时快照；重查须通过当前性验证。原始查询和运行记录属于数据库数据，不应提交到仓库。
+
+## 一致性、失败与容量
+
+混合检索先检查发布版本与模型空间，然后提交只读事务释放发布锁，再真实编码查询；编码后重新获取目录/向量状态锁，复核目录修订、文本/向量代数、审核指纹、实际 chunks、完整向量集及空间身份。变化时 409，且不保存部分结果。两路排名及写入检索历史在同一个受锁保护的事务内完成。
+
+关键词一路没有命中或没有分词时，混合模式明确保留空候选列表，可仅有向量一路；纯关键词模式没有可检索词则 422。分类无资料或空的有效发布返回 `no_evidence`。模型 token 超限/输出校验失败返回 422，运行环境不可用或编码忙返回 503；模型失败不降级成一份声称成功的混合结果。
+
+两种新接口均沿用向量阶段最多 2000 chunks、5000 条知识/200 万字符的整体核验预算，超限明确 503。只在分类范围内打分，但当前性及容量检查针对整份发布。文本快照允许的 20000 chunks 不等同于本接口容量。同步精确排序及 BM25 在锁内运行，会短暂串行化目录写入；当前没有 ANN、后台队列或生产吞吐量结论。
+
+运行依赖、固定缓存、默认关闭的本地模型、构建/发布步骤见 [VECTOR_API.md](VECTOR_API.md)。前端新增 `services/hybrid.ts` 两个调用及历史类型；模式切换页面、Docker/浏览器端和真实领域验收仍待交接。
+
+## 验证与回滚
+
+轻量测试使用明确的合成正文及坐标 fixture，覆盖手算 BM25/RRF、原始分数与融合的独立性、候选边界、两路缺席、稳定排序、chunk 偏移、分类、权限、历史、失效/破坏快照、编码期间版本竞争、无部分日志及容量拒绝。真实模型专项沿用 `test_vector_runtime.py` 的显式 opt-in，增加固定 BGE small 的 HTTP 混合检索和进程重启后的两路排名/历史复核。运行命令见 [VECTOR_API.md](VECTOR_API.md)。这些用例验证实际模型/数据库/API 贯通，合成问题不代表领域质量或论文实验结论。
+
+撤回本功能代码即可撤回两个新增接口；数据库无需降级。旧后端历史响应模型不认识 `keyword_chunks/hybrid`，回滚时应限制旧历史端点或先以备份保留新模式记录，避免把新记录交给旧模型解析；不要为回滚而无备份删除检索历史。

docs/TASKS.md

@@ -70,3 +70,12 @@
 - 管理接口统一要求 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 方案选择仍有待完成。
+
+
+## 2026-10-16 T07：合并后的 RRF 混合检索
+
+- 本地模型评估和向量持久化/精确检索分别经 PR #11/#13 合入 `dev`；#9/#10 是被替代的旧 AGit 入口。读取已确认合并结果后，从最新 `dev` 创建 `feat/backend-rrf-retrieval`，遵循每阶段 PR 合并后才开发下一阶段。
+- 新增公开 chunk 正文 BM25 与 RRF 接口，三种 chunk 模式限定同一文本快照及分类；原条目关键词接口保留独立口径。固定 k=60、等权、每路候选预算，记录两路完整候选排名/原始分数、贡献、模型空间、切分/BM25/分词参数与出处。前端仅同步调用与历史类型。
+- 模型编码在发布锁外，完成后重新检查审核语料、文本/向量代数与空间；两路排序和记录保存受同一事务保护。竞争、模型或写入失败不保存部分混合结果；2000 chunks 容量与原向量阶段一致。没有新增数据库迁移。完整字段、比较口径及回滚见 [HYBRID_API.md](HYBRID_API.md)。
+- 常规测试 `pytest -q` **180 passed, 1 skipped**（真实模型专项须显式启用）；新增 27 项覆盖手算排序/融合、候选截断/缺席/同分、相同粒度及分类、原文、历史、权限、编码期间锁释放和版本竞争、损坏/失效快照、失败回滚与容量。`ruff check`、`ruff format --check` 和前端 `npm run build` 通过。真实模型专项在独立 CPU 环境另执行 **1 passed**：固定 BGE small 经真实 HTTP 完成向量/混合/关键词 chunk 查询，进程重启后两路候选排名、结果和历史保持一致。首次专项请求因 candidate_k 小于默认 top_k 被按契约拒绝，修正测试请求后通过；未改变该输入约束。
+- Issue #5 本轮复查无新评论，后端实现已合入，前端登录/CSRF/角色页面及浏览器流程仍待交接；Docker 容器、核验领域语料/问题集、正式模型冻结与领域评测同样未验收。所有新增自动化资料均为合成测试，不作为文旅事实或论文分数。下一阶段须等待本轮 PR 真正合入 `dev`。

frontend/src/api.ts

@@ -1,5 +1,6 @@
 import { apiRequest } from './services/session'
 import type { VectorRetrievalResult } from './services/vectors'
+import type { ChunkKeywordResult, HybridResult } from './services/hybrid'
 
 export const getHealth = () => apiRequest<{status:string}>('/api/health')
 export const askQuestion = (question:string) => apiRequest<{answer:string;status:string;sources:{title:string;source_url?:string|null}[]}>('/api/chat',{method:'POST',body:JSON.stringify({question})})
@@ -15,6 +16,6 @@ export function reviewKnowledge(id:number,payload:{expected_version:number;verif
 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 apiRequest<RetrievalResult>('/api/retrieval',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(payload)})}
-export type RetrievalRunResult = RetrievalResult | VectorRetrievalResult
+export type RetrievalRunResult = RetrievalResult | VectorRetrievalResult | ChunkKeywordResult | HybridResult
 export function getRetrievalRun(runId:string){return apiRequest<RetrievalRunResult>(`/api/retrieval/runs/${encodeURIComponent(runId)}`)}
 export function getRetrievalRuns(page=1,pageSize=20){return apiRequest<{items:RetrievalRunResult[];total:number;page:number;page_size:number}>(`/api/retrieval/runs?page=${page}&page_size=${pageSize}`)}

frontend/src/services/hybrid.ts

@@ -0,0 +1,19 @@
+import { apiRequest } from './session'
+import type { RetrievalHit, RetrievalResult } from '../api'
+import type { IndexVersion } from './indexes'
+import type { EmbeddingSpace } from './vectors'
+
+type ChunkHit = RetrievalHit & { chunk_id: string; chunk_ordinal: number }
+type ChunkConfig = { ranking_unit: 'chunk'; lexical: RetrievalResult['config']; chunking: IndexVersion['config']; tie_break: 'knowledge_id,chunk_ordinal,chunk_id' }
+type ChunkSnapshot = { run_id: string; query: string; status: 'ok' | 'no_evidence'; top_k: number; category: string | null; query_terms: string[]; text_index_version_id: string; text_generation: number; catalog_revision: number; corpus_fingerprint: string; corpus_size: number; total_hits: number; elapsed_ms: number; created_at: string }
+export type ChunkKeywordResult = ChunkSnapshot & { mode: 'keyword_chunks'; results: ChunkHit[]; config: ChunkConfig }
+export type RankedCandidate = { chunk_id: string; rank: number; score: number }
+export type HybridResult = ChunkSnapshot & {
+  mode: 'hybrid'; vector_version_id: string; generation: number; lexical_total_hits: number; vector_total_hits: number;
+  lexical_candidates: RankedCandidate[]; vector_candidates: RankedCandidate[];
+  results: (ChunkHit & { lexical_rank: number | null; lexical_score: number | null; vector_rank: number | null; vector_score: number | null; lexical_contribution: number; vector_contribution: number })[];
+  config: ChunkConfig & { scorer: 'rrf-v1'; rrf_k: 60; weights: [1, 1]; candidate_k: number; vector_scorer: 'pgvector-cosine-exact-v1'; space: EmbeddingSpace }
+}
+type Request = { query: string; top_k?: number; category?: string }
+export const retrieveKeywordChunks = (payload: Request & { mode?: 'keyword_chunks' }) => apiRequest<ChunkKeywordResult>('/api/retrieval/chunks', { method: 'POST', body: JSON.stringify(payload) })
+export const retrieveHybrid = (payload: Request & { mode?: 'hybrid'; candidate_k?: number }) => apiRequest<HybridResult>('/api/retrieval/hybrid', { method: 'POST', body: JSON.stringify(payload) })