From 2c160f0ac93e5b3c52f7bce5ff086e77e78d526f Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sun, 4 Oct 2026 02:17:32 +0800 Subject: [PATCH] =?UTF-8?q?=E8=BF=90=E8=A1=8C=E6=97=B6=E9=97=B4=E7=BA=BF?= =?UTF-8?q?=E7=9C=8B=E5=BE=97=E5=88=B0=E5=8D=95=E6=9D=A1=E6=9F=A5=E8=AF=A2?= =?UTF-8?q?=E7=9A=84=20SQL=20=E6=A3=80=E6=9F=A5=EF=BC=9A=E8=B0=83=E7=94=A8?= =?UTF-8?q?=E5=B7=A5=E5=85=B7=E8=8A=82=E7=82=B9=E3=80=81Agent=20=E5=92=8C?= =?UTF-8?q?=E5=8D=8F=E4=BD=9C=E5=9B=A2=E9=98=9F=E9=87=8C=E7=9A=84=E6=9F=A5?= =?UTF-8?q?=E5=BA=93=E8=B0=83=E7=94=A8=EF=BC=8Ctool.end=20=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=E5=8D=95=E7=8B=AC=E5=B8=A6=E4=B8=8A=E8=BF=99=E6=AC=A1?= =?UTF-8?q?=E6=9F=A5=E8=AF=A2=E7=9A=84=E6=A3=80=E6=9F=A5=E7=BB=93=E6=9E=9C?= =?UTF-8?q?=EF=BC=88=E5=8F=AA=E7=95=99=E7=BB=99=E4=BA=BA=E7=9C=8B=E7=9A=84?= =?UTF-8?q?=E5=AD=97=E6=AE=B5=EF=BC=89=EF=BC=9Bpreview=20=E5=8F=AA?= =?UTF-8?q?=E6=9C=89=E7=BB=93=E6=9E=9C=E7=9A=84=E5=89=8D=E4=B8=80=E6=88=AA?= =?UTF-8?q?=EF=BC=8C=E6=A3=80=E6=9F=A5=E7=BB=93=E6=9E=9C=E4=BB=A5=E5=89=8D?= =?UTF-8?q?=E9=80=9A=E5=B8=B8=E8=A2=AB=E6=88=AA=E6=8E=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/engine/nodes/llm.py | 3 ++ backend/app/engine/nodes/metrics.py | 8 ++--- backend/app/engine/nodes/multi.py | 8 +++-- backend/app/engine/nodes/tools.py | 5 ++- backend/app/engine/sql_problems.py | 27 +++++++++++++++- backend/tests/test_report_sqlcheck_cells.py | 34 +++++++++++++++++++++ 6 files changed, 75 insertions(+), 10 deletions(-) diff --git a/backend/app/engine/nodes/llm.py b/backend/app/engine/nodes/llm.py index 42852f23..5d0c51e4 100644 --- a/backend/app/engine/nodes/llm.py +++ b/backend/app/engine/nodes/llm.py @@ -47,6 +47,7 @@ from app.providers.factory import ( bind_tools_safely, get_chat_model, ) +from app.engine.sql_problems import event_checks from app.tools.datasource import QUERY_PREFIX, run_versions from app.tools.registry import ( ToolArgsError, @@ -518,6 +519,8 @@ async def run_agent(state: GraphState, ctx: NodeContext) -> dict[str, Any]: **({"query_artifact": query["artifact"]} if query else {}), # 查询当时数据源的表结构快照：和查询快照一样落在封存范围内，证据接口从这里认它 **({"schema_artifact": query["schema_artifact"]} if query and query.get("schema_artifact") else {}), + # 这次查询的 SQL 检查结果单独带上：preview 截到前 2000 个字符，checks 通常在截掉的那段里 + **({"checks": checks} if ok and name.startswith(QUERY_PREFIX) and (checks := event_checks(content)) else {}), **extra, ) outcome: dict[str, Any] = {"content": content, "ok": ok, "duration_ms": elapsed} diff --git a/backend/app/engine/nodes/metrics.py b/backend/app/engine/nodes/metrics.py index 152f51f6..0d9f7105 100644 --- a/backend/app/engine/nodes/metrics.py +++ b/backend/app/engine/nodes/metrics.py @@ -33,7 +33,7 @@ from app.engine.expressions import ( substitute, ) from app.engine.labels import field_label -from app.engine.sql_problems import QueryProblem, merge_problems, node_of, query_problems, reason_text +from app.engine.sql_problems import CHECK_FIELDS, QueryProblem, merge_problems, node_of, query_problems, reason_text from app.engine.state import GraphState, template_context @@ -410,10 +410,6 @@ def _sql_check_reason(failed: list[tuple[str, list[QueryProblem]]]) -> str: return reason_text(list(dict.fromkeys(text for _, found in failed for p in found for text in p.problems))) -#: 运行日志里每条 SQL 检查结果交给界面的字段（data/sqlcheck.SqlCheck.as_dict 去掉给模型的那句 for_model） -_CHECK_FIELDS = ("code", "level", "message", "table", "column", "relation_id", "sql_excerpt") - - def _emit_sql_check_logs(metrics: list[dict[str, Any]], failed_by: dict[str, list[QueryProblem]], state: GraphState, ctx: NodeContext) -> None: """没通过 SQL 检查的来源查询，一条一行运行日志（metric_sql_check），列出受它影响的指标。 @@ -441,7 +437,7 @@ def _emit_sql_check_logs(metrics: list[dict[str, Any]], failed_by: dict[str, lis ctx.emit(EventType.LOG, level="warn", code="metric_sql_check", metric=ids[0], metrics=ids, metric_names=[names.get(i, i) for i in ids], reason=reason, problems=problem.problems, - checks=[{k: c[k] for k in _CHECK_FIELDS if k in c} for c in problem.checks], + checks=[{k: c[k] for k in CHECK_FIELDS if k in c} for c in problem.checks], query_artifact=artifact, **({"source_node": source} if source else {}), **({"source_field": "args.sql"} if sql_in_args else {}), diff --git a/backend/app/engine/nodes/multi.py b/backend/app/engine/nodes/multi.py index 0a3bea51..31fb757f 100644 --- a/backend/app/engine/nodes/multi.py +++ b/backend/app/engine/nodes/multi.py @@ -27,7 +27,8 @@ from app.engine.toolcalls import ( ) from app.providers import catalog from app.providers.factory import ModelSpec, bind_tools_safely, get_chat_model -from app.tools.datasource import run_versions +from app.engine.sql_problems import event_checks +from app.tools.datasource import QUERY_PREFIX, run_versions from app.tools.registry import ( ToolArgsError, ToolContext, @@ -419,9 +420,12 @@ async def run_supervisor(state: GraphState, ctx: NodeContext) -> dict[str, Any]: except Exception as e: # noqa: BLE001 content = f"工具失败：{describe_exception(e)}" failure = {"error": content, "detail": raw_detail(e)} + # 查库的 SQL 检查结果单独带上：preview 截到前 1500 个字符，checks 通常在截掉的那段里 + checks = event_checks(content) if not failure and tname.startswith(QUERY_PREFIX) else [] ctx.emit(EventType.TOOL_ERROR if failure else EventType.TOOL_END, tool=tname, agent=name, call_id=cid, preview=content[:1500], - duration_ms=int((time.perf_counter() - started) * 1000), **failure) + duration_ms=int((time.perf_counter() - started) * 1000), **failure, + **({"checks": checks} if checks else {})) calls.append({"tool": tname, "args": targs, "result": content[:2000]}) messages.append(ToolMessage(content=content, tool_call_id=cid)) diff --git a/backend/app/engine/nodes/tools.py b/backend/app/engine/nodes/tools.py index 58c95886..deb63718 100644 --- a/backend/app/engine/nodes/tools.py +++ b/backend/app/engine/nodes/tools.py @@ -18,6 +18,7 @@ from app.engine.state import GraphState from app.engine.toolcalls import ToolTimeout, limit_fields, limit_of, run_bounded from app.sandbox.base import SandboxLimits from app.sandbox.manager import sandbox_manager +from app.engine.sql_problems import event_checks from app.tools.datasource import DATA_UNAVAILABLE, QUERY_PREFIX, SCHEMA_PREFIX, run_versions from app.tools.registry import ( ToolArgsError, @@ -184,10 +185,12 @@ async def run_tool(state: GraphState, ctx: NodeContext) -> dict[str, Any]: # 只认数据源工具交回的查询：别的工具（MCP、自定义）拼出同样形状的 JSON，里面的工件 id 不是我们落的 query = query_entry_fields(result) if evidence_on and name.startswith(QUERY_PREFIX) else None schema = query.get("schema_artifact") if query else None + # 这次查询的 SQL 检查结果单独带上：preview 只有前 2000 个字符，checks 排在 rows 后面，通常被截掉 + checks = event_checks(result) if name.startswith(QUERY_PREFIX) else [] ctx.emit(EventType.TOOL_END, tool=name, duration_ms=elapsed, preview=json.dumps(result, ensure_ascii=False, default=str)[:2000], artifact=snapshot_id, **({"query_artifact": query["artifact"]} if query else {}), - **({"schema_artifact": schema} if schema else {})) + **({"schema_artifact": schema} if schema else {}), **({"checks": checks} if checks else {})) updates: dict[str, Any] = {"nodes": {ctx.node.id: result}} if evidence_on: diff --git a/backend/app/engine/sql_problems.py b/backend/app/engine/sql_problems.py index 1b1f9372..e67916eb 100644 --- a/backend/app/engine/sql_problems.py +++ b/backend/app/engine/sql_problems.py @@ -12,6 +12,7 @@ """ from __future__ import annotations +import json from collections.abc import Callable, Iterable from dataclasses import dataclass, field from typing import Any @@ -87,6 +88,30 @@ def error_checks(artifact: Any, *, loader: Callable[[str], Any] | None = None) - return [c for p in query_problems(artifact, loader=loader) for c in p.checks] +#: 运行事件、证据接口交给界面的检查字段（data/sqlcheck.SqlCheck.as_dict 去掉给模型的那句 for_model） +CHECK_FIELDS = ("code", "level", "message", "table", "column", "relation_id", "sql_excerpt") + + +def event_checks(result: Any) -> list[dict[str, Any]]: + """数据源查询工具交回的结果（dict 或 JSON 文本）里的 SQL 检查结果，只留给人看的字段；没有就是空列表。 + + tool.end 的 preview 只是结果的前 2000 个字符，checks 排在 rows 后面，通常被截掉。事件里单独带一份， + 运行时间线才看得到这条查询的检查结果。调用方只对数据源查询工具（db_query__）问这里。 + """ + if isinstance(result, str): + text = result.lstrip() + if not text.startswith("{") or '"checks"' not in text: + return [] + try: + result = json.loads(text) + except ValueError: + return [] + checks = result.get("checks") if isinstance(result, dict) else None + if not isinstance(checks, list): + return [] + return [{k: c[k] for k in CHECK_FIELDS if k in c} for c in checks if isinstance(c, dict) and c.get("code")] + + def node_of(artifact: str, ledger: Iterable[Any] | None) -> str | None: """证据台账里交回这份查询快照的节点。台账里没有（老运行、Agent 没交台账）返回 None。""" for entry in ledger or []: @@ -122,5 +147,5 @@ def merge_problems(found: Iterable[QueryProblem]) -> list[QueryProblem]: return list(out.values()) -__all__ = ["MERGE_SOURCE", "QueryProblem", "error_checks", "merge_problems", "node_of", "problem_clause", "query_problems", +__all__ = ["CHECK_FIELDS", "MERGE_SOURCE", "QueryProblem", "error_checks", "event_checks", "merge_problems", "node_of", "problem_clause", "query_problems", "reason_text"] diff --git a/backend/tests/test_report_sqlcheck_cells.py b/backend/tests/test_report_sqlcheck_cells.py index cd5b2f12..1c9ffdf3 100644 --- a/backend/tests/test_report_sqlcheck_cells.py +++ b/backend/tests/test_report_sqlcheck_cells.py @@ -158,3 +158,37 @@ def test_a_flagged_metric_whose_snapshot_is_gone_still_counts(monkeypatch): "sql_check_sources": [{"artifact": "gone", "codes": ["fanout_sum"]}]}] assert sql_check_gaps(metrics) == [f"指标「订单金额」{REASON}"] assert sql_check_gaps([{"id": "n", "name": "订单数", "value": 2}]) == [] + + +# -------------------------------------------------------------------------- +# 运行时间线：tool.end 单独带上这次查询的 SQL 检查结果 +# -------------------------------------------------------------------------- + + +async def test_tool_end_carries_the_sql_checks_of_the_query(scenic_source, engine_up, monkeypatch): + """preview 只是结果的前 2000 个字符，checks 排在 rows 后面，通常被截掉：界面看不到这条查询的检查结果。""" + Writer(monkeypatch, text="本周订单金额 [[v:Q1.r0.gmv]]。") + run = await run_manager.start(graph=direct(scenic_source, FANOUT_SQL, ""), input_payload={}) + row = await wait(run.id) + [end] = [e.data for e in await events(row.id, "tool.end", "fetch")] + [check] = end["checks"] + assert check["code"] == "fanout_sum" and check["level"] == "error" and check["table"] == "orders" + assert PROBLEM in check["message"] and "for_model" not in check + + clean = await run_manager.start(graph=direct(scenic_source, CLEAN_SQL, ""), input_payload={}) + row = await wait(clean.id) + [end] = [e.data for e in await events(row.id, "tool.end", "fetch")] + assert "checks" not in end + + +def test_agent_and_team_tool_events_read_checks_the_same_way(): + """Agent 节点和协作团队里的查库调用交回的是 JSON 文本：同一个读法，取不到就不加。""" + import json + + from app.engine.sql_problems import event_checks + + text = json.dumps({"rows": [[1]], "checks": [{**FANOUT, "for_model": "改写"}]}, ensure_ascii=False) + assert event_checks(text) == [{k: v for k, v in FANOUT.items()}] + assert event_checks({"rows": [], "checks": [FANOUT]}) == [FANOUT] + assert event_checks("查询失败：no such table") == [] and event_checks(None) == [] + assert event_checks('{"rows": [], "checks": "坏的"}') == []