first published

.env.example

@@ -1,9 +1,36 @@
+# neo4j配置信息
 NEO4J_URI=http://121.43.55.7:7474
 NEO4J_USER=neo4j
-NEO4J_PASSWORD=change-me
+NEO4J_PASSWORD=skyversation
 NEO4J_DATABASE=neo4j
-DEEPSEEK_API_KEY=sk-xxxx
+
+# deepseek配置信息
+DEEPSEEK_API_KEY=sk-8a1d2d5b63f945c1a11c2760de81da61
 DEEPSEEK_BASE_URL=https://api.deepseek.com
-DEEPSEEK_MODEL=deepseek-v4-flash
+DEEPSEEK_MODEL=deepseek-v4-pro
+STEP1_ANALYSIS_MAX_WORKERS=3
+
+# redis配置信息
 REDIS_URL=redis://127.0.0.1:6379/0
 REDIS_CONTEXT_TTL_SECONDS=1296000
+
+# DMS配置信息
+DMS_BASE_URL=http://121.43.55.7:2101/
+DMS_USERNAME=user_wangxi
+DMS_PASSWORD=Wx@123456
+
+# Step2: cosine similarity must be strictly greater than this value
+STEP2_RELATION_SIMILARITY_THRESHOLD=0.8
+
+# Web/API 服务监听地址与端口；命令行参数可覆盖
+SERVICE_HOST=0.0.0.0
+SERVICE_PORT=8000
+
+# 模板及关系目录；相对路径基于项目根目录，支持绝对路径
+METADATA_TEMPLATE_DIR=data/templates/metadata
+RELATION_DIR=data/templates
+
+# 本地嵌入模型下载与运行配置
+EMBEDDING_MODEL_DIR=models/Qwen3-Embedding-0.6B
+EMBEDDING_MODEL_REPO=Qwen/Qwen3-Embedding-0.6B
+EMBEDDING_MODEL_SOURCE=modelscope

README.md

@@ -248,7 +248,7 @@ uv run python scripts/ask.py
 生产问答在元图谱 `meta.schema_version == 2` 时走以下分支：
 
 ```text
-问题理解与本轮元图谱快照 → 选择相关数据评估 → JSON 查询计划
+问题理解与本轮元图谱快照 → 选择相关数据评估 → 简单查询直接构造 / 完整模型规划
 → 计划确认 → Schema 校验与只读查询编译 → Neo4j 查询 → 可溯源回答
 ```
 
@@ -269,6 +269,33 @@ uv run python scripts/ask.py
 | `api_log/` | 请求、执行结果和错误日志 |
 | `reference/`、`data/test_data/` | 原始参考资料与测试数据 |
 
+### 问答速度优化
+
+生产分支对可完整表达为单数据集的计数、属性查询和简单列表，在问题理解后直接构造计划，
+减少一次规划模型调用；复杂条件、跨关系或用户修改计划时回退完整规划。人工确认与实际查询保留。
+理解和规划仅输出必要的紧凑 JSON，省略空字段、默认值和重复字段，保留全部业务条件。
+
+`QA_FAST_PLAN_ENABLED` 与 `QA_COMPACT_INTERMEDIATE_ENABLED` 默认均为 true，
+可分别在环境配置中设为 false 关闭；改代码/配置后重启服务即可，无需更新图谱。
+请求日志新增阶段耗时、模型流调用记录和快速路径命中原因。说明与验证入口见
+[Step3 速度优化](docs/step3-qa-agent.md)。
+
+### 确认答案后缓存
+
+答案下的“答案正确，加入缓存”用于人工确认；保存后可“撤销缓存”。
+查询计划的确认和自动确认不会保存答案缓存。重复提问精确命中后显示“已复用你确认的历史答案”，
+直接复用答案和来源。永久保存（不设置TTL），数据版本发布成功后删除全部旧数据版本缓存，按会话、上下文、数据/Schema版本和模型/策略隔离；
+新建会话或改变上下文可能不命中，相似问题仍走正常流程。
+
+错误、截断、待澄清、修改过条件及明确依赖历史指代/相对时间的回答不提供缓存按钮。
+关闭“历史复用”可绕过读取。`QA_ANSWER_CACHE_ENABLED`默认true；
+`QA_ANSWER_CACHE_TTL_SECONDS`已停用，即使旧环境仍配置该值也不设置过期时间；
+
+问答缓存清理只操作独立索引，不影响Redis会话上下文的过期策略。服务启动时将尚存的当前版本旧TTL缓存转为永久；已过期消失的记录无法恢复。发布完成后立即清理旧数据版本；Redis故障记录在更新报告的`answer_cache_cleanup`，服务启动及运行期间每60秒重试。原子版本校验阻止旧回答在版本切换后重新入库；用户仍可主动撤销缓存。
+
+修改提示词/权限策略时递增`QA_ANSWER_CACHE_POLICY_VERSION`（默认1），重启后生效。
+详细边界与接口见[Step3文档](docs/step3-qa-agent.md)、[API说明](docs/api-reference.md)。
+
 ### 独立构图与历史工具
 
 日常维护使用上面的统一更新入口。`scripts/build_graph.py --check-only` 是独立 Step2 预检：读取 `data/production/`、metadata 模板和 relation；历史默认主键文件 `data/graph_keys.json` 已移除，使用该入口须通过 `--keys <配置路径>` 提供主键配置，**不会跟随统一发布清单选择生产目录**。完整独立构图也不负责切换统一发布清单，不应与统一更新流程混用。

docs/api-reference.md

