diff --git a/backend/app/api/evidence.py b/backend/app/api/evidence.py index 741dab46..5deb102d 100644 --- a/backend/app/api/evidence.py +++ b/backend/app/api/evidence.py @@ -909,9 +909,30 @@ async def evidence_metric(run_id: str, node: str, metric: str) -> dict[str, Any] "redacted": {"columns": masked, "note": MASK_NOTE} if masked else {"columns": []}} -def _open_segment(sealed: _Sealed, wanted: str | None, segment_id: str - ) -> tuple[_Report, tuple[dict[str, Any], dict[str, Any], dict[str, Any]]]: - """选报告、确认文档完好、找到片段：(报告, (块, 句子, 片段))。片段接口和推断来源接口共用同一套错误码。""" +ANALYSIS_NOT_FOUND = "evidence_analysis_not_found" + + +@router.get("/{run_id}/evidence/analyses/{alias}") +async def evidence_analysis(run_id: str, alias: str, report: str | None = None) -> dict[str, Any]: + """点开报告末尾系统写的「分析范围」里的一个编号（A1…）：那份分析结论的出处链（判定 → 方法 → 计算 → 输入 → 复算 → + 封存），和点开正文里它的判定语是同一条（_scope_chain，导出页的分析范围也按它）。正文没引用的结论（排除的、算不了的）也 + 点得开。只认这份报告证据目录里的分析结论；选哪份报告、文档完不完好同片段接口（?report=<节点 id>）。""" + sealed = await _sealed(run_id) + chosen = _open_report(sealed, report) + doc = chosen.doc or {} + chain = await _scope_chain(alias, doc, sealed, chosen.node_id) + if chain is None: + raise CodedHTTPException(404, "这份报告的证据目录里没有这份分析结论", ANALYSIS_NOT_FOUND) + covered = sealed.trusted and all(step.get("sealed") for step in chain if "sealed" in step) + masked = _chain_masked(chain) + return {"run_id": run_id, "report": {"node_id": chosen.node_id, "doc_artifact": chosen.doc_artifact}, "alias": alias, + "rendered": next((step.get("rendered") for step in chain if step.get("step") == LEDGER_KIND), None), + "chain": chain, "seal": {"sealed": sealed.sealed, "ok": sealed.seal["ok"], "covered": covered}, + "redacted": {"columns": masked, "note": MASK_NOTE} if masked else {"columns": []}} + + +def _open_report(sealed: _Sealed, wanted: str | None) -> _Report: + """选报告、确认文档完好。片段接口、推断来源接口、分析范围里的编号共用同一套错误码。""" reports = _named_reports(sealed, wanted) if not reports: raise CodedHTTPException(404, "本次运行没有封存的报告文档，没有可查看的片段", REPORT_NOT_FOUND) @@ -920,7 +941,14 @@ def _open_segment(sealed: _Sealed, wanted: str | None, segment_id: str raise CodedHTTPException(409, "报告文档与哈希不一致，疑似被修改，不能作为证据展示", DOC_TAMPERED) if chosen.doc is None: raise CodedHTTPException(404, "无法读取报告文档", DOC_MISSING) - found = find_segment(chosen.doc, segment_id) + return chosen + + +def _open_segment(sealed: _Sealed, wanted: str | None, segment_id: str + ) -> tuple[_Report, tuple[dict[str, Any], dict[str, Any], dict[str, Any]]]: + """选报告、确认文档完好、找到片段：(报告, (块, 句子, 片段))。片段接口和推断来源接口共用同一套错误码。""" + chosen = _open_report(sealed, wanted) + found = find_segment(chosen.doc or {}, segment_id) if found is None: raise CodedHTTPException(404, "报告中没有这个片段", SEGMENT_NOT_FOUND) return chosen, found diff --git a/backend/app/bundle/verifier.js b/backend/app/bundle/verifier.js index 767828fe..a0ad030a 100644 --- a/backend/app/bundle/verifier.js +++ b/backend/app/bundle/verifier.js @@ -198,6 +198,8 @@ /** cited 已经加好引号（空值写「空值（NULL）」）；expected 和引用的值写法一样时不写（文字取值） */ numShownDiffers: (shown, cited, expected) => `正文显示「${shown}」，引用的值是${cited}` + (expected == null ? '' : `，按显示规则应显示为「${expected}」`), + /** 判定语的引用不带值（判定语就是结论里的那句话）：只点名正文显示的字和结论里的判定语，不写「引用的值是空值」 */ + numPhraseDiffers: (shown, expected) => `正文显示「${shown}」，分析结论里的判定语是「${expected}」`, numShownUnknown: (why) => `无法核对显示文字：${why}`, shownFormat: (fmt) => `显示格式「${fmt}」无法识别`, shownConv: (conv) => `换算「${conv}」无法识别`, @@ -1619,6 +1621,9 @@ info.expected = expected const shown = String(seg.text ?? '') if (pyStrip(shown) === pyStrip(expected)) return { ...info, result: PASS, shownOk: true } + if (info.cite.kind === 'analysis' && !Object.prototype.hasOwnProperty.call(info.cite, 'value')) { + return { ...info, result: FAIL, reason: T.numPhraseDiffers(shown, expected) } + } const value = info.cite.value const cited = typeof value === 'number' ? numberText(value, info.citedToken) : value == null ? null : textOf(value) return { ...info, result: FAIL, reason: T.numShownDiffers(shown, cited == null ? T.nullValue : `「${cited}」`, 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/bundle_samples.py b/backend/tests/bundle_samples.py index 7aada838..c821973f 100644 --- a/backend/tests/bundle_samples.py +++ b/backend/tests/bundle_samples.py @@ -38,6 +38,10 @@ 原样导出照样全部通过 - prose：同经营月报的口径卡和流水，正文是写作者自己写在引用旁边的普通文字（「运费统一[[m:…]]」「合计——[[m:…]]」 「SKU-[[v:…]]」「共[[m:…]]家」这类）：原样导出全部通过、总览是绿的，紧挨着核对的规则不误判正常写法 +- analysis：异动周报（分析能力一轮）。Agent 在虚构零售库 shop 上查了两周的总数（Q1）和按地区两周的销售额（Q2），做了按地区的 + 变化拆解（A1，带总数核对）、波动基线（A2，两周历史不够，算不了）、两期可比性检查（A3），结论都是 app.analysis 真算的； + 报告引用 A1 的判定语、具名的数（占比）、结果表的格，正文后面是系统写的「分析范围」（engine/analysis_scope 按虚构的数据 + 目录算出没拆的渠道）。结论和输入快照随文件附带、导出时复算过，原样导出全部通过；分析范围里的编号点得开（A2、A3 正文没引用） """ from __future__ import annotations @@ -313,6 +317,125 @@ def formats_bundle(*, card: dict[str, Any] = FORMATS_CARD, table: dict[str, Any] return build(f, events, bound) +#: 分析样本（analysis）的两份查询：虚构零售库 shop 两周的总数（Q1）、四个地区两周的销售额（Q2） +AN_TOTALS = {"columns": ["week", "gmv"], "column_types": {"week": "text", "gmv": "number"}, + "rows": [["2026-W41", 2220], ["2026-W42", 2026]], "source": "shop", + "sql": "SELECT week, SUM(amount) AS gmv FROM orders GROUP BY week ORDER BY week", "truncated": False, + "row_count": 2} +AN_REGIONS = {"columns": ["region", "week", "gmv"], "column_types": {"region": "text", "week": "text", "gmv": "number"}, + "rows": [["华东", "2026-W41", 820], ["华东", "2026-W42", 700], ["华北", "2026-W41", 600], + ["华北", "2026-W42", 560], ["华南", "2026-W41", 700], ["华南", "2026-W42", 666], + ["西南", "2026-W41", 100], ["西南", "2026-W42", 100]], "source": "shop", + "sql": "SELECT region, week, SUM(amount) AS gmv FROM orders GROUP BY region, week ORDER BY region, week", + "truncated": False, "row_count": 8} +#: Agent 的三次分析调用（工具名、请求），照分析工具交给 app.analysis.run 的写法 +AN_CALLS = [ + ("analysis_decompose", {"op": "decompose", "mode": "additive", "hypothesis": "下降是不是集中在某个地区", + "params": {"query": "Q2", "by": ["region"], "period": "week", "base": "2026-W41", + "target": "2026-W42", "measure": "gmv", "label": "销售额", + "headline": {"base": "Q1.r0.gmv", "target": "Q1.r1.gmv"}}}), + ("analysis_baseline", {"op": "baseline", "mode": None, "hypothesis": "这周的下降算不算异常", + "params": {"query": "Q1", "period": "week", "measure": "gmv", "label": "销售额"}}), + ("analysis_comparable", {"op": "comparable", "mode": None, "hypothesis": "两周能不能直接比", + "params": {"query": "Q2", "period": "week", "base": "2026-W41", "target": "2026-W42", + "measure": "gmv"}}), +] +#: 异动周报：引用 A1 的判定语、具名的数（占比）、结果表的格；A2、A3 只在系统写的分析范围里 +AN_REPORT = ("## 异动周报\n\n[[a:A1]]。华东占总变化的 [[a:A1.top_share|pct]]，华东的变化是 [[v:A1.r0.delta]]。[[see:A1]]\n\n" + "各地区的变化见下表。\n\n[[table:A1 cols=region,base,target,delta rows=0-3]]") + + +def _shop_notes() -> dict[str, Any]: + """虚构零售库 shop 的数据目录（orders 是事实表；地区、渠道是属性列，码值表列全）：分析范围按它算没拆的候选维度。""" + from app.data.catalog import make_item + + def column(label: str, measure: str, codes: tuple[str, ...] = ()) -> dict[str, Any]: + out = {"label": make_item(label, "human"), "measure": make_item(measure, "human")} + if codes: + out["codes"] = make_item({c: "" for c in codes}, "profile", "verified") + return out + + return {"orders": {"kind": make_item("fact", "human"), "columns": { + "region": column("地区", "attribute", ("华东", "华北", "华南", "西南")), + "channel": column("渠道", "attribute", ("线上", "线下")), + "week": column("周", "attribute"), "amount": column("实收金额", "flow")}}} + + +def analysis_bundle() -> dict[str, Any]: + """异动周报（analysis）的页面数据：结论按两份快照真算（app.analysis.run），Agent 节点的事件照分析工具真实的写法 + （tool.end 带结论工件和判定那一段），报告文档按写作目录真实组装，分析范围按虚构的数据目录真实算出。""" + import asyncio + from datetime import date + + from app import analysis + from app.analysis.contract import effective_settings, ledger_entry + from app.engine import analysis_scope + from app.tools.analysis import event_view, input_snapshot + + run = {**RUN, "id": "9c2e4a6b8d0f4e1a3c5b7d9f1e3a5c7b", "workflow_id": "wf-sample-anomaly", "workflow_name": "异动周报", + "started_at": "2026-10-19T01:00:00+00:00", "finished_at": "2026-10-19T01:01:30+00:00"} + graph = {"nodes": [{"id": n, "type": t, "data": {"label": label}} for n, t, label in [ + ("start", "input", "开始"), ("bot", "agent", "问数据"), ("write", "report", "写周报"), ("out", "output", "成果")]]} + files: dict[str, bytes] = {} + + def put(obj: Any) -> str: + artifact, raw = stored(obj) + files[artifact] = raw + return artifact + + def load(artifact: str) -> Any: + raw = files.get(artifact) + return json.loads(raw) if raw is not None else None + + queries = [(put(snap), snap) for snap in (AN_TOTALS, AN_REGIONS)] + # 照分析工具交给算子的写法：查询快照原样加上出处；目录事实（列的叫法、度量类型）按虚构的数据目录给，结论里记下 + snapshots = {f"Q{i}": input_snapshot(snap, ref=f"Q{i}", artifact=sha, node_id="bot") + for i, (sha, snap) in enumerate(queries, 1)} + context = {"today": date(2026, 10, 19), "catalogs": {"shop": _shop_notes()}} + ledger: list[dict[str, Any]] = [ + {"kind": "query", "node_id": "bot", "exec": 1, "artifact": sha, "call_id": f"call_{i}", "tool": "db_query__shop", + "source": "shop", "columns": snap["columns"], "rows": len(snap["rows"]), "truncated": False} + for i, (sha, snap) in enumerate(queries)] + results = [] + for n, (tool, request) in enumerate(AN_CALLS, len(queries)): + result = dict(asyncio.run(analysis.run(request, snapshots.get, settings=effective_settings(None), context=context))) + artifact = put(result) + results.append((tool, f"call_{n}", artifact, result)) + ledger.append(ledger_entry(result, node_id="bot", exec_no=1, artifact=artifact, call_id=f"call_{n}")) + catalog = build_catalog(nodes={}, ledger=ledger, loader=load) + doc = compose_doc(AN_REPORT, catalog, node_id="write", run_id=run["id"], loader=load) + doc["analysis_scope"] = analysis_scope.build(catalog, sources={"shop": {"tables": _shop_notes(), "masked": []}}, + load=load) + doc_id = put(doc) + + events: list[tuple[int, str, str | None, Any]] = [] + + def add(etype: str, node_id: str | None, data: dict[str, Any]) -> None: + events.append((len(events) + 1, etype, node_id, data)) + + add("run.started", None, {"nodes": len(graph["nodes"]), "resumed": False, "memory_scope": "default", + "collection": "default", "approval_default": None, "tool_trust": "formal", + "agent_limits": {}, "data_versions": None, "replay_protocol": 2}) + add("node.started", "bot", {"node_type": "agent", "label": "问数据"}) + for i, (sha, snap) in enumerate(queries): + add("tool.end", "bot", {"tool": "db_query__shop", "call_id": f"call_{i}", "duration_ms": 9, "query_artifact": sha, + "preview": json.dumps({"columns": snap["columns"], "rows": snap["rows"][:2]}, + ensure_ascii=False)}) + for k, (tool, call_id, artifact, result) in enumerate(results, 1): + add("tool.end", "bot", {"tool": tool, "call_id": call_id, "duration_ms": 6, "analysis_artifact": artifact, + "preview": json.dumps({"analysis_artifact": artifact}), "analysis": event_view(result, f"A{k}")}) + add("node.finished", "bot", {"duration_ms": 240, "attempt": 1, "preview": {"text": "销售额下降，主要来自华东。"}, + "evidence": ledger}) + add("llm.end", "write", {"model": "mock", "output": "（写作草稿）"}) + add("report.checked", "write", {"doc_artifact": doc_id, "ok": True}) + add("node.finished", "write", {"duration_ms": 20, "attempt": 1, "preview": {"text": "（报告正文）", "doc_artifact": doc_id}}) + add("issuance", "out", {"tier": "formal", "mode": "citations"}) + add("node.finished", "out", {"duration_ms": 2, "attempt": 1, "preview": {"周报": "（报告正文）"}}) + add("run.finished", None, {"output": {"周报": "（报告正文）"}}) + return build_bundle(run, events, manifest_seq=len(events), manifest_hash=manifest_hash(events), read=files.get, + exported_at=EXPORTED_AT, graph=graph, source_masks={}) + + def bundles() -> dict[str, dict[str, Any]]: """{页面名: 页面数据}""" out: dict[str, dict[str, Any]] = {} @@ -368,6 +491,7 @@ def bundles() -> dict[str, dict[str, Any]]: out["formats"] = formats_bundle() out["formats-unit"] = formats_bundle(card=FORMATS_UNIT_CARD) out["prose"] = formats_bundle(text=PROSE) + out["analysis"] = analysis_bundle() return out diff --git a/backend/tests/test_analysis_scope_api.py b/backend/tests/test_analysis_scope_api.py new file mode 100644 index 00000000..9727f874 --- /dev/null +++ b/backend/tests/test_analysis_scope_api.py @@ -0,0 +1,101 @@ +"""报告末尾「分析范围」里的编号点得开（GET /runs/{id}/evidence/analyses/{alias}）：那份分析结论的出处链和点开正文里它的判定语 +是同一条（api/evidence._scope_chain），正文没引用的结论（算不了的、可比性检查）也点得开；证据目录里没有的编号 404。 + +真运行、真算子：照 tests/test_analysis_e2e_claims.py 的虚构零售库 shop（数据目录标了地区、渠道是属性），Agent 查两次库、 +做一次变化拆解、一次波动基线（两周历史不够，算不了）、一次可比性检查；写作模型只引用拆解的判定语。模型都换成脚本。 +""" +from __future__ import annotations + +from httpx import ASGITransport, AsyncClient +from langchain_core.messages import AIMessage, ToolMessage + +from app.analysis import contract +from app.core import artifact_store +from app.engine.evidence import iter_segments +from app.main import app +from tests.test_analysis_e2e_claims import ( # noqa: F401 - 夹具按名字引入 + ARGS, BY_REGION, DECOMPOSE, TOOL, TOTALS, _finish, _graph, _report, engine_up, shop, +) + +BASELINE = contract.TOOL_NAMES["baseline"] +COMPARABLE = contract.TOOL_NAMES["comparable"] +TEXT = "## 本周销售\n\n[[a:A1]]。[[see:A1]]" +CHAIN = ["analysis", "analysis_method", "analysis_calc", "analysis_input", "analysis_recompute", "analysis_seal"] + + +def _script(monkeypatch): + from app.providers import mock_model + + calls = [(TOOL, {"sql": TOTALS}), (TOOL, {"sql": BY_REGION}), (DECOMPOSE, ARGS), + (BASELINE, {"query": "Q1", "period": "week", "measure": "gmv", "label": "销售额"}), + (COMPARABLE, {"query": "Q2", "period": "week", "base": "2026-W41", "target": "2026-W42", "measure": "gmv"})] + + def decide(model, messages): + if not model.tools: + return AIMessage(content=TEXT) + done = sum(isinstance(m, ToolMessage) for m in messages) + if done < len(calls): + name, args = calls[done] + return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": f"call_{done}"}]) + return AIMessage(content="销售额下降，主要来自华东。") + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide) + + +def _graph_with_tools(): + graph = _graph() + bot = next(n for n in graph["nodes"] if n["id"] == "bot") + bot["data"]["config"]["tools"] = [TOOL, DECOMPOSE, BASELINE, COMPARABLE] + return graph + + +def _steps(chain): + """同一步连着几条（几份输入、输入后面接的查询步骤）算一步。""" + steps = [s["step"] for s in chain] + return [s for i, s in enumerate(steps) if s.startswith("analysis") and steps[i - 1:i] != [s]] + + +async def test_scope_aliases_open_the_same_chain_as_the_phrase_in_the_text(monkeypatch): + _script(monkeypatch) + row = await _finish(_graph_with_tools()) + assert row.status == "succeeded", row.error + _, doc, _ = await _report(row) + scope = doc["analysis_scope"] + assert [i["alias"] for i in scope["decomposed"]] == ["A1"] + assert [i["alias"] for i in scope["refused"]] == ["A2"] and [i["alias"] for i in scope["phenomena"]] == ["A3"] + phrase = next(s for s in iter_segments(doc) if s.get("ref") == "a:A1") + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + cited = (await client.get(f"/api/runs/{row.id}/evidence/segments/{phrase['id']}")).json() + body = (await client.get(f"/api/runs/{row.id}/evidence/analyses/A1")).json() + # 和点开正文里的判定语同一条链（同一份结论、同样复算一致），只是没有这一处引用的那几样 + assert _steps(body["chain"]) == CHAIN + assert [s["step"] for s in body["chain"]] == [s["step"] for s in cited["chain"]] + assert body["rendered"] == phrase["text"] == body["chain"][0]["phrase"] + assert body["seal"]["covered"] is True and body["report"]["node_id"] == "write" + + # 正文没引用的也点得开：算不了的波动基线照实给原因，可比性检查照样复算 + refused = (await client.get(f"/api/runs/{row.id}/evidence/analyses/A2", params={"report": "write"})).json() + judged = refused["chain"][0] + assert (judged["op"], judged["status"], judged["reason_code"]) == ("baseline", "refused", "history_short") + assert judged["phrase"] == scope["refused"][0]["phrase"] + assert next(s for s in refused["chain"] if s["step"] == "analysis_recompute")["status"] == "same" + comparable = (await client.get(f"/api/runs/{row.id}/evidence/analyses/A3")).json() + assert comparable["chain"][0]["method_label"] == contract.method_label("comparable") + + # 目录里没有的编号、不是分析结论的编号（Q1）：404，说清是哪种 + for alias in ("A9", "Q1"): + missing = await client.get(f"/api/runs/{row.id}/evidence/analyses/{alias}") + assert missing.status_code == 404 and missing.json()["code"] == "evidence_analysis_not_found" + + +async def test_a_tampered_report_does_not_open_its_scope(monkeypatch): + _script(monkeypatch) + row = await _finish(_graph_with_tools()) + checked, _, _ = await _report(row) + # 报告文档被改过（哈希对不上）：和片段接口同一个错误码，不给出处 + path = artifact_store._path_of(checked["doc_artifact"]) + path.write_bytes(path.read_bytes().replace("本周销售".encode(), "本周营业".encode(), 1)) + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + res = await client.get(f"/api/runs/{row.id}/evidence/analyses/A1") + assert res.status_code == 409 and res.json()["code"] == "evidence_doc_tampered" 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 写的假设不带" diff --git a/backend/tests/test_bundle_build.py b/backend/tests/test_bundle_build.py index 4172ade7..2afb0d42 100644 --- a/backend/tests/test_bundle_build.py +++ b/backend/tests/test_bundle_build.py @@ -470,7 +470,7 @@ def test_the_sample_pages_for_the_browser_check_are_sealed_for_real(): got = {name: json.loads(data_block(page)) for name, page in rendered.items()} assert set(got) == {"base", "two-reports", "metric-only", "legacy", "failed", "rejected", "continued", "degraded", "missing-doc", "missing-card", "nan", "stray", "masked-card", "masked-old-doc", "masked-source", - "formats", "formats-unit", "prose"} + "formats", "formats-unit", "prose", "analysis"} for name, bundle in got.items(): assert browser_manifest(bundle["events"]) == bundle["meta"]["manifest_hash"], name @@ -557,3 +557,15 @@ def test_the_sample_pages_for_the_browser_check_are_sealed_for_real(): # 口径卡单位带尾随空白：服务端把「元 」原样接在数后面，正文里的销售额带着那个空白 shown = {s.get("ref"): s["text"] for s in segments(got["formats-unit"])} assert shown["m:gmv"] == "74,613.65元 " and shown["m:gmv|万"] == "7.46万元 " + # 异动周报：三份分析结论（导出时都复算一致）和两份输入快照随文件附带；文档带着系统写的分析范围，正文没引用的 A2、A3 + # 只在范围里点名；分析工具的 tool.end 结论随文件附带了，带规范字符串 + analysed = got["analysis"] + kinds = sorted((a["kind"], a.get("node_id"), "b64" in a) for a in analysed["artifacts"]) + assert kinds == [("analysis_result", "bot", True)] * 3 + [("query", "bot", True)] * 2 + [("report_doc", "write", True)] + assert {a["recheck"]["status"] for a in analysed["artifacts"] if a["kind"] == "analysis_result"} == {"same"} + doc = json.loads(base64.b64decode(next(a["b64"] for a in analysed["artifacts"] if a["kind"] == "report_doc"))) + scope = doc["analysis_scope"] + assert [[i["alias"] for i in scope[k]] for k in ("phenomena", "decomposed", "refused")] == [["A3"], ["A1"], ["A2"]] + assert scope["not_decomposed"] == ["渠道"] and scope["decomposed"][0]["dims"] == ["地区"] + assert {s["ref"] for s in segments(analysed) if s["ref"].startswith("a:")} == {"a:A1.top_share|pct"} + assert all("canonical" in e for e in analysed["events"] if e["type"] == "tool.end") diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index 30ed5507..3947004e 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -856,6 +856,14 @@ export const api = { get( `/runs/${encodeURIComponent(runId)}/evidence/segments/${encodeURIComponent(segmentId)}${qs({ report: opts?.report })}`, opts), + /** + * 报告末尾「分析范围」里的一个编号（A1）：那份分析结论的出处链，和点开正文里它的判定语同一条。正文没引用的结论也点得开； + * 证据目录里没有这份结论 404。report 同片段接口 + */ + analysis: (runId: string, alias: string, opts?: RequestOptions & { report?: string }) => + get<{ report?: { node_id?: string; doc_artifact?: string }; alias?: string; rendered?: string | null + chain?: EvidenceSegmentDetail['chain']; seal?: EvidenceSegmentDetail['seal']; redacted?: EvidenceSegmentDetail['redacted'] }>( + `/runs/${encodeURIComponent(runId)}/evidence/analyses/${encodeURIComponent(alias)}${qs({ report: opts?.report })}`, opts), /** * 审计表：每个有状态的片段按状态分组（有出处 / 无证据 / 可疑实体 / 旧运行猜测），附上封存核对的结果。 * groups 只要这几组（none、suspicious…） diff --git a/frontend/src/canvas/Inspector.tsx b/frontend/src/canvas/Inspector.tsx index d0a099e8..e67938bc 100644 --- a/frontend/src/canvas/Inspector.tsx +++ b/frontend/src/canvas/Inspector.tsx @@ -15,7 +15,15 @@ import { formatShortcut } from '../lib/keys' import { INPUT_FIELD_TEXT, JUDGE_FIELD_LABEL, JUDGE_ON_UNSUPPORTED_HINT, JUDGE_ON_UNSUPPORTED_LABEL, JUDGE_SETTING_TEXT, MERGE_TEXT, CATALOG_UI_TEXT, NODE_TYPE_LABEL, PROVIDER_REF_TEXT, SUBGRAPH_UPGRADE_HELP, judgeByModelText, upgradeNewerText } from '../lib/terms' import { TemplateText } from './TemplateText' import { MERGE_ALIAS, isQueryNode, mergeAlias } from './mergeInputs' -import type { NodeType, Provider, ValidationIssue, WorkflowVersion } from '../types' +import { + ITEM_ID, TWO_QUERY_SKIP, allowsHeadline, allowsParent, allowsTwoQueries, itemsOf, nextItemId, paramSpecs, paramText, + paramValue, withMode, withOp, type ParamSpec, +} from './analysisItems' +import { + ANALYSIS_MODES, ANALYSIS_MODE_LABEL, ANALYSIS_OPS, ANALYSIS_OP_LABEL, ANALYSIS_SETTING_KEYS, ANALYSIS_SETTING_LABEL, + ANALYSIS_TEXT, isAnalysisOp, settingRangeText, settingValueProblem, type AnalysisSettingKey, +} from '../lib/termsAnalysis' +import type { AnalysisItem, NodeType, Provider, ValidationIssue, WorkflowVersion } from '../types' /** 检查器里一条落到字段上的问题 */ type FieldIssue = ValidationIssue & { at: FieldRef | null } @@ -151,6 +159,34 @@ function NodeRefsInput({ nodeId, type, value, onChange }: { ) } +/** 上游能当查询输入的节点（选了数据库查询工具的「调用工具」节点、合并查询）：合并查询、分析卡共用 */ +function useUpstreamQueries(nodeId: string): { value: string; label: string; hint?: string }[] { + const nodes = useStudio((s) => s.nodes) + const edges = useStudio((s) => s.edges) + return useMemo(() => { + const parents = new Map() + for (const e of edges) parents.set(e.target, [...(parents.get(e.target) ?? []), e.source]) + const seen = new Set() + const stack = [...(parents.get(nodeId) ?? [])] + while (stack.length) { + const id = stack.pop()! + if (seen.has(id) || id === nodeId) continue + seen.add(id) + stack.push(...(parents.get(id) ?? [])) + } + return nodes + .filter((n) => seen.has(n.id) && isQueryNode(n)) + .map((n) => { + const label = String(n.data?.label || n.id) + const tool = String((n.data?.config as Record | undefined)?.tool ?? '') + return { + value: n.id, label: label === n.id ? label : `${label}（${n.id}）`, + hint: n.data?.nodeType === 'merge' ? NODE_TYPE_LABEL.merge : tool, + } + }) + }, [nodes, edges, nodeId]) +} + interface MergeRow { alias: string; node: string } /** config.inputs（{别名: 节点 id}）→ 行。对象的键顺序就是行的顺序 */ @@ -178,28 +214,7 @@ function MergeInputList({ nodeId, value, issues, onChange }: { const nodes = useStudio((s) => s.nodes) const edges = useStudio((s) => s.edges) const uid = useId() - const options = useMemo(() => { - const parents = new Map() - for (const e of edges) parents.set(e.target, [...(parents.get(e.target) ?? []), e.source]) - const seen = new Set() - const stack = [...(parents.get(nodeId) ?? [])] - while (stack.length) { - const id = stack.pop()! - if (seen.has(id) || id === nodeId) continue - seen.add(id) - stack.push(...(parents.get(id) ?? [])) - } - return nodes - .filter((n) => seen.has(n.id) && isQueryNode(n)) - .map((n) => { - const label = String(n.data?.label || n.id) - const tool = String((n.data?.config as Record | undefined)?.tool ?? '') - return { - value: n.id, label: label === n.id ? label : `${label}（${n.id}）`, - hint: n.data?.nodeType === 'merge' ? NODE_TYPE_LABEL.merge : tool, - } - }) - }, [nodes, edges, nodeId]) + const options = useUpstreamQueries(nodeId) const external = useMemo(() => mergeRowsOf(value), [value]) const [rows, setRows] = useState(external) @@ -295,6 +310,319 @@ function MergeInputList({ nodeId, value, issues, onChange }: { ) } +const AT = ANALYSIS_TEXT.node + +/** 选一个上游查询节点。已经选了、后来删掉或挪到下游的节点照样列出来并说明，好让人看见并改掉 */ +function QueryPick({ id, value, options, onChange }: { + id: string; value: string; options: { value: string; label: string }[]; onChange: (v: string) => void +}) { + const known = options.some((o) => o.value === value) + return ( + + ) +} + +/** + * 分析卡的分析项（engine/schema.analysis_issues 的配置）：每一项一个标识、一种分析方法（有模式的再选模式）、要分析的上游 + * 查询节点（或基期、本期两份查询），再按方法摆参数格（analysisItems.paramSpecs）；可加、比率的变化拆解另有总数核对、下钻。 + * 阈值不在这里写（统一写在阈值覆盖里）。服务端按 items[i].<键> 报的问题落在那一格下面，定位按 [data-item] [data-sub] 落进来 + */ +function AnalysisItemList({ nodeId, value, issues, onChange }: { + nodeId: string; value: any; issues: FieldIssue[]; onChange: (v: any) => void +}) { + const uid = useId() + const options = useUpstreamQueries(nodeId) + const items = itemsOf(value) + const commit = (next: AnalysisItem[]) => onChange(next) + const update = (i: number, next: AnalysisItem) => commit(items.map((it, j) => (j === i ? next : it))) + const setParam = (i: number, key: string, v: unknown) => { + const params = { ...(items[i].params ?? {}) } + if (v === undefined) delete params[key] + else params[key] = v + update(i, { ...items[i], params }) + } + const add = () => commit([...items, { id: nextItemId(items), params: {} }]) + return ( +
+ {items.map((item, i) => { + const own = issues.filter((x) => x.at?.index === i) + const at = (sub: string) => own.filter((x) => x.at?.sub === sub || x.at?.sub?.startsWith(`${sub}.`)) + const lines = (sub: string) => at(sub).map((x, k) => ) + const op = isAnalysisOp(item.op) ? item.op : null + const modes = op ? ANALYSIS_MODES[op] : undefined + const params = (item.params && typeof item.params === 'object' ? item.params : {}) as Record + const two = !!op && allowsTwoQueries(op) && ('base_query' in params || 'target_query' in params) + const id = String(item.id ?? '') + const dup = !!id && items.some((other, j) => j !== i && String(other.id ?? '') === id) + const local = !id || !ITEM_ID.test(id) ? AT.idInvalid : dup ? AT.idDuplicate(id) : '' + const tone = own.some((x) => x.level === 'error') || local ? 'error' : own.length ? 'warning' : undefined + const before = items.slice(0, i).filter((it) => it.op === 'decompose' && it.id) + const parent = params.parent && typeof params.parent === 'object' ? params.parent as { of?: string; key?: unknown } : null + const field = (spec: ParamSpec) => { + const fid = `${uid}-${i}-p-${spec.key}` + const label = AT.param[spec.key] ?? spec.key + const text = paramText(spec.kind, params[spec.key]) + const set = (t: string) => setParam(i, spec.key, paramValue(spec.kind, t)) + let control + if (spec.kind === 'basis' || spec.kind === 'grain') { + const opts = spec.kind === 'basis' ? AT.basis : AT.grain + control = ( + + ) + } else if (spec.kind === 'pairs') { + control =