diff --git a/KNOWLEDGE_RAG_STAGE12.md b/KNOWLEDGE_RAG_STAGE12.md new file mode 100644 index 0000000..1830fe4 --- /dev/null +++ b/KNOWLEDGE_RAG_STAGE12.md @@ -0,0 +1,92 @@ +# 阶段十二：知识检索与 RAG 复杂企划落地与离线评估 + +能力主题键：`CodeSense:knowledge-rag:stage12` + +## 系统地图（先于实现） + +```text +AssignmentKnowledgePoint + | + v +assignment-scoped adapter (services/knowledge_rag.py) + | + v +KnowledgeDocument -> chunk -> embed -> candidate retrieve -> rerank -> citation + | | + v v + prompt context --------------------------------------> /api/ask_question + | + v + JSON / SSE receipt +``` + +当前主干已经有可替换的切分、嵌入、召回、重排和引用契约，但此前它仍是每次请求重建的内存流水线，不是可单独验证的最小向量索引；固定离线问题集和关键词回退也尚未落地。 + +## 观察与可证伪假设 + +### 已确认观察 + +1. `services/knowledge_rag.py::retrieve_assignment_knowledge()` 只读取当前作业的 `AssignmentKnowledgePoint`，并在适配器层最多取 8 条记录。 +2. `services/knowledge_pipeline.py::OfflineKnowledgePipeline.search()` 每次从文档重新切分并计算 embedding，没有独立的向量写入/查询对象，因此无法单独测量索引规模、top-k 命中和回退路径。 +3. 当前没有固定问题集；现有测试能验证排序、引用和安全回退，但不能报告固定样本的 Recall@k 或向量检索耗时。 +4. 查询没有向量命中时没有独立的标题/正文关键词回退；已有作业记录仍按旧优先级结果返回，空作业才进入 `NO_KNOWLEDGE_EVIDENCE`。 + +### 可证伪假设 + +在不改变数据库结构、权限、部署和学生端 JSON/SSE 字段的前提下，引入“每次请求隔离的稀疏内存向量索引 + 有界 top-k + 关键词回退”，可以找回旧 8 条截断会漏掉的后置匹配证据，并让固定问题集中至少一条已知证据命中，同时保持引用完整性和安全无结果回退；索引构建和查询的本地 P95 应保持在可接受的毫秒级范围内。 + +若实验显示索引构建或候选扩展造成明显延迟，保留原有 assignment-scoped 读取和 answer-only 回退，不引入跨请求缓存或外部向量服务。 + +## 目标与边界 + +- 目标：提供可替换的最小内存向量库、top-k 检索、引用输出和关键词回退；用固定 JSON 问题集报告 Recall@1/Recall@k、回退次数和延迟。 +- 真实行为改进：当前最多 8 条候选会漏掉后置但匹配的作业知识点；新索引把读取范围扩大到一个明确的有界上限，再在索引内做 top-k，仍最多返回 8 条证据。 +- 保留：作业隔离、学生私有评分隔离、既有 JSON/SSE 字段、引用 ID、无结果和数据源异常时的 answer-only 回退。 +- 不做：数据库表/字段、权限、生产密钥、部署配置、跨作业索引、外部 embedding、向量数据库和不可逆迁移。 + +## 计划改动 + +1. 新增 `services/knowledge_vector_store.py`：标准库 `NgramCountEmbedder`、显式 `write()` 的 `InMemoryVectorStore`、标题/正文 `KeywordFallbackRetriever` 和 `HybridKnowledgeIndex`。 +2. 在 `knowledge_rag` 中将作业范围候选读取限制为 `MAX_INDEX_DOCUMENTS=64`，构建请求内隔离索引并返回 `retrieval_mode`、`indexed_chunk_count` 等只增不破坏既有字段的指标；最终证据仍由 `MAX_EVIDENCE=8` 限制。 +3. 新增 `tests/fixtures/knowledge_rag_eval.json` 和 `python -m services.knowledge_eval` 评估命令；补充向量命中、关键词回退、top-k、索引隔离、固定评估和异常回退测试。 +4. 保留旧的优先级回退行为，避免未匹配问题改变学生端结果；只有空索引才进入原有 no-result 回退。 + +## 验证计划 + +- 修复前：运行当前基线测试，并用后置第 10 条唯一命中样本证明原有 8 条候选边界会漏检。 +- 修复后：运行定向测试、固定评估集、全量测试、`git diff --check`；报告 Recall@1、Recall@k、fallback 次数和索引查询/构建耗时。 +- 故障实验：让向量索引构建或查询抛出异常，确认原有 answer-only 回退，不向学生暴露内部异常或伪造引用。 + +## AI 建议边界 + +本阶段先基于源码和可运行基线形成观察、假设和指标；实现阶段不接入外部 AI、真实凭据或外部向量服务。若后续比较模型/向量数据库方案，只作为待决策事项，不能替代本地测试和固定评估证据。 + +## 实际结果 + +- 修复前基线：阶段 11 定向命令 `python -m pytest tests/test_knowledge_pipeline.py tests/test_knowledge_rag.py -q --disable-warnings` 为 `13 passed`；其中 10 条作业知识点只读取前 8 条，唯一匹配的第 10 条证据被漏掉。 +- 本次审查指定定向命令：`python -m pytest tests/test_knowledge_eval.py tests/test_knowledge_vector_store.py tests/test_knowledge_rag.py -q --disable-warnings` 为 `19 passed`，退出码 0，37.93s。 +- 全量回归（前一提交 `d370b5f`）：`python -m pytest -q --disable-warnings` 为 `681 passed`，退出码 0，13:26；该结果未在本次标签修订后重跑，warnings 为既有项目噪音和测试环境输出。 +- 固定评估命令：`python -m services.knowledge_eval`；5 个固定问题，4 个有标注问题，最后一题的两个相关文档现在均为真实的数组下标文档（`array-boundary`、`array-indexing`），不再把排序稳定性文档误标为相关文档。按“每题前 k 个去重文档命中数 / 该题相关文档数”计算 Recall@1=0.875、Recall@k=0.875；模式为 3 次 vector、1 次 keyword fallback、1 次 no-result，期望模式不匹配数为 0；共索引 16 个切片。 +- 固定评估的普通问题耗时：切分/索引构建 mean 0.341ms、P95 1.477ms；查询 mean 0.045ms、P95 0.094ms；合计 mean 0.387ms、P95 1.532ms。 +- 固定 64 文档性能样本、64 个切片、100 次查询、top-k=8：构建 0.946ms；查询 mean 0.392ms、P95 0.660ms；总耗时 40.161ms，按运行摊销 0.402ms/次。 +- 行为回归：查询第 10 条唯一知识点时从旧的“前 8 条漏检”变为返回 `assignment-kp:10`；最终结果仍最多 8 条。超过 64 条时第 65 条及以后保持明确的资源边界。 +- 故障实验：向量 embedding 抛出异常时返回 `KNOWLEDGE_RETRIEVAL_UNAVAILABLE`，学生端继续 answer-only，不暴露内部异常或伪造引用。 +- `git diff --check`：通过。 + +### 维护者抽查答复 + +- ID10 能命中、ID70 不能进入结果，是因为 `MAX_INDEX_DOCUMENTS=64` 控制从数据库读取并建立索引的候选文档数：ID10 在前 64 条范围内，ID70 超出范围会被有意截断。`MAX_EVIDENCE=8` 是另一层边界，只控制最终返回给调用方的证据最多 8 条；它不决定是否建立索引。 +- 查询“数据库迁移”时，空索引走 `no_result`，返回 `NO_KNOWLEDGE_EVIDENCE`，保持原有 answer-only 回退；索引含数组/指针文档但没有匹配时走 `priority_fallback`，按稳定优先级返回已有证据，避免有作业知识记录时把学生端结果突然变成无证据。只有索引为空时才进入 no-result 分支。 + +## 事实、推断与未解决问题 + +- 已确认事实：索引只存在于单次调用创建的 `HybridKnowledgeIndex`；未写入数据库、磁盘、Redis 或跨请求全局缓存；候选读取上限为 64，最终证据上限为 8；固定集和回归测试均在无生产凭据的隔离环境运行。 +- 仅属推断：字符 bigram 稀疏向量能减少中文单字误匹配，但不等同于语义 embedding；固定集 Recall=0.875 只说明这 5 个样本，不能代表线上知识库质量。 +- 未解决：每次请求仍会重建索引；64 条是保守资源上限，不是线上容量结论；未验证真实 Redis、外部 AI、生产数据库、并发压力、浏览器全流程或持久化向量库迁移。 + +## 取舍、回滚与后续建议 + +- 采纳：标准库 sparse vector、请求内索引、固定离线集和显式模式指标，原因是可复现、可注入、无需新凭据和迁移。 +- 拒绝：跨请求缓存、外部向量数据库、真实 embedding API 和数据库字段，原因是会扩大数据隔离、权限、成本、部署和回滚边界。 +- 回滚：删除 `knowledge_vector_store.py` 和评估文件，并将 `knowledge_rag.py` 的索引调用恢复为 `knowledge_pipeline.search()`；数据库和部署不需要回滚。 +- 后续：若要提高语义召回，先用脱敏固定集比较候选模型与成本/延迟，再单独评审持久化索引、失效策略、权限和数据迁移；不要把本 PR 的内存原型直接当作生产向量库。 diff --git a/services/knowledge_eval.py b/services/knowledge_eval.py new file mode 100644 index 0000000..cdbbf68 --- /dev/null +++ b/services/knowledge_eval.py @@ -0,0 +1,216 @@ +"""Fixed, offline evaluation for the stage 12 knowledge retrieval prototype.""" + +from __future__ import annotations + +from collections import Counter +import json +from math import ceil +from pathlib import Path +import time +from typing import Any + +from services.knowledge_pipeline import KnowledgeDocument, ParagraphChunker +from services.knowledge_vector_store import HybridKnowledgeIndex + + +DEFAULT_FIXTURE = ( + Path(__file__).resolve().parents[1] + / "tests" + / "fixtures" + / "knowledge_rag_eval.json" +) + + +def load_fixture(path: str | Path = DEFAULT_FIXTURE): + payload = json.loads(Path(path).read_text(encoding="utf-8")) + documents = tuple( + KnowledgeDocument( + document_id=str(item["document_id"]), + title=str(item["title"]), + content=str(item["content"]), + source_type=str(item.get("source_type", "offline-eval")), + priority=float(item.get("priority", 0.0)), + metadata=dict(item.get("metadata", {})), + ) + for item in payload["documents"] + ) + return documents, tuple(payload["queries"]), dict(payload.get("performance", {})) + + +def _p95(values: list[float]) -> float: + if not values: + return 0.0 + ordered = sorted(values) + return ordered[max(0, min(len(ordered) - 1, ceil(len(ordered) * 0.95) - 1))] + + +def _unique_document_ids(candidates, *, limit: int | None = None) -> list[str]: + """Deduplicate chunk hits before calculating document-level recall.""" + + document_ids = [] + seen = set() + for candidate in candidates: + document_id = candidate.chunk.document_id + if document_id in seen: + continue + seen.add(document_id) + document_ids.append(document_id) + if limit is not None and len(document_ids) >= limit: + break + return document_ids + + +def _build_performance_documents(spec: dict[str, Any]) -> tuple[KnowledgeDocument, ...]: + """Create a deterministic fixed-size sample without external files.""" + + count = max(1, int(spec.get("document_count", 64))) + return tuple( + KnowledgeDocument( + document_id=f"performance-{index:03d}", + title=f"性能样本 {index:03d}", + content="固定样本用于边界条件检索和索引性能测量。", + source_type="offline-performance", + priority=1.0, + ) + for index in range(count) + ) + + +def evaluate_fixture( + path: str | Path = DEFAULT_FIXTURE, + *, + clock=time.perf_counter, +) -> dict[str, Any]: + """Run the fixed query set and report recall, modes, and latency.""" + + documents, queries, performance_spec = load_fixture(path) + by_id = {document.document_id: document for document in documents} + chunker = ParagraphChunker() + mode_counts: Counter[str] = Counter() + build_latencies = [] + query_latencies = [] + total_latencies = [] + recall_at_1_values = [] + recall_at_k_values = [] + relevant_case_count = 0 + total_indexed_chunks = 0 + mode_mismatch_count = 0 + case_results = [] + + for case in queries: + scoped_ids = case.get("document_ids") + scoped_documents = ( + [by_id[item] for item in scoped_ids if item in by_id] + if scoped_ids is not None + else list(documents) + ) + total_started_at = clock() + build_started_at = clock() + chunks = tuple( + chunk + for document in scoped_documents + for chunk in chunker.split(document) + ) + index = HybridKnowledgeIndex(chunks) + build_latencies.append((clock() - build_started_at) * 1000.0) + total_indexed_chunks += index.indexed_chunk_count + query_started_at = clock() + result = index.search(case["query"], top_k=int(case.get("top_k", 3))) + query_latencies.append((clock() - query_started_at) * 1000.0) + total_latencies.append((clock() - total_started_at) * 1000.0) + mode_counts[result.mode] += 1 + expected_mode = case.get("expected_mode") + if expected_mode and result.mode != expected_mode: + mode_mismatch_count += 1 + + top_k = int(case.get("top_k", 3)) + actual_ids = _unique_document_ids(result.candidates, limit=top_k) + expected_ids = set(case.get("relevant_document_ids", [])) + case_result = { + "query": case["query"], + "expected_mode": expected_mode, + "actual_mode": result.mode, + "retrieved_document_ids": actual_ids, + } + if not expected_ids: + case_result["recall_at_1"] = None + case_result["recall_at_k"] = None + case_results.append(case_result) + continue + relevant_case_count += 1 + recall_at_1 = len(set(actual_ids[:1]) & expected_ids) / len(expected_ids) + recall_at_k = len(set(actual_ids[:top_k]) & expected_ids) / len(expected_ids) + recall_at_1_values.append(recall_at_1) + recall_at_k_values.append(recall_at_k) + case_result["recall_at_1"] = round(recall_at_1, 3) + case_result["recall_at_k"] = round(recall_at_k, 3) + case_results.append(case_result) + + performance_documents = _build_performance_documents(performance_spec) + performance_total_started_at = clock() + performance_build_started_at = clock() + performance_chunks = tuple( + chunk + for document in performance_documents + for chunk in chunker.split(document) + ) + performance_index = HybridKnowledgeIndex(performance_chunks) + performance_build_ms = (clock() - performance_build_started_at) * 1000.0 + performance_query_latencies = [] + performance_runs = max(1, int(performance_spec.get("runs", 100))) + performance_query = str(performance_spec.get("query", "边界条件")) + performance_top_k = max(1, int(performance_spec.get("top_k", 8))) + for _ in range(performance_runs): + query_started_at = clock() + performance_index.search(performance_query, top_k=performance_top_k) + performance_query_latencies.append((clock() - query_started_at) * 1000.0) + performance_total_ms = (clock() - performance_total_started_at) * 1000.0 + + return { + "query_count": len(queries), + "relevant_query_count": relevant_case_count, + "recall_at_1": round(sum(recall_at_1_values) / relevant_case_count, 3) + if relevant_case_count + else 0.0, + "recall_at_k": round(sum(recall_at_k_values) / relevant_case_count, 3) + if relevant_case_count + else 0.0, + "mode_counts": dict(sorted(mode_counts.items())), + "expected_mode_mismatch_count": mode_mismatch_count, + "mean_build_latency_ms": round(sum(build_latencies) / len(build_latencies), 3) + if build_latencies + else 0.0, + "p95_build_latency_ms": round(_p95(build_latencies), 3), + "mean_query_latency_ms": round(sum(query_latencies) / len(query_latencies), 3) + if query_latencies + else 0.0, + "p95_query_latency_ms": round(_p95(query_latencies), 3), + "mean_total_latency_ms": round(sum(total_latencies) / len(total_latencies), 3) + if total_latencies + else 0.0, + "p95_total_latency_ms": round(_p95(total_latencies), 3), + "indexed_chunks_total": total_indexed_chunks, + "case_results": case_results, + "performance_sample": { + "document_count": len(performance_documents), + "chunk_count": len(performance_chunks), + "runs": performance_runs, + "top_k": performance_top_k, + "build_ms": round(performance_build_ms, 3), + "query_mean_ms": round(sum(performance_query_latencies) / performance_runs, 3), + "query_p95_ms": round(_p95(performance_query_latencies), 3), + "total_ms": round(performance_total_ms, 3), + "total_per_run_ms": round( + performance_total_ms / performance_runs, + 3, + ), + }, + } + + +def main() -> None: + print(json.dumps(evaluate_fixture(), ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/services/knowledge_rag.py b/services/knowledge_rag.py index 6394a7a..48178e4 100644 --- a/services/knowledge_rag.py +++ b/services/knowledge_rag.py @@ -1,10 +1,11 @@ """Bounded, assignment-scoped knowledge retrieval for student answers. -This is an explicit-evidence baseline, not a vector database. It reads only -the knowledge points attached to the current assignment and returns stable, -non-sensitive citations plus a deterministic no-result state. Keeping the -retriever here makes a future index interchangeable without changing the -student-facing answer route. +This is an explicit-evidence, request-scoped vector prototype, not a +persistent vector database. It reads only the knowledge points attached to +the current assignment and returns stable, non-sensitive citations plus a +deterministic fallback state. Keeping the index adapter here makes a future +persistent implementation interchangeable without changing the student-facing +answer route. """ from __future__ import annotations @@ -14,11 +15,14 @@ from models import AssignmentKnowledgePoint, KnowledgePointScore, db from services.knowledge_pipeline import KnowledgeDocument, build_offline_pipeline +from services.knowledge_vector_store import HybridKnowledgeIndex, NgramCountEmbedder MAX_EVIDENCE = 8 +MAX_INDEX_DOCUMENTS = 64 logger = logging.getLogger(__name__) knowledge_pipeline = build_offline_pipeline() +knowledge_vector_embedder = NgramCountEmbedder() NO_KNOWLEDGE_EVIDENCE = { "code": "NO_KNOWLEDGE_EVIDENCE", "message": "当前作业没有已标注知识点，回答仅基于题目和代码。", @@ -34,7 +38,16 @@ def _created_at_value(record): return created_at.isoformat() if created_at else None -def _result(status, evidence, candidate_count, started_at, fallback=None): +def _result( + status, + evidence, + candidate_count, + started_at, + fallback=None, + *, + retrieval_mode=None, + indexed_chunk_count=0, +): hit_count = len(evidence) latency_ms = max(0.0, (time.perf_counter() - started_at) * 1000.0) metrics = { @@ -55,6 +68,8 @@ def _result(status, evidence, candidate_count, started_at, fallback=None): "retrieval_error_fallback": bool( fallback and fallback.get("code") == RETRIEVAL_UNAVAILABLE["code"] ), + "retrieval_mode": retrieval_mode or ("unavailable" if fallback else "unknown"), + "indexed_chunk_count": int(indexed_chunk_count), } return { "status": status, @@ -74,16 +89,24 @@ def retrieve_assignment_knowledge( The retrieval is intentionally assignment-scoped and does not inspect a student's private ``KnowledgePointScore`` rows. With a query, the - replaceable offline pipeline first scores lexical overlap and then uses - the teacher/AI-maintained weight and stable row ID as deterministic - tie-breakers. An empty query preserves priority order. + replaceable offline index first scores sparse-vector overlap, then tries a + title/body keyword fallback, and finally uses the teacher/AI-maintained + weight and stable row ID for the legacy priority fallback. An empty query + preserves priority order. """ started_at = time.perf_counter() try: assignment_id = int(assignment_id) except (TypeError, ValueError): - return _result("no_result", [], 0, started_at, NO_KNOWLEDGE_EVIDENCE.copy()) + return _result( + "no_result", + [], + 0, + started_at, + NO_KNOWLEDGE_EVIDENCE.copy(), + retrieval_mode="no_result", + ) try: bounded_limit = max(1, min(int(limit), MAX_EVIDENCE)) @@ -94,7 +117,7 @@ def retrieve_assignment_knowledge( AssignmentKnowledgePoint.weight.desc(), AssignmentKnowledgePoint.id.asc(), ) - .limit(bounded_limit) + .limit(MAX_INDEX_DOCUMENTS) .all() ) except Exception: @@ -108,6 +131,7 @@ def retrieve_assignment_knowledge( 0, started_at, RETRIEVAL_UNAVAILABLE.copy(), + retrieval_mode="unavailable", ) documents = [] @@ -131,11 +155,34 @@ def retrieve_assignment_knowledge( ) ) - citations = knowledge_pipeline.search( - query, - documents, - limit=bounded_limit, - ) + try: + chunks = tuple( + chunk + for document in documents + for chunk in knowledge_pipeline.chunker.split(document) + ) + index = HybridKnowledgeIndex( + chunks, + embedder=knowledge_vector_embedder, + ) + search_result = index.search(query, top_k=bounded_limit) + citations = tuple( + knowledge_pipeline.citation_builder.build(candidate, rank) + for rank, candidate in enumerate(search_result.candidates, start=1) + ) + except Exception: + db.session.rollback() + logger.exception( + "knowledge vector index failed; using safe answer-only fallback" + ) + return _result( + "unavailable", + [], + len(records), + started_at, + RETRIEVAL_UNAVAILABLE.copy(), + retrieval_mode="unavailable", + ) evidence = [ { "evidence_id": citation.evidence_id, @@ -149,8 +196,23 @@ def retrieve_assignment_knowledge( ] if not evidence: - return _result("no_result", [], len(records), started_at, NO_KNOWLEDGE_EVIDENCE.copy()) - return _result("grounded", evidence, len(records), started_at) + return _result( + "no_result", + [], + len(records), + started_at, + NO_KNOWLEDGE_EVIDENCE.copy(), + retrieval_mode=search_result.mode, + indexed_chunk_count=search_result.indexed_chunk_count, + ) + return _result( + "grounded", + evidence, + len(records), + started_at, + retrieval_mode=search_result.mode, + indexed_chunk_count=search_result.indexed_chunk_count, + ) def build_knowledge_prompt_context(retrieval): diff --git a/services/knowledge_vector_store.py b/services/knowledge_vector_store.py new file mode 100644 index 0000000..1160992 --- /dev/null +++ b/services/knowledge_vector_store.py @@ -0,0 +1,216 @@ +"""Small, request-scoped vector retrieval and keyword fallback primitives. + +The store is deliberately in-memory and sparse. It writes only the chunks +passed by the caller, computes cosine similarity with a replaceable embedder, +and never shares index state between requests. That makes it suitable for an +offline evaluation and a safe adapter prototype before selecting a persistent +vector service. +""" + +from __future__ import annotations + +from collections import Counter +from dataclasses import dataclass +import math +import re +from typing import Mapping, Sequence + +from services.knowledge_pipeline import ( + KnowledgeChunk, + RetrievalCandidate, + StablePriorityReranker, + TextEmbedder, + _stable_chunk_key, +) + + +@dataclass(frozen=True) +class RetrievalResult: + """A bounded result set plus the path that produced it.""" + + mode: str + candidates: tuple[RetrievalCandidate, ...] + indexed_chunk_count: int + + +def _cosine_similarity(left: Mapping[str, float], right: Mapping[str, float]) -> float: + """Calculate cosine similarity for two sparse, non-negative vectors.""" + + if not left or not right: + return 0.0 + dot = sum(value * right.get(key, 0.0) for key, value in left.items()) + left_norm = math.sqrt(sum(value * value for value in left.values())) + right_norm = math.sqrt(sum(value * value for value in right.values())) + if not left_norm or not right_norm: + return 0.0 + return dot / (left_norm * right_norm) + + +class NgramCountEmbedder: + """Transparent sparse vectors with CJK bigrams and whole ASCII tokens.""" + + _TOKEN_RE = re.compile(r"[A-Za-z0-9_]+|[\u4e00-\u9fff]+") + + def embed(self, text: str) -> Mapping[str, float]: + tokens = [] + for match in self._TOKEN_RE.findall(str(text or "").lower()): + if re.fullmatch(r"[\u4e00-\u9fff]+", match): + tokens.extend( + match[index:index + 2] + for index in range(max(1, len(match) - 1)) + ) + else: + tokens.append(match) + counts = Counter(tokens) + total = sum(counts.values()) or 1 + return {token: count / total for token, count in counts.items()} + + +class InMemoryVectorStore: + """Write and query one isolated sparse vector index.""" + + def __init__(self, embedder: TextEmbedder | None = None): + self.embedder = embedder or NgramCountEmbedder() + self._chunks: tuple[KnowledgeChunk, ...] = () + self._embeddings: dict[str, Mapping[str, float]] = {} + + @property + def chunks(self) -> tuple[KnowledgeChunk, ...]: + return self._chunks + + def write(self, chunks: Sequence[KnowledgeChunk]) -> int: + """Replace this index with caller-owned chunks and their embeddings.""" + + self._chunks = tuple(chunks) + self._embeddings = { + chunk.chunk_id: self.embedder.embed(chunk.text) + for chunk in self._chunks + } + return len(self._chunks) + + def search(self, query: str, *, top_k: int = 8) -> tuple[RetrievalCandidate, ...]: + """Return only positive cosine matches, in deterministic top-k order.""" + + query_embedding = self.embedder.embed(query) + if not query_embedding: + return () + + candidates = [] + query_terms = set(query_embedding) + for chunk in self._chunks: + embedding = self._embeddings.get(chunk.chunk_id, {}) + score = _cosine_similarity(query_embedding, embedding) + if score <= 0.0: + continue + candidates.append( + RetrievalCandidate( + chunk=chunk, + score=score, + matched_terms=tuple(sorted(query_terms & set(embedding))), + ) + ) + + ordered = sorted( + candidates, + key=lambda candidate: ( + -candidate.score, + -candidate.chunk.priority, + *_stable_chunk_key(candidate.chunk), + ), + ) + return tuple(ordered[: max(1, int(top_k))]) + + +class KeywordFallbackRetriever: + """Find title/body matches when the vector index has no positive hit.""" + + def retrieve( + self, + query: str, + chunks: Sequence[KnowledgeChunk], + *, + top_k: int = 8, + ) -> tuple[RetrievalCandidate, ...]: + query_terms = set(NgramCountEmbedder().embed(query)) + if not query_terms: + return () + + candidates = [] + for chunk in chunks: + # Titles are intentionally included here: metadata-only labels can + # be useful fallback evidence even when the body vector is sparse. + searchable_terms = set( + NgramCountEmbedder().embed(f"{chunk.title} {chunk.text}") + ) + matched = tuple(sorted(query_terms & searchable_terms)) + if not matched: + continue + candidates.append( + RetrievalCandidate( + chunk=chunk, + score=len(matched) / len(query_terms), + matched_terms=matched, + ) + ) + + return tuple( + StablePriorityReranker().rerank({}, candidates)[: max(1, int(top_k))] + ) + + +class HybridKnowledgeIndex: + """Combine vector top-k, keyword fallback, and legacy priority fallback.""" + + def __init__( + self, + chunks: Sequence[KnowledgeChunk], + *, + embedder: TextEmbedder | None = None, + ): + self.vector_store = InMemoryVectorStore(embedder) + self.vector_store.write(chunks) + self.keyword_fallback = KeywordFallbackRetriever() + + @property + def indexed_chunk_count(self) -> int: + return len(self.vector_store.chunks) + + def search(self, query: str, *, top_k: int = 8) -> RetrievalResult: + """Search with explicit mode reporting and a compatibility fallback.""" + + vector_candidates = self.vector_store.search(query, top_k=top_k) + if vector_candidates: + return RetrievalResult("vector", vector_candidates, self.indexed_chunk_count) + + keyword_candidates = self.keyword_fallback.retrieve( + query, + self.vector_store.chunks, + top_k=top_k, + ) + if keyword_candidates: + return RetrievalResult( + "keyword_fallback", + keyword_candidates, + self.indexed_chunk_count, + ) + + # Preserve the stage 11 behavior for an existing assignment with an + # empty or unmatched question: return stable priority evidence instead + # of changing the student-facing no-result contract. + priority_candidates = tuple( + sorted( + ( + RetrievalCandidate(chunk=chunk, score=0.0) + for chunk in self.vector_store.chunks + ), + key=lambda candidate: ( + -candidate.chunk.priority, + *_stable_chunk_key(candidate.chunk), + ), + )[: max(1, int(top_k))] + ) + return RetrievalResult( + "priority_fallback" if priority_candidates else "no_result", + priority_candidates, + self.indexed_chunk_count, + ) diff --git a/tests/fixtures/knowledge_rag_eval.json b/tests/fixtures/knowledge_rag_eval.json new file mode 100644 index 0000000..dce0378 --- /dev/null +++ b/tests/fixtures/knowledge_rag_eval.json @@ -0,0 +1,71 @@ +{ + "documents": [ + { + "document_id": "array-boundary", + "title": "数组边界", + "content": "检查下标范围，访问元素前确认索引合法。", + "source_type": "offline-eval", + "priority": 1 + }, + { + "document_id": "pointer-lifetime", + "title": "指针生命周期", + "content": "释放后不要解引用指针，使用前确认对象仍然有效。", + "source_type": "offline-eval", + "priority": 1 + }, + { + "document_id": "sorting-stability", + "title": "排序稳定性", + "content": "稳定排序会保持相等元素的原有相对顺序。", + "source_type": "offline-eval", + "priority": 1 + }, + { + "document_id": "array-indexing", + "title": "数组下标", + "content": "数组下标从零开始，访问元素前确认索引合法。", + "source_type": "offline-eval", + "priority": 1 + } + ], + "queries": [ + { + "query": "下标范围", + "relevant_document_ids": ["array-boundary"], + "expected_mode": "vector", + "top_k": 3 + }, + { + "query": "释放后解引用", + "relevant_document_ids": ["pointer-lifetime"], + "expected_mode": "vector", + "top_k": 3 + }, + { + "query": "边界", + "relevant_document_ids": ["array-boundary"], + "expected_mode": "keyword_fallback", + "top_k": 3 + }, + { + "query": "数据库迁移", + "document_ids": [], + "relevant_document_ids": [], + "expected_mode": "no_result", + "top_k": 3 + }, + { + "query": "下标", + "relevant_document_ids": ["array-boundary", "array-indexing"], + "expected_mode": "vector", + "top_k": 1 + } + ], + "performance": { + "document_count": 64, + "runs": 100, + "query": "边界条件", + "top_k": 8 + } +} diff --git a/tests/test_knowledge_eval.py b/tests/test_knowledge_eval.py new file mode 100644 index 0000000..5bc79e5 --- /dev/null +++ b/tests/test_knowledge_eval.py @@ -0,0 +1,118 @@ +import itertools +import json + +from services.knowledge_eval import evaluate_fixture + + +def test_fixed_knowledge_eval_reports_vector_fallback_and_no_result_modes(): + metrics = evaluate_fixture() + + assert metrics["query_count"] == 5 + assert metrics["relevant_query_count"] == 4 + assert metrics["recall_at_1"] == 0.875 + assert metrics["recall_at_k"] == 0.875 + assert metrics["mode_counts"] == { + "keyword_fallback": 1, + "no_result": 1, + "vector": 3, + } + assert metrics["expected_mode_mismatch_count"] == 0 + assert metrics["mean_build_latency_ms"] >= 0 + assert metrics["mean_query_latency_ms"] >= 0 + assert metrics["mean_total_latency_ms"] >= metrics["mean_query_latency_ms"] + assert metrics["performance_sample"]["document_count"] == 64 + assert metrics["performance_sample"]["chunk_count"] == 64 + assert metrics["case_results"][-1]["retrieved_document_ids"] == [ + "array-boundary" + ] + assert metrics["case_results"][-1]["recall_at_k"] == 0.5 + + +def _write_fixture(tmp_path, documents, queries): + path = tmp_path / "knowledge_eval.json" + path.write_text( + json.dumps( + { + "documents": documents, + "queries": queries, + "performance": {"document_count": 1, "runs": 1}, + }, + ensure_ascii=False, + ), + encoding="utf-8", + ) + return path + + +def test_recall_uses_relevant_document_denominator(tmp_path): + path = _write_fixture( + tmp_path, + [ + { + "document_id": "array", + "title": "数组", + "content": "数组边界。", + }, + { + "document_id": "pointer", + "title": "指针", + "content": "指针生命周期。", + }, + ], + [ + { + "query": "数组", + "relevant_document_ids": ["array", "pointer"], + "expected_mode": "vector", + "top_k": 1, + } + ], + ) + + metrics = evaluate_fixture(path) + + assert metrics["recall_at_1"] == 0.5 + assert metrics["recall_at_k"] == 0.5 + + +def test_recall_deduplicates_multiple_chunks_from_one_document(tmp_path): + path = _write_fixture( + tmp_path, + [ + { + "document_id": "long-array", + "title": "数组", + "content": " ".join("数组边界。" for _ in range(100)), + }, + { + "document_id": "pointer", + "title": "指针", + "content": "指针生命周期。", + }, + ], + [ + { + "query": "数组", + "relevant_document_ids": ["long-array", "pointer"], + "expected_mode": "vector", + "top_k": 3, + } + ], + ) + + metrics = evaluate_fixture(path) + + assert metrics["recall_at_k"] == 0.5 + assert metrics["case_results"][0]["retrieved_document_ids"] == ["long-array"] + + +def test_evaluation_total_latency_includes_index_build_time(): + ticks = itertools.count() + + def fake_clock(): + return next(ticks) / 1000 + + metrics = evaluate_fixture(clock=fake_clock) + + assert metrics["mean_build_latency_ms"] > 0 + assert metrics["mean_total_latency_ms"] > metrics["mean_query_latency_ms"] diff --git a/tests/test_knowledge_rag.py b/tests/test_knowledge_rag.py index 5f8e0ae..a8882b7 100644 --- a/tests/test_knowledge_rag.py +++ b/tests/test_knowledge_rag.py @@ -184,7 +184,7 @@ def test_retriever_keeps_numeric_record_order_for_equal_priority(knowledge_conte assert unmatched_query["evidence"][0]["evidence_id"] == "assignment-kp:2" -def test_retriever_only_reranks_the_bounded_candidate_pool(knowledge_context): +def test_retriever_indexes_beyond_legacy_eight_but_keeps_final_top_k(knowledge_context): app, _, assignment_id = knowledge_context with app.app_context(): db.session.add_all( @@ -212,9 +212,35 @@ def test_retriever_only_reranks_the_bounded_candidate_pool(knowledge_context): query="unique-ten", ) - assert retrieval["metrics"]["candidate_count"] == 8 - assert len(retrieval["evidence"]) == 8 - assert all(item["evidence_id"] != "assignment-kp:10" for item in retrieval["evidence"]) + assert retrieval["metrics"]["candidate_count"] == 10 + assert len(retrieval["evidence"]) == 1 + assert retrieval["evidence"][0]["evidence_id"] == "assignment-kp:10" + assert retrieval["metrics"]["retrieval_mode"] == "vector" + + +def test_retriever_keeps_explicit_index_document_cap(knowledge_context): + app, _, assignment_id = knowledge_context + with app.app_context(): + db.session.add_all( + [ + AssignmentKnowledgePoint( + id=index, + assignment_id=assignment_id, + knowledge_point=f"base-{index}", + weight=1.0, + ) + for index in range(1, 71) + ] + ) + db.session.commit() + retrieval = retrieve_assignment_knowledge( + assignment_id, + query="base-70", + ) + + assert retrieval["metrics"]["candidate_count"] == 64 + assert retrieval["metrics"]["indexed_chunk_count"] == 64 + assert all(item["evidence_id"] != "assignment-kp:70" for item in retrieval["evidence"]) def test_ask_question_sse_includes_retrieval_receipt(knowledge_context, monkeypatch): @@ -296,6 +322,29 @@ def filter_by(self, **_kwargs): assert "暂时不可用" in knowledge_rag.render_knowledge_receipt(retrieval) +def test_retriever_returns_safe_fallback_when_vector_index_fails( + knowledge_context, monkeypatch +): + app, _, assignment_id = knowledge_context + + class BrokenEmbedder: + def embed(self, _text): + raise RuntimeError("offline index failure") + + with app.app_context(): + AssignmentKnowledgePoint.add_to_assignment( + assignment_id, + "array", + weight=1.0, + ) + monkeypatch.setattr(knowledge_rag, "knowledge_vector_embedder", BrokenEmbedder()) + retrieval = retrieve_assignment_knowledge(assignment_id, query="数组") + + assert retrieval["status"] == "unavailable" + assert retrieval["fallback"]["code"] == "KNOWLEDGE_RETRIEVAL_UNAVAILABLE" + assert retrieval["metrics"]["retrieval_mode"] == "unavailable" + + def test_ask_question_continues_with_answer_only_when_knowledge_source_is_unavailable( knowledge_context, monkeypatch ): diff --git a/tests/test_knowledge_vector_store.py b/tests/test_knowledge_vector_store.py new file mode 100644 index 0000000..bcee58a --- /dev/null +++ b/tests/test_knowledge_vector_store.py @@ -0,0 +1,88 @@ +from services.knowledge_pipeline import KnowledgeDocument, ParagraphChunker +from services.knowledge_vector_store import HybridKnowledgeIndex + + +def _chunks(*documents): + chunker = ParagraphChunker() + return tuple( + chunk + for document in documents + for chunk in chunker.split(document) + ) + + +def test_vector_index_writes_and_returns_deterministic_top_k(): + index = HybridKnowledgeIndex( + _chunks( + KnowledgeDocument( + "pointer", + "指针生命周期", + "释放后不要解引用指针。", + "offline-eval", + priority=2, + ), + KnowledgeDocument( + "array", + "数组边界", + "检查下标范围后再访问元素。", + "offline-eval", + priority=1, + ), + ) + ) + + first = index.search("下标范围", top_k=1) + second = index.search("下标范围", top_k=1) + + assert first == second + assert first.mode == "vector" + assert first.indexed_chunk_count == 2 + assert [item.chunk.document_id for item in first.candidates] == ["array"] + + +def test_keyword_fallback_can_search_a_title_not_present_in_body_vector(): + index = HybridKnowledgeIndex( + _chunks( + KnowledgeDocument( + "array", + "数组边界", + "检查下标范围。", + "offline-eval", + ) + ) + ) + + result = index.search("边界", top_k=1) + + assert result.mode == "keyword_fallback" + assert [item.chunk.document_id for item in result.candidates] == ["array"] + + +def test_hybrid_index_keeps_unmatched_query_legacy_priority_fallback(): + index = HybridKnowledgeIndex( + _chunks( + KnowledgeDocument("low", "低优先级", "数组。", "offline-eval", priority=1), + KnowledgeDocument("high", "高优先级", "指针。", "offline-eval", priority=2), + ) + ) + + result = index.search("数据库迁移", top_k=2) + + assert result.mode == "priority_fallback" + assert [item.chunk.document_id for item in result.candidates] == ["high", "low"] + + +def test_hybrid_indexes_do_not_share_request_state(): + first = HybridKnowledgeIndex( + _chunks(KnowledgeDocument("first", "第一份", "alpha", "offline-eval")) + ) + second = HybridKnowledgeIndex( + _chunks(KnowledgeDocument("second", "第二份", "beta", "offline-eval")) + ) + + assert [item.chunk.document_id for item in first.search("beta").candidates] == [ + "first" + ] + assert [item.chunk.document_id for item in second.search("beta").candidates] == [ + "second" + ]