diff --git a/backend/app/engine/nodes/report.py b/backend/app/engine/nodes/report.py index efffa7d7..cb9b5b4f 100644 --- a/backend/app/engine/nodes/report.py +++ b/backend/app/engine/nodes/report.py @@ -37,7 +37,7 @@ import asyncio import re import time from collections import Counter -from collections.abc import Iterable +from collections.abc import Iterable, Mapping from typing import Any from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage @@ -234,13 +234,18 @@ def failed_queries(state: GraphState, spec: GraphSpec, node: GraphNode) -> list[ 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 + calls = [c for c in result.get("tool_calls") or [] + if isinstance(c, dict) and not (c.get("repeat") or c.get("skipped") or c.get("denied")) + and is_platform_query(str(c.get("tool") or ""))] + for i, call in enumerate(calls): 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("查询失败")): + if not _query_failed(call): continue args = call.get("args") if isinstance(call.get("args"), dict) else {} + if _superseded(name, str(args.get("sql") or ""), calls[i + 1:]): + # 写错了一条、下一条马上改对了：这次失败已被替代，不算「没有查到」（不然写作模型会写「还有一次取数失败」， + # 复核据此判有缺口） + continue 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], @@ -248,6 +253,37 @@ def failed_queries(state: GraphState, spec: GraphSpec, node: GraphNode) -> list[ return found +def _query_failed(call: Mapping[str, Any]) -> bool: + said = str(call.get("result") or "").strip() + return call.get("ok") is False or said.startswith("查询失败") + + +def _tables(sql: str) -> set[str] | None: + """这条 SQL 读的表（sql_semantics.facets 的 tables，按比较用的写法）；解析不了的为 None。""" + from app.data.sql_semantics import facets + + f = facets(sql) + if f.parse_error or not f.tables: + return None + return {str(p.key) for p in f.tables} + + +def _superseded(tool: str, sql: str, later: list[dict[str, Any]]) -> bool: + """同一个节点里这次失败之后，还有一条成功的平台查询替代了它：同一个数据源（同一个查询工具）、读的表覆盖了失败那条读的 + 表；失败那条解析不了（取不出表）的，同一个数据源之后有成功的查询就算。""" + wanted = _tables(sql) + for call in later: + if str(call.get("tool") or "") != tool or _query_failed(call): + continue + if wanted is None: + return True + args = call.get("args") if isinstance(call.get("args"), dict) else {} + got = _tables(str(args.get("sql") or "")) + if got is not None and wanted <= got: + return True + return False + + 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']}）" diff --git a/backend/tests/test_report_failed_queries.py b/backend/tests/test_report_failed_queries.py index 8dc03c5f..eb59b080 100644 --- a/backend/tests/test_report_failed_queries.py +++ b/backend/tests/test_report_failed_queries.py @@ -52,3 +52,59 @@ async def test_without_failed_queries_the_prompt_is_unchanged(monkeypatch): 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"] + + +# -------------------------------------------------------------------------- +# 失败之后又改对了的那一次不算「没有查到」 +# -------------------------------------------------------------------------- + + +def _failed(calls: list[dict]) -> list[dict]: + from app.engine.nodes.report import failed_queries + from app.engine.schema import GraphSpec + + g = graph("[]") + spec = GraphSpec.model_validate(g) + state = {"nodes": {"agent1": {"text": "x", "tool_calls": calls}}} + return failed_queries(state, spec, spec.node_map()["write"]) + + +BAD = {"tool": "db_query__shop", "args": {"sql": "SELECT line AS 产线, SUM(qty) AS 产量 FROM output GROUP BY SUM(qty)"}, + "ok": True, "result": "查询失败：aggregate functions are not allowed in GROUP BY"} + + +def _good(sql: str, tool: str = "db_query__shop") -> dict: + return {"tool": tool, "args": {"sql": sql}, "ok": True, "result": '{"columns": ["产线", "产量"], "rows": [["甲线", 10]]}'} + + +def test_a_failure_fixed_by_the_next_query_is_superseded(): + """取数 Agent 写错一条（GROUP BY 里放了聚合），下一条马上改对、读的表覆盖了失败那条：这次失败已被替代，不进写作提示。""" + fixed = _good("SELECT line AS 产线, SUM(qty) AS 产量 FROM output GROUP BY line") + assert _failed([BAD, fixed]) == [] + # 解析不了的失败 SQL：同一个数据源之后有成功的查询，也算被替代 + garbled = {**BAD, "args": {"sql": "SELEC line FROM"}} + assert _failed([garbled, fixed]) == [] + + +def test_a_failure_not_covered_by_a_later_query_stays(): + """之后成功的查询读的是别的表（没覆盖失败那条读的表）、或者成功在失败之前：照旧列进「没有查到的数据」。""" + other = _good("SELECT COUNT(*) AS n FROM stations") + [left] = _failed([BAD, other]) + assert "aggregate functions" in left["reason"] + before = _good("SELECT line AS 产线, SUM(qty) AS 产量 FROM output GROUP BY line") + assert len(_failed([before, BAD])) == 1, "成功在失败之前，不算替代了它" + elsewhere = _good("SELECT line, SUM(qty) FROM output GROUP BY line", tool="db_query__crm") + assert len(_failed([BAD, elsewhere])) == 1, "别的数据源上的成功不算" + + +async def test_when_every_query_failed_the_writer_still_says_none_succeeded(monkeypatch): + calls = ("[{'tool': 'db_query__shop', 'args': {'sql': 'SELECT line, SUM(qty) FROM output GROUP BY SUM(qty)'}, " + "'ok': True, 'result': '查询失败：aggregate functions are not allowed'}, " + "{'tool': 'db_query__shop', 'args': {'sql': 'SELECT line, SUM(qty) FROM output GROUP BY 1'}, " + "'ok': True, 'result': '查询失败：connection reset'}]") + seen = script(monkeypatch, "没有查到产量：查询失败。") + run = await finish(graph(calls)) + first = _prompt(seen[0]) + assert "上游有 2 次数据查询失败了" in first and "这次一次成功的查询都没有" in first + [warned] = [e for e in await events(run.id, "log") if (e.data or {}).get("code") == "report_queries_failed"] + assert warned.data["failed"] == 2