From 5f616616250151d9edc058b2eebbacb1d3ac4ecb Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Wed, 21 Oct 2026 17:53:56 +0800 Subject: [PATCH] =?UTF-8?q?=E5=88=86=E6=9E=90=E5=B7=A5=E5=85=B7=E7=9A=84?= =?UTF-8?q?=20tool.end=20=E5=8F=A6=E5=B8=A6=E4=B8=80=E5=B0=8F=E6=AE=B5?= =?UTF-8?q?=E5=88=A4=E5=AE=9A=EF=BC=88tools/analysis.event=5Fview=EF=BC=9A?= =?UTF-8?q?=E7=BC=96=E5=8F=B7=E3=80=81=E7=AE=97=E5=AD=90=E3=80=81=E6=A8=A1?= =?UTF-8?q?=E5=BC=8F=E3=80=81=E7=8A=B6=E6=80=81=E3=80=81=E5=88=A4=E5=AE=9A?= =?UTF-8?q?=E7=A0=81=E3=80=81=E6=96=B9=E5=90=91=E3=80=81=E5=88=A4=E5=AE=9A?= =?UTF-8?q?=E8=AF=AD=E3=80=81=E7=AE=97=E4=B8=8D=E4=BA=86=E7=9A=84=E5=8E=9F?= =?UTF-8?q?=E5=9B=A0=E7=A0=81=EF=BC=89=EF=BC=9Apreview=20=E5=8F=AA?= =?UTF-8?q?=E6=98=AF=E7=BB=93=E8=AE=BA=20JSON=20=E7=9A=84=E5=89=8D?= =?UTF-8?q?=E4=B8=80=E6=88=AA=EF=BC=88=E6=88=AA=E5=88=B0=202000=20?= =?UTF-8?q?=E4=B8=AA=E5=AD=97=E7=AC=A6=EF=BC=89=EF=BC=8C=E5=8F=AF=E5=8A=A0?= =?UTF-8?q?=E6=8B=86=E8=A7=A3=E3=80=81=E5=8F=AF=E6=AF=94=E6=80=A7=E6=A3=80?= =?UTF-8?q?=E6=9F=A5=E7=9A=84=E5=88=A4=E5=AE=9A=E8=AF=AD=E9=83=BD=E5=9C=A8?= =?UTF-8?q?=E6=88=AA=E6=8E=89=E7=9A=84=E9=82=A3=E6=AE=B5=E9=87=8C=EF=BC=8C?= =?UTF-8?q?=E9=97=AE=E6=95=B0=E6=8D=AE=E6=97=B6=E9=97=B4=E7=BA=BF=E7=85=A7?= =?UTF-8?q?=E8=BF=99=E4=B8=80=E6=AE=B5=E6=98=BE=E7=A4=BA=EF=BC=8C=E4=B8=8D?= =?UTF-8?q?=E4=BB=8E=20preview=20=E9=87=8C=E6=8B=BC=E3=80=82=E7=BC=96?= =?UTF-8?q?=E5=8F=B7=E5=92=8C=E4=BA=A4=E5=9B=9E=E6=A8=A1=E5=9E=8B=E7=9A=84?= =?UTF-8?q?=E3=80=8C=E3=80=90=E5=88=86=E6=9E=90=20An=E3=80=91=E3=80=8D?= =?UTF-8?q?=E5=90=8C=E4=B8=80=E4=B8=AA=EF=BC=88AnalysisScope.next=5Falias?= =?UTF-8?q?=EF=BC=8C=E8=AE=B0=E5=8F=B0=E8=B4=A6=E6=97=B6=20add=5Fresult=20?= =?UTF-8?q?=E7=B4=A7=E6=8E=A5=E7=9D=80=E6=8C=89=E5=AE=83=E7=BC=96=EF=BC=89?= =?UTF-8?q?=EF=BC=8C=E5=86=99=E9=94=99=E7=9A=84=E8=B0=83=E7=94=A8=E4=B8=8D?= =?UTF-8?q?=E5=B8=A6=E8=BF=99=E4=B8=80=E6=AE=B5=E3=80=81=E4=B8=8D=E5=8D=A0?= =?UTF-8?q?=E7=BC=96=E5=8F=B7=EF=BC=9B=E5=8D=8F=E4=BD=9C=E6=88=90=E5=91=98?= =?UTF-8?q?=E6=8C=89=E8=87=AA=E5=B7=B1=E7=9A=84=E5=88=86=E6=9E=90=E5=AE=B9?= =?UTF-8?q?=E5=99=A8=E7=BC=96=E5=8F=B7=EF=BC=9BAgent=20=E5=86=99=E7=9A=84?= =?UTF-8?q?=E5=81=87=E8=AE=BE=E4=B8=8D=E5=B8=A6=E3=80=82=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=EF=BC=9A=E7=BC=96=E5=8F=B7=E4=B8=8E=E4=BA=A4=E5=9B=9E=E6=A8=A1?= =?UTF-8?q?=E5=9E=8B=E7=9A=84=E4=B8=80=E8=87=B4=E3=80=81=E5=88=A4=E5=AE=9A?= =?UTF-8?q?=E8=AF=AD=E5=8F=AA=E5=9C=A8=E8=BF=99=E4=B8=80=E6=AE=B5=E9=87=8C?= =?UTF-8?q?=E3=80=81=E7=AE=97=E4=B8=8D=E4=BA=86=E7=9A=84=E5=B8=A6=E5=8E=9F?= =?UTF-8?q?=E5=9B=A0=E7=A0=81=E3=80=81=E6=88=90=E5=91=98=E8=87=AA=E5=B7=B1?= =?UTF-8?q?=E7=BC=96=E5=8F=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/engine/nodes/llm.py | 9 ++- backend/app/engine/nodes/multi.py | 5 +- backend/app/tools/analysis.py | 19 ++++- backend/tests/test_analysis_tool_events.py | 88 ++++++++++++++++++++++ 4 files changed, 116 insertions(+), 5 deletions(-) create mode 100644 backend/tests/test_analysis_tool_events.py diff --git a/backend/app/engine/nodes/llm.py b/backend/app/engine/nodes/llm.py index 7dff9e3f..2b90393e 100644 --- a/backend/app/engine/nodes/llm.py +++ b/backend/app/engine/nodes/llm.py @@ -55,6 +55,7 @@ from app.tools.analysis import ( SCOPE_KEY as ANALYSIS_SCOPE_KEY, AnalysisScope, call_error_text, + event_view, input_snapshot, is_analysis_tool, payload_of, @@ -550,8 +551,11 @@ async def run_agent(state: GraphState, ctx: NodeContext) -> dict[str, Any]: query = query_entry_fields(content) if evidence_on and ok and is_platform_query(name) else None if query and not await snapshot_owned(query["artifact"], ctx.run.run_id, name): query = None - # 分析工具交回的结论工件：同查询快照，记进 tool.end（封存范围内），证据接口、时间线从这里认这一步的结论 + # 分析工具交回的结论工件：同查询快照，记进 tool.end（封存范围内），证据接口、时间线从这里认这一步的结论。另带一小段 + # 判定（event_view：编号、算子、判定码、判定语……）：preview 只是结论的前一截，判定语多半被截掉。编号和记台账时 + # 交回模型的「【分析 An】」是同一个（_record_call 紧接着按同一个容器编），不记台账的执行没有编号 analysis = payload_of(content) if ok and is_analysis_tool(name) else None + alias = scope.next_alias() if analysis and scope is not None and evidence_on else None ctx.emit( EventType.TOOL_END if ok else EventType.TOOL_ERROR, tool=name, @@ -560,7 +564,8 @@ async def run_agent(state: GraphState, ctx: NodeContext) -> dict[str, Any]: preview=content[:2000], artifact=snapshot_id, **({"query_artifact": query["artifact"]} if query else {}), - **({"analysis_artifact": analysis["artifact"]} if analysis else {}), + **({"analysis_artifact": analysis["artifact"], "analysis": event_view(analysis["result"], alias)} + if analysis else {}), # 查询当时数据源的表结构快照：和查询快照一样落在封存范围内，证据接口从这里认它 **({"schema_artifact": query["schema_artifact"]} if query and query.get("schema_artifact") else {}), # 这次查询的 SQL 检查结果单独带上：preview 截到前 2000 个字符，checks 通常在截掉的那段里 diff --git a/backend/app/engine/nodes/multi.py b/backend/app/engine/nodes/multi.py index da37a869..11892067 100644 --- a/backend/app/engine/nodes/multi.py +++ b/backend/app/engine/nodes/multi.py @@ -38,6 +38,7 @@ from app.tools.analysis import ( SCOPE_KEY as ANALYSIS_SCOPE_KEY, AnalysisScope, call_error_text, + event_view, input_snapshot, is_analysis_tool, payload_of, @@ -461,7 +462,9 @@ async def run_supervisor(state: GraphState, ctx: NodeContext) -> dict[str, Any]: agent=name, call_id=cid, preview=content[:1500], duration_ms=int((time.perf_counter() - started) * 1000), **failure, **({"checks": checks} if checks else {}), - **({"analysis_artifact": analysis["artifact"]} if analysis else {})) + # 判定的那一小段同 Agent 节点（event_view）；编号是成员自己的分析容器里的，紧接着 _numbered 按它编 + **({"analysis_artifact": analysis["artifact"], + "analysis": event_view(analysis["result"], scope.next_alias())} if analysis else {})) if analysis is not None and analysis["result"].get("status") == "refused": ctx.emit(EventType.LOG, level="warn", code=ANALYSIS_REFUSED, agent=name, call_id=cid, tool=tname, reason_code=analysis["result"].get("reason_code"), diff --git a/backend/app/tools/analysis.py b/backend/app/tools/analysis.py index fc122fdc..b70e0c99 100644 --- a/backend/app/tools/analysis.py +++ b/backend/app/tools/analysis.py @@ -86,9 +86,13 @@ class AnalysisScope: def add_query(self, alias: str, snapshot: InputSnapshot) -> None: self.queries[alias] = snapshot + def next_alias(self) -> str: + """下一份结论在本节点的编号：工具执行完、记台账之前，tool.end 先按它写上这份结论的编号（event_view）。""" + return f"{contract.CATALOG_PREFIX}{len(self.results) + 1}" + def add_result(self, result: AnalysisResult, artifact: str) -> str: """记一份结论，返回它在本节点的编号（A1、A2……，按记下的先后）。""" - alias = f"{contract.CATALOG_PREFIX}{len(self.results) + 1}" + alias = self.next_alias() self.results[alias] = result self.artifacts[alias] = artifact return alias @@ -363,6 +367,17 @@ def reply_text(alias: str, result: AnalysisResult) -> str: return REPLY_HEAD.format(alias=alias) + "\n" + json.dumps(body, ensure_ascii=False, default=str) +def event_view(result: AnalysisResult, alias: str | None = None) -> dict[str, Any]: + """分析工具的 tool.end 上带的那一小段（tool.end 的 analysis）：编号、算子、模式、状态、判定码、方向、判定语、算不了的原因码。 + preview 只是结论 JSON 的前一截（截到 2000 个字符），判定语多半在截掉的那段里；时间线照这一段显示，不从 preview 里拼。 + 编号是这份结论在本节点的编号（「【分析 An】」那个，AnalysisScope.next_alias），不记台账的执行里没有编号，为 None。 + 判定语是系统渲染的（遮罩列不当维度、度量，里面没有遮罩值）；Agent 写的假设不带。""" + verdict = result.get("verdict") or {} + return {"alias": alias, "op": result.get("op"), "mode": result.get("mode"), "status": result.get("status"), + "verdict": verdict.get("code"), "direction": verdict.get("direction"), "phrase": result.get("phrase") or "", + "reason_code": result.get("reason_code")} + + def hypothesis_of(value: Any) -> str | None: """agent（或模板作者）写的「这次在验证什么」：去掉首尾空白、截到 HYPOTHESIS_MAX 字；空的为 None。""" text = str(value or "").strip() @@ -371,6 +386,6 @@ def hypothesis_of(value: Any) -> str | None: __all__ = [ "ANALYSIS_TOOLS", "CALL_ERROR_HEAD", "REFUSED_CODE", "REPLY_HEAD", "SCOPE_KEY", "AnalysisScope", "call_error_text", "context_of", - "hypothesis_of", "input_snapshot", "is_analysis_tool", "masked_of", "payload_of", "prepare", "reply_text", + "event_view", "hypothesis_of", "input_snapshot", "is_analysis_tool", "masked_of", "payload_of", "prepare", "reply_text", "run_snapshot", "scope_of", "settings_of", "store", "tool_context", ] diff --git a/backend/tests/test_analysis_tool_events.py b/backend/tests/test_analysis_tool_events.py new file mode 100644 index 00000000..f41a1237 --- /dev/null +++ b/backend/tests/test_analysis_tool_events.py @@ -0,0 +1,88 @@ +"""分析工具的 tool.end 带的那一小段判定（app/tools/analysis.event_view）：时间线照它显示算子、判定语和编号，不从 preview 里拼。 + +preview 只是结论 JSON 的前一截（截到 2000 个字符），可加拆解、可比性检查的判定语都在截掉的那段里；tool.end 的 analysis 另带 +编号、算子、模式、状态、判定码、方向、判定语、算不了的原因码。编号和交回模型的「【分析 An】」是同一个；写错的调用（tool.error） +不带这一段、不占编号。接真算子（不用替身），库是虚构的零售库 shop，模型换成脚本（tests/test_analysis_tools.Script）。 +""" +from __future__ import annotations + +import json + +from langchain_core.messages import ToolMessage + +from app.analysis import contract +from app.core import artifact_store +from app.engine.runner import run_manager +from app.tools.analysis import event_view +from tests.test_analysis_tools import ( # noqa: F401 - 夹具按名字引入 + DECOMPOSE, DECOMPOSE_ARGS, SQL, TOOL, Script, agent_graph, engine_up, events, node, run_graph, shop, wait, +) + +BASELINE = contract.TOOL_NAMES["baseline"] +COMPARABLE = contract.TOOL_NAMES["comparable"] +WRONG = {**DECOMPOSE_ARGS, "measure": "sales"} +#: 整体的周序列（Q2）：波动基线、两周的可比性读它 +TOTALS = "SELECT week, SUM(amount) AS gmv FROM orders GROUP BY week ORDER BY week" +WEEKLY = {"query": "Q2", "period": "week", "measure": "gmv", "label": "销售额"} +TWO_WEEKS = {"query": "Q2", "period": "week", "base": "2026-W41", "target": "2026-W42", "measure": "gmv"} + + +def _replies(script: Script) -> list[str]: + return [m.content for m in script.seen[-1] if isinstance(m, ToolMessage)] + + +def test_event_view_has_the_judgment_without_the_hypothesis(): + result = {"op": "decompose", "mode": "additive", "status": "ok", "reason_code": None, "phrase": "销售额下降 194", + "verdict": {"code": "driver", "subjects": [["华东"]], "direction": "down"}, "hypothesis": "华东拖累", + "table": {"columns": ["region"], "rows": [["华东"]]}} + assert event_view(result, "A2") == {"alias": "A2", "op": "decompose", "mode": "additive", "status": "ok", + "verdict": "driver", "direction": "down", "phrase": "销售额下降 194", + "reason_code": None} + refused = {"op": "baseline", "mode": None, "status": "refused", "reason_code": "history_short", + "phrase": "波动基线没有做成：只有 1 期历史，至少要 8 期", "verdict": None} + assert event_view(refused) == {"alias": None, "op": "baseline", "mode": None, "status": "refused", "verdict": None, + "direction": None, "phrase": "波动基线没有做成：只有 1 期历史，至少要 8 期", + "reason_code": "history_short"} + + +async def test_tool_end_carries_the_phrase_and_the_same_number_the_model_got(monkeypatch, shop): + script = Script(monkeypatch, [(TOOL, {"sql": SQL}), (TOOL, {"sql": TOTALS}), (DECOMPOSE, WRONG), + (DECOMPOSE, DECOMPOSE_ARGS), (BASELINE, WEEKLY), (COMPARABLE, TWO_WEEKS)]) + row = await run_graph(agent_graph(tools=(TOOL, DECOMPOSE, BASELINE, COMPARABLE))) + ends = [e for e in await events(row.id, "tool.end", "bot") if e.get("analysis_artifact")] + assert [e["analysis"]["alias"] for e in ends] == ["A1", "A2", "A3"] + # 交回模型的编号同一套：写错的那次不占编号 + replies = [r.split("\n", 1)[0] for r in _replies(script) if r.startswith("【分析")] + assert replies == ["【分析 A1】", "【分析 A2】", "【分析 A3】"] + for end in ends: + result = artifact_store.load(end["analysis_artifact"]) + assert end["analysis"] == event_view(result, end["analysis"]["alias"]) + decompose, baseline, comparable = (e["analysis"] for e in ends) + assert (decompose["op"], decompose["mode"], decompose["status"], decompose["verdict"]) == ( + "decompose", "additive", "ok", "driver") + assert decompose["phrase"].startswith("销售额下降") and decompose["phrase"] not in ends[0]["preview"], \ + "可加拆解的判定语在 preview 截掉的那段里，只能从这一段读" + assert (baseline["status"], baseline["reason_code"]) == ("refused", "history_short") + assert baseline["phrase"].startswith(contract.OP_LABELS["baseline"] + "没有做成") + assert comparable["op"] == "comparable" and comparable["phrase"].startswith("两期") + # 写错的调用：tool.error 带调用错误的代码，不带判定 + [error] = [e for e in await events(row.id, "tool.error", "bot")] + assert error["call_error"] == "unknown_column" and "analysis" not in error + + +async def test_a_team_member_numbers_its_own_analyses(monkeypatch, shop): + Script(monkeypatch, [(TOOL, {"sql": SQL}), (DECOMPOSE, DECOMPOSE_ARGS)], + route={"assignments": [{"agent": "分析员", "instruction": "拆一下销售额的变化"}], "done": False}) + nodes = [node("start", "input"), + node("team", "supervisor", goal="销售额为什么下降", approval="never", max_rounds=2, agents=[ + {"name": "分析员", "system": "你负责拆解", "tools": [TOOL, DECOMPOSE], "max_steps": 4}]), + node("out", "output", fields=[{"name": "a", "value": "{{ nodes.team.text }}"}])] + run = await run_manager.start(graph={"nodes": nodes, "edges": [{"source": "start", "target": "team"}, + {"source": "team", "target": "out"}]}, + input_payload={}) + row = await wait(run.id) + assert row.status == "succeeded", row.error + [end] = [e for e in await events(row.id, "tool.end", "team") if e.get("analysis_artifact")] + assert end["agent"] == "分析员" + assert (end["analysis"]["alias"], end["analysis"]["op"], end["analysis"]["verdict"]) == ("A1", "decompose", "driver") + assert json.dumps(end["analysis"], ensure_ascii=False).count("下降是不是集中在某个地区") == 0, "Agent 写的假设不带"