diff --git a/KNOWLEDGE_RAG_STAGE14.md b/KNOWLEDGE_RAG_STAGE14.md new file mode 100644 index 0000000..fa4c172 --- /dev/null +++ b/KNOWLEDGE_RAG_STAGE14.md @@ -0,0 +1,113 @@ +# 阶段十四：知识检索与 RAG｜嵌入替换、成本上限与可观测优化 + +能力主题键：`CodeSense:knowledge-rag:stage14` + +## 系统地图 + +```text +AssignmentKnowledgePoint rows + | + v +privacy filter -> chunker -> request-scoped VersionedKnowledgeIndex + | + v + embedding provider boundary + (cjk_ngram default / token candidate) + | + +--> call and cost budget + +--> bounded latency counters + v + vector -> keyword -> priority fallback + | + v + safe citations -> answer-only fallback + | + +--> public bounded metrics + +--> offline quality comparison +``` + +## 先验观察、假设与成功指标 + +### 已确认观察 + +1. `services/knowledge_rag.py::retrieve_assignment_knowledge()` 在当前作业知识点范围内按请求构建索引，阶段十三已提供隐私过滤、版本控制、限流、deadline 和质量监控。 +2. 学生端默认固定使用 `NgramCountEmbedder`；`services/knowledge_pipeline.py` 已有可独立运行的 `TokenCountEmbedder`，但此前没有统一 provider 选择、成本边界或对比报告。 +3. `services/knowledge_eval.py` 已有固定离线集和 recall/latency 统计，可以作为 provider 替换前后的可复现基线。 +4. 对阶段十三主线合并结果做 API 回归时发现：内部 `timeout`、`rate_limited` 状态被 `services/knowledge_evidence.py` 的旧白名单投影成 `unknown`，导致 answer-only 回退的状态丢失。 + +### 可证伪假设 + +在不改数据库结构、权限、部署和核心 JSON/SSE 字段的前提下，把嵌入器放入显式 registry，并对每个请求限制调用次数和预估成本，可以支持离线 provider 替换，同时让 provider、调用数、成本和预算回退可观测；若候选 provider 在固定集上的 `recall_at_k` 不低于基线，则可以作为后续候选。修复公共投影白名单后，超时和限流状态应原样到达 API，而不是变成 `unknown`。 + +### 成功指标 + +- 固定集基线 `recall_at_k=0.875`；候选 provider 的 `recall_at_k` 不低于基线，且通过成本门槛。 +- 当前本地 provider 的预估成本为 `0`；RAG 请求默认成本上限为 `0`，未来付费 provider 必须显式登记单次成本并在上限内运行。 +- 单个请求最多为 64 个索引切片加 1 次查询调用，即 `65` 次 embedding；超限在调用前拒绝并回到 answer-only。 +- provider、调用数、成本、预算拒绝和延迟只以有界计数返回，不保存问题文本、答案或学生隐私。 +- `timeout`、`rate_limited` 继续返回原有 HTTP 200 answer-only 路径，并保留明确回退码。 + +## 变更范围 + +- `services/knowledge_optimization.py`：新增 provider spec/registry、请求级 `BudgetedEmbedder`、调用/成本/延迟快照和环境成本上限 `KNOWLEDGE_RAG_EMBEDDING_MAX_COST`。 +- `services/knowledge_optimization_eval.py`：使用现有固定集比较 `cjk_ngram` 与 `token`，输出 recall、延迟、调用数、成本和质量/成本门禁结果。 +- `services/knowledge_eval.py`：为固定评估增加可注入 embedder，不改变默认评估结果。 +- `services/knowledge_rag.py`：默认仍使用 `cjk_ngram`；显式设置 `KNOWLEDGE_RAG_EMBEDDER=token` 才切换，返回新增的安全 embedding 指标，预算拒绝沿用 unavailable/answer-only 回退。 +- `services/knowledge_evidence.py`：补齐 `timeout`、`rate_limited` 状态、回退码和公共安全指标白名单，避免可靠性状态被投影成 `unknown`。 +- `routes/api.py`：在既有无查询内容日志中补充 provider、调用数、成本和预算状态。 +- `tests/test_knowledge_optimization.py`、`tests/test_knowledge_rag.py`、`tests/test_knowledge_evidence.py`：覆盖预算拒绝、provider 切换、质量/成本对比和可靠性状态投影。 + +不做：数据库表/字段、生产权限、真实 Redis、外部 embedding API、持久化向量库、跨作业缓存、部署变更、学生端既有字段删除或改名。当前 registry 只登记两个标准库离线 provider；它是替换边界和评估工具，不代表已接入生产模型。 + +## 学习总结与 AI 辅助边界 + +本阶段的关键学习是把“模型替换”和“模型一定更好”分开：registry 只解决构造和边界，固定集才负责比较质量，预算封装负责成本和失败前置，回退路径负责兼容性。嵌入调用应按请求隔离，不能把预算计数放到跨请求全局对象，否则一次演练会耗尽后续请求额度。 + +采纳的 AI 辅助建议：保留旧 provider 为默认值；使用固定离线集和可注入时钟/故障对象；把质量门禁和成本门禁写成可测试数据；把超限作为安全回退而不是把异常传播给学生端。未采纳的扩大方案：直接接入外部模型、Redis 全局计费或持久化索引，因为这些会引入凭据、权限、部署、跨实例一致性和数据迁移决策。 + +## 验证命令与结果 + +环境：`pr-stage14-rag-optimization` 隔离 worktree，使用共享 `student-eval` Python；未使用生产数据库、生产 Redis、外部 AI 或生产凭据。 + +基线（改动前）： + +```text +python -m services.knowledge_eval +recall_at_1=0.875, recall_at_k=0.875 +mode_counts=keyword_fallback:1, no_result:1, vector:3 +``` + +阶段十四对比： + +```text +python -m services.knowledge_optimization_eval +cjk_ngram: recall_at_k=0.875, calls=181, estimated_cost=0.0, + quality_gate=True, cost_gate=True, selected=True +token: recall_at_k=0.875, calls=181, estimated_cost=0.0, + quality_gate=True, cost_gate=True, selected=False +``` + +测试与静态检查： + +```text +python -m pytest tests/test_knowledge_evidence.py tests/test_knowledge_reliability.py tests/test_knowledge_rag.py tests/test_knowledge_optimization.py tests/test_knowledge_eval.py -q --disable-warnings +46 passed, exit code 0 + +python -c "from pathlib import Path; [compile(p.read_text(encoding='utf-8'), str(p), 'exec') for p in [Path('services/knowledge_optimization.py'), Path('services/knowledge_optimization_eval.py'), Path('services/knowledge_eval.py'), Path('services/knowledge_rag.py'), Path('services/knowledge_evidence.py')]]" +git diff --check +通过 +``` + +另外，阶段十三既有可靠性集合在本阶段回归中包含限流和超时 API 用例；修复前两例因公共投影返回 `unknown` 失败，修复后与阶段十四测试一起为 `46 passed`。本地日志中的 Redis 未安装/连接失败属于测试环境降级提示，不是本阶段新增失败。 + +## 事实、推断与未解决问题 + +- 已确认事实：默认 provider 未改变；`token` 通过环境变量显式切换；固定集两个 provider 均达到 `recall_at_k=0.875`；成本和调用预算由请求级封装执行；可靠性状态现在能通过公共投影；所有改动不触碰数据库 schema、权限或部署。 +- 仅属推断：固定集只有 5 个查询，不能推出生产数据上的模型优劣；标准库 provider 的零成本不等于外部模型零成本；单次本地延迟不能代表生产 SLA。 +- 未解决：没有接入真实 embedding 模型、生产账单或多实例全局配额；没有管理端质量趋势页面；当前 registry 仅包含离线候选；阶段十三的单进程限流、请求内索引和未验证的真实 Redis/生产数据库边界仍然成立。 + +## 回滚与后续建议 + +回滚只需撤回本阶段新增的两个 optimization 模块、相关测试/文档，并将 `knowledge_eval.py` 的可选 embedder、`knowledge_rag.py` 的请求级 wrapper 和 `knowledge_evidence.py` 的状态白名单恢复到本 PR 前版本；不需要数据库或部署回滚。阶段十三的 `timeout`/`rate_limited` 投影修复应保留，除非维护者明确恢复旧 API 行为。 + +后续建议：先由维护者提供脱敏、规模更大的评估集和真实成本口径，再决定是否登记外部 embedding provider；若需要跨实例配额，应单独评审 Redis/账单/权限方案，不把本阶段的进程内计数直接扩展成生产全局配额。 diff --git a/KNOWLEDGE_RAG_STAGE15.md b/KNOWLEDGE_RAG_STAGE15.md new file mode 100644 index 0000000..e789050 --- /dev/null +++ b/KNOWLEDGE_RAG_STAGE15.md @@ -0,0 +1,52 @@ +# 阶段十五：真实学生入口集成与失败可恢复路径 + +能力主题键：`CodeSense:knowledge-rag:stage15` + +## 真实入口 + +本次选择学生代码工作台的聊天入口： + +```text +学生打开 /submit/ + -> submit_code.html 的 sendToAI() + -> POST /api/code_advice（SSE） + -> assignment-scoped knowledge retrieval + -> AI done/error 事件 + -> 页面显示回答和知识证据回执 +``` + +阶段十四已经让 `/api/code_advice` 使用可预算的 RAG provider，并在成功的 `done` 事件中返回安全的 `knowledge_evidence`。本阶段关注真实使用中的失败边界：流式 AI 在已经输出部分内容后中断时，旧前端会吞掉 `error` 事件，只留下空白状态，且用户看不到本次检索是否成功。 + +## 改动 + +- `routes/api.py`：代码建议 SSE 的空回答、LLM 流中断和未知异常都通过同一失败事件构造器返回；如果本次请求有作业知识证据，则以现有安全投影同时放在事件顶层和 `data` 中。没有作业上下文时不新增跨作业检索数据。 +- `templates/submit_code.html`：不再吞掉 SSE 错误。前端停止继续消费失败流，将错误交给统一失败处理，向学生显示可理解的错误信息，并保留已经得到的知识证据回执。正常 `done` 和旧的非流式响应路径保持不变。 +- `tests/test_code_advice_knowledge.py`：补充真实 `/api/code_advice` SSE 中断集成测试，验证回答失败时证据状态仍可见。 +- `tests/test_knowledge_evidence_integration.py`：验证学生提交页包含证据回执挂载点、流中断提示和错误状态传递钩子。 + +## 事实与边界 + +已验证：真实 Flask 登录会话、作业权限、作业级知识检索、`/api/code_advice` SSE 成功路径和中断路径；中断时不会把私有评分或 prompt 放入公开事件。 + +未验证：真实外部 AI provider、浏览器端真实网络断线、生产 Redis、生产数据库和部署环境。浏览器仍可能在 TCP 断开时无法取得服务端错误事件，此时由页面已有的网络异常提示兜底。 + +## 回滚与复现 + +回滚本次提交即可恢复原前端和 SSE 事件行为；不涉及数据库结构、权限、部署和核心成功响应字段。 + +在隔离 worktree 中运行： + +```powershell +python -m pytest tests/test_code_advice_knowledge.py tests/test_knowledge_evidence_integration.py -q --disable-warnings +python -m pytest -q --disable-warnings +``` + +实际结果：集成测试 `7 passed`；全量测试 `749 passed`，退出码 0，耗时约 17 分 17 秒。Python 静态编译、`git diff --check` 和 10 个不含 Jinja 占位符的页面脚本块语法校验均通过；含 Jinja 变量的 1 个脚本块需在 Flask 渲染后校验。 + +测试结果和 commit SHA 也会写入 PR 描述及任务台验证结果。 + +## 后续风险 + +1. 真实浏览器断网和代理中断仍需补端到端演练。 +2. 生产 provider 的成本、超时和全局配额仍属于阶段十四未决边界。 +3. 任务完成仍需 PR 评审和隔离验证通过。 diff --git a/routes/api.py b/routes/api.py index e534998..e3dbb48 100644 --- a/routes/api.py +++ b/routes/api.py @@ -408,7 +408,8 @@ def _retrieve_knowledge_context(assignment_id, query="", *, limit=MAX_EVIDENCE): "knowledge_rag status=%s candidates=%s hits=%s latency_ms=%.2f " "citation_completeness=%.3f no_result_fallback=%s " "retrieval_error_fallback=%s retrieval_mode=%s indexed_chunks=%s " - "fallback_code=%s", + "fallback_code=%s embedding_provider=%s embedding_calls=%s " + "embedding_cost=%s embedding_budget_exceeded=%s", retrieval["status"], metrics.get("candidate_count", 0), metrics.get("hit_count", 0), @@ -419,6 +420,10 @@ def _retrieve_knowledge_context(assignment_id, query="", *, limit=MAX_EVIDENCE): metrics.get("retrieval_mode", "unknown"), metrics.get("indexed_chunk_count", 0), (retrieval.get("fallback") or {}).get("code"), + metrics.get("embedding_provider"), + metrics.get("embedding_calls", 0), + metrics.get("embedding_estimated_cost", 0.0), + metrics.get("embedding_budget_exceeded", False), ) return retrieval @@ -1253,6 +1258,19 @@ def generate(): 'type': 'start', 'message': '正在根据你的问题分析代码...' }) + + def stream_error_event(code, message): + """Keep the evidence receipt available on a failed stream.""" + payload = { + 'type': 'error', + 'error': code, + 'message': message, + } + if knowledge_fields: + payload.update(knowledge_fields) + payload['data'] = dict(knowledge_fields) + return payload + try: for content in shared_client.chat_stream( messages, @@ -1269,11 +1287,10 @@ def generate(): full_content = ''.join(chunks) if not full_content: - yield sse_event({ - 'type': 'error', - 'error': 'AI_EMPTY_RESPONSE', - 'message': 'AI服务未返回有效内容，请稍后重试', - }) + yield sse_event(stream_error_event( + 'AI_EMPTY_RESPONSE', + 'AI服务未返回有效内容，请稍后重试', + )) return yield sse_event({ 'type': 'done', @@ -1287,18 +1304,16 @@ def generate(): **knowledge_fields, }) except LLMServiceError as exc: - yield sse_event({ - 'type': 'error', - 'error': exc.code, - 'message': 'AI服务流式输出中断，请稍后重试', - }) + yield sse_event(stream_error_event( + exc.code, + 'AI服务流式输出中断，请稍后重试', + )) except Exception as exc: current_app.logger.warning('代码建议流式输出失败: %s', type(exc).__name__) - yield sse_event({ - 'type': 'error', - 'error': 'AI_STREAM_FAILED', - 'message': 'AI服务流式输出失败，请稍后重试', - }) + yield sse_event(stream_error_event( + 'AI_STREAM_FAILED', + 'AI服务流式输出失败，请稍后重试', + )) return sse_response(generate()) diff --git a/services/knowledge_eval.py b/services/knowledge_eval.py index cdbbf68..6dd92d5 100644 --- a/services/knowledge_eval.py +++ b/services/knowledge_eval.py @@ -1,4 +1,4 @@ -"""Fixed, offline evaluation for the stage 12 knowledge retrieval prototype.""" +"""Fixed, offline evaluation for the knowledge retrieval prototype.""" from __future__ import annotations @@ -80,8 +80,9 @@ def evaluate_fixture( path: str | Path = DEFAULT_FIXTURE, *, clock=time.perf_counter, + embedder=None, ) -> dict[str, Any]: - """Run the fixed query set and report recall, modes, and latency.""" + """Run the fixed query set with an optional replaceable embedder.""" documents, queries, performance_spec = load_fixture(path) by_id = {document.document_id: document for document in documents} @@ -111,7 +112,7 @@ def evaluate_fixture( for document in scoped_documents for chunk in chunker.split(document) ) - index = HybridKnowledgeIndex(chunks) + index = HybridKnowledgeIndex(chunks, embedder=embedder) build_latencies.append((clock() - build_started_at) * 1000.0) total_indexed_chunks += index.indexed_chunk_count query_started_at = clock() @@ -154,7 +155,7 @@ def evaluate_fixture( for document in performance_documents for chunk in chunker.split(document) ) - performance_index = HybridKnowledgeIndex(performance_chunks) + performance_index = HybridKnowledgeIndex(performance_chunks, embedder=embedder) performance_build_ms = (clock() - performance_build_started_at) * 1000.0 performance_query_latencies = [] performance_runs = max(1, int(performance_spec.get("runs", 100))) @@ -166,7 +167,7 @@ def evaluate_fixture( performance_query_latencies.append((clock() - query_started_at) * 1000.0) performance_total_ms = (clock() - performance_total_started_at) * 1000.0 - return { + result = { "query_count": len(queries), "relevant_query_count": relevant_case_count, "recall_at_1": round(sum(recall_at_1_values) / relevant_case_count, 3) @@ -206,6 +207,9 @@ def evaluate_fixture( ), }, } + if embedder is not None and hasattr(embedder, "snapshot"): + result["embedding_usage"] = embedder.snapshot() + return result def main() -> None: diff --git a/services/knowledge_evidence.py b/services/knowledge_evidence.py index 817a92e..053d0fa 100644 --- a/services/knowledge_evidence.py +++ b/services/knowledge_evidence.py @@ -16,9 +16,12 @@ MAX_EVIDENCE = 8 MAX_CANDIDATES = 64 MAX_INDEXED_CHUNKS = 64 +MAX_EMBEDDING_CALLS = 128 MAX_LATENCY_MS = 600_000.0 -_ALLOWED_STATUSES = frozenset({"grounded", "no_result", "unavailable"}) +_ALLOWED_STATUSES = frozenset( + {"grounded", "no_result", "unavailable", "timeout", "rate_limited"} +) _ALLOWED_AUDIENCES = frozenset({"teacher", "admin"}) _ALLOWED_RETRIEVAL_MODES = frozenset( { @@ -27,6 +30,8 @@ "priority_fallback", "no_result", "unavailable", + "timeout", + "rate_limited", "unknown", } ) @@ -37,10 +42,14 @@ _FALLBACK_CODES = { "no_result": "NO_KNOWLEDGE_EVIDENCE", "unavailable": "KNOWLEDGE_RETRIEVAL_UNAVAILABLE", + "timeout": "KNOWLEDGE_RETRIEVAL_TIMEOUT", + "rate_limited": "KNOWLEDGE_RETRIEVAL_RATE_LIMITED", } _FALLBACK_MESSAGES = { "no_result": "当前作业没有已标注知识点，回答仅基于题目和代码。", "unavailable": "知识证据暂时不可用，回答仅基于题目和代码。", + "timeout": "知识证据检索超时，回答仅基于题目和代码。", + "rate_limited": "知识证据请求过于频繁，回答仅基于题目和代码。", } _STATUS_COPY = { "grounded": { @@ -58,6 +67,16 @@ "summary": "知识证据暂时不可用，但基础指导仍可继续。", "next_step": "继续查看基础指导，稍后重试证据检索。", }, + "timeout": { + "status_label": "证据检索超时", + "summary": "知识证据检索超时，但基础指导仍可继续。", + "next_step": "继续查看基础指导，稍后重试证据检索。", + }, + "rate_limited": { + "status_label": "证据请求受限", + "summary": "知识证据请求过于频繁，但基础指导仍可继续。", + "next_step": "稍后重试证据检索。", + }, "unknown": { "status_label": "证据状态不可用", "summary": "当前无法确认知识证据状态。", @@ -200,8 +219,8 @@ def build_public_knowledge_retrieval( metrics = raw_metrics if isinstance(raw_metrics, Mapping) else {} if status == "unknown": retrieval_mode = "unknown" - elif status == "unavailable": - retrieval_mode = "unavailable" + elif status in {"unavailable", "timeout", "rate_limited"}: + retrieval_mode = status else: retrieval_mode = _safe_retrieval_mode(metrics.get("retrieval_mode")) @@ -257,6 +276,32 @@ def build_public_knowledge_retrieval( metrics.get("indexed_chunk_count"), maximum=MAX_INDEXED_CHUNKS, ), + "embedding_provider": _safe_text( + metrics.get("embedding_provider"), + limit=64, + ) or None, + "embedding_calls": _safe_nonnegative_int( + metrics.get("embedding_calls"), + maximum=MAX_EMBEDDING_CALLS, + ), + "embedding_estimated_cost": _safe_nonnegative_float( + metrics.get("embedding_estimated_cost") + ), + "embedding_budget_exceeded": ( + bool(metrics.get("embedding_budget_exceeded")) + if isinstance(metrics.get("embedding_budget_exceeded"), bool) + else False + ), + "retrieval_timeout_fallback": ( + bool(metrics.get("retrieval_timeout_fallback")) + if isinstance(metrics.get("retrieval_timeout_fallback"), bool) + else status == "timeout" + ), + "rate_limit_fallback": ( + bool(metrics.get("rate_limit_fallback")) + if isinstance(metrics.get("rate_limit_fallback"), bool) + else status == "rate_limited" + ), }, "fallback": fallback, } diff --git a/services/knowledge_optimization.py b/services/knowledge_optimization.py new file mode 100644 index 0000000..9d0e270 --- /dev/null +++ b/services/knowledge_optimization.py @@ -0,0 +1,190 @@ +"""Bounded, replaceable embedding providers for the knowledge RAG adapter. + +The registry is intentionally offline-first. It makes provider selection +explicit, records only bounded cost/latency counters, and rejects a request +before a configured budget can be exceeded. A future model provider can be +added behind the same contract without changing the student-facing route. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import os +import threading +import time +from typing import Callable, Mapping + +from services.knowledge_pipeline import TextEmbedder, TokenCountEmbedder +from services.knowledge_vector_store import NgramCountEmbedder + + +DEFAULT_EMBEDDING_MAX_COST = 0.0 +MAX_EMBEDDING_COST = 1000.0 + + +class EmbeddingBudgetExceeded(RuntimeError): + """Raised before an embedding call would exceed its configured budget.""" + + +@dataclass(frozen=True) +class EmbeddingProviderSpec: + """Construction and offline cost metadata for one embedding provider.""" + + name: str + factory: Callable[[], TextEmbedder] + estimated_cost_per_call: float = 0.0 + + def __post_init__(self): + if not str(self.name).strip(): + raise ValueError("embedding provider name is required") + if self.estimated_cost_per_call < 0: + raise ValueError("embedding provider cost cannot be negative") + + +class BudgetedEmbedder: + """Wrap an embedder with per-request call, cost, and latency accounting.""" + + def __init__( + self, + embedder: TextEmbedder, + *, + provider_name: str, + estimated_cost_per_call: float = 0.0, + max_calls: int | None = None, + max_estimated_cost: float | None = None, + clock=time.perf_counter, + ): + self._embedder = embedder + self.provider_name = str(provider_name) + self.estimated_cost_per_call = max(0.0, float(estimated_cost_per_call)) + self.max_calls = None if max_calls is None else max(1, int(max_calls)) + self.max_estimated_cost = ( + None + if max_estimated_cost is None + else max(0.0, float(max_estimated_cost)) + ) + self._clock = clock + self._calls = 0 + self._failed_calls = 0 + self._budget_rejections = 0 + self._estimated_cost = 0.0 + self._total_latency_ms = 0.0 + self._lock = threading.Lock() + + def _reserve(self) -> None: + with self._lock: + next_cost = self._estimated_cost + self.estimated_cost_per_call + if self.max_calls is not None and self._calls >= self.max_calls: + self._budget_rejections += 1 + raise EmbeddingBudgetExceeded( + f"embedding call budget exceeded for {self.provider_name}" + ) + if ( + self.max_estimated_cost is not None + and next_cost > self.max_estimated_cost + 1e-12 + ): + self._budget_rejections += 1 + raise EmbeddingBudgetExceeded( + f"embedding cost budget exceeded for {self.provider_name}" + ) + self._calls += 1 + self._estimated_cost = next_cost + + def embed(self, text: str) -> Mapping[str, float]: + self._reserve() + started_at = self._clock() + try: + return self._embedder.embed(text) + except Exception: + with self._lock: + self._failed_calls += 1 + raise + finally: + elapsed_ms = max(0.0, (self._clock() - started_at) * 1000.0) + with self._lock: + self._total_latency_ms += elapsed_ms + + def snapshot(self) -> dict[str, float | int | str | bool | None]: + """Return bounded metrics without retaining text or query content.""" + + with self._lock: + return { + "provider": self.provider_name, + "calls": self._calls, + "failed_calls": self._failed_calls, + "budget_rejections": self._budget_rejections, + "estimated_cost": round(self._estimated_cost, 8), + "total_latency_ms": round(self._total_latency_ms, 3), + "max_calls": self.max_calls, + "max_estimated_cost": self.max_estimated_cost, + "budget_exceeded": bool(self._budget_rejections), + } + + +class KnowledgeEmbeddingRegistry: + """Small explicit registry for offline and future embedding providers.""" + + def __init__(self, specs: tuple[EmbeddingProviderSpec, ...] = ()): + self._specs: dict[str, EmbeddingProviderSpec] = {} + for spec in specs: + self.register(spec) + + def register(self, spec: EmbeddingProviderSpec) -> None: + name = str(spec.name).strip() + if name in self._specs: + raise ValueError(f"embedding provider already registered: {name}") + self._specs[name] = spec + + def names(self) -> tuple[str, ...]: + return tuple(sorted(self._specs)) + + def build( + self, + name: str, + *, + max_calls: int | None = None, + max_estimated_cost: float | None = None, + clock=time.perf_counter, + ) -> BudgetedEmbedder: + normalized = str(name).strip() + try: + spec = self._specs[normalized] + except KeyError as exc: + available = ", ".join(self.names()) or "" + raise ValueError( + f"unknown embedding provider {normalized!r}; available: {available}" + ) from exc + return BudgetedEmbedder( + spec.factory(), + provider_name=spec.name, + estimated_cost_per_call=spec.estimated_cost_per_call, + max_calls=max_calls, + max_estimated_cost=max_estimated_cost, + clock=clock, + ) + + +def build_default_embedding_registry() -> KnowledgeEmbeddingRegistry: + """Build the local registry; both providers are deterministic and free.""" + + return KnowledgeEmbeddingRegistry( + ( + EmbeddingProviderSpec("cjk_ngram", NgramCountEmbedder), + EmbeddingProviderSpec("token", TokenCountEmbedder), + ) + ) + + +def default_embedding_max_cost() -> float: + """Read a bounded per-request cost cap, defaulting to free local models.""" + + try: + value = float( + os.environ.get( + "KNOWLEDGE_RAG_EMBEDDING_MAX_COST", + DEFAULT_EMBEDDING_MAX_COST, + ) + ) + except (TypeError, ValueError): + value = DEFAULT_EMBEDDING_MAX_COST + return max(0.0, min(MAX_EMBEDDING_COST, value)) diff --git a/services/knowledge_optimization_eval.py b/services/knowledge_optimization_eval.py new file mode 100644 index 0000000..e83f550 --- /dev/null +++ b/services/knowledge_optimization_eval.py @@ -0,0 +1,79 @@ +"""Offline quality/cost comparison for the stage 14 embedding seam.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from services.knowledge_eval import DEFAULT_FIXTURE, evaluate_fixture +from services.knowledge_optimization import build_default_embedding_registry + + +DEFAULT_PROVIDERS = ("cjk_ngram", "token") + + +def compare_embedders( + path: str | Path = DEFAULT_FIXTURE, + *, + providers: tuple[str, ...] = DEFAULT_PROVIDERS, + max_calls: int = 512, + max_estimated_cost: float = 0.0, + quality_floor: float = 1.0, +) -> dict[str, Any]: + """Compare providers on the fixed set and apply quality/cost gates.""" + + registry = build_default_embedding_registry() + reports = {} + baseline_name = providers[0] if providers else None + for provider_name in providers: + embedder = registry.build( + provider_name, + max_calls=max_calls, + max_estimated_cost=max_estimated_cost, + ) + metrics = evaluate_fixture(path, embedder=embedder) + reports[provider_name] = { + "recall_at_1": metrics["recall_at_1"], + "recall_at_k": metrics["recall_at_k"], + "mean_query_latency_ms": metrics["mean_query_latency_ms"], + "p95_query_latency_ms": metrics["p95_query_latency_ms"], + "mean_total_latency_ms": metrics["mean_total_latency_ms"], + "embedding_usage": metrics["embedding_usage"], + } + + baseline_recall = ( + reports[baseline_name]["recall_at_k"] if baseline_name in reports else 0.0 + ) + selected_provider = None + for provider_name in providers: + report = reports[provider_name] + usage = report["embedding_usage"] + quality_ok = report["recall_at_k"] >= baseline_recall * float(quality_floor) + cost_ok = ( + not usage["budget_exceeded"] + and usage["estimated_cost"] <= float(max_estimated_cost) + 1e-12 + ) + report["quality_gate"] = quality_ok + report["cost_gate"] = cost_ok + report["selected"] = selected_provider is None and quality_ok and cost_ok + if report["selected"]: + selected_provider = provider_name + + return { + "fixture": str(Path(path)), + "baseline_provider": baseline_name, + "quality_floor": float(quality_floor), + "max_calls": max(1, int(max_calls)), + "max_estimated_cost": float(max_estimated_cost), + "providers": reports, + "selected_provider": selected_provider, + } + + +def main() -> None: + print(json.dumps(compare_embedders(), ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/services/knowledge_rag.py b/services/knowledge_rag.py index 655adf0..c5f54de 100644 --- a/services/knowledge_rag.py +++ b/services/knowledge_rag.py @@ -11,9 +11,16 @@ from __future__ import annotations import logging +import os import time from models import AssignmentKnowledgePoint, KnowledgePointScore, db +from services.knowledge_optimization import ( + BudgetedEmbedder, + EmbeddingBudgetExceeded, + build_default_embedding_registry, + default_embedding_max_cost, +) from services.knowledge_pipeline import KnowledgeDocument, build_offline_pipeline from services.knowledge_reliability import ( KnowledgePrivacyFilter, @@ -31,6 +38,7 @@ logger = logging.getLogger(__name__) knowledge_pipeline = build_offline_pipeline() knowledge_vector_embedder = NgramCountEmbedder() +knowledge_embedding_registry = build_default_embedding_registry() NO_KNOWLEDGE_EVIDENCE = { "code": "NO_KNOWLEDGE_EVIDENCE", "message": "当前作业没有已标注知识点，回答仅基于题目和代码。", @@ -51,6 +59,35 @@ knowledge_quality_monitor = KnowledgeQualityMonitor() +def _request_embedder(): + """Create a request-scoped, budgeted embedder while preserving the default.""" + + configured_name = os.environ.get("KNOWLEDGE_RAG_EMBEDDER", "").strip() + max_calls = MAX_INDEX_DOCUMENTS + 1 # at most 64 chunks plus one query + max_cost = default_embedding_max_cost() + if configured_name: + return knowledge_embedding_registry.build( + configured_name, + max_calls=max_calls, + max_estimated_cost=max_cost, + ) + provider_name = ( + "cjk_ngram" + if isinstance(knowledge_vector_embedder, NgramCountEmbedder) + else "custom" + ) + return BudgetedEmbedder( + knowledge_vector_embedder, + provider_name=provider_name, + max_calls=max_calls, + max_estimated_cost=max_cost, + ) + + +def _embedding_usage(embedder): + return embedder.snapshot() if embedder is not None else None + + def _created_at_value(record): created_at = getattr(record, "created_at", None) return created_at.isoformat() if created_at else None @@ -68,6 +105,7 @@ def _result( index_revision=None, privacy_filtered_count=0, quality_monitor=None, + embedding_usage=None, ): hit_count = len(evidence) latency_ms = max(0.0, (time.perf_counter() - started_at) * 1000.0) @@ -99,6 +137,18 @@ def _result( "indexed_chunk_count": int(indexed_chunk_count), "index_revision": int(index_revision) if index_revision is not None else None, "privacy_filtered_count": int(privacy_filtered_count), + "embedding_provider": ( + embedding_usage.get("provider") if embedding_usage else None + ), + "embedding_calls": int(embedding_usage.get("calls", 0)) + if embedding_usage + else 0, + "embedding_estimated_cost": float( + embedding_usage.get("estimated_cost", 0.0) + ) if embedding_usage else 0.0, + "embedding_budget_exceeded": bool( + embedding_usage and embedding_usage.get("budget_exceeded") + ), } (quality_monitor or knowledge_quality_monitor).record( status=status, @@ -236,15 +286,17 @@ def retrieve_assignment_knowledge( privacy_filtered_count += 1 documents.append(document) + active_embedder = None try: chunks = tuple( chunk for document in documents for chunk in knowledge_pipeline.chunker.split(document) ) + active_embedder = _request_embedder() index = VersionedKnowledgeIndex( chunks, - embedder=knowledge_vector_embedder, + embedder=active_embedder, deadline=deadline, ) search_result = index.search(query, top_k=bounded_limit) @@ -263,6 +315,21 @@ def retrieve_assignment_knowledge( retrieval_mode="timeout", privacy_filtered_count=privacy_filtered_count, quality_monitor=quality_monitor, + embedding_usage=_embedding_usage(active_embedder), + ) + except EmbeddingBudgetExceeded: + logger.warning("knowledge embedding budget exceeded; using safe fallback") + db.session.rollback() + return _result( + "unavailable", + [], + len(records), + started_at, + RETRIEVAL_UNAVAILABLE.copy(), + retrieval_mode="unavailable", + privacy_filtered_count=privacy_filtered_count, + quality_monitor=quality_monitor, + embedding_usage=_embedding_usage(active_embedder), ) except Exception: db.session.rollback() @@ -278,6 +345,7 @@ def retrieve_assignment_knowledge( retrieval_mode="unavailable", privacy_filtered_count=privacy_filtered_count, quality_monitor=quality_monitor, + embedding_usage=_embedding_usage(active_embedder), ) evidence = [ { @@ -303,6 +371,7 @@ def retrieve_assignment_knowledge( index_revision=index.revision.number, privacy_filtered_count=privacy_filtered_count, quality_monitor=quality_monitor, + embedding_usage=_embedding_usage(active_embedder), ) return _result( "grounded", @@ -314,6 +383,7 @@ def retrieve_assignment_knowledge( index_revision=index.revision.number, privacy_filtered_count=privacy_filtered_count, quality_monitor=quality_monitor, + embedding_usage=_embedding_usage(active_embedder), ) diff --git a/templates/submit_code.html b/templates/submit_code.html index 6ce217a..d9e9e32 100644 --- a/templates/submit_code.html +++ b/templates/submit_code.html @@ -1320,9 +1320,10 @@
提交历史
// 流式处理 const reader = response.body.getReader(); const decoder = new TextDecoder(); - let buffer = ''; - let assistantMessage = ''; - let messageElement = null; + let buffer = ''; + let assistantMessage = ''; + let messageElement = null; + let streamErrorPayload = null; while (true) { const { done, value } = await reader.read(); @@ -1368,15 +1369,31 @@
提交历史
} updateStreamingMessage(messageElement, assistantMessage); } else if (data.error) { - throw new Error(data.error); - } - } catch (e) { - console.error('解析SSE数据失败:', e); - } - } - } - } - } else { + streamErrorPayload = data; + break; + } + } catch (e) { + console.error('解析SSE数据失败:', e); + streamErrorPayload = { + type: 'error', + error: 'AI_RESPONSE_INVALID', + message: 'AI服务响应格式异常，请稍后重试' + }; + break; + } + } + } + if (streamErrorPayload) break; + } + + if (streamErrorPayload) { + const streamError = new Error( + streamErrorPayload.message || streamErrorPayload.error || 'AI服务暂时不可用' + ); + streamError.payload = streamErrorPayload; + throw streamError; + } + } else { // 非流式响应（兼容旧版本） const data = await response.json(); if (data.success) { @@ -1390,10 +1407,19 @@
提交历史
addMessage('assistant', '抱歉，出现了错误：' + (data.message || '未知错误')); } } - } catch (error) { - console.error('AI请求失败:', error); - addMessage('assistant', '抱歉，连接AI服务失败，请稍后再试。'); - } + } catch (error) { + console.error('AI请求失败:', error); + const errorWrapper = addMessage( + 'assistant', + error.message || '抱歉，连接AI服务失败，请稍后再试。' + ); + const errorContent = errorWrapper && errorWrapper.querySelector('.ai-message-content'); + const errorPayload = error.payload || {}; + renderKnowledgeReceipt( + errorContent, + errorPayload.knowledge_evidence || (errorPayload.data && errorPayload.data.knowledge_evidence) + ); + } } // 创建流式消息元素 diff --git a/tests/test_code_advice_knowledge.py b/tests/test_code_advice_knowledge.py index e4efeac..5ac1f9b 100644 --- a/tests/test_code_advice_knowledge.py +++ b/tests/test_code_advice_knowledge.py @@ -6,6 +6,7 @@ from config import TestingConfig as _TestingConfig from models import Assignment, AssignmentKnowledgePoint, User, db from routes import api as api_routes +from services.llm_client import LLMServiceError @pytest.fixture @@ -96,6 +97,17 @@ def chat_stream(self, messages, **kwargs): return iter(["先检查边界条件。", "再手动追踪一次。"]) +class _InterruptedSharedClient(_FakeSharedClient): + def chat_stream(self, messages, **kwargs): + type(self).captured_messages = messages + + def stream(): + yield "先检查边界条件。" + raise LLMServiceError("STREAM_INTERRUPTED") + + return stream() + + def test_chat_advice_is_grounded_and_emits_evidence_only_on_done( code_advice_knowledge_context, monkeypatch ): @@ -233,3 +245,33 @@ def broken_retrieval(*args, **kwargs): assert done["type"] == "done" assert done["knowledge_evidence"]["status"] == "unavailable" assert "先检查边界条件" in done["answer"] + + +def test_code_advice_stream_failure_preserves_evidence_receipt( + code_advice_knowledge_context, monkeypatch +): + _, client, assignment_id = code_advice_knowledge_context + monkeypatch.setattr(api_routes, "retrieve_assignment_knowledge", _grounded_retrieval) + monkeypatch.setattr( + "services.llm_client.SharedLLMClient", + lambda: _InterruptedSharedClient(), + ) + + response = client.post( + "/api/code_advice", + json={ + "code": "int main(){return 0;}", + "assignment_id": assignment_id, + "question": "数组边界为什么重要？", + }, + headers={"Accept": "text/event-stream"}, + ) + + assert response.status_code == 200 + events = _events(response) + assert [event["type"] for event in events] == ["start", "delta", "error"] + error = events[-1] + assert error["error"] == "STREAM_INTERRUPTED" + assert error["knowledge_retrieval"]["status"] == "grounded" + assert error["knowledge_evidence"]["evidence"][0]["citation"] == "[K1]" + assert error["data"]["knowledge_evidence"]["status"] == "grounded" diff --git a/tests/test_knowledge_evidence.py b/tests/test_knowledge_evidence.py index 66e6dd8..de23911 100644 --- a/tests/test_knowledge_evidence.py +++ b/tests/test_knowledge_evidence.py @@ -133,6 +133,35 @@ def test_unavailable_preserves_only_the_known_fallback_code(): assert "稍后重试" in view["next_step"] +@pytest.mark.parametrize( + ("status", "fallback_code"), + [ + ("timeout", "KNOWLEDGE_RETRIEVAL_TIMEOUT"), + ("rate_limited", "KNOWLEDGE_RETRIEVAL_RATE_LIMITED"), + ], +) +def test_public_projection_preserves_reliability_fallback_states( + status, fallback_code +): + public = build_public_knowledge_retrieval( + { + "status": status, + "evidence": [], + "fallback": {"code": fallback_code, "message": "safe fallback"}, + "metrics": {"retrieval_mode": status}, + } + ) + + assert public["status"] == status + assert public["metrics"]["retrieval_mode"] == status + assert public["fallback"]["code"] == fallback_code + + view = build_knowledge_evidence_view(public) + assert view["status"] == status + assert view["fallback_code"] == fallback_code + assert view["has_evidence"] is False + + @pytest.mark.parametrize("audience", ["teacher", "admin"]) def test_teacher_and_admin_receive_bounded_diagnostics(audience): view = build_knowledge_evidence_view( diff --git a/tests/test_knowledge_evidence_integration.py b/tests/test_knowledge_evidence_integration.py index 280f8e9..e850a70 100644 --- a/tests/test_knowledge_evidence_integration.py +++ b/tests/test_knowledge_evidence_integration.py @@ -141,6 +141,11 @@ def test_assignment_submission_api_and_code_advice_share_safe_evidence_projectio assert "数组边界" in assignment_page.get_data(as_text=True) assert "数组边界" in submit_page.get_data(as_text=True) assert "不是本次评分依据" in submission_page.get_data(as_text=True) + submit_html = submit_page.get_data(as_text=True) + assert 'data-knowledge-evidence-receipt="true"' in submit_html + assert "streamErrorPayload" in submit_html + assert "streamErrorPayload.message" in submit_html + assert "error.payload" in submit_html api_view = evidence_api.json["data"]["knowledge_evidence"] done = _events(advice)[-1] diff --git a/tests/test_knowledge_optimization.py b/tests/test_knowledge_optimization.py new file mode 100644 index 0000000..f5ec2c7 --- /dev/null +++ b/tests/test_knowledge_optimization.py @@ -0,0 +1,58 @@ +import pytest + +from services.knowledge_optimization import ( + BudgetedEmbedder, + EmbeddingBudgetExceeded, + build_default_embedding_registry, +) +from services.knowledge_optimization_eval import compare_embedders + + +class _CountingEmbedder: + def __init__(self): + self.calls = 0 + + def embed(self, text): + self.calls += 1 + return {str(text): 1.0} + + +def test_budgeted_embedder_rejects_call_before_cost_cap_is_exceeded(): + provider = _CountingEmbedder() + embedder = BudgetedEmbedder( + provider, + provider_name="test-paid", + estimated_cost_per_call=0.02, + max_calls=3, + max_estimated_cost=0.02, + ) + + assert embedder.embed("first") == {"first": 1.0} + with pytest.raises(EmbeddingBudgetExceeded): + embedder.embed("second") + + usage = embedder.snapshot() + assert provider.calls == 1 + assert usage["calls"] == 1 + assert usage["estimated_cost"] == 0.02 + assert usage["budget_rejections"] == 1 + assert usage["budget_exceeded"] is True + + +def test_comparison_reports_quality_cost_and_selected_provider(): + report = compare_embedders() + + assert report["baseline_provider"] == "cjk_ngram" + assert report["selected_provider"] == "cjk_ngram" + assert set(report["providers"]) == {"cjk_ngram", "token"} + for provider_report in report["providers"].values(): + assert provider_report["recall_at_k"] >= 0.0 + assert provider_report["embedding_usage"]["calls"] > 0 + assert provider_report["embedding_usage"]["estimated_cost"] == 0.0 + assert provider_report["quality_gate"] is True + assert provider_report["cost_gate"] is True + + +def test_registry_rejects_unknown_provider(): + with pytest.raises(ValueError, match="unknown embedding provider"): + build_default_embedding_registry().build("not-installed") diff --git a/tests/test_knowledge_rag.py b/tests/test_knowledge_rag.py index a8882b7..252675c 100644 --- a/tests/test_knowledge_rag.py +++ b/tests/test_knowledge_rag.py @@ -345,6 +345,31 @@ def embed(self, _text): assert retrieval["metrics"]["retrieval_mode"] == "unavailable" +def test_rag_can_select_token_provider_and_exposes_bounded_usage( + knowledge_context, monkeypatch +): + app, _, assignment_id = knowledge_context + with app.app_context(): + AssignmentKnowledgePoint.add_to_assignment( + assignment_id, + "array", + weight=1.0, + ) + db.session.commit() + monkeypatch.setenv("KNOWLEDGE_RAG_EMBEDDER", "token") + retrieval = retrieve_assignment_knowledge( + assignment_id, + query="数组", + ) + + assert retrieval["status"] == "grounded" + metrics = retrieval["metrics"] + assert metrics["embedding_provider"] == "token" + assert metrics["embedding_calls"] >= 2 + assert metrics["embedding_estimated_cost"] == 0.0 + assert metrics["embedding_budget_exceeded"] is False + + def test_ask_question_continues_with_answer_only_when_knowledge_source_is_unavailable( knowledge_context, monkeypatch ):