@@ -43,3 +43,24 @@ SSE 使用 POST，可参考 [html/index.html](../html/index.html) 中的 `askStr
 ## 数据质量
 
 接口 `GET /api/data-quality` 返回总数及各文件计数；`?details=true&data_version=<版本>` 返回完整问题记录，版本变化返回409。明细沿用现有服务访问边界，能访问此接口的用户可看到问题记录的全部字段。
+
+## 人工确认答案缓存
+
+`POST /threads/{thread_id}/answer-cache`：确认该答案正确并保存缓存。
+`DELETE /threads/{thread_id}/answer-cache`：撤销对应缓存（需entry_id）。
+
+请求仅允许：
+`{"answer_id":"答案ID","checkpoint_id":"该答案的checkpoint ID","entry_id":null}`。
+POST成功返回status、entry_id、approved_at；DELETE用该entry_id撤销。
+接口从服务端checkpoint读取原答案，不接收答案、问题或查询条件正文。
+记录失效、未完成、不符合缓存规则或版本变化返回409；缓存不可用返回503。
+额外字段或缺失标识返回422。
+
+/ask、/ask/stream的done和resume完整响应新增cache_review：
+eligible、reason、answer_id、checkpoint_id、cached、hit、entry_id、approved_at。
+eligible=true时前端展示“答案正确，加入缓存”，成功后可撤销。
+cached/hit=true显示历史答案来源。人工计划确认和auto_confirm不等于缓存确认。
+
+索引仅限同一thread_id、精确问题和上下文，包含数据/Schema/模型/策略版本，永久保存，数据版本发布成功后物理删除全部旧数据版本缓存。
+关闭reuse_check绕过缓存读取。缓存命中直接返回已确认答案及来源，不调用模型/Neo4j。
+目前无真实鉴权，不能据此宣称具有用户身份认证或允许跨用户共享缓存。

docs/step3-qa-agent.md

@@ -60,3 +60,92 @@ uv run python scripts/test_step2_production.py
 hash和字段校验均通过。没有发送真实业务问题或评估内容到外部模型，尚不能据此
 宣称真实问答准确率提高；需在查询工具适配后执行业务问题集验收。
 
+
+## 简单问题快速规划与中间输出（2026-09-07）
+
+生产 Schema v2 在一次问题理解中同时提取可选的 `simple_query`。仅支持单数据集的
+节点去重计数、属性查询和明细列表，最多6个明确文本等于/包含的 AND 条件。
+候选必须声明完整、与原始 data_needs 单数据集/无关系选择一致，筛选值必须来自本轮问题，
+已抽取的实体及筛选槽位不能遗漏，字段和结构必须通过现有只读编译器校验。
+计数、关系或字段口径不明确时应返回 null。模型对完整性的判断仍需真实问题验收。
+
+命中时程序构造计划，省去单独规划模型调用；仍经过中文计划确认、版本检查、
+实际 Neo4j 查询、空结果候选及溯源回答。跨关系、时间/数值条件、分组排序、复杂表达、
+历史指代、未知字段、缺失候选、修改反馈都回退完整规划。每次理解重置候选，旧会话不继承。
+
+理解与完整规划均要求紧凑 JSON，不输出额外分析文字；省略空数组/空槽位、默认 limit=50、
+重复 fields/depends。所有实际筛选、关联、分组、排序、返回字段和统计口径必须保留。
+简单计数候选最小结构为：
+`{"kind":"count","dataset":"人员信息","complete":true}`。
+最终答案仍按原有规则生成，不对 JSON 使用可能导致截断的硬性 token 上限。
+
+可选环境变量（未设置时默认 true，支持 true/false/1/0）：
+
+| 变量 | 作用 |
+|---|---|
+| `QA_FAST_PLAN_ENABLED` | 单数据集快速规划 |
+| `QA_COMPACT_INTERMEDIATE_ENABLED` | 理解与完整规划的紧凑输出 |
+
+两个都设为 false 可恢复原理解/规划提示词及两次调用路径。修改配置或代码后重启服务，
+无需重新取数或构图。
+
+`api_log` 中 state.trace 的 understand/production_plan 新增 duration_ms；
+production_plan.mode 为 simple 或 llm，reason 说明命中或回退原因，validation_attempts
+记录完整计划结构校验次数。llm_calls 每项记录一个 SDK 流的耗时、首次 chunk/文本等待、
+输入/输出字符数及可用的 usage。JSON 解析重试计为不同流；SDK 内部网络重试只能计入耗时，
+不能据此声称记录了精确网络请求次数。usage 缺失为 null，不把字符数当 token 数。
+耗时不包含人工等待；节点耗时减去流耗时包含上下文、SDK创建等本地处理。
+
+离线回归：
+`uv run --frozen python -X utf8 -m unittest scripts.test_simple_plan scripts.test_qa_question_context scripts.test_value_clarification scripts.test_plan_presentation scripts.test_runtime_workflow -q`。
+
+真实测试脚本 `scripts/benchmark_qa_planning.py` 交替关闭/开启两个优化，记录理解和规划耗时；
+它会调用真实配置的外部模型，须取得授权后运行。加 `--execute` 执行受控只读图谱查询，
+仅保存结果行数、截断标记及按字段/操作归一的结果哈希，不保存原始查询行。
+输出默认在 `.runtime/qa/planning-benchmark.json`。哈希比较需确认同一 build_id、
+同一口径且结果未截断；随机模型输出和无排序明细可能不同。单次样本不能代表 P95。
+
+## 用户确认的独立问答缓存（2026-09-07）
+
+网页版完成生产图谱回答后，在答案下显示“答案正确，加入缓存”。只有点击此按钮，
+服务端才把该答案加入独立 Redis 索引；查询计划的“确认”或自动确认不会写入缓存。
+保存成功后可“撤销缓存”，缓存命中显示“已复用你确认的历史答案”。
+
+索引使用 `ka:qa:approved:v1:` 前缀的 Redis String，与上下文 checkpoint 分离。
+键包含问题文本哈希、thread_id 隔离范围、最近对话摘要哈希、数据版本、Schema版本、
+模型名及缓存策略版本。只去除问题两端空白，不去除否定、数字、标点或实体内部空格；
+不做相似度直接复用。新提问先作一次有超时限制的 GET；未命中继续原问答流程。
+同一问题连续重复时沿用原问题的上下文签名；中途问过其他问题会重新计算签名，
+因此可能不命中。带明确历史指代或相对时间的问题保守地不缓存。
+
+命中后直接复用用户确认的答案、查询计划和来源，跳过模型及图谱查询，
+但仍追加本轮历史并保存新的 checkpoint；关闭页面“历史复用”可强制正常问答。
+目前API没有真实鉴权，缓存仅按会话隔离，清空/新建会话不会共享已有缓存。
+未来接入鉴权后须把实际用户/租户/权限版本纳入键，不能使用当前固定 user_id=api 共享。
+
+仅接收答案ID及checkpoint ID，不接收浏览器提交的答案正文。服务端读取该会话不可变
+checkpoint并校验完成状态，所以旧答案按钮也不会误确认后来的另一条回答。
+过期checkpoint无法确认。错误、闲聊、不支持、截断/缺失结果、待澄清、
+修改过计划/替换过候选、依赖上下文/相对时间及超过256KB的记录不入缓存。
+保存查询结果与来源仅用于复用已确认内容，不把评估报告当作事实。
+
+缓存永久保存（不设置TTL），数据版本发布成功后删除全部旧数据版本缓存。写入通过Redis原子脚本核对当前数据版本，不设置过期时间。数据或Schema更新、模型名、
+策略版本变化均不再命中旧键。数据版本发布成功后，仅扫描独立问答索引并删除全部非当前数据版本记录，不影响会话上下文及其他Redis数据。清理失败记录警告，服务启动及运行期间每60秒重试；启动时将尚存的当前版本旧TTL记录转为永久。
+撤销使用比较后删除，避免并发时删除另一条新确认记录。
+缓存读超时/损坏时回退正常问答；写入/撤销失败会明确提示，不显示成功。
+底层Redis整体不可用仍会影响原有checkpoint持久化，不承诺整个服务脱离Redis运行。
+
+| 配置 | 默认 | 用途 |
+|---|---|---|
+| `QA_ANSWER_CACHE_ENABLED` | true | 独立答案缓存及确认入口开关 |
+| `QA_ANSWER_CACHE_TTL_SECONDS` | 已停用 | 旧配置被忽略；问答缓存永久保存 |
+| `QA_ANSWER_CACHE_POLICY_VERSION` | 1 | 提示词/权限/回答策略变化时主动递增以失效旧缓存 |
+
+修改代码或配置后重启服务，无需取数或构图。现有缓存不会自动认定旧上下文中的答案正确，
+历史答案必须由用户通过对应按钮确认。
+
+主要实现：`src/step4_web/answer_cache.py`、`api.py`、
+`html/answer_cache.js`、`html/answer_cache.css`；根页iframe版本同步更新。
+离线验证：
+`uv run --frozen python -X utf8 -m unittest scripts.test_answer_cache -q`；
+页面：`node scripts/test_answer_cache_ui.cjs`（需Playwright和Edge）。

docs/技术方案.md

@@ -668,3 +668,11 @@ knowledge_agent/
 ### 空结果候选多选（2026-09-07）
 
 clarify_value 支持一个或多个候选。前端候选按钮可切换选中状态，“都不是”与实际候选互斥；接口透传确认类型，普通计划确认仍为单选。后端接受逗号、中文逗号、顿号、分号分隔的序号/完整值及 JSON 字符串数组，严格解析且不做模糊代选。同一查询条件选择多个真实值时编译为 `in`，不同条件分别替换并继续使用 AND 组合，未选择的筛选和关系保持不变。
+
+### 查询延迟优化（2026-09-07）
+
+生产 Schema v2 问题理解增加可选 simple_query：单数据集计数/属性/列表的完整候选通过字段、原词、槽位覆盖和只读编译校验后，确定性构造计划；复杂/不完整候选及修改反馈继续完整规划。人工确认、版本隔离、查询和候选澄清保持原流程。理解与完整规划改为紧凑 JSON，省略空数组、默认值及重复字段，禁止遗漏条件。两个优化由 QA_FAST_PLAN_ENABLED / QA_COMPACT_INTERMEDIATE_ENABLED 独立控制，默认 true。trace 保存节点耗时、模型流次数、首次输出及字符/可用 token 统计；尚未取得真实性能结果。详见 [Step3](step3-qa-agent.md)。
+
+### 人工确认问答缓存（2026-09-07）
+
+新增独立Redis索引ka:qa:approved:v1，仅在前端点击“答案正确，加入缓存”后保存；确认查询计划不等于确认答案。按精确问题、会话、上下文、数据/Schema/模型/策略版本匹配，永久保存，数据版本发布成功后物理删除全部旧数据版本缓存。缓存命中跳过模型和图谱，保留来源并追加会话。用户可撤销；超时未命中回原流程。checkpoint和答案ID绑定，拒绝客户端答案正文；错误、截断、未完成、改过条件/候选及明确上下文/相对时间问题不缓存。现阶段不跨会话共享、不按语义相似度直接复用；详见Step3及API说明。

html/answer_cache.css

@@ -0,0 +1,6 @@
+.answer-cache-actions { display:flex; flex-wrap:wrap; align-items:center; gap:8px; margin-top:12px; padding-top:10px; border-top:1px solid #e2e8f0; font-size:12px; }
+.answer-cache-status { color:#64748b; overflow-wrap:anywhere; }
+.answer-cache-button { cursor:pointer; border:1px solid #94a3b8; border-radius:7px; padding:6px 10px; background:#fff; color:#334155; font:inherit; }
+.answer-cache-button:hover { background:#f1f5f9; }
+.answer-cache-button:disabled { cursor:wait; opacity:.6; }
+.answer-cache-button:focus-visible { outline:2px solid #2563eb; outline-offset:2px; }

html/answer_cache.js

@@ -0,0 +1,56 @@
+/* Explicit, per-answer cache approval. No background approval requests. */
+(function () {
+  function attach(bubble, payload, threadId, baseUrl) {
+    const review = payload.cache_review;
+    if (!review) return;
+    bubble.querySelector('.answer-cache-actions')?.remove();
+    const bar = document.createElement('div');
+    bar.className = 'answer-cache-actions';
+    const status = document.createElement('span');
+    status.className = 'answer-cache-status';
+    status.setAttribute('role', 'status');
+    status.setAttribute('aria-live', 'polite');
+    bar.appendChild(status);
+    bubble.appendChild(bar);
+    if (!review.eligible && !review.cached) {
+      status.textContent = review.reason ? '暂不缓存：' + review.reason : '';
+      return;
+    }
+    let cached = Boolean(review.cached);
+    let entryId = review.entry_id;
+    const button = document.createElement('button');
+    button.type = 'button';
+    button.className = 'answer-cache-button';
+    const label = () => {
+      button.textContent = cached ? '撤销缓存' : '答案正确，加入缓存';
+      button.setAttribute('aria-label', button.textContent);
+    };
+    status.textContent = review.hit ? '已复用你确认的历史答案' : '确认答案正确后，可缓存供本会话重复提问使用';
+    label();
+    button.addEventListener('click', async () => {
+      if (button.disabled) return;
+      button.disabled = true;
+      status.textContent = cached ? '正在撤销…' : '正在加入缓存…';
+      try {
+        const response = await fetch(baseUrl + '/threads/' + encodeURIComponent(threadId) + '/answer-cache', {
+          method: cached ? 'DELETE' : 'POST',
+          headers: {'Content-Type': 'application/json'},
+          body: JSON.stringify({answer_id: review.answer_id, checkpoint_id: review.checkpoint_id,
+                                entry_id: entryId || null})
+        });
+        const result = await response.json();
+        if (!response.ok || result.status !== 'ok') throw new Error(result.detail || result.message || '操作失败，请重试');
+        cached = !cached;
+        entryId = cached ? result.entry_id : null;
+        status.textContent = cached ? '已确认并加入缓存' : '已撤销，后续提问将重新处理';
+      } catch (error) {
+        status.textContent = error.message || '缓存操作失败，请重试';
+      } finally {
+        label();
+        button.disabled = false;
+      }
+    });
+    bar.appendChild(button);
+  }
+  window.AnswerCacheUI = {attach};
+})();

html/index.html

@@ -361,6 +361,8 @@
   <link rel="stylesheet" href="answer_markdown.css?v=20260904-qa">
   <script src="vendor/marked.umd.js?v=17.0.5"></script>
   <script src="answer_markdown.js?v=20260904-qa"></script>
+  <link rel="stylesheet" href="answer_cache.css?v=20260907-cache">
+  <script src="answer_cache.js?v=20260907-cache"></script>
   <script>
     const $ = (id) => document.getElementById(id);
     const chatEl = $("chat");
@@ -612,6 +614,7 @@
             const payload = JSON.parse(data);
             if (typeof payload.answer === "string") answerText = payload.answer;
             renderAnswer(answerEl, answerText);
+            AnswerCacheUI.attach(assistantBubble, payload, threadId, apiBase());
             finishProgressFlow(progressEl, "完成");
             setStatus(`完成，耗时 ${payload.elapsed_sec || 0} 秒`);
           } catch (_) {}
@@ -676,7 +679,8 @@
       } else if (payload.status === "ok") {
         hideConfirm();
         if (activeProgressEl) finishProgressFlow(activeProgressEl, "完成");
-        appendMessage("assistant", payload.answer || "");
+        const answerBubble = appendMessage("assistant", payload.answer || "");
+        AnswerCacheUI.attach(answerBubble, payload, threadId, apiBase());
         setStatus(`完成，耗时 ${payload.elapsed_sec || 0} 秒`);
       } else {
         hideConfirm();

html/knowledge_graph_3d.html

@@ -301,7 +301,7 @@
         <div class="spacer"></div>
         <button id="chatToggle" class="chat-toggle">收起</button>
       </div>
-      <iframe id="chatFrame" src="index.html?v=20260907-multiselect" title="申勤物业知识助手"></iframe>
+      <iframe id="chatFrame" src="index.html?v=20260907-cache" title="申勤物业知识助手"></iframe>
     </div>
   </div>
 

scripts/benchmark_qa_planning.py

@@ -0,0 +1,92 @@
+"""Compare planning with both optimizations off/on; real LLM calls require authorization.
+
+--execute additionally runs schema-validated read-only Neo4j queries. No raw query
+results are saved. Planning times exclude confirmation, query execution and answer.
+"""
+from __future__ import annotations
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+from time import perf_counter
+from unittest.mock import patch
+from step3_qa_agent.agent import nodes, production, llm
+from step3_qa_agent.retrieval.production_query import compile_query
+from step2_graph_building.runtime import release_snapshot
+
+def query_fingerprint(plan, context):
+    if release_snapshot()['data_version'] != context['build_id']:
+        raise RuntimeError('数据版本变化，取消对比')
+    driver = production.get_driver()
+    results = []
+    for step in plan['steps']:
+        spec = step['params']
+        compiled = compile_query(spec, context['schema'], context['build_id'])
+        records = driver.execute_query(compiled.query, **compiled.params).records
+        aliases = {n['alias']: n['type'] for n in spec['nodes']}
+        columns = [(str((aliases[c['alias']], c.get('op', 'field'), c.get('field'))), c['as'])
+                   for c in spec.get('select', []) + spec.get('aggregates', [])]
+        rows = [sorted((identity, row.get(label)) for identity, label in columns) for row in records]
+        encoded = sorted(json.dumps(row, ensure_ascii=False, sort_keys=True, default=str) for row in rows)
+        results.append({'rows': len(records), 'truncated': len(records) > compiled.limit,
+                        'sha256': hashlib.sha256(json.dumps(encoded, ensure_ascii=False).encode()).hexdigest()})
+    if release_snapshot()['data_version'] != context['build_id']:
+        raise RuntimeError('查询期间数据版本变化，取消对比')
+    return results
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--question', action='append')
+    parser.add_argument('--repeat', type=int, default=1)
+    parser.add_argument('--execute', action='store_true', help='实际执行受控只读查询，仅保存结果摘要')
+    parser.add_argument('--output', default='.runtime/qa/planning-benchmark.json')
+    args = parser.parse_args()
+    if args.repeat < 1:
+        parser.error('--repeat 必须大于0')
+    questions = args.question or ['共有多少名员工？', '岗位名称包含保安的员工有多少名？',
+                                   '张三的岗位名称是什么？', '按岗位名称分别统计员工人数']
+    output = Path(args.output)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    report = {'scope': 'understand + production_plan only; both switches off vs on',
+              'executed_read_only_queries': args.execute, 'runs': []}
+    original_chat = llm.get_chat
+    def bounded_chat(*a, **kw):
+        model = original_chat(*a, **kw)
+        model.request_timeout = 90
+        model.max_retries = 0
+        return model
+    for repeat in range(args.repeat):
+        for index, question in enumerate(questions):
+            for enabled in ([False, True] if (repeat + index) % 2 == 0 else [True, False]):
+                row = {'question': question, 'repeat': repeat, 'optimizations_enabled': enabled}
+                started = perf_counter()
+                try:
+                    with patch.dict(os.environ, {
+                            'QA_FAST_PLAN_ENABLED': str(enabled).lower(),
+                            'QA_COMPACT_INTERMEDIATE_ENABLED': str(enabled).lower()}), \
+                         patch.object(llm, 'get_chat', side_effect=bounded_chat):
+                        state = {'question': question}
+                        state.update(nodes.understand(state))
+                        state.update(production.plan(state))
+                    row.update(total_ms=round((perf_counter() - started) * 1000, 2),
+                               build_id=state['qa_context']['build_id'], trace=state['trace'],
+                               plan=state['plan'], candidate=state.get('simple_query'))
+                    if args.execute:
+                        row['query_results'] = query_fingerprint(state['plan'], state['qa_context'])
+                    row['status'] = 'ok'
+                except Exception as exc:
+                    row.update(status='error', error_type=type(exc).__name__,
+                               total_ms=round((perf_counter() - started) * 1000, 2))
+                report['runs'].append(row)
+                output.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding='utf-8')
+                trace = row.get('trace', [])
+                print(json.dumps({'question': question, 'optimized': enabled, 'status': row['status'],
+                                  'ms': row['total_ms'], 'mode': trace[-1].get('mode') if trace else None,
+                                  'reason': trace[-1].get('reason') if trace else None,
+                                  'llm_streams': sum(len(t.get('llm_calls', [])) for t in trace)}, ensure_ascii=False), flush=True)
+    print('Saved: ' + str(output), flush=True)
+    return 1 if any(r['status'] != 'ok' for r in report['runs']) else 0
+
+if __name__ == '__main__':
+    raise SystemExit(main())

scripts/test_answer_cache.py

@@ -0,0 +1,299 @@
+"""Approved cache: no external Redis, LLM or Neo4j connections."""
+from copy import deepcopy
+import json
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+from fastapi import HTTPException
+from langgraph.checkpoint.memory import InMemorySaver
+from step3_qa_agent.agent import nodes, production
+from step3_qa_agent.agent.graph import build_agent_graph
+from step4_web import api
+from step4_web.answer_cache import (ApprovedAnswerCache, prepare_request, cache_key,
+                                    eligibility, restored_state, review_info, ACTIVE_VERSION, APPROVE_CURRENT)
+from scripts.test_simple_plan import CONTEXT, state as simple_state
+from step3_qa_agent.agent.simple_plan import build_simple_plan
+
+RELEASE = {'data_version': 'v1', 'schema_version': 'schema1'}
+Q = '共有多少名员工？'
+
+class FakeRedis:
+    def __init__(self):
+        self.entries = {ACTIVE_VERSION: 'v1'}
+        self.ttls = {}
+        self.writes = 0
+    async def get(self, key):
+        return self.entries.get(key)
+    async def set(self, key, value, ex=None):
+        self.entries[key] = value
+        self.ttls[key] = ex
+        self.writes += 1
+    async def eval(self, script, count, key, *args):
+        if script == APPROVE_CURRENT:
+            fence, raw, version = args
+            if self.entries.get(fence) != version:
+                return 0
+            await self.set(key, raw)
+            return 1
+        raw, = args
+        if self.entries.get(key) == raw:
+            del self.entries[key]
+            return 1
+        return 0
+
+def completed():
+    s = simple_state()
+    s.update(answer='共有2名员工，按工号去重。', answer_id='answer1', fit=True,
+             cache_request=prepare_request('t1', Q, {}, RELEASE),
+             subgraph={'results': {'s1': {'rows': [{'记录数': 2}], 'truncated': False,
+                        'aggregate': True, 'data_version': 'v1'}}, 'suggestions': {'options': []}},
+             trace=[{'node': 'understand'}])
+    s['plan'] = build_simple_plan(s)[0]
+    return s
+
+class CacheTests(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.redis = FakeRedis()
+        self.cache = ApprovedAnswerCache(self.redis)
+
+    async def test_lookup_never_populates_cache(self):
+        self.assertIsNone(await self.cache.lookup(completed()['cache_request']))
+        self.assertEqual(self.redis.writes, 0)
+
+    async def test_only_approved_answer_hits_permanently(self):
+        s = completed()
+        entry = await self.cache.approve(s)
+        loaded = await self.cache.lookup(s['cache_request'])
+        self.assertEqual(loaded['payload']['answer'], s['answer'])
+        self.assertTrue(loaded['approved'])
+        self.assertIsNone(self.redis.ttls[cache_key(s['cache_request'])])
+        self.assertNotIn('expires_at', entry)
+        self.assertTrue(cache_key(s['cache_request']).startswith('ka:qa:approved:v1:'))
+        self.assertEqual(entry['entry_id'], 'answer1')
+
+    async def test_old_inflight_approval_rejected_after_version_switch(self):
+        self.redis.entries[ACTIVE_VERSION] = 'v2'
+        with self.assertRaises(ValueError):
+            await self.cache.approve(completed())
+        self.assertEqual(self.redis.writes, 0)
+
+    async def test_scope_version_schema_context_and_question_are_separate(self):
+        s = completed()
+        await self.cache.approve(s)
+        for field in ('scope', 'data_version', 'schema_version', 'context', 'question'):
+            request = {**s['cache_request'], field: 'different'}
+            self.assertIsNone(await self.cache.lookup(request))
+
+    async def test_no_semantic_or_negation_matching(self):
+        s = completed()
+        await self.cache.approve(s)
+        for q in ('员工共有几人？', '没有证书的员工有多少？', '共有多少名员工!'):
+            self.assertIsNone(await self.cache.lookup(prepare_request('t1', q, {}, RELEASE)))
+
+    async def test_contextual_and_relative_questions_are_not_cacheable(self):
+        for question in ('他们有多少人', '这些员工的岗位', '今年入职人数'):
+            s = completed()
+            s['question'] = question
+            s['cache_request']['question'] = question
+            self.assertTrue(eligibility(s))
+            with self.assertRaises(ValueError):
+                await self.cache.approve(s)
+        self.assertEqual(self.redis.writes, 0)
+
+    async def test_failed_truncated_modified_and_partial_answers_rejected(self):
+        variations = []
+        s = completed(); s['fit'] = False; variations.append(s)
+        s = completed(); s['category'] = '闲聊'; variations.append(s)
+        s = completed(); s['plan_feedback'] = '换成项目'; variations.append(s)
+        s = completed(); s['subgraph']['results']['s1']['truncated'] = True; variations.append(s)
+        s = completed(); s['subgraph']['results']['s1']['data_version'] = 'old'; variations.append(s)
+        s = completed(); s['subgraph']['results'] = {}; variations.append(s)
+        s = completed(); s['subgraph']['error'] = 'failed'; variations.append(s)
+        s = completed(); s['subgraph']['suggestions']['options'] = ['candidate']; variations.append(s)
+        s = completed(); s['trace'].append({'node': 'value_choice'}); variations.append(s)
+        s = completed(); s['answer'] = ''; variations.append(s)
+        for s in variations:
+            self.assertTrue(eligibility(s))
+            with self.assertRaises(ValueError):
+                await self.cache.approve(s)
+        self.assertEqual(self.redis.writes, 0)
+
+    async def test_corrupt_unapproved_entries_miss(self):
+        s = completed()
+        entry = await self.cache.approve(s)
+        key = cache_key(s['cache_request'])
+        for value in ('broken', '[]', json.dumps({**entry, 'approved': False})):
+            self.redis.entries[key] = value
+            self.assertIsNone(await self.cache.lookup(s['cache_request']))
+
+    async def test_cache_read_timeout_falls_back(self):
+        import asyncio
+        async def slow(key):
+            await asyncio.sleep(1)
+        self.cache.timeout_seconds = .001
+        with patch.object(self.redis, 'get', side_effect=slow):
+            self.assertIsNone(await self.cache.lookup(completed()['cache_request']))
+
+    async def test_revoke_and_old_revoke_cannot_delete_new_approval(self):
+        s = completed()
+        await self.cache.approve(s)
+        s['answer_id'] = 'new-answer'
+        await self.cache.approve(s)
+        with self.assertRaises(ValueError):
+            await self.cache.revoke(s['cache_request'], 'answer1')
+        self.assertIsNotNone(await self.cache.lookup(s['cache_request']))
+        await self.cache.revoke(s['cache_request'], 'new-answer')
+        self.assertIsNone(await self.cache.lookup(s['cache_request']))
+
+    async def test_repeat_preserves_original_context_but_other_context_misses(self):
+        first = prepare_request('t1', Q, {}, RELEASE)
+        prev = {'rounds': [{'question': Q, 'answer': '2', 'schema_build_id': 'v1',
+                            'cache_context': first['context']}],
+                'messages': [{'role': 'user', 'content': Q}, {'role': 'assistant', 'content': '2'}]}
+        self.assertEqual(prepare_request('t1', Q, prev, RELEASE), first)
+        prev['rounds'].append({'question': '只看保安', 'answer': '1'})
+        self.assertNotEqual(prepare_request('t1', Q, prev, RELEASE), first)
+
+    async def test_oversize_not_eligible_and_never_written(self):
+        s = completed(); s['answer'] = 'x' * 300000
+        self.assertFalse(review_info(s, 'cp')['eligible'])
+        with self.assertRaises(ValueError):
+            await self.cache.approve(s)
+        self.assertEqual(self.redis.writes, 0)
+
+    async def test_neo4j_records_keep_named_columns(self):
+        from neo4j import Record
+        s = completed()
+        s['subgraph']['results']['s1']['rows'] = [Record({'记录数': 2})]
+        await self.cache.approve(s)
+        hit = await self.cache.lookup(s['cache_request'])
+        self.assertEqual(hit['payload']['subgraph']['results']['s1']['rows'], [{'记录数': 2}])
+
+class ApiCacheTests(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.redis = FakeRedis()
+        self.cache = ApprovedAnswerCache(self.redis)
+        s = simple_state()
+        self.planner = patch.object(production, 'chat_json', return_value=build_simple_plan(s)[0])
+        self.patches = [
+            patch.object(api, '_graph', build_agent_graph(InMemorySaver())),
+            patch.object(api, '_answer_cache', self.cache),
+            patch.object(api, 'release_snapshot', return_value=RELEASE),
+            patch.object(api, '_write_api_log'),
+            patch.object(nodes, 'load_question_context', return_value=deepcopy(CONTEXT)),
+            patch.object(nodes, 'select_assessments', side_effect=lambda ctx, *a, **kw: ctx),
+            patch.object(nodes, 'chat_json', return_value={'category': '图谱检索',
+                'slots': {}, 'data_needs': s['simple_query_needs']}),
+            self.planner,
+            patch.object(production, 'chat_text', return_value='共有2名员工，按工号去重。'),
+            patch.object(production, 'release_snapshot', return_value={'data_version': 'v1'}),
+            patch.object(production, 'get_driver', return_value=SimpleNamespace(
+                execute_query=lambda *a, **kw: SimpleNamespace(records=[{'记录数': 2}])))]
+        self.mocks = [p.start() for p in self.patches]
+        self.addCleanup(lambda: [p.stop() for p in reversed(self.patches)])
+
+    async def ask(self, question=Q, reuse=True, auto=True):
+        events = [e async for e in api._astream_run('t1', question, auto, reuse)]
+        return events[-1]
+
+    def body(self, state, **extra):
+        review = state['cache_review']
+        return api.AnswerReviewRequest(answer_id=review['answer_id'],
+                                       checkpoint_id=review['checkpoint_id'], **extra)
+
+    async def test_full_flow_requires_click_then_repeat_uses_no_models(self):
+        first = (await self.ask())['state']
+        self.assertTrue(first['cache_review']['eligible'])
+        self.assertEqual(self.redis.writes, 0)
+        result = await api.approve_answer('t1', self.body(first))
+        self.assertEqual(self.redis.writes, 1)
+        calls = [self.mocks[i].call_count for i in (6, 7, 8)]
+        hit = (await self.ask())['state']
+        self.assertTrue(hit['cache_review']['hit'])
+        self.assertEqual(hit['answer'], first['answer'])
+        self.assertEqual([self.mocks[i].call_count for i in (6, 7, 8)], calls)
+        self.assertEqual(len(hit['rounds']), 2)
+        self.assertNotEqual(hit['answer_id'], first['answer_id'])
+        await api.revoke_answer('t1', self.body(hit, entry_id=result['entry_id']))
+        fresh = (await self.ask())['state']
+        self.assertFalse(fresh['cache_review']['hit'])
+
+    async def test_reuse_switch_bypasses_cache(self):
+        first = (await self.ask())['state']
+        await api.approve_answer('t1', self.body(first))
+        next_answer = (await self.ask(reuse=False))['state']
+        self.assertFalse(next_answer['cache_review']['hit'])
+        self.assertEqual(self.mocks[6].call_count, 2)
+
+    async def test_approval_binds_immutable_old_checkpoint(self):
+        first = (await self.ask())['state']
+        await self.ask('人员有多少？')
+        await api.approve_answer('t1', self.body(first))
+        record = await self.cache.lookup(first['cache_request'])
+        self.assertEqual(record['payload']['answer_id'], first['answer_id'])
+
+    async def test_wrong_answer_or_thread_rejected(self):
+        first = (await self.ask())['state']
+        wrong = self.body(first).model_copy(update={'answer_id': 'forged'})
+        for thread, body in [('t1', wrong), ('another', self.body(first))]:
+            with self.assertRaises(HTTPException):
+                await api.approve_answer(thread, body)
+        self.assertEqual(self.redis.writes, 0)
+
+    async def test_stale_approval_rejected(self):
+        first = (await self.ask())['state']
+        with patch.object(api, 'release_snapshot', return_value={**RELEASE, 'data_version': 'v2'}):
+            with self.assertRaises(HTTPException) as exc:
+                await api.approve_answer('t1', self.body(first))
+        self.assertEqual(exc.exception.status_code, 409)
+        self.assertEqual(self.redis.writes, 0)
+
+    async def test_manual_plan_confirmation_is_not_answer_approval(self):
+        event = await self.ask(auto=False)
+        self.assertEqual(event['type'], 'confirm')
+        self.assertEqual(self.redis.writes, 0)
+        done = [e async for e in api._astream_resume('t1', '确认')][-1]['state']
+        self.assertTrue(done['cache_review']['eligible'])
+        self.assertEqual(self.redis.writes, 0)
+        await api.approve_answer('t1', self.body(done))
+        self.assertEqual(self.redis.writes, 1)
+
+    async def test_client_cannot_supply_answer_content(self):
+        from pydantic import ValidationError
+        with self.assertRaises(ValidationError):
+            api.AnswerReviewRequest(answer_id='a', checkpoint_id='b', answer='forged')
+
+    async def test_write_failure_does_not_report_success(self):
+        first = (await self.ask())['state']
+        with patch.object(self.redis, 'set', side_effect=TimeoutError):
+            with self.assertRaises(HTTPException) as exc:
+                await api.approve_answer('t1', self.body(first))
+        self.assertEqual(exc.exception.status_code, 503)
+
+
+    async def test_modified_plan_stays_ineligible_after_confirmation(self):
+        await self.ask(auto=False)
+        edits = [e async for e in api._astream_resume('t1', '请修改查询条件')]
+        self.assertEqual(edits[-1]['type'], 'confirm')
+        final = [e async for e in api._astream_resume('t1', '确认')][-1]['state']
+        self.assertTrue(final['cache_modified'])
+        self.assertFalse(final['cache_review']['eligible'])
+        with self.assertRaises(HTTPException):
+            await api.approve_answer('t1', self.body(final))
+
+    async def test_policy_change_invalidates_old_approval(self):
+        first = (await self.ask())['state']
+        with patch.dict('os.environ', {'QA_ANSWER_CACHE_POLICY_VERSION': 'changed'}):
+            with self.assertRaises(HTTPException):
+                await api.approve_answer('t1', self.body(first))
+
+    async def test_revoke_race_not_reported_as_success(self):
+        first = (await self.ask())['state']
+        result = await api.approve_answer('t1', self.body(first))
+        with patch.object(self.redis, 'eval', return_value=0):
+            with self.assertRaises(HTTPException) as exc:
+                await api.revoke_answer('t1', self.body(first, entry_id=result['entry_id']))
+        self.assertEqual(exc.exception.status_code, 409)
+
+if __name__ == '__main__':
+    unittest.main(verbosity=2)

scripts/test_answer_cache_retention.py

@@ -0,0 +1,82 @@
+"""Offline cache lifecycle contracts; never connects to business services."""
+import json
+import unittest
+from unittest.mock import patch
+from step4_web import answer_cache_retention as retention
+from step4_web.answer_cache import PREFIX, ACTIVE_VERSION, DELETE_IF_SAME
+
+class Redis:
+    def __init__(self):
+        self.entries = {}
+        self.ttls = {}
+    def set(self, key, raw):
+        self.entries[key] = raw
+        self.ttls.pop(key, None)
+    def get(self, key):
+        return self.entries.get(key)
+    def scan_iter(self, match, count):
+        return (k for k in list(self.entries) if k.startswith(PREFIX))
+    def eval(self, script, count, key, raw, *args):
+        if self.entries.get(key) != raw:
+            return 0
+        if script == DELETE_IF_SAME:
+            del self.entries[key]
+            self.ttls.pop(key, None)
+        else:
+            self.set(key, args[0])
+        return 1
+
+class RetentionTests(unittest.TestCase):
+    def setUp(self):
+        self.redis = Redis()
+        for name, version in [('old1', 'v0'), ('old2', 'v-1'), ('current', 'v1')]:
+            self.redis.set(PREFIX + name, json.dumps({'request': {'data_version': version},
+                                                    'approved': True, 'expires_at': 100}))
+            self.redis.ttls[PREFIX + name] = 60
+        self.redis.set('checkpoint:thread1', 'keep')
+        self.redis.set('other:cache', 'keep')
+    def test_all_old_versions_removed_current_migrated_context_untouched(self):
+        report = retention.clean_entries(self.redis, 'v1')
+        self.assertEqual(report, {'removed': 2, 'preserved': 1})
+        self.assertNotIn(PREFIX + 'old1', self.redis.entries)
+        self.assertNotIn(PREFIX + 'old2', self.redis.entries)
+        self.assertNotIn(PREFIX + 'current', self.redis.ttls)
+        self.assertNotIn('expires_at', json.loads(self.redis.get(PREFIX + 'current')))
+        self.assertEqual(self.redis.get('checkpoint:thread1'), 'keep')
+        self.assertEqual(self.redis.get('other:cache'), 'keep')
+        self.assertEqual(self.redis.get(ACTIVE_VERSION), 'v1')
+    def test_completed_version_skips_scan_but_new_version_cleans(self):
+        retention.clean_entries(self.redis, 'v1')
+        with patch.object(self.redis, 'scan_iter', side_effect=AssertionError):
+            self.assertTrue(retention.clean_entries(self.redis, 'v1')['already_clean'])
+        self.assertEqual(retention.clean_entries(self.redis, 'v2')['removed'], 1)
+    def test_failure_leaves_retry_pending_and_fences_old_writes(self):
+        with patch.object(self.redis, 'eval', side_effect=TimeoutError):
+            with self.assertRaises(TimeoutError):
+                retention.clean_entries(self.redis, 'v1')
+        self.assertIsNone(self.redis.get(retention.CLEANED_VERSION))
+        self.assertEqual(self.redis.get(ACTIVE_VERSION), 'v1')
+        self.assertEqual(retention.clean_entries(self.redis, 'v1')['removed'], 2)
+    def test_changed_entry_not_deleted_by_stale_scan(self):
+        original = self.redis.eval
+        def race(script, count, key, raw, *args):
+            if key == PREFIX + 'old1':
+                self.redis.set(key, json.dumps({'request': {'data_version': 'v1'}}))
+            return original(script, count, key, raw, *args)
+        with patch.object(self.redis, 'eval', side_effect=race):
+            retention.clean_entries(self.redis, 'v1')
+        self.assertIn(PREFIX + 'old1', self.redis.entries)
+    def test_unreadable_own_entry_removed(self):
+        self.redis.set(PREFIX + 'broken', 'broken')
+        self.assertEqual(retention.clean_entries(self.redis, 'v1')['removed'], 3)
+    def test_outage_reports_warning(self):
+        with patch.object(retention, 'get_settings', side_effect=TimeoutError):
+            self.assertIn('warning', retention.cleanup_answer_cache_for_release('v1'))
+    def test_busy_publication_defers_cleanup(self):
+        with patch('step4_web.data_update.update_lock', side_effect=RuntimeError), patch.object(
+                retention, 'cleanup_answer_cache_for_release') as cleanup:
+            self.assertIn('warning', retention.retry_cleanup())
+            cleanup.assert_not_called()
+
+if __name__ == '__main__':
+    unittest.main()

scripts/test_answer_cache_ui.cjs

@@ -0,0 +1,83 @@
+const assert = require('node:assert/strict');
+const fs = require('node:fs');
+const path = require('node:path');
+const {chromium} = require('playwright');
+const root = path.resolve(__dirname, '../html');
+(async () => {
+  const browser = await chromium.launch({headless:true,channel:'msedge'});
+  try {
+    const page = await browser.newPage({viewport:{width:1280,height:900}});
+    const requests = [];
+    let failNext = false;
+    await page.route('**/*', async route => {
+      const url = new URL(route.request().url());
+      if (url.pathname.endsWith('/answer-cache')) {
+        requests.push({method:route.request().method(), path:url.pathname, body:route.request().postDataJSON()});
+        if (failNext) { failNext=false; return route.fulfill({status:409,json:{detail:'数据已更新，请重新提问后确认'}}); }
+        return route.fulfill({json:{status:'ok',entry_id:'entry1'}});
+      }
+      if (url.pathname === '/api/data-quality') return route.fulfill({json:{data_version:'v',total:0,datasets:{}}});
+      if (url.pathname === '/api/runtime') return route.fulfill({json:{data_version:'v',schema_version:'s'}});
+      if (url.pathname.startsWith('/output/')) return route.fulfill({json:{nodes:[],links:[],relations:[],meta:{}}});
+      const file = path.join(root, url.pathname === '/' ? 'knowledge_graph_3d.html' : url.pathname);
+      if (!file.startsWith(root) || !fs.existsSync(file)) return route.fulfill({status:404,body:''});
+      return route.fulfill({body:fs.readFileSync(file),contentType:file.endsWith('.js')?'application/javascript':file.endsWith('.css')?'text/css':'text/html'});
+    });
+    await page.goto('http://qa.test/');
+    assert.ok((await page.locator('#chatFrame').getAttribute('src')).includes('20260907-cache'));
+    const frame = await (await page.locator('#chatFrame').elementHandle()).contentFrame();
+    await frame.waitForFunction(() => typeof askStream === 'function' && Boolean(window.AnswerCacheUI));
+    await frame.evaluate(async () => {
+      const original=window.fetch;
+      const payload={status:'ok',answer:'## 员工人数\n\n共有2名员工。',elapsed_sec:.01,
+        cache_review:{eligible:true,answer_id:'answer1',checkpoint_id:'checkpoint1',cached:false,hit:false}};
+      const sse='event: answer_chunk\ndata: '+JSON.stringify({text:'处理中'})+'\n\n' +
+                'event: done\ndata: '+JSON.stringify(payload)+'\n\n';
+      window.fetch=async () => new Response(sse);
+      try {await askStream('员工有多少人',true,true);} finally {window.fetch=original;}
+    });
+    assert.equal(requests.length,0,'rendering an answer must not cache it');
+    const button=frame.locator('.answer-cache-button').last();
+    await button.click();
+    await frame.waitForFunction(() => document.querySelector('.answer-cache-button')?.textContent==='撤销缓存');
+    assert.equal(requests.length,1);
+    assert.equal(requests[0].method,'POST');
+    assert.deepEqual(requests[0].body,{answer_id:'answer1',checkpoint_id:'checkpoint1',entry_id:null});
+    const originalThreadPath=requests[0].path;
+    await frame.evaluate(() => { sessionStorage.setItem('kg_thread_id','new-thread'); });
+    await button.click();
+    await frame.waitForFunction(() => document.querySelector('.answer-cache-button')?.textContent==='答案正确，加入缓存');
+    assert.equal(requests[1].method,'DELETE');
+    assert.equal(requests[1].path,originalThreadPath,'old answer keeps its own thread scope');
+    assert.equal(requests[1].body.entry_id,'entry1');
+    failNext=true;
+    await button.click();
+    await frame.waitForFunction(() => document.querySelector('.answer-cache-status').textContent.includes('数据已更新'));
+    assert.equal(await button.innerText(),'答案正确，加入缓存');
+    assert.equal(await button.isEnabled(),true);
+    await button.click();
+    await frame.waitForFunction(() => document.querySelector('.answer-cache-button').textContent==='撤销缓存');
+    // Resume rendering gets the same explicit approval control.
+    await frame.evaluate(async () => {
+      const original=window.fetch;
+      window.fetch=async () => new Response(JSON.stringify({status:'ok',answer:'确认后完成',
+        cache_review:{eligible:true,answer_id:'answer2',checkpoint_id:'checkpoint2',cached:false}}));
+      try {await resumeThread('确认');} finally {window.fetch=original;}
+    });
+    assert.equal(await frame.locator('.answer-cache-button').count(),2);
+    await frame.evaluate(() => {
+      const bubble=appendMessage('assistant','历史确认答案');
+      AnswerCacheUI.attach(bubble,{cache_review:{eligible:true,answer_id:'hit',checkpoint_id:'cphit',
+        cached:true,hit:true,entry_id:'entry1'}},'t1',apiBase());
+      const incomplete=appendMessage('assistant','不完整结果');
+      AnswerCacheUI.attach(incomplete,{cache_review:{eligible:false,reason:'结果截断'}},'t1',apiBase());
+    });
+    assert.equal(await frame.getByText('已复用你确认的历史答案',{exact:true}).count(),1);
+    assert.equal(await frame.getByText('暂不缓存：结果截断',{exact:true}).count(),1);
+    assert.equal(await frame.locator('.answer-cache-button').count(),3);
+    await page.setViewportSize({width:390,height:844});
+    await frame.locator('.answer-cache-button').last().scrollIntoViewIfNeeded();
+    assert.ok(await frame.locator('.answer-cache-button').last().isVisible());
+    console.log('PASS root iframe: explicit approval only, bound answer/checkpoint/thread, revoke, stale-error retry, resume, hit badge, excluded answer, mobile');
+  } finally {await browser.close();}
+})().catch(e=>{console.error(e);process.exitCode=1;});

scripts/test_runtime_workflow.py

@@ -57,7 +57,8 @@ class UpdateTests(unittest.TestCase):
         self.write_template('人员信息',['工号','姓名'])
         self.write_template('项目信息',['项目编号','负责人编号'])
         self.write_relation()
-        self.patches=[patch.object(update,'ROOT',self.root),
+        self.patches=[patch.object(update,'cleanup_answer_cache_for_release',return_value={'removed': 0}),
+                      patch.object(update,'ROOT',self.root),
                       patch.object(update,'MANIFEST',self.root/'output/current_release.json'),
                       patch.object(runtime,'ROOT',self.root),
                       patch.object(runtime,'MANIFEST',self.root/'output/current_release.json'),
@@ -109,6 +110,22 @@ class UpdateTests(unittest.TestCase):
         return update.update_data(mode,fetcher=self.fetch,assessor=kwargs.get('assessor',self.assess),
                                   analyzer=self.analyze,driver=self.driver)
 
+    def test_cache_cleanup_only_after_successful_publication(self):
+        def cleaned(version):
+            manifest = json.loads(update.MANIFEST.read_text(encoding='utf-8'))
+            self.assertEqual(manifest['data_version'], version)
+            return {'warning': 'Redis unavailable; retry pending'}
+        with patch.object(update, 'cleanup_answer_cache_for_release', side_effect=cleaned) as cleanup:
+            report = self.run_update()
+            self.assertTrue(report['ok'])
+            self.assertIn('warning', report['answer_cache_cleanup'])
+            cleanup.assert_called_once_with(report['data_version'])
+        with patch.object(update, 'cleanup_answer_cache_for_release') as cleanup:
+            with patch.object(self, 'fetch', side_effect=ValueError('fetch failed')):
+                with self.assertRaises(ValueError):
+                    self.run_update()
+            cleanup.assert_not_called()
+
     def test_dms_updates_assessment_but_preserves_schema_bytes(self):
         first=self.run_update()
         before=runtime.release_snapshot()

scripts/test_simple_plan.py

@@ -0,0 +1,307 @@
+"""Offline correctness and model-call regression for the simple planning path."""
+from copy import deepcopy
+from types import SimpleNamespace
+import os
+import unittest
+from unittest.mock import patch
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.types import Command
+from step3_qa_agent.agent import nodes, production, llm
+from step3_qa_agent.agent.graph import build_agent_graph
+from step3_qa_agent.agent.simple_plan import build_simple_plan
+from step3_qa_agent.agent.timing import measure_llm_calls, llm_calls
+from step3_qa_agent.retrieval.production_query import compile_query
+
+SCHEMA = {'meta': {'schema_version': 2}, 'nodes': [
+    {'id': '人员信息', 'active': True, 'attributes': ['工号', '姓名', '岗位名称', '部门'],
+     'identity_fields': ['工号'], 'multivalue_fields': {'岗位名称': 1}},
+    {'id': '项目信息', 'attributes': ['项目编号', '项目名称']}],
+    'relations': [{'id': 'works', 'source': '人员信息', 'target': '项目信息', 'type': '服务于'}]}
+CONTEXT = {'schema': SCHEMA, 'build_id': 'v1', 'catalog': [], 'assessment_text': '',
+           'selected_datasets': ['人员信息'], 'loaded_datasets': [], 'selection_source': 'understanding',
+           'issues': []}
+
+def state(question='共有多少名员工？'):
+    return {'question': question, 'category': '图谱检索', 'qa_context': deepcopy(CONTEXT),
+            'slots': {'anchors': [], 'filters': {}, 'aggregation': '汇总'},
+            'simple_query_needs': {'datasets': ['人员信息'], 'relations': []},
+            'simple_query': {'kind': 'count', 'dataset': '人员信息', 'filters': [],
+                             'fields': [], 'limit': 50, 'complete': True}}
+
+class SimplePlanTests(unittest.TestCase):
+    def setUp(self):
+        self.env = patch.dict(os.environ, {'QA_FAST_PLAN_ENABLED': 'true'})
+        self.env.start()
+        self.addCleanup(self.env.stop)
+
+    def test_count_is_distinct_node_query_not_schema_count(self):
+        s = state()
+        plan, _ = build_simple_plan(s)
+        spec = plan['steps'][0]['params']
+        compiled = compile_query(spec, SCHEMA, 'v1')
+        self.assertIn('count(DISTINCT n._kg_id)', compiled.query)
+        self.assertEqual(compiled.params['build'], 'v1')
+        with patch.object(production, 'chat_json') as model:
+            result = production.plan(s)
+        model.assert_not_called()
+        self.assertEqual(result['trace'][-1]['mode'], 'simple')
+        self.assertEqual(result['trace'][-1]['llm_calls'], [])
+
+    def test_property_query_keeps_anchor_and_provenance(self):
+        s = state('张三的岗位名称是什么？')
+        s['slots'] = {'anchors': [{'raw': '张三'}]}
+        s['simple_query'].update(kind='list', fields=['姓名', '岗位名称'],
+             filters=[{'field': '姓名', 'op': 'eq', 'value': '张三'}])
+        plan, _ = build_simple_plan(s)
+        compiled = compile_query(plan['steps'][0]['params'], SCHEMA, 'v1')
+        self.assertEqual(compiled.params['v0'], '张三')
+        self.assertIn('_sources', compiled.query)
+        self.assertIn('any(x IN CASE', compiled.query)
+
+    def test_list_preserves_all_and_filters_and_limit(self):
+        s = state('列出运营部岗位名称包含保安的员工姓名')
+        s['slots'] = {'filters': {'部门': '运营部', '岗位': '保安'}}
+        s['simple_query'].update(kind='list', fields=['工号', '姓名'], limit=20,
+            filters=[{'field': '部门', 'op': 'eq', 'value': '运营部'},
+                     {'field': '岗位名称', 'op': 'contains', 'value': '保安'}])
+        plan, _ = build_simple_plan(s)
+        compiled = compile_query(plan['steps'][0]['params'], SCHEMA, 'v1')
+        self.assertEqual((compiled.params['v0'], compiled.params['v1']), ('运营部', '保安'))
+        self.assertEqual(compiled.params['result_limit'], 21)
+        self.assertIn(' AND ', compiled.query)
+
+    def test_complex_questions_reject_even_when_model_claims_complete(self):
+        for question in ['他们有多少人', '按部门统计人数', '每个项目多少人', '保安或电工有多少',
+                         '没有证书的员工', '去年入职人数', '工资大于5000人数',
+                         '前10名员工', '删除所有员工', '比较项目人数']:
+            with self.subTest(question=question):
+                self.assertIsNone(build_simple_plan(state(question))[0])
+
+    def test_relation_and_invalid_selection_never_disappear(self):
+        for needs in [{'datasets': ['人员信息'], 'relations': ['unknown']},
+                      {'datasets': ['人员信息', '项目信息'], 'relations': ['works']},
+                      {'datasets': ['人员信息', 'unknown'], 'relations': []}]:
+            s = state()
+            s['simple_query_needs'] = needs
+            self.assertIsNone(build_simple_plan(s)[0])
+
+    def test_dropped_anchor_or_filter_is_rejected(self):
+        for slots in [{'anchors': [{'raw': '张三'}]}, {'filters': {'岗位': '保安'}},
+                      {'time_range': {'start': '2026-01'}}, {'filters': {'岗位': ['保安', '电工']}}]:
+            s = state('张三是保安吗')
+            s['slots'] = slots
+            self.assertIsNone(build_simple_plan(s)[0])
+
+    def test_unknown_field_and_nonliteral_value_fall_back(self):
+        for condition in [{'field': '不存在', 'op': 'eq', 'value': '员工'},
+                          {'field': '岗位名称', 'op': 'eq', 'value': '保安'},
+                          {'field': '岗位名称', 'op': 'ne', 'value': '员工'}]:
+            s = state()
+            s['simple_query']['filters'] = [condition]
+            self.assertIsNone(build_simple_plan(s)[0])
+
+    def test_malformed_candidates_fail_closed(self):
+        for candidate in [None, [], {}, {'complete': True},
+                          {**state()['simple_query'], 'fields': None},
+                          {**state()['simple_query'], 'fields': [['姓名']]},
+                          {**state()['simple_query'], 'limit': True},
+                          {**state()['simple_query'], 'limit': 201},
+                          {**state()['simple_query'], 'filters': [None]},
+                          {**state()['simple_query'], 'extra': 'ignored?'},
+                          {**state()['simple_query'], 'kind': 'sum'},
+                          {**state()['simple_query'], 'complete': 'true'}]:
+            s = state()
+            s['simple_query'] = candidate
+            self.assertIsNone(build_simple_plan(s)[0])
+
+    def test_feedback_always_uses_full_planner(self):
+        original, _ = build_simple_plan(state())
+        for key in ['plan_feedback', 'run_feedback', 'slots_feedback', 'feedback']:
+            s = state()
+            s[key] = '改成只查保安'
+            with patch.object(production, 'chat_json', return_value=original) as model:
+                result = production.plan(s)
+            model.assert_called_once()
+            self.assertEqual(result['trace'][-1]['mode'], 'llm')
+            self.assertEqual(result['trace'][-1]['reason'], 'feedback_requires_replanning')
+
+    def test_disabled_and_legacy_fall_back(self):
+        original, _ = build_simple_plan(state())
+        with patch.dict(os.environ, {'QA_FAST_PLAN_ENABLED': 'false'}):
+            with patch.object(production, 'chat_json', return_value=original) as model:
+                self.assertEqual(production.plan(state())['trace'][-1]['reason'], 'disabled')
+                model.assert_called_once()
+        s = state()
+        s['qa_context']['schema']['meta']['schema_version'] = 1
+        self.assertIsNone(build_simple_plan(s)[0])
+
+    def model_output(self):
+        s = state()
+        return {'category': '图谱检索', 'slots': s['slots'], 'data_needs': s['simple_query_needs'],
+                'simple_query': s['simple_query']}
+
+    def test_real_graph_skips_second_model_and_still_confirms(self):
+        with patch.object(nodes, 'load_question_context', return_value=deepcopy(CONTEXT)), \
+             patch.object(nodes, 'select_assessments', side_effect=lambda ctx, *a, **kw: ctx), \
+             patch.object(nodes, 'chat_json', return_value=self.model_output()) as understanding, \
+             patch.object(production, 'chat_json') as planner:
+            graph = build_agent_graph(InMemorySaver())
+            result = graph.invoke({'question': '共有多少名员工？'},
+                                  {'configurable': {'thread_id': 'simple'}})
+        understanding.assert_called_once()
+        planner.assert_not_called()
+        self.assertEqual(result['__interrupt__'][0].value['type'], 'confirm_plan')
+        self.assertEqual(result['trace'][-1]['mode'], 'simple')
+
+    def test_confirm_edit_replans_and_then_executes(self):
+        original, _ = build_simple_plan(state())
+        driver = SimpleNamespace(execute_query=lambda *a, **kw: SimpleNamespace(records=[{'记录数': 2}]))
+        with patch.object(nodes, 'load_question_context', return_value=deepcopy(CONTEXT)), \
+             patch.object(nodes, 'select_assessments', side_effect=lambda ctx, *a, **kw: ctx), \
+             patch.object(nodes, 'chat_json', return_value=self.model_output()), \
+             patch.object(production, 'chat_json', return_value=original) as planner, \
+             patch.object(production, 'get_driver', return_value=driver), \
+             patch.object(production, 'release_snapshot', return_value={'data_version': 'v1'}), \
+             patch.object(production, 'chat_text', return_value='共2条记录'):
+            graph = build_agent_graph(InMemorySaver())
+            config = {'configurable': {'thread_id': 'edit'}}
+            graph.invoke({'question': '共有多少名员工？'}, config)
+            result = graph.invoke(Command(resume='请重新核对计数口径'), config)
+            self.assertIn('__interrupt__', result)
+            planner.assert_called_once()
+            result = graph.invoke(Command(resume='确认'), config)
+            self.assertEqual(result['answer'], '共2条记录')
+            self.assertEqual(result['subgraph']['results']['s1']['rows'], [{'记录数': 2}])
+
+    def test_understanding_resets_old_candidate_and_switch_restores_prompt(self):
+        for enabled in ['true', 'false']:
+            s = state()
+            with patch.dict(os.environ, {'QA_FAST_PLAN_ENABLED': enabled}), \
+                 patch.object(nodes, 'load_question_context', return_value=deepcopy(CONTEXT)), \
+                 patch.object(nodes, 'select_assessments', side_effect=lambda ctx, *a, **kw: ctx), \
+                 patch.object(nodes, 'chat_json', return_value={'category': '闲聊'}) as model:
+                result = nodes.understand(s)
+            self.assertIsNone(result['simple_query'])
+            self.assertIsNone(result['simple_query_needs'])
+            self.assertEqual('simple_query' in model.call_args.args[0], enabled == 'true')
+
+    def test_full_planner_still_retries_invalid_plan(self):
+        s = state()
+        s['simple_query'] = None
+        original, _ = build_simple_plan(state())
+        with patch.object(production, 'chat_json', side_effect=[{'steps': []}, original]) as model:
+            result = production.plan(s)
+        self.assertEqual(model.call_count, 2)
+        self.assertEqual(result['trace'][-1]['validation_attempts'], 2)
+
+    def test_minimal_candidate_matches_explicit_defaults(self):
+        full = state()
+        minimal = state()
+        minimal['simple_query'] = {'kind': 'count', 'dataset': '人员信息', 'complete': True}
+        self.assertEqual(build_simple_plan(full), build_simple_plan(minimal))
+        plan, _ = build_simple_plan(minimal)
+        step = plan['steps'][0]
+        self.assertNotIn('fields', step)
+        self.assertNotIn('depends', step)
+        self.assertNotIn('select', step['params'])
+        self.assertNotIn('filters', step['params'])
+        self.assertNotIn('limit', step['params'])
+
+    def test_compact_full_plan_preserves_filters_and_does_not_add_empty_fields(self):
+        s = state('按部门分别统计运营部员工')
+        s['simple_query'] = None
+        plan = {'steps': [{'step_id': 's1', 'tool': '图谱查询', 'params': {
+            'nodes': [{'alias': 'n', 'type': '人员信息'}],
+            'filters': [{'alias': 'n', 'field': '部门', 'op': 'eq', 'value': '运营部'}],
+            'select': [{'alias': 'n', 'field': '部门', 'as': '部门'}],
+            'aggregates': [{'alias': 'n', 'op': 'count', 'as': '人数'}]}}]}
+        with patch.dict(os.environ, {'QA_COMPACT_INTERMEDIATE_ENABLED': 'true'}), \
+             patch.object(production, 'chat_json', return_value=plan) as model:
+            result = production.plan(s)
+        self.assertEqual(result['plan'], plan)
+        self.assertIn('省略所有空数组', model.call_args.args[0])
+        self.assertNotIn('"history"', model.call_args.args[1])
+        self.assertNotIn('"feedback"', model.call_args.args[1])
+        compiled = compile_query(plan['steps'][0]['params'], SCHEMA, 'v1')
+        self.assertEqual(compiled.params['v0'], '运营部')
+        self.assertIn('n.`部门` AS `部门`', compiled.query)
+
+    def test_both_switches_off_restore_original_prompt(self):
+        s = state()
+        original, _ = build_simple_plan(s)
+        with patch.dict(os.environ, {'QA_FAST_PLAN_ENABLED': 'false',
+                                     'QA_COMPACT_INTERMEDIATE_ENABLED': 'false'}), \
+             patch.object(nodes, 'load_question_context', return_value=deepcopy(CONTEXT)), \
+             patch.object(nodes, 'select_assessments', side_effect=lambda ctx, *a, **kw: ctx), \
+             patch.object(nodes, 'chat_json', return_value=self.model_output()) as understand_model, \
+             patch.object(production, 'chat_json', return_value=original) as plan_model:
+            s.update(nodes.understand(s))
+            production.plan(s)
+        self.assertNotIn('simple_query', understand_model.call_args.args[0])
+        self.assertNotIn('仅输出紧凑JSON', understand_model.call_args.args[0])
+        self.assertIn('不需要的数组填[]', plan_model.call_args.args[0])
+
+    def test_invalid_switch_is_not_silently_accepted(self):
+        from step2_graph_building.config import get_qa_fast_plan_enabled
+        with patch.dict(os.environ, {'QA_FAST_PLAN_ENABLED': 'maybe'}):
+            with self.assertRaises(ValueError):
+                get_qa_fast_plan_enabled()
+
+    def test_graph_records_one_vs_two_real_wrapper_streams(self):
+        import json
+        original, _ = build_simple_plan(state())
+        output = self.model_output()
+        class Model:
+            def stream(self, messages):
+                body = output if '问题理解器' in messages[0]['content'] else original
+                yield SimpleNamespace(content=json.dumps(body, ensure_ascii=False), usage_metadata=None)
+        for enabled, expected in [('true', 1), ('false', 2)]:
+            with patch.dict(os.environ, {'QA_FAST_PLAN_ENABLED': enabled,
+                                         'QA_COMPACT_INTERMEDIATE_ENABLED': enabled}), \
+                 patch.object(nodes, 'load_question_context', return_value=deepcopy(CONTEXT)), \
+                 patch.object(nodes, 'select_assessments', side_effect=lambda ctx, *a, **kw: ctx), \
+                 patch.object(llm, 'get_chat', return_value=Model()) as client:
+                result = build_agent_graph(InMemorySaver()).invoke({'question': '共有多少名员工？'},
+                    {'configurable': {'thread_id': enabled}})
+            self.assertEqual(client.call_count, expected)
+            self.assertEqual(sum(len(t['llm_calls']) for t in result['trace']), expected)
+            self.assertEqual(result['__interrupt__'][0].value['type'], 'confirm_plan')
+
+class TimingTests(unittest.TestCase):
+    def test_json_retries_and_usage_are_measured(self):
+        class Model:
+            def __init__(self):
+                self.count = 0
+            def stream(self, messages):
+                self.count += 1
+                yield SimpleNamespace(content='', usage_metadata=None)
+                yield SimpleNamespace(content='invalid' if self.count == 1 else '{"ok":true}',
+                                      usage_metadata={'input_tokens': 4, 'output_tokens': 2})
+        with measure_llm_calls() as calls, patch.object(llm, 'get_chat', return_value=Model()):
+            self.assertEqual(llm.chat_json('system', 'user'), {'ok': True})
+        self.assertEqual(len(calls), 2)
+        self.assertEqual(calls[-1]['usage']['input_tokens'], 4)
+        self.assertGreaterEqual(calls[-1]['duration_ms'], calls[-1]['first_text_ms'])
+        self.assertNotIn('system', str(calls))
+        self.assertIsNone(llm_calls.get())
+
+    def test_nested_collectors_do_not_leak(self):
+        with measure_llm_calls() as outer:
+            with measure_llm_calls() as inner:
+                self.assertIs(llm_calls.get(), inner)
+            self.assertIs(llm_calls.get(), outer)
+        self.assertIsNone(llm_calls.get())
+
+    def test_stream_failure_is_measured_and_propagated(self):
+        def fail(messages):
+            raise RuntimeError('transport failed')
+            yield
+        with measure_llm_calls() as calls, patch.object(llm, 'get_chat',
+                return_value=SimpleNamespace(stream=fail)):
+            with self.assertRaises(RuntimeError):
+                llm.chat_json('system', 'user')
+        self.assertEqual(calls[0]['status'], 'error')
+        self.assertIsNone(calls[0]['first_text_ms'])
+
+if __name__ == '__main__':
+    unittest.main(verbosity=2)

src/step2_graph_building/config.py

@@ -106,6 +106,34 @@ def get_relation_path(dotenv_path: str | Path | None = None) -> Path:
     return path
 
 
+def _optional_bool(name: str, default: str = 'true') -> bool:
+    _load()
+    value = os.getenv(name, default).strip().lower()
+    if value not in {'true', 'false', '1', '0'}:
+        raise ValueError(f'{name} 必须为 true/false 或 1/0')
+    return value in {'true', '1'}
+
+
+def get_qa_answer_cache_policy() -> tuple[str, str]:
+    _load()
+    return os.getenv('QA_ANSWER_CACHE_POLICY_VERSION', '1'), os.getenv('DEEPSEEK_MODEL', '')
+
+
+def get_qa_answer_cache_config() -> bool:
+    # Approved answers live until a data-version change or explicit revocation.
+    return _optional_bool('QA_ANSWER_CACHE_ENABLED')
+
+
+def get_qa_fast_plan_enabled() -> bool:
+    """Use validated single-dataset candidates before calling the full planner."""
+    return _optional_bool('QA_FAST_PLAN_ENABLED')
+
+
+def get_qa_compact_intermediate_enabled() -> bool:
+    """Generate only required JSON fields in production intermediate responses."""
+    return _optional_bool('QA_COMPACT_INTERMEDIATE_ENABLED')
+
+
 def get_embedding_model_dir() -> Path:
     _load()
     raw = os.getenv("EMBEDDING_MODEL_DIR", "models/Qwen3-Embedding-0.6B").strip()

src/step3_qa_agent/agent/llm.py

@@ -4,10 +4,12 @@ from __future__ import annotations
 
 import json
 import re
+from time import perf_counter
 
 from langchain_openai import ChatOpenAI
 
 from step2_graph_building.config import get_settings
+from .timing import llm_calls
 
 
 SYSTEM_SAFETY = (
@@ -54,6 +56,35 @@ def _chunk_text(chunk) -> str:
     return str(c or "")
 
 
+def _measured_stream(llm, messages):
+    """Record one SDK stream; internal transport retries are included in duration."""
+    started = perf_counter()
+    record = {'first_chunk_ms': None, 'first_text_ms': None, 'output_chars': 0,
+              'input_chars': sum(len(m['content']) for m in messages), 'usage': None}
+    calls = llm_calls.get()
+    if calls is not None:
+        calls.append(record)
+    try:
+        for chunk in llm.stream(messages):
+            elapsed = round((perf_counter() - started) * 1000, 2)
+            if record['first_chunk_ms'] is None:
+                record['first_chunk_ms'] = elapsed
+            piece = _chunk_text(chunk)
+            if piece and record['first_text_ms'] is None:
+                record['first_text_ms'] = elapsed
+            record['output_chars'] += len(piece)
+            usage = getattr(chunk, 'usage_metadata', None)
+            if isinstance(usage, dict) and usage:
+                record['usage'] = usage
+            yield chunk
+        record['status'] = 'ok'
+    except Exception:
+        record['status'] = 'error'
+        raise
+    finally:
+        record['duration_ms'] = round((perf_counter() - started) * 1000, 2)
+
+
 def chat_json(system: str, user: str, model: str | None = None,
               on_chunk=None) -> dict:
     """JSON 输出：底层用流式收集（结果等价，方便后续透传进度）。"""
@@ -61,7 +92,7 @@ def chat_json(system: str, user: str, model: str | None = None,
     last_err = None
     for _ in range(3):
         text = ""
-        for chunk in llm.stream([
+        for chunk in _measured_stream(llm, [
             {"role": "system", "content": system + "\n" + SYSTEM_SAFETY},
             {"role": "user", "content": user},
         ]):
@@ -82,7 +113,7 @@ def chat_text(system: str, user: str, model: str | None = None,
     """文本输出：底层用流式收集；on_chunk 可回调每个增量（API 透传进度用）。"""
     llm = get_chat(model=model, temperature=0.3)
     text = ""
-    for chunk in llm.stream([
+    for chunk in _measured_stream(llm, [
         {"role": "system", "content": system + "\n" + SYSTEM_SAFETY},
         {"role": "user", "content": user},
     ]):

src/step3_qa_agent/agent/nodes.py

@@ -7,6 +7,7 @@ import json
 import inspect
 import re
 from datetime import date
+from time import perf_counter
 
 from langgraph.types import interrupt
 from langgraph.config import get_stream_writer
@@ -20,6 +21,9 @@ from .derived import RULES, apply_derived_expr, extract_condition
 from .llm import chat_json, chat_text
 from .schema_context import build_schema_text
 from .question_context import load_question_context, question_schema_text, select_assessments
+from .simple_plan import SIMPLE_QUERY_GUIDE
+from .timing import measure_llm_calls
+from step2_graph_building.config import get_qa_fast_plan_enabled, get_qa_compact_intermediate_enabled
 
 
 INTENT_EDGES: dict[str, list[str]] = {
@@ -70,6 +74,7 @@ def _norm_month(v: str) -> str:
 
 def understand(state: dict) -> dict:
     """问题理解：一次 LLM 调用完成 闲聊/图谱分类 + 图谱检索槽位抽取（替代 classify+fill_slots）。"""
+    started = perf_counter()
     ctx = _history_context(state)
     context = load_question_context()
     sys_p = f"""你是申勤物业问题理解器。先判断用户问题是闲聊还是需要基于业务知识图谱检索回答；
@@ -96,19 +101,32 @@ data_needs 规则：
 {question_schema_text(context)}
 【可用数据评估目录（本阶段不注入评估正文）】
 {json.dumps(context["catalog"], ensure_ascii=False)}"""
+    fast_enabled = (context['schema'].get('meta', {}).get('schema_version') == 2
+                    and get_qa_fast_plan_enabled())
+    if fast_enabled:
+        sys_p += '\n' + SIMPLE_QUERY_GUIDE
+    if context['schema'].get('meta', {}).get('schema_version') == 2 and get_qa_compact_intermediate_enabled():
+        sys_p += ('\n输出要求：仅输出紧凑JSON，不输出分析、解释或Markdown。'
+                  'slots仅保留有值的键，空anchors/filters、null时间、空筛选值无需输出。'
+                  'data_needs保留datasets和relations；不得为了缩短输出遗漏条件、实体或返回要求。')
     user_p = f"用户问题：{state['question']}"
     if state.get("slots_feedback"):
         user_p += f"\n用户对上次抽取结果的修改意见：{state['slots_feedback']}"
-    out = chat_json(sys_p, user_p)
+    with measure_llm_calls() as calls:
+        out = chat_json(sys_p, user_p)
     cat = "图谱检索" if str(out.get("category", "")).strip() in ("图谱检索", "知识图谱", "检索") else "闲聊"
     slots = out.get("slots") if isinstance(out.get("slots"), dict) else {}
     needs = out.get("data_needs") if cat == "图谱检索" else {"datasets": [], "relations": []}
     context = select_assessments(context, needs, question=state["question"], slots=slots)
     return {"category": cat, "slots": slots, "qa_context": context,
+            "simple_query": out.get('simple_query') if fast_enabled and cat == '图谱检索' else None,
+            "simple_query_needs": needs if fast_enabled and cat == '图谱检索' else None,
             "plan_feedback": "", "run_feedback": "", "feedback": "", "iterations": 0,
             "answer": "", "chat_answer": "", "subgraph": {}, "plan": {},
             "messages": state.get("messages", []) + [{"role": "user", "content": state["question"]}],
             "trace": state.get("trace", []) + [{"node": "understand", "category": cat,
+                                                "duration_ms": round((perf_counter() - started) * 1000, 2),
+                                                "llm_calls": calls,
                                                 "slots": slots,
                                                 "schema_build_id": context["build_id"],
                                                 "selected_datasets": context["selected_datasets"],

src/step3_qa_agent/agent/production.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 import json
 from copy import deepcopy
+from time import perf_counter
 from .value_clarification import find_candidates, selected_options
 from langgraph.types import interrupt
 from langgraph.config import get_stream_writer
@@ -11,6 +12,9 @@ from ..retrieval.production_query import compile_query
 from .llm import chat_json, chat_text
 from .plan_presentation import describe_plan
 from .question_context import question_schema_text, assessment_context_text
+from .simple_plan import build_simple_plan
+from .timing import measure_llm_calls
+from step2_graph_building.config import get_qa_fast_plan_enabled, get_qa_compact_intermediate_enabled
 
 QUERY_GUIDE = """
 只输出JSON {"steps":[{"step_id":"s1","tool":"图谱查询","params":{
@@ -32,21 +36,57 @@ eq/in/contains筛选同时支持标量与多值字段，按任一元素命中；
 """
 
 
+COMPACT_QUERY_GUIDE = """
+仅输出紧凑JSON，不写分析/解释/Markdown：
+{"steps":[{"step_id":"s1","tool":"图谱查询","params":{"nodes":[{"alias":"n","type":"节点id"}],"select":[{"alias":"n","field":"真实字段","as":"列名"}]}}]}。
+params可选项：
+relations:[{source:"别名",target:"别名",id:"关系id"}]；
+filters:[{alias:"别名",field:"真实字段",op:"eq|ne|contains|in|gt|gte|lt|lte|is_null|not_null",value:条件值}]；
+aggregates:[{alias:"别名",op:"count|count_distinct|sum|avg|min|max",field:"字段",as:"列名"}]；
+order_by:[{field:"返回列名",direction:"asc|desc"}]；limit:1至200整数，默认50。
+省略所有空数组、默认limit、重复fields和depends；不得省略实际条件、关系、返回字段及统计口径。
+最多5步，每步独立执行，无步骤引用；每步1至4节点、最多3关系，所有节点连通。
+select与aggregates合用表示分组；只计数时省略select。count按节点id去重且省略field，
+人数只能计人员节点，不能计证书节点；count_distinct统计字段不同值。
+sum/avg/min/max只允许单节点的单值数值字段，禁止跨关系累加。
+eq/in/contains按多值字段任一元素命中；contains为文本包含，不能代表否定。
+数字比较用数字值；is_null/not_null省略value。明细上限不限制聚合前计数。
+复合概念拆为真实属性（如中级电工：级别=中级 AND 证书名称=电工），不猜同义值、不遗漏条件。
+字段仅用元图谱真实字段，不输出Cypher/SQL/脚本；评估报告不能充当事实答案。
+不支持则输出{"steps":[],"unsupported_reason":"简短说明"}。
+"""
+
+
 def context_text(state):
     ctx=state['qa_context']
     return question_schema_text(ctx)+'\n'+assessment_context_text(ctx)
 
 
 def plan(state):
-    prompt='你是生产知识图谱查询规划器。根据本轮问题、元图谱和评估制定最少的只读查询。\n'+QUERY_GUIDE
+    started = perf_counter()
+    proposed, reason = build_simple_plan(state) if get_qa_fast_plan_enabled() else (None, 'disabled')
+    if proposed is not None:
+        validate_plan(proposed, state['qa_context'])
+        return {'plan': proposed, 'run_feedback': '',
+                'trace': state.get('trace', []) + [{'node': 'production_plan', 'plan': proposed,
+                    'mode': 'simple', 'reason': reason, 'llm_calls': [],
+                    'duration_ms': round((perf_counter() - started) * 1000, 2)}]}
+    compact = get_qa_compact_intermediate_enabled()
+    guide = COMPACT_QUERY_GUIDE if compact else QUERY_GUIDE
+    prompt='你是生产知识图谱查询规划器。根据本轮问题、元图谱和评估制定最少的只读查询。\n'+guide
     prompt+='\n'+context_text(state)
     history=[{'question':r.get('question'), 'answer':str(r.get('answer',''))[:600]}
              for r in (state.get('rounds') or [])[-3:]]
     payload={'question':state['question'], 'slots':state.get('slots',{}),
              'history':history, 'feedback':state.get('plan_feedback') or state.get('run_feedback','')}
+    if compact:
+        payload = {key: value for key, value in payload.items() if value not in ('', {}, [])}
     proposed={}
+    calls = []
     for attempt in range(2):
-        proposed=chat_json(prompt,json.dumps(payload,ensure_ascii=False))
+        with measure_llm_calls() as attempt_calls:
+            proposed=chat_json(prompt,json.dumps(payload,ensure_ascii=False))
+        calls.extend(attempt_calls)
         try:
             validate_plan(proposed,state['qa_context'])
             break
@@ -55,7 +95,9 @@ def plan(state):
                 proposed={'steps':[], 'unsupported_reason':'查询计划未通过结构校验，需重新明确问题。'}
             else:
                 prompt+='\n上次计划错误：'+str(exc)
-    return {'plan':proposed,'run_feedback':'','trace':state.get('trace',[])+[{'node':'production_plan','plan':proposed}]}
+    return {'plan':proposed,'run_feedback':'','trace':state.get('trace',[])+[{'node':'production_plan','plan':proposed,
+        'mode': 'llm', 'reason': reason, 'llm_calls': calls, 'validation_attempts': attempt + 1,
+        'duration_ms': round((perf_counter() - started) * 1000, 2)}]}
 
 
 def validate_plan(proposed,context):
@@ -80,7 +122,8 @@ def confirm(state):
     message=describe_plan(state['plan'], state['qa_context']['schema'])
     reply=interrupt({'type':'confirm_plan','message':message})
     ok=str(reply).strip().lower() in {'确认','是','继续','执行','ok','yes','y'}
-    return {'plan_confirm':ok, 'plan_feedback':'' if ok else str(reply)}
+    return {'plan_confirm':ok, 'plan_feedback':'' if ok else str(reply),
+            'cache_modified': state.get('cache_modified', False) or not ok}
 
 
 def execute(state):
@@ -141,7 +184,7 @@ def clarify(state):
                 condition.update(op='eq' if len(unique_values) == 1 else 'in',
                                  value=unique_values[0] if len(unique_values) == 1 else unique_values)
             validate_plan(updated, state['qa_context'])
-            return {'plan': updated, 'plan_confirm': True,
+            return {'plan': updated, 'plan_confirm': True, 'cache_modified': True,
                     'trace': state.get('trace', []) + [{'node': 'value_choice', 'choices': choices}]}
         message = ('请勾选一个或多个候选，也可输入多个序号或完整值（用逗号分隔），'
                    '或回复“都不是”。“确认”不会自动选择候选。')
@@ -181,7 +224,9 @@ def answer(state):
                        on_chunk=on_chunk)
     rounds=list(state.get('rounds',[]))
     rounds.append({'question':state['question'],'answer':text,
-                   'schema_build_id':state['qa_context']['build_id'],'anchors':[]})
+                   'schema_build_id':state['qa_context']['build_id'],'anchors':[],
+                   'cache_context': (state.get('cache_request') or {}).get('context'),
+                   'answer_id': state.get('answer_id')})
     return {'answer':text,'rounds':rounds,
             'messages':state.get('messages',[])+[{'role':'assistant','content':text}]}
 

src/step3_qa_agent/agent/simple_plan.py

@@ -0,0 +1,109 @@
+"""Conservative single-dataset plans from the current understanding result."""
+from __future__ import annotations
+import re
+from ..retrieval.production_query import compile_query
+
+SIMPLE_QUERY_GUIDE = """
+生产图谱还需输出 simple_query（复杂或不确定时为 null）。仅当本轮问题完整地对应
+一个数据集的计数、属性查询或明细列表，且所有条件都是明确文本等于/包含的 AND 组合时，输出：
+{"kind":"count|list","dataset":"节点id","filters":[{"field":"真实字段","op":"eq|contains","value":"用户原词"}],
+ "fields":["明细返回字段"],"limit":50,"complete":true}。
+count 按该数据集节点主键去重计数，不能把证书数当人数。
+空 filters、空 fields 和默认 limit=50 可省略。例如总数查询仅需
+{"kind":"count","dataset":"节点id","complete":true}，完整性声明 complete 不可省略。
+list 给出真实返回字段（最多8个，列表带标识字段），limit 为1至200，默认50。
+只允许原问题明确出现的筛选值，不猜测同义词；不得漏掉任何条件来简化问题。
+data_needs 必须只选择这个数据集且 relations=[]。complete=true 仅表示所有条件和返回要求已完整表达。
+跨表/关系、时间条件、数值比较、否定/或条件、分组/排序/比较/计算、历史指代、
+需参考评估正文才能确定口径或字段的情况，均输出 null，交给后续规划器。
+不要输出分析说明，不生成完整 steps 或 Cypher。闲聊时 simple_query=null。
+"""
+
+# Full planning is required even if the model incorrectly proposes a shortcut.
+COMPLEX_QUESTION = re.compile(
+    r'他们|她们|它们|这些|那些|上述|上面|之前|刚才|该项目|该人员|继续|分别|每个|各个|'
+    r'按.+(?:统计|分组|汇总)|分组|排序|排名|前\s*\d+|最高|最低|最多|最少|'
+    r'平均|合计|总额|占比|比例|比较|对比|相比|大于|小于|超过|不足|至少|至多|'
+    r'不含|不等于|不是|没有|未|排除|除了|或者|或|以及|同时|'
+    r'今年|去年|本月|上月|本周|今天|昨天|最近|截至|期间|\d{4}[年/-]|'
+    r'修改|删除|新增|更新|写入|设置|忽略|指令|提示词|[<>]'
+)
+
+def build_simple_plan(state):
+    """Return (plan, reason); invalid/incomplete candidates always fall back."""
+    context = state.get('qa_context') or {}
+    schema = context.get('schema') or {}
+    if schema.get('meta', {}).get('schema_version') != 2:
+        return None, 'legacy_schema'
+    if any(state.get(k) for k in ('plan_feedback', 'run_feedback', 'slots_feedback', 'feedback')):
+        return None, 'feedback_requires_replanning'
+    candidate = state.get('simple_query')
+    if not isinstance(candidate, dict) or candidate.get('complete') is not True:
+        return None, 'no_complete_candidate'
+    if (set(candidate) - {'kind', 'dataset', 'filters', 'fields', 'limit', 'complete'}
+            or not {'kind', 'dataset', 'complete'} <= set(candidate)):
+        return None, 'invalid_candidate_shape'
+    candidate = {'filters': [], 'fields': [], 'limit': 50, **candidate}
+    question = state.get('question', '')
+    if not isinstance(question, str) or COMPLEX_QUESTION.search(question):
+        return None, 'complex_question'
+    slots = state.get('slots') or {}
+    if slots.get('time_range') or slots.get('aggregation') in ('对比', '排序'):
+        return None, 'complex_slots'
+    dataset = candidate['dataset']
+    if not isinstance(dataset, str) or context.get('selected_datasets') != [dataset]:
+        return None, 'dataset_selection_mismatch'
+    if context.get('selection_source') != 'understanding':
+        return None, 'missing_semantic_selection'
+    needs = state.get('simple_query_needs')
+    if not isinstance(needs, dict) or needs.get('datasets') != [dataset] or needs.get('relations') != []:
+        return None, 'relation_or_ambiguous_selection'
+    matches = [n for n in schema.get('nodes', []) if n.get('id') == dataset]
+    if len(matches) != 1 or matches[0].get('active') is False:
+        return None, 'unknown_or_inactive_dataset'
+    fields, filters = candidate['fields'], candidate['filters']
+    if not isinstance(fields, list) or not all(isinstance(f, str) for f in fields):
+        return None, 'invalid_fields'
+    if len(fields) != len(set(fields)) or len(fields) > 8:
+        return None, 'invalid_fields'
+    if type(candidate['limit']) is not int or not 1 <= candidate['limit'] <= 200:
+        return None, 'invalid_limit'
+    if candidate['kind'] not in ('count', 'list'):
+        return None, 'unsupported_kind'
+    if (candidate['kind'] == 'count' and fields) or (candidate['kind'] == 'list' and not fields):
+        return None, 'invalid_projection'
+    if not isinstance(filters, list) or len(filters) > 6:
+        return None, 'invalid_filters'
+    compiled_filters = []
+    for condition in filters:
+        if not isinstance(condition, dict) or set(condition) != {'field', 'op', 'value'}:
+            return None, 'invalid_filters'
+        value = condition['value']
+        if condition['op'] not in ('eq', 'contains') or not isinstance(value, str) or not value.strip():
+            return None, 'unsupported_filter'
+        if value not in question:
+            return None, 'filter_value_not_in_question'
+        compiled_filters.append({'alias': 'n', **condition})
+    # Cross-check legacy slots to catch dropped named entities and explicit filters.
+    values = [f['value'] for f in filters]
+    anchors = slots.get('anchors') or []
+    slot_filters = slots.get('filters') or {}
+    if not isinstance(anchors, list) or not isinstance(slot_filters, dict):
+        return None, 'invalid_slots'
+    required = [a.get('raw') for a in anchors if isinstance(a, dict)]
+    required.extend(v for v in slot_filters.values() if v not in ('', None))
+    if any(not isinstance(v, str) or v not in values for v in required):
+        return None, 'uncovered_slot'
+    spec = {'nodes': [{'alias': 'n', 'type': dataset}], 'relations': [],
+            'filters': compiled_filters,
+            'select': [{'alias': 'n', 'field': f, 'as': f} for f in fields],
+            'aggregates': [{'alias': 'n', 'op': 'count', 'as': '记录数'}] if candidate['kind'] == 'count' else [],
+            'order_by': [], 'limit': candidate['limit']}
+    try:
+        compile_query(spec, schema, context['build_id'])
+    except (ValueError, TypeError, KeyError):
+        return None, 'compiler_rejected'
+    spec = {key: value for key, value in spec.items() if value != []}
+    if spec['limit'] == 50:
+        del spec['limit']
+    return {'steps': [{'step_id': 's1', 'tool': '图谱查询', 'params': spec}]}, 'single_dataset_' + candidate['kind']

src/step3_qa_agent/agent/state.py

@@ -7,6 +7,10 @@ from typing import Any, TypedDict
 
 class AgentState(TypedDict, total=False):
     question: str
+    answer_id: str               # 当前答案身份，绑定人工缓存确认
+    cache_request: dict[str, Any]
+    cache_hit: dict[str, Any]
+    cache_modified: bool         # 本轮计划/候选是否经用户修改
     user_id: str                 # 提问人身份（API 时由调用方传入）
     category: str                # 闲聊 / 图谱检索
     chat_answer: str
@@ -17,6 +21,8 @@ class AgentState(TypedDict, total=False):
     enable_reuse: bool            # 是否允许多轮答案复用判断（默认 True）
     qa_context: dict[str, Any]   # 本轮元图谱快照、语义选择的数据评估和可用性记录
     slots: dict[str, Any]        # LLM 原始槽位
+    simple_query: dict[str, Any] | None  # 本轮单数据集查询候选；每问重置
+    simple_query_needs: dict[str, Any] | None  # 原始语义选择，拒绝未知/被过滤的关系
     slots_feedback: str          # 用户对槽位的修改意见（回到槽位填充）
     confirm: bool                # 槽位是否已确认
     grounded: dict[str, Any]     # 接地后槽位（锚点→主键、意图→边、时间）

src/step3_qa_agent/agent/timing.py

@@ -0,0 +1,14 @@
+"""Per-node stream measurements isolated across concurrent requests."""
+from contextlib import contextmanager
+from contextvars import ContextVar
+
+llm_calls = ContextVar('qa_llm_calls', default=None)
+
+@contextmanager
+def measure_llm_calls():
+    calls = []
+    token = llm_calls.set(calls)
+    try:
+        yield calls
+    finally:
+        llm_calls.reset(token)

src/step4_web/answer_cache.py

@@ -0,0 +1,183 @@
+"""User-approved answer cache, separate from LangGraph checkpoints.
+
+Exact matching only. Conversation/data/schema/policy changes produce different
+keys. Raw question/answer content is never accepted from the approval request.
+"""
+from __future__ import annotations
+import asyncio
+from copy import deepcopy
+from collections.abc import Mapping
+from datetime import datetime, timezone
+import hashlib
+import json
+import re
+
+from redis.exceptions import RedisError
+from step2_graph_building.config import get_qa_answer_cache_policy
+from step3_qa_agent.agent.production import validate_plan
+
+PREFIX = 'ka:qa:approved:v1:'
+ACTIVE_VERSION = 'ka:qa:approved:active_data_version'
+APPROVE_CURRENT = "if redis.call('GET',KEYS[2]) ~= ARGV[2] then return 0 end redis.call('SET',KEYS[1],ARGV[1]); return 1"
+MAX_BYTES = 256_000
+CONTEXTUAL = re.compile(r'他们|她们|它们|这些|那些|上述|上面|之前|刚才|该项目|该人员|继续|上次|这个|那个|这批|那批')
+RELATIVE = re.compile(r'今天|昨天|明天|今年|去年|本月|上月|下月|本周|上周|最近|截至|现在几岁')
+DELETE_IF_SAME = "if redis.call('GET',KEYS[1]) == ARGV[1] then return redis.call('DEL',KEYS[1]) else return 0 end"
+
+def digest(value):
+    return hashlib.sha256(json.dumps(value, ensure_ascii=False, sort_keys=True, default=str,
+                                     separators=(',', ':')).encode()).hexdigest()
+
+def prepare_request(thread_id, question, previous, release):
+    question = question.strip()  # Do not drop punctuation, negation, digits or entity whitespace.
+    rounds = previous.get('rounds') or []
+    history = [{'question': r.get('question'), 'answer': r.get('answer'),
+                'anchors': r.get('anchors'), 'version': r.get('schema_build_id')} for r in rounds[-3:]]
+    context = digest({'rounds': history, 'messages': (previous.get('messages') or [])[-6:]})
+    # Repeating the immediately preceding identical question retains its original
+    # context, so the answer we just confirmed is reusable without context drift.
+    if (rounds and str(rounds[-1].get('question', '')).strip() == question
+            and rounds[-1].get('schema_build_id') == release['data_version']
+            and rounds[-1].get('cache_context')):
+        context = rounds[-1]['cache_context']
+    policy, model = get_qa_answer_cache_policy()
+    return {'question': question, 'scope': digest(thread_id), 'context': context,
+            'data_version': release['data_version'], 'schema_version': release['schema_version'],
+            'policy': policy, 'model': model}
+
+def cache_key(request):
+    return PREFIX + digest(request)
+
+def eligibility(state):
+    request = state.get('cache_request') or {}
+    if not request or request.get('question') != str(state.get('question', '')).strip():
+        return '缺少本轮问题记录'
+    if CONTEXTUAL.search(request['question']) or RELATIVE.search(request['question']):
+        return '依赖上下文或相对时间的问题暂不缓存'
+    context = state.get('qa_context') or {}
+    if state.get('category') != '图谱检索' or context.get('schema', {}).get('meta', {}).get('schema_version') != 2:
+        return '仅缓存完成的生产图谱回答'
+    if context.get('build_id') != request['data_version']:
+        return '数据版本已变化，请重新提问'
+    if not state.get('answer') or not state.get('fit') or not state.get('answer_id'):
+        return '回答尚未完整完成'
+    if state.get('cache_modified') or any(state.get(k) for k in ('plan_feedback', 'run_feedback', 'slots_feedback')):
+        return '修改过查询条件的回答暂不缓存'
+    subgraph = state.get('subgraph') or {}
+    if subgraph.get('error') or subgraph.get('unsupported_reason') or subgraph.get('clarification_declined'):
+        return '错误或未确认查询结果不缓存'
+    trace = state.get('trace') or []
+    last_understand = max((i for i, item in enumerate(trace) if item.get('node') == 'understand'), default=0)
+    if any(item.get('node') == 'value_choice' for item in trace[last_understand:]):
+        return '替换过候选条件的回答暂不缓存'
+    suggestions = subgraph.get('suggestions') or {}
+    if suggestions.get('options') or suggestions.get('error') or suggestions.get('limited'):
+        return '仍有澄清或不完整候选，暂不缓存'
+    plan = state.get('plan') or {}
+    try:
+        validate_plan(plan, context)
+    except (ValueError, TypeError, KeyError):
+        return '查询计划无效'
+    steps = plan.get('steps') or []
+    results = subgraph.get('results') or {}
+    if not steps or {s['step_id'] for s in steps} != set(results):
+        return '查询结果不完整'
+    for result in results.values():
+        if (result.get('truncated') or result.get('error') or result.get('rows') is None
+                or result.get('data_version') != request['data_version']):
+            return '结果截断、失败或版本不一致'
+    return ''
+
+def payload_from_state(state):
+    # Keep source provenance and the exact approved plan. Exclude old messages and
+    # trace to bound entry size and prevent restoring another turn's context.
+    def plain(value):
+        if isinstance(value, Mapping):
+            return {key: plain(item) for key, item in value.items()}
+        if isinstance(value, (list, tuple)):
+            return [plain(item) for item in value]
+        return value
+    payload = {key: plain(state.get(key)) for key in
+               ('question', 'answer', 'category', 'slots', 'qa_context', 'plan', 'subgraph', 'fit', 'answer_id')}
+    for result in payload['subgraph']['results'].values():
+        result['rows'] = [dict(row) for row in result['rows']]
+    return payload
+
+def review_info(state, checkpoint_id, enabled=True):
+    reason = eligibility(state) if enabled else '问答缓存未启用'
+    if not reason and len(json.dumps(payload_from_state(state), ensure_ascii=False, default=str).encode()) > MAX_BYTES - 4096:
+        reason = '回答及来源较大，暂不缓存'
+    hit = state.get('cache_hit') or {}
+    return {'eligible': not reason, 'reason': reason, 'answer_id': state.get('answer_id'),
+            'checkpoint_id': checkpoint_id, 'cached': bool(hit), 'hit': bool(hit),
+            'entry_id': hit.get('entry_id'), 'approved_at': hit.get('approved_at')}
+
+class ApprovedAnswerCache:
+    def __init__(self, redis, timeout_seconds=0.25):
+        self.redis = redis
+        self.timeout_seconds = timeout_seconds
+
+    async def lookup(self, request):
+        if CONTEXTUAL.search(request['question']) or RELATIVE.search(request['question']):
+            return None
+        try:
+            raw = await asyncio.wait_for(self.redis.get(cache_key(request)), self.timeout_seconds)
+            if not raw or len(raw) > MAX_BYTES:
+                return None
+            entry = json.loads(raw)
+            if (not isinstance(entry, dict) or entry.get('approved') is not True
+                    or entry.get('request') != request
+                    or not isinstance(entry.get('payload'), dict)):
+                return None
+            state = {**entry['payload'], 'cache_request': request}
+            if eligibility(state):
+                return None
+            return entry
+        except (TimeoutError, RedisError, ValueError, TypeError, KeyError, AttributeError):
+            return None  # A cache outage must not prevent the normal QA flow.
+
+    async def approve(self, state):
+        reason = eligibility(state)
+        if reason:
+            raise ValueError(reason)
+        request = state['cache_request']
+        entry = {'approved': True, 'entry_id': state['answer_id'], 'request': request,
+                 'approved_at': datetime.now(timezone.utc).isoformat(),
+                 'payload': payload_from_state(state)}
+        raw = json.dumps(entry, ensure_ascii=False, default=str, separators=(',', ':'))
+        if len(raw.encode()) > MAX_BYTES:
+            raise ValueError('回答及来源较大，暂不缓存')
+        saved = await asyncio.wait_for(self.redis.eval(APPROVE_CURRENT, 2, cache_key(request),
+            ACTIVE_VERSION, raw, request['data_version']), self.timeout_seconds)
+        if not saved:
+            raise ValueError('缓存数据版本已变化或尚未就绪，请重新查询后重试')
+        return entry
+
+    async def revoke(self, request, entry_id):
+        key = cache_key(request)
+        raw = await asyncio.wait_for(self.redis.get(key), self.timeout_seconds)
+        if not raw:
+            return
+        entry = json.loads(raw)
+        if entry.get('entry_id') != entry_id:
+            raise ValueError('缓存已被另一条确认记录替换，请重新查询')
+        removed = await asyncio.wait_for(self.redis.eval(DELETE_IF_SAME, 1, key, raw), self.timeout_seconds)
+        if not removed:
+            raise ValueError('缓存已变化，请重新查询后操作')
+
+def restored_state(entry, previous, request, answer_id):
+    state = deepcopy(entry['payload'])
+    question, answer = request['question'], state['answer']
+    state.update(question=question, answer_id=answer_id, cache_request=request,
+                 cache_hit={'entry_id': entry['entry_id'], 'approved_at': entry['approved_at']},
+                 chat_answer='', simple_query=None, simple_query_needs=None, cache_modified=False,
+                 plan_feedback='', run_feedback='', slots_feedback='', feedback='',
+                 grounded={}, capability={}, reuse={}, iterations=0,
+                 rounds=list(previous.get('rounds') or []) + [
+                     {'question': question, 'answer': answer, 'schema_build_id': request['data_version'],
+                      'anchors': [], 'cache_context': request['context'], 'answer_id': answer_id}],
+                 messages=list(previous.get('messages') or []) + [
+                     {'role': 'user', 'content': question}, {'role': 'assistant', 'content': answer}],
+                 trace=list(previous.get('trace') or []) + [{'node': 'answer_cache', 'hit': True,
+                                                             'llm_calls': []}])
+    return state

src/step4_web/answer_cache_retention.py

@@ -0,0 +1,62 @@
+"""Clean only approved answers, serialized with publication by update_lock."""
+from __future__ import annotations
+import asyncio
+import json
+from redis import Redis
+from step2_graph_building.config import get_settings
+from step2_graph_building.runtime import release_snapshot
+from .answer_cache import PREFIX, ACTIVE_VERSION, DELETE_IF_SAME
+
+CLEANED_VERSION = 'ka:qa:approved:cleaned_data_version:v2'
+PERSIST_IF_SAME = "if redis.call('GET',KEYS[1]) == ARGV[1] then redis.call('SET',KEYS[1],ARGV[2]); return 1 else return 0 end"
+
+def clean_entries(redis, version):
+    if not isinstance(version, str) or not version:
+        raise ValueError('缺少当前数据版本')
+    # Fence old in-flight approvals before deleting anything.
+    redis.set(ACTIVE_VERSION, version)
+    completed = redis.get(CLEANED_VERSION)
+    if completed in (version, version.encode()):
+        return {'removed': 0, 'preserved': 0, 'already_clean': True}
+    report = {'removed': 0, 'preserved': 0}
+    for key in redis.scan_iter(match=PREFIX + '*', count=200):
+        raw = redis.get(key)
+        if raw is None:
+            continue
+        try:
+            entry = json.loads(raw)
+            current = entry.get('request', {}).get('data_version') == version
+        except (ValueError, TypeError, AttributeError):
+            current = False  # Invalid entries in our dedicated index are unusable.
+        if not current:
+            report['removed'] += redis.eval(DELETE_IF_SAME, 1, key, raw)
+        else:
+            entry.pop('expires_at', None)
+            migrated = json.dumps(entry, ensure_ascii=False, default=str, separators=(',', ':'))
+            report['preserved'] += redis.eval(PERSIST_IF_SAME, 1, key, raw, migrated)
+    redis.set(CLEANED_VERSION, version)
+    return report
+
+def cleanup_answer_cache_for_release(version):
+    """Caller holds update_lock; a Redis outage must not undo publication."""
+    try:
+        with Redis.from_url(get_settings().redis_url, socket_timeout=2,
+                            socket_connect_timeout=2) as redis:
+            return clean_entries(redis, version)
+    except Exception as exc:
+        return {'warning': '旧问答缓存清理未完成，将自动重试：' + type(exc).__name__}
+
+def retry_cleanup():
+    from .data_update import update_lock
+    try:
+        with update_lock():
+            return cleanup_answer_cache_for_release(release_snapshot()['data_version'])
+    except (RuntimeError, OSError, ValueError, KeyError) as exc:
+        return {'warning': '问答缓存清理暂缓：' + type(exc).__name__}
+
+async def cleanup_loop():
+    while True:
+        report = await asyncio.to_thread(retry_cleanup)
+        if report.get('warning'):
+            print(report['warning'], flush=True)
+        await asyncio.sleep(60)

src/step4_web/api.py

@@ -13,33 +13,37 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import re
 import time
 import traceback
-from contextlib import asynccontextmanager
+from contextlib import asynccontextmanager, suppress
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import AsyncIterator
 from uuid import uuid4
 from zoneinfo import ZoneInfo
 
-from fastapi import FastAPI
+from fastapi import FastAPI, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import FileResponse, StreamingResponse
 from fastapi.staticfiles import StaticFiles
 from step3_qa_agent.agent.redis_checkpointer import PlainRedisSaver
 from langgraph.types import Command
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 from step3_qa_agent.agent.graph import build_agent_graph
-from step2_graph_building.config import get_service_address, get_settings
+from step2_graph_building.config import get_service_address, get_settings, get_qa_answer_cache_config
+from step2_graph_building.runtime import release_snapshot
+from step4_web.answer_cache import ApprovedAnswerCache, prepare_request, restored_state, review_info, digest
+from step4_web.answer_cache_retention import retry_cleanup, cleanup_loop
 from step4_web.data_quality import router as data_quality_router
 
 
 @asynccontextmanager
 async def lifespan(_: FastAPI):
-    global _graph
+    global _graph, _answer_cache
 
     from step2_graph_building.runtime import release_snapshot
     from step2_graph_building.db import get_driver
@@ -54,10 +58,22 @@ async def lifespan(_: FastAPI):
         settings.redis_url, ttl_seconds=settings.redis_context_ttl_seconds
     ) as checkpointer:
         _graph = build_agent_graph(checkpointer=checkpointer)
+        cache_enabled = get_qa_answer_cache_config()
+        _answer_cache = ApprovedAnswerCache(checkpointer.redis) if cache_enabled else None
+        cleanup_report = await asyncio.to_thread(retry_cleanup)
+        if cleanup_report.get('warning'):
+            print(cleanup_report['warning'], flush=True)
+        cleanup_task = asyncio.create_task(cleanup_loop())
         print("Redis 异步 checkpointer 已就绪", flush=True)
 
-        yield
+        try:
+            yield
+        finally:
+            cleanup_task.cancel()
+            with suppress(asyncio.CancelledError):
+                await cleanup_task
     _graph = None
+    _answer_cache = None
 
 
 app = FastAPI(title="申勤物业知识助手", version="0.1.0", lifespan=lifespan)
@@ -71,6 +87,7 @@ app.add_middleware(
 
 # 全局共享图：状态按 thread_id 隔离，由 PlainRedisSaver 持久化到原生 Redis
 _graph = None
+_answer_cache = None
 API_LOG_DIR = Path(__file__).resolve().parents[2] / "api_log"
 HTML_DIR = Path(__file__).resolve().parents[2] / "html"
 OUTPUT_DIR = Path(__file__).resolve().parents[2] / "output"
@@ -101,6 +118,7 @@ async def health():
 
 # 节点 → 中文处理阶段说明（SSE progress 事件携带，前端可直接展示）
 NODE_LABELS: dict[str, str] = {
+    "answer_cache": "正在查询已确认的历史答案",
     "production_plan": "正在根据元图谱和数据评估规划查询",
     "production_confirm": "等待确认查询计划",
     "production_run": "正在查询最新业务图谱",
@@ -199,6 +217,7 @@ def _answer_payload(thread_id: str, st: dict, elapsed: float) -> dict:
         "subgraph": st.get("subgraph", {}),
         "rounds": st.get("rounds", []),
         "elapsed_sec": round(elapsed, 2),
+        "cache_review": st.get("cache_review"),
     }
 
 
@@ -215,7 +234,33 @@ async def _astream_run(thread_id: str, query: str,
     """
     t0 = time.monotonic()
     config = _config(thread_id)
-    inp: object = {"question": query, "user_id": "api", "enable_reuse": enable_reuse}
+    graph = _get_graph()
+    inp: object = {"question": query, "user_id": "api", "enable_reuse": enable_reuse,
+                   "answer_id": uuid4().hex, "cache_request": {}, "cache_hit": {}, "cache_modified": False}
+    if _answer_cache is not None:
+        previous = await graph.aget_state(config)
+        try:
+            release = release_snapshot()
+            request = prepare_request(thread_id, query, previous.values or {}, release)
+            inp['cache_request'] = request
+            # A paused workflow must be resumed normally; never overwrite its checkpoint.
+            if enable_reuse and not previous.next:
+                yield {"type": "node", "node": "answer_cache", "payload": {}}
+                entry = await _answer_cache.lookup(request)
+                current = release_snapshot()
+                if (entry and current['data_version'] == request['data_version']
+                        and current['schema_version'] == request['schema_version']):
+                    state = restored_state(entry, previous.values or {}, request, inp['answer_id'])
+                    await graph.aupdate_state(config, state, as_node="production_answer")
+                    snapshot = await graph.aget_state(config)
+                    state['cache_review'] = review_info(state,
+                        snapshot.config['configurable']['checkpoint_id'])
+                    _write_api_log(endpoint, thread_id, {"query": query, "cache_hit": True},
+                                   state, round(time.monotonic() - t0, 3))
+                    yield {"type": "done", "state": state}
+                    return
+        except (OSError, ValueError, KeyError):
+            inp['cache_request'] = {}  # Local version metadata unavailable: normal path.
     while True:
         interrupted: object | None = None
         async for item in _get_graph().astream(inp, config, stream_mode=["updates", "custom"]):
@@ -247,7 +292,9 @@ async def _astream_run(thread_id: str, query: str,
             return
         inp = Command(resume="确认")
     st = await _get_graph().aget_state(config)
-    state = st.values or {}
+    state = dict(st.values or {})
+    checkpoint_id = (getattr(st, 'config', None) or {}).get('configurable', {}).get('checkpoint_id')
+    state['cache_review'] = review_info(state, checkpoint_id, _answer_cache is not None)
     _write_api_log(endpoint, thread_id, {
         "query": query,
         "auto_confirm": auto_confirm,
@@ -303,7 +350,9 @@ async def _astream_resume(thread_id: str, reply: str,
                "options": intr.get("options", [])}
         return
     st = await _get_graph().aget_state(config)
-    state = st.values or {}
+    state = dict(st.values or {})
+    checkpoint_id = (getattr(st, 'config', None) or {}).get('configurable', {}).get('checkpoint_id')
+    state['cache_review'] = review_info(state, checkpoint_id, _answer_cache is not None)
     _write_api_log(endpoint, thread_id, {"reply": reply}, state,
                    round(time.monotonic() - t0, 3))
     yield {"type": "done", "state": state}
@@ -416,6 +465,61 @@ async def resume(thread_id: str, body: ResumeRequest):
     return _answer_payload(thread_id, last_state, time.monotonic() - t0)
 
 
+class AnswerReviewRequest(BaseModel):
+    model_config = {"extra": "forbid"}
+    answer_id: str = Field(min_length=1, max_length=128)
+    checkpoint_id: str = Field(min_length=1, max_length=128)
+    entry_id: str | None = Field(default=None, max_length=128)
+
+
+async def _review_snapshot(thread_id, body):
+    if _answer_cache is None:
+        raise HTTPException(503, "问答缓存未启用")
+    config = _config(thread_id)
+    config['configurable']['checkpoint_id'] = body.checkpoint_id
+    snapshot = await _get_graph().aget_state(config)
+    state = dict(snapshot.values or {})
+    if (snapshot.next or state.get('answer_id') != body.answer_id
+            or (state.get('cache_request') or {}).get('scope') != digest(thread_id)):
+        raise HTTPException(409, "该答案记录不可用，请重新提问")
+    return state
+
+
+@app.post("/threads/{thread_id}/answer-cache")
+async def approve_answer(thread_id: str, body: AnswerReviewRequest):
+    state = await _review_snapshot(thread_id, body)
+    request = state.get('cache_request') or {}
+    current = release_snapshot()
+    current_key = prepare_request(thread_id, state['question'], {}, current)
+    if any(request.get(k) != current_key[k] for k in ('data_version', 'schema_version', 'policy', 'model')):
+        raise HTTPException(409, "数据已更新，请重新提问后确认")
+    try:
+        entry = await _answer_cache.approve(state)
+    except ValueError as exc:
+        raise HTTPException(409, str(exc)) from exc
+    except Exception as exc:
+        raise HTTPException(503, "缓存暂时不可用，未能确认保存，请重试") from exc
+    current_key = prepare_request(thread_id, state['question'], {}, release_snapshot())
+    if any(request.get(k) != current_key[k] for k in ('data_version', 'schema_version', 'policy', 'model')):
+        raise HTTPException(409, "数据或配置已更新，本次缓存不会用于新版本")
+    return {"status": "ok", "entry_id": entry['entry_id'], "approved_at": entry['approved_at'],
+            "message": "已确认并加入缓存"}
+
+
+@app.delete("/threads/{thread_id}/answer-cache")
+async def revoke_answer(thread_id: str, body: AnswerReviewRequest):
+    state = await _review_snapshot(thread_id, body)
+    if not body.entry_id:
+        raise HTTPException(422, "缺少缓存记录标识")
+    try:
+        await _answer_cache.revoke(state['cache_request'], body.entry_id)
+    except ValueError as exc:
+        raise HTTPException(409, str(exc)) from exc
+    except Exception as exc:
+        raise HTTPException(503, "暂时无法撤销缓存，请重试") from exc
+    return {"status": "ok", "message": "已撤销缓存"}
+
+
 @app.get("/threads/{thread_id}/history")
 async def history(thread_id: str):
     st = await _get_graph().aget_state(_config(thread_id))

src/step4_web/data_update.py

@@ -18,6 +18,7 @@ from step2_graph_building.config import get_metadata_template_dir, get_relation_
 from step2_graph_building.db import get_driver
 from step2_graph_building.runtime import ROOT, MANIFEST, release_snapshot
 from .release_retention import prune_releases
+from .answer_cache_retention import cleanup_answer_cache_for_release
 
 
 def template_fingerprint():
@@ -167,6 +168,9 @@ def update_data(mode='auto', *, fetcher=None, assessor=None, analyzer=None, driv
                 driver.execute_query('MATCH (n:_Step2Record) WHERE n._kg_build <> $build DETACH DELETE n', build=build_id)
             except Exception:
                 report['warning'] = '发布成功，旧图清理失败；查询仍只读取当前版本'
+            report['answer_cache_cleanup'] = cleanup_answer_cache_for_release(build_id)
+            if report['answer_cache_cleanup'].get('warning'):
+                print(report['answer_cache_cleanup']['warning'], flush=True)
             print('数据版本发布完成', flush=True)
             return report
         except Exception as exc: