运行时间线看得到单条查询的 SQL 检查:调用工具节点、Agent 和协作团队里的查库调用,tool.end 事件单独带上这次查询的检查结果(只留给人看的字段);preview 只有结果的前一截,检查结果以前通常被截掉

This commit is contained in:
ThearchyHelios
2026-10-04 02:17:32 +08:00
parent c133967bf9
commit 2c160f0ac9
6 changed files with 75 additions and 10 deletions
+3
View File
@@ -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}
+2 -6
View File
@@ -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 {}),
+6 -2
View File
@@ -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))
+4 -1
View File
@@ -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:
+26 -1
View File
@@ -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"]
@@ -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": "坏的"}') == []