分析工具的 tool.end 另带一小段判定(tools/analysis.event_view:编号、算子、模式、状态、判定码、方向、判定语、算不了的原因码):preview 只是结论 JSON 的前一截(截到 2000 个字符),可加拆解、可比性检查的判定语都在截掉的那段里,问数据时间线照这一段显示,不从 preview 里拼。编号和交回模型的「【分析 An】」同一个(AnalysisScope.next_alias,记台账时 add_result 紧接着按它编),写错的调用不带这一段、不占编号;协作成员按自己的分析容器编号;Agent 写的假设不带。测试:编号与交回模型的一致、判定语只在这一段里、算不了的带原因码、成员自己编号

This commit is contained in:
ThearchyHelios
2026-10-21 17:53:56 +08:00
parent 843be9a556
commit 5f61661625
4 changed files with 116 additions and 5 deletions
+7 -2
View File
@@ -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 通常在截掉的那段里
+4 -1
View File
@@ -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"),
+17 -2
View File
@@ -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",
]
@@ -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 写的假设不带"