From 06286deefa69367e0ad21ebe6d80cff8e55e38e8 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 19 Oct 2026 08:23:29 +0800 Subject: [PATCH] =?UTF-8?q?=E4=B8=8A=E6=B8=B8=E6=9F=A5=E5=BA=93=E5=85=A8?= =?UTF-8?q?=E5=A4=B1=E8=B4=A5=E6=97=B6=E6=8A=A5=E5=91=8A=E7=85=A7=E5=AE=9E?= =?UTF-8?q?=E8=AF=B4=E6=B2=A1=E6=9F=A5=E5=88=B0=EF=BC=9A=E6=8A=A5=E5=91=8A?= =?UTF-8?q?=E8=8A=82=E7=82=B9=E5=86=99=E4=BD=9C=E5=89=8D=E4=BB=8E=E4=B8=8A?= =?UTF-8?q?=E6=B8=B8=20Agent=20=E7=9A=84=E8=B0=83=E7=94=A8=E8=AE=B0?= =?UTF-8?q?=E5=BD=95=E9=87=8C=E6=89=BE=E5=87=BA=E5=A4=B1=E8=B4=A5=E7=9A=84?= =?UTF-8?q?=E5=B9=B3=E5=8F=B0=E6=9F=A5=E8=AF=A2=EF=BC=88engine/nodes/repor?= =?UTF-8?q?t.failed=5Fqueries=EF=BC=89=EF=BC=8C=E5=86=99=E4=BD=9C=E6=8F=90?= =?UTF-8?q?=E7=A4=BA=E9=87=8C=E5=A4=9A=E4=B8=80=E8=8A=82=E3=80=8C=E6=B2=A1?= =?UTF-8?q?=E6=9C=89=E6=9F=A5=E5=88=B0=E7=9A=84=E6=95=B0=E6=8D=AE=E3=80=8D?= =?UTF-8?q?=E5=88=97=E5=87=BA=E5=93=AA=E5=87=A0=E6=AC=A1=E3=80=81=E5=8E=9F?= =?UTF-8?q?=E5=9B=A0=E7=AC=AC=E4=B8=80=E8=A1=8C=EF=BC=9B=E4=B8=80=E6=AC=A1?= =?UTF-8?q?=E6=88=90=E5=8A=9F=E7=9A=84=E6=9F=A5=E8=AF=A2=E9=83=BD=E6=B2=A1?= =?UTF-8?q?=E6=9C=89=E6=97=B6=E4=B8=8D=E8=AE=B8=E5=86=99=E5=8D=95=E5=85=83?= =?UTF-8?q?=E6=A0=BC=E5=BC=95=E7=94=A8=E3=80=81=E7=85=A7=E5=AE=9E=E5=86=99?= =?UTF-8?q?=E6=B2=A1=E6=9F=A5=E5=88=B0=E5=92=8C=E6=80=8E=E4=B9=88=E6=8E=92?= =?UTF-8?q?=E6=9F=A5=EF=BC=8C=E9=87=8D=E5=86=99=E8=AF=B7=E6=B1=82=E9=87=8C?= =?UTF-8?q?=E5=90=8C=E4=B8=80=E5=8F=A5=EF=BC=9B=E8=AE=B0=E4=B8=80=E6=9D=A1?= =?UTF-8?q?=20report=5Fqueries=5Ffailed=20=E6=8F=90=E9=86=92=E3=80=82?= =?UTF-8?q?=E4=BB=A5=E5=89=8D=E5=86=99=E4=BD=9C=E8=80=85=E7=85=A7=E7=9D=80?= =?UTF-8?q?=E4=B8=8A=E6=B8=B8=E6=96=87=E5=AD=97=E9=87=8C=E7=9A=84=20SQL=20?= =?UTF-8?q?=E5=92=8C=E5=88=97=E5=90=8D=E5=86=99=E5=87=BA=E5=BC=95=E7=94=A8?= =?UTF-8?q?=E4=B8=8D=E5=AD=98=E5=9C=A8=E7=9A=84=20Q1=E2=80=A6=EF=BC=8C?= =?UTF-8?q?=E9=87=8D=E5=86=99=E4=B8=80=E6=AC=A1=E8=BF=98=E6=98=AF=E9=82=A3?= =?UTF-8?q?=E6=A0=B7=EF=BC=8C=E7=AD=94=E6=A1=88=E9=87=8C=E7=95=99=E7=9D=80?= =?UTF-8?q?=E6=B2=A1=E5=A1=AB=E4=B8=8A=E7=9A=84=20=E2=9F=A6=3Fv:Q1?= =?UTF-8?q?=E2=80=A6=E2=9F=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/engine/nodes/report.py | 80 +++++++++++++++++++-- backend/tests/test_report_failed_queries.py | 54 ++++++++++++++ 2 files changed, 128 insertions(+), 6 deletions(-) create mode 100644 backend/tests/test_report_failed_queries.py diff --git a/backend/app/engine/nodes/report.py b/backend/app/engine/nodes/report.py index ff5ed473..ca4fff39 100644 --- a/backend/app/engine/nodes/report.py +++ b/backend/app/engine/nodes/report.py @@ -208,6 +208,63 @@ def report_fields(state: GraphState, spec: GraphSpec, node: GraphNode) -> list[d return found +#: 写作提示里「没有查到的数据」一节最多列几次、每次的原因留多长：只是让写作者知道哪部分没查到、为什么，不是排错日志 +FAILED_SHOWN = 8 +FAILED_REASON_CHARS = 160 + + +def failed_queries(state: GraphState, spec: GraphSpec, node: GraphNode) -> list[dict[str, Any]]: + """上游 Agent 查库失败的那几次：调用记录（产出的 tool_calls）里平台查询工具交回「查询失败…」的。 + + 失败的查询不进证据台账，证据目录里没有它；写作者只在上游的文字里看到 SQL 和列名，会照着写出引用不存在的 Q1… 的 + 标记，重写一次也常常还是那样，最后把解析不了的标记原样交给看的人（2026-10 部署上撞过：数据源连不上，六次查询全失败， + 答案里全是没填上的 ⟦?v:Q1…⟧）。照实告诉写作者哪几次没查到、为什么，让它照实说没查到。范围和 report_fields 一致： + 祖先节点，写了 evidence_from 就只收那几个。原因只取报错的第一行：工具交回时已经按这次运行的遮罩规则处理过（datasource + 的 mask_error_text），和上游 Agent 自己看到的是同一句。""" + from app.capabilities.registry import is_platform_query + + sources = _setting(spec, node, "evidence_from", "ancestors") + if isinstance(sources, str) and sources != "ancestors": + sources = [sources] + keep = {str(s) for s in sources} if isinstance(sources, list) else None + allowed = _ancestors(spec, node.id) + outputs = state.get("nodes") or {} + found: list[dict[str, Any]] = [] + for other in spec.nodes: + result = outputs.get(other.id) + if other.id not in allowed or (keep is not None and other.id not in keep) or not isinstance(result, dict): + continue + for call in result.get("tool_calls") or []: + if not isinstance(call, dict) or call.get("repeat") or call.get("skipped") or call.get("denied"): + continue + name, said = str(call.get("tool") or ""), str(call.get("result") or "").strip() + if not is_platform_query(name) or not (call.get("ok") is False or said.startswith("查询失败")): + continue + args = call.get("args") if isinstance(call.get("args"), dict) else {} + reason = said.splitlines()[0] if said else "没有交回结果" + found.append({"node_id": other.id, "title": other.title or other.id, "tool": name, + "sql": " ".join(str(args.get("sql") or "").split())[:200], + "reason": reason[:FAILED_REASON_CHARS]}) + return found + + +def _failed_section(failed: list[dict[str, Any]], *, any_query: bool) -> str: + """写作提示里「没有查到的数据」一节，和照实说没查到的要求。一次成功的查询都没有时，不许写单元格引用。""" + lines = [f"- 「{f['title']}」{f['tool']}" + (f"：{f['sql']}" if f["sql"] else "") + f"（{f['reason']}）" + for f in failed[:FAILED_SHOWN]] + if len(failed) > FAILED_SHOWN: + lines.append(f"- 另有 {len(failed) - FAILED_SHOWN} 次，同样没有结果") + rule = (FAILED_NONE_RULE if not any_query else FAILED_SOME_RULE) + return f"## 没有查到的数据\n上游有 {len(failed)} 次数据查询失败了，证据目录里没有它们的结果：\n" + "\n".join(lines) + f"\n\n{rule}" + + +#: 一次成功的查询都没有时交给写作者的要求（写作提示和重写请求里同一句） +FAILED_NONE_RULE = ("这次一次成功的查询都没有：不要写 [[v:…]] 单元格引用，也不要写任何数字；照实写没有查到数据、失败的原因，" + "以及怎么排查（比如检查数据源连接）。") +#: 有成功的查询、也有失败的时 +FAILED_SOME_RULE = "上面这几次没有结果可以引用：不要为它们写引用、也不要估数，照实说这部分没有查到。" + + async def model_window(session: Any, spec: ModelSpec, model_id: str) -> int | None: """写作模型的上下文窗口（token）：模型接入里给这个模型登记的 context 优先，其次模型目录 （providers/catalog.py）；都没有返回 None，由 catalog_budget 取保守值。""" @@ -531,16 +588,18 @@ def _named(violations: list[dict[str, Any]], limit: int = 3) -> str: return "、".join(f"「{n}」" for n in names) -def _repair_request(violations: list[dict[str, Any]]) -> str: +def _repair_request(violations: list[dict[str, Any]], failed_rule: str | None = None) -> str: return ( "你写的报告没有通过系统核对，问题如下：\n" f"{describe_violations(violations)}\n\n" "请重写整篇报告：所有数字都用引用标记写，只引用证据目录里有的指标和输入；" - "不要解释改了什么，只输出新的报告正文。" + + (f"{failed_rule}" if failed_rule else "") + + "不要解释改了什么，只输出新的报告正文。" ) -def _messages(ctx: NodeContext, state: GraphState, writer: WriterCatalog, claims: str = "off") -> list[BaseMessage]: +def _messages(ctx: NodeContext, state: GraphState, writer: WriterCatalog, claims: str = "off", + failed: str | None = None) -> list[BaseMessage]: system = ctx.render_str(ctx.cfg("system", ""), state) instructions = ctx.render_str(ctx.cfg("instructions", ""), state).strip() \ or "根据下面的证据写一份简洁的报告，先总后分。" @@ -554,7 +613,7 @@ def _messages(ctx: NodeContext, state: GraphState, writer: WriterCatalog, claims return [ SystemMessage(content="\n\n".join(p for p in (system, ROLE) if p)), HumanMessage(content=( - f"{instructions}\n\n{writer.text}\n\n{rules}\n\n" + f"{instructions}\n\n{writer.text}\n\n" + (f"{failed}\n\n" if failed else "") + f"{rules}\n\n" "只输出报告正文（Markdown），不要写撰写说明。" )), ] @@ -678,7 +737,16 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]: budget=writer.budget, window=window, queries=writer.queries, omitted=len(writer.omitted), compact=writer.compact, compact_rows=writer.compact_rows, nodes=[{k: r[k] for k in ("node_id", "title", "total", "omitted", "compact")} for r in writer.nodes]) - messages = _messages(ctx, state, writer, claims) + failed = failed_queries(upstream, spec, ctx.node) # type: ignore[arg-type] + any_query = any(entry.get("kind") == "query" for entry in catalog.values()) + failed_rule = (FAILED_SOME_RULE if any_query else FAILED_NONE_RULE) if failed else None + if failed: + ctx.emit(EventType.LOG, level="warn", code="report_queries_failed", + message=f"上游有 {len(failed)} 次数据查询失败" + ("，一次成功的都没有" if not any_query else "") + + "：已告诉写作模型照实说没有查到，不为它们写引用", + failed=len(failed), any_query=any_query) + messages = _messages(ctx, state, writer, claims, + _failed_section(failed, any_query=any_query) if failed else None) started = time.perf_counter() spent: list[dict[str, Any]] = [] nudged = False @@ -726,7 +794,7 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]: ctx.emit(EventType.LOG, level="warn", code="report_repair", message=f"报告有 {len(blocking)} 处未通过核对（{_named(blocking)}），" f"已要求模型重写（第 {repairs} 次）") - messages = [*messages, response, HumanMessage(content=_repair_request(blocking))] + messages = [*messages, response, HumanMessage(content=_repair_request(blocking, failed_rule))] response, raw = await draft(messages) doc = await compose(raw) diff --git a/backend/tests/test_report_failed_queries.py b/backend/tests/test_report_failed_queries.py new file mode 100644 index 00000000..8dc03c5f --- /dev/null +++ b/backend/tests/test_report_failed_queries.py @@ -0,0 +1,54 @@ +"""上游查库全失败时，报告照实说没查到（engine/nodes/report.failed_queries）。 + +2026-10 部署上撞过：数据源连不上，Agent 的六次查询全部「查询失败」，证据目录里一次查询都没有；写作模型照着上游文字里的 +SQL 和列名写出 [[v:Q1.r0.…]]，重写一次还是那样，答案里全是没填上的 ⟦?v:Q1…⟧。现在写作提示里多一节「没有查到的数据」， +一次成功的都没有时不许写单元格引用、照实写没查到和原因；重写请求里同一句。上游节点的调用记录照 Agent 产出的形状写 +（tool_calls：tool、args、ok、result），数据源和 SQL 都是虚构的。 +""" +from __future__ import annotations + +from tests.test_report_node import engine_up, events, finish, node, script # noqa: F401 + +FAILED_CALLS = ("[{'tool': 'db_query__shop', 'args': {'sql': 'SELECT COUNT(*) AS n FROM orders'}, 'ok': True, " + "'result': '查询失败：connect() got an unexpected keyword argument'}, " + "{'tool': 'db_query__shop', 'args': {'sql': 'SELECT COUNT(*) AS n FROM orders'}, 'repeat': True}, " + "{'tool': 'db_schema__shop', 'args': {}, 'ok': True, 'result': '表 orders …'}]") + + +def graph(calls: str = FAILED_CALLS) -> dict: + nodes = [ + node("start", "input", fields=[{"name": "q", "default": "订单有多少"}]), + node("agent1", "transform", mode="expression", + expression="{'text': '三条查询都失败了：连不上数据源', 'tool_calls': " + calls + "}"), + node("write", "report", instructions="回答：{{ input.q }}"), + node("out", "output", fields=[{"name": "答案", "value": "{{ nodes.write.text }}"}]), + ] + chain = ["start", "agent1", "write", "out"] + return {"nodes": nodes, "edges": [{"source": a, "target": b} for a, b in zip(chain, chain[1:])]} + + +def _prompt(messages) -> str: + return "\n".join(str(m.content) for m in messages) + + +async def test_the_writer_is_told_the_queries_failed_and_not_to_cite(monkeypatch): + seen = script(monkeypatch, "订单有 [[v:Q1.r0.n]] 单。", "没有查到订单数：数据源连接失败，请检查数据源连接。") + run = await finish(graph()) + first = _prompt(seen[0]) + assert "## 没有查到的数据" in first and "上游有 1 次数据查询失败了" in first, first + assert "db_query__shop：SELECT COUNT(*) AS n FROM orders（查询失败：connect() got an unexpected keyword argument）" in first + assert "这次一次成功的查询都没有：不要写 [[v:…]] 单元格引用" in first + assert "db_schema__shop" not in first.split("## 没有查到的数据")[1].split("\n\n")[0], "看表结构不算查询失败" + # 第一稿照旧引用了不存在的 Q1：重写请求里同一句要求 + assert len(seen) == 2 and "这次一次成功的查询都没有" in str(seen[1][-1].content) + assert run.output["答案"] == "没有查到订单数：数据源连接失败，请检查数据源连接。" + [warned] = [e for e in await events(run.id, "log") if (e.data or {}).get("code") == "report_queries_failed"] + assert warned.data["failed"] == 1 and warned.data["any_query"] is False + + +async def test_without_failed_queries_the_prompt_is_unchanged(monkeypatch): + seen = script(monkeypatch, "没有数据。") + calls = "[{'tool': 'db_schema__shop', 'args': {}, 'ok': True, 'result': '表 orders …'}]" + run = await finish(graph(calls)) + assert "没有查到的数据" not in _prompt(seen[0]) + assert not [e for e in await events(run.id, "log") if (e.data or {}).get("code") == "report_queries_failed"]