report.failed_queries 不再把「写错一条、下一条就改对了」当取数失败:同一个节点里,失败之后还有一条同一数据源的成功平台查询、读的表覆盖了失败那条的表(按 sql_semantics.facets 取表;失败那条解析不了就按同一数据源之后有成功的查询认),这次失败算已被替代,不进写作提示、不记 report_queries_failed;一次成功的都没有时照旧写「这次一次成功的查询都没有」

This commit is contained in:
ThearchyHelios
2026-10-19 16:47:07 +08:00
parent 1e87504e5c
commit 5dfe2e7fd1
2 changed files with 97 additions and 5 deletions
+41 -5
View File
@@ -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']})"
@@ -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