mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:31:58 +00:00
ThearchyHelios
651 lines
38 KiB
Python
651 lines
38 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
from collections import Counter
|
||
from collections.abc import Iterable, Mapping
|
||
from dataclasses import dataclass
|
||
from typing import Any
|
||
|
||
from app.analysis.contract import OP_LABELS, TOOL_NAMES
|
||
from app.core.events import EventType
|
||
from app.engine.issuance import extract_numbers, trace_numbers
|
||
from app.engine.toolcalls import leaked_markup
|
||
from app.tools.analysis import CALL_ERROR_HEAD, REFUSED_CODE
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 跑完之后的复核层。
|
||
#
|
||
# 编排跑完只说明「流程走完了」，不说明「答得对」。库里能查到大量 status=succeeded
|
||
# 的运行，过程中检索降级了、工具报错了、agent 步数用满了，而交到用户手上的答案
|
||
# 对此只字不提——最糟的一次（run dd9927e6）交出去的干脆是一句内部管道文案。
|
||
#
|
||
# 出具体系（issuance.py）本来就是干这个的，但它挂在输出节点的可选契约上，还要配
|
||
# 口径卡节点提供受控指标集；问数据页生成的图基本都是 input → agent → output，
|
||
# 走不到那条路。所以这里补一层轻的：
|
||
#
|
||
# 1. scan() 纯规则扫一遍事件流，零成本。干净的运行到此为止，不调模型。
|
||
# 2. review() 只在扫出信号时调一次模型，重组答案并把异常说破。
|
||
# 3. 复核模型看不到原始数据，所以它写出来的任何新数字都是凭空来的
|
||
# —— check_numbers() 机械拦截，对不上就整条退回原答案。
|
||
# --------------------------------------------------------------------------
|
||
|
||
BROKEN = "broken" # 答案不可信
|
||
DEGRADED = "degraded" # 能用，但有缺口要说明
|
||
BENIGN = "benign" # 记在轨迹里就够了，不值得为它叫一次模型
|
||
|
||
_SEVERITY_RANK = {BENIGN: 0, DEGRADED: 1, BROKEN: 2}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Signal:
|
||
"""一条「这次运行有什么不对」。detail 同时给模型和用户看，所以写人话。"""
|
||
|
||
kind: str
|
||
detail: str
|
||
severity: str = DEGRADED
|
||
#: agent 为什么提前收尾（engine/guards.py：steps / stall / budget_tokens / budget_usd / context）。
|
||
#: 问数据页据此决定给不给「放宽步数重跑」：只有步数用满时放宽步数才对症
|
||
reason: str | None = None
|
||
|
||
def as_dict(self) -> dict[str, str]:
|
||
out = {"kind": self.kind, "detail": self.detail, "severity": self.severity}
|
||
return {**out, "reason": self.reason} if self.reason else out
|
||
|
||
|
||
# warn 日志按 code 分类，而不是靠匹配中文消息串——消息是给人读的，随时会改，
|
||
# 把复核的判断挂在它上面等于给自己埋一颗定时炸弹。各 emit 点带上 code。
|
||
#
|
||
# 没有 code 的（老运行留在库里的事件）一律按 degraded 处理：宁可多说一句，
|
||
# 也好过把真问题当噪音吞掉。
|
||
WARN_CODES: dict[str, tuple[str, str]] = {
|
||
# 对答案内容零影响，轨迹里有就够了
|
||
"stream_fallback": ("stream_fallback", BENIGN),
|
||
# 安全属性上的降级，不是答案内容的问题；轨迹里已经说清楚了
|
||
"isolation_fallback": ("isolation_fallback", BENIGN),
|
||
# 答案本身就没写完
|
||
"step_limit": ("step_limit", BROKEN),
|
||
"empty_completion": ("empty_completion", BROKEN),
|
||
# 能用但有缺口
|
||
# 步数用尽，但收尾轮让模型基于已查到的部分收了口：答案是完整的一段话，
|
||
# 只是覆盖面不全——这是缺口不是断裂，所以不判 BROKEN。
|
||
# 连带效果是不再触发自动重跑，那正是想要的：同样配置再跑一遍还撞同一堵墙
|
||
"step_limit_settled": ("step_limit_settled", DEGRADED),
|
||
# 收尾轮自己挂了。影响由同时发出的 step_limit(BROKEN) 承担，这条是说清原因
|
||
"settle_failed": ("settle_failed", DEGRADED),
|
||
"loop_limit": ("loop_limit", DEGRADED),
|
||
"tool_args_fixed": ("tool_args_fixed", DEGRADED),
|
||
"node_retry": ("node_retry", DEGRADED),
|
||
"validate_retry": ("validate_retry", DEGRADED),
|
||
"sandbox_reset": ("sandbox_reset", DEGRADED),
|
||
"retrieve_degraded": ("retrieve_degraded", DEGRADED),
|
||
# 模型把工具调用写成了文字。纠正一次好了的，答案是真查过的，只记缺口；
|
||
# 没好的那次节点已经判失败，node_failed 会按 BROKEN 算
|
||
"tool_markup_leak": ("tool_markup_leak", DEGRADED),
|
||
# 协作团队用完轮数、按降档交付：交来的是成员原话，不是调度者认可的结论
|
||
"team_exhausted": ("team_exhausted", BROKEN),
|
||
# 调度者收了工，但最后派出去的成员失败了，成果取自更早的那份：有缺口，不是断裂
|
||
"team_last_failed": ("team_last_failed", DEGRADED),
|
||
# 校验节点的修复编出了原文没有的值，这次修复被作废
|
||
"repair_invented": ("repair_invented", DEGRADED),
|
||
# 报告撰写节点：写作者写了裸数字或引用了不存在的指标，被要求重写。重写好了的
|
||
# 答案是核对过的，只记缺口；没好的那次按 on_violation 失败或把违规标在报告里
|
||
"report_repair": ("report_repair", DEGRADED),
|
||
# 报告撰写节点的上游没有可引用的口径卡指标：报告里的数都没有出处
|
||
"report_no_evidence": ("report_no_evidence", DEGRADED),
|
||
# 证据目录超出篇幅，有的查询只列了编号和摘要：写作模型没看全证据，报告的覆盖面有缺口
|
||
"report_catalog_trimmed": ("report_catalog_trimmed", DEGRADED),
|
||
# 目录有省略时写了「没有查到」这类缺失断言、重写后仍在：报告照常产出并标注，这几句结论依据不足
|
||
"report_absence_claim": ("report_absence_claim", DEGRADED),
|
||
# 报告直接引用了取自遮罩列的格子、重写后仍在（「违规处置」为标注）：照常产出并标注，另记一条缺口（有出具契约的
|
||
# 成果按缺口降档）
|
||
"report_masked_cell": ("report_masked_cell", DEGRADED),
|
||
# 口径卡的指标集、报告文档没能落进工件库：答案照常，只是点不开证据
|
||
"evidence_store_failed": ("evidence_store_failed", DEGRADED),
|
||
# agent 的结构化字段：模型报的值和查询快照对不上（已按快照取值），或者出处核对不了
|
||
# （记成空值，不兜底成 0）。答案能用，缺口要说清是哪一种
|
||
"agent_field_mismatch": ("agent_field_mismatch", DEGRADED),
|
||
"agent_field_unverified": ("agent_field_unverified", DEGRADED),
|
||
# 结构化抽取两次都被截断（输出额度在推理阶段就用完了）：字段都记空值，答案能用，缺口要说清是额度问题
|
||
"agent_field_truncated": ("agent_field_truncated", DEGRADED),
|
||
# 结论句裁判（claims: judge）：触顶、没跑成的句子记为未裁判，出具不能判完整——是缺口，要说明
|
||
"judge_limit": ("judge_limit", DEGRADED),
|
||
"judge_failed": ("judge_failed", DEGRADED),
|
||
# 裁判模型的输出到了上限被截断（推理模型的思考也算在额度里）：同 judge_failed 是缺口，记成它自己的 kind——说得清
|
||
# 是输出额度的问题，该换的是模型
|
||
"judge_truncated": ("judge_truncated", DEGRADED),
|
||
# 证据不支持的句子交回写作者改过一次（和 report_repair 一样只记缺口）；改写稿冒出新问题没采用的，
|
||
# 原稿里那几句仍是证据不支持
|
||
"report_rewrite": ("report_rewrite", DEGRADED),
|
||
"report_rewrite_rejected": ("report_rewrite_rejected", DEGRADED),
|
||
# 估不出裁判的金额、裁判和写作是同一个模型：配置上的提醒，答案本身没受影响
|
||
"judge_unpriced": ("judge_unpriced", BENIGN),
|
||
"judge_same_model": ("judge_same_model", BENIGN),
|
||
# 双盲复算、反方审稿的核查模型和被核查的模型相同：同 judge_same_model，只关乎独立性，结论本身没受影响
|
||
"dual_check_same_model": ("dual_check_same_model", BENIGN),
|
||
"challenge_same_model": ("challenge_same_model", BENIGN),
|
||
# 核查 agent（engine/verify_agent.py，双盲复算、反方审稿共用）估不出金额：同 judge_unpriced，配置上的提醒
|
||
"verify_unpriced": ("verify_unpriced", BENIGN),
|
||
# 核查没做完、没通过：盲算触顶、两路不一致、指标无法复算；反方审稿触顶、推翻了结论、有句子没能反查、只查了
|
||
# 部分数据源。答案照常交付（出具契约另按缺口降档），是缺口不是断裂，按各自的 kind 记，界面说得清是哪一种
|
||
"dual_check_limit": ("dual_check_limit", DEGRADED),
|
||
"dual_check_disagreed": ("dual_check_disagreed", DEGRADED),
|
||
"dual_check_unchecked": ("dual_check_unchecked", DEGRADED),
|
||
"challenge_limit": ("challenge_limit", DEGRADED),
|
||
"challenge_refuted": ("challenge_refuted", DEGRADED),
|
||
"challenge_unreviewed": ("challenge_unreviewed", DEGRADED),
|
||
"challenge_sources_lost": ("challenge_sources_lost", DEGRADED),
|
||
# 核查 agent 自己触顶、按已有结果收尾：多半和节点自己的 *_limit 说的是同一次触顶，照样是缺口，记成它自己的 kind，
|
||
# 不落成笼统的 warn——步数上限（max_steps）只有它说
|
||
"verify_limit": ("verify_limit", DEGRADED),
|
||
# 核查 agent 的输出到了上限被截断（推理模型的思考也算在额度里）：没交成结果，由节点按无法复算、未能反查记缺口。
|
||
# 同 verify_limit，是缺口不是断裂，记成它自己的 kind：说得清是输出额度的问题，换个模型就好
|
||
"verify_truncated": ("verify_truncated", DEGRADED),
|
||
# 查询区间里这张表没有数据（data/coverage.py）：查询结果整列是 0，口径卡的指标取自这些列。答案照常交付（出具契约另按
|
||
# 缺口降档），是缺口不是断裂；但换一张有这段数据的表再建一次大概率就好，所以能给出重来的指示（REBUILD）
|
||
"query_no_data": ("query_no_data", DEGRADED),
|
||
"metric_no_data": ("metric_no_data", DEGRADED),
|
||
# 分析能力一轮（00-总方案 §3.7）：只写说明、不自动重来（NOTE_ONLY）。答案照常交付，缺口照实说一句就够了：
|
||
# 报告里判断词没挂对分析结论、原因没有依据，重写后仍有（check 标在报告里，strict 另记出具缺口）
|
||
"report_analysis_claims": ("report_analysis_claims", DEGRADED),
|
||
# 一份分析算不了（历史不够、本期不完整、截断……）：算不了本身是一条结论，写作者照实写原因。一次运行里几次算不了的
|
||
# 并成一条「有 n 次分析没做成」（_analysis_signal），不一次一条
|
||
REFUSED_CODE: (REFUSED_CODE, DEGRADED),
|
||
# 口径卡上用到两次查询的指标没做成可比性检查（探查失败、算不了）：指标照常算，两期能不能直接比要人自己核对
|
||
"metric_comparability_skipped": ("metric_comparability_skipped", DEGRADED),
|
||
}
|
||
#: 分析调用写错了、后来同一个分析工具一次都没做成（tool.error 带 call_error，见 scan）：这几次分析没做成，同 analysis_refused
|
||
#: 是缺口不是断裂。它不是日志代码，是 scan 从工具事件里认出来的信号种类
|
||
ANALYSIS_CALL_FAILED = "analysis_call_failed"
|
||
|
||
# 这些 kind 重跑一次大概率能好：原因说得清，且不是「问题本身没问明白」。
|
||
# 工具参数错了、步数不够、图没校验过，都属于改一下再来一遍就行的。
|
||
RETRYABLE = {"step_limit", "tool_error", "empty_completion", "node_failed"}
|
||
#: 是缺口（DEGRADED）、却值得重建一次的信号：查询区间里这张表没有数据——图选错了表，换一张有这段数据的表大概率就好。
|
||
#: 答案照常交付，复核照样可以给出「按复核重来」的指示（should_retry、chat/turn.retry_after_review 都认）
|
||
REBUILD = {"query_no_data", "metric_no_data"}
|
||
|
||
#: 只写说明、不自动重来的信号（00-总方案 §3.7）：都是缺口（DEGRADED）、不进 RETRYABLE / REBUILD。一次运行的信号全是
|
||
#: 这几种时复核不叫模型，说明就照信号原话写（review）：答案照常、报告里已经标出，叫一次模型只是把同一句话换个说法
|
||
NOTE_ONLY = frozenset({"report_analysis_claims", REFUSED_CODE, "metric_comparability_skipped", ANALYSIS_CALL_FAILED})
|
||
#: 「有 n 次分析没做成」那条信号里最多点名几次，其余写「……」
|
||
_ANALYSIS_SHOWN = 3
|
||
|
||
#: 串行工具模式下塞给未执行调用的占位说明。它漏到用户面前当过答案，
|
||
#: 所以复核要认得出来。从 nodes.llm 导入会绕成循环，这里按内容识别。
|
||
_PLACEHOLDER_MARKS = ("本轮只执行了第一个工具",)
|
||
|
||
|
||
def answer_text(output: Mapping[str, Any] | None) -> str:
|
||
"""把 output 各字段拼成答案正文。和前端 answerText 同一套规则。"""
|
||
if not output:
|
||
return ""
|
||
parts = [
|
||
value if isinstance(value, str) else json.dumps(value, ensure_ascii=False)
|
||
for key, value in output.items()
|
||
if not key.startswith("_") and value not in (None, "")
|
||
]
|
||
return "\n".join(parts).strip()
|
||
|
||
|
||
def _event_field(event: Any, name: str, default: Any = None) -> Any:
|
||
"""事件既可能是 ORM 行也可能是 dict，两种都要吃得下。"""
|
||
if isinstance(event, Mapping):
|
||
return event.get(name, default)
|
||
return getattr(event, name, default)
|
||
|
||
|
||
def scan(events: Iterable[Any], output: Mapping[str, Any] | None = None) -> list[Signal]:
|
||
"""纯规则扫一遍运行事件，挑出会影响答案可信度的那些。
|
||
|
||
不调模型、不碰网络。返回空列表就意味着这次运行干净，可以直接交付。
|
||
"""
|
||
signals: list[Signal] = []
|
||
seen: set[tuple[str, str]] = set()
|
||
|
||
def add(kind: str, detail: str, severity: str, reason: str | None = None) -> None:
|
||
if severity == BENIGN:
|
||
return
|
||
key = (kind, detail)
|
||
if key in seen:
|
||
return
|
||
seen.add(key)
|
||
kinds.add(kind)
|
||
signals.append(Signal(kind, detail, severity, reason))
|
||
|
||
kinds: set[str] = set()
|
||
|
||
rows = list(events)
|
||
|
||
# 工具报错之后，同一个工具在后面又跑成了，说明自愈接住了——那是缺口不是断裂。
|
||
# 不区分的话，参数自动纠正这条路每修一次就报一次「答案不可信」。
|
||
recovered: set[str] = set()
|
||
failed_at: dict[str, int] = {}
|
||
#: 节点名：node.started 带着画布上的名字，报失败时写名字，不写节点 id
|
||
titles: dict[str, str] = {}
|
||
for i, e in enumerate(rows):
|
||
etype = str(_event_field(e, "type", ""))
|
||
data = _event_field(e, "data") or {}
|
||
if etype == EventType.NODE_STARTED and data.get("label"):
|
||
titles.setdefault(str(_event_field(e, "node_id") or ""), str(data["label"]))
|
||
tool = str(data.get("tool") or "")
|
||
if etype == EventType.TOOL_ERROR and tool:
|
||
failed_at.setdefault(tool, i)
|
||
elif etype == EventType.TOOL_END and tool and tool in failed_at:
|
||
recovered.add(tool)
|
||
|
||
#: 每个报告节点最后一次 report.checked：「改过一次」的信号按最终结果定档（_settled）
|
||
finals = _final_checks(rows)
|
||
#: 报过的节点失败 (节点 id, 原话)：子工作流节点跟着子图里那一步失败时认得出是同一件事
|
||
failures: list[tuple[str, str]] = []
|
||
#: 没做成的分析：算不了的（analysis_refused 日志）、调用写错了一直没改对的（带 call_error 的 tool.error）。各并成一条信号，
|
||
#: 排在第一次出现的位置（slots：信号种类 → (那时信号清单有多长, 第几条事件)）
|
||
refused: list[tuple[str, Mapping[str, Any]]] = []
|
||
unfixed: list[tuple[str, Mapping[str, Any]]] = []
|
||
slots: dict[str, tuple[int, int]] = {}
|
||
for seq, e in enumerate(rows):
|
||
etype = str(_event_field(e, "type", ""))
|
||
data = _event_field(e, "data") or {}
|
||
node_id = _event_field(e, "node_id") or ""
|
||
|
||
if etype == EventType.TOOL_ERROR and data.get("call_error"):
|
||
# 分析调用写错了（contract.CallError：列不存在、比率当可加、这次执行的分析次数用完……）：不记结论，原话交回 Agent
|
||
# 照着改。后来同一个分析工具做成了，那是它改对了，不是缺口；一直没做成的，那几次分析没做成——缺口不是断裂，
|
||
# 只写说明、不重来（NOTE_ONLY）：重来一遍，Agent 照样可能不做这项分析
|
||
if str(data.get("tool") or "") not in recovered:
|
||
slots.setdefault(ANALYSIS_CALL_FAILED, (len(signals), seq))
|
||
unfixed.append((str(node_id), data))
|
||
|
||
elif etype == EventType.TOOL_ERROR:
|
||
tool = str(data.get("tool") or "工具")
|
||
err = str(data.get("error") or "").strip()[:200]
|
||
if tool in recovered:
|
||
add("tool_args_fixed", f"工具 {tool} 第一次调用失败（{err}），重试后成功", DEGRADED)
|
||
else:
|
||
add("tool_error", f"工具 {tool} 调用失败：{err}", BROKEN)
|
||
|
||
elif etype == EventType.NODE_FAILED and data.get("reason") == "rejected":
|
||
# 人的驳回（engine/compiler 给被驳回的那一步的 node.failed 带 reason=rejected，总方案 X9）：不是故障，不说「执行
|
||
# 失败」；也不进 RETRYABLE——重跑只会把同一件事再问一遍人。答案没走完是缺口，记 DEGRADED（没有成果时
|
||
# empty_output 另按 BROKEN 说）
|
||
err = str(data.get("error") or "").strip()[:200]
|
||
if any(inner.startswith(f"{node_id}/") and said == err for inner, said in failures):
|
||
continue # 子工作流里那一步被驳回、子工作流节点跟着收尾：同一件事，只报那一步（同下面的执行失败）
|
||
failures.append((str(node_id), err))
|
||
add("node_rejected", f"节点「{_step_name(str(node_id), titles)}」被驳回：{err}", DEGRADED)
|
||
|
||
elif etype == EventType.NODE_FAILED:
|
||
err = str(data.get("error") or "").strip()[:200]
|
||
if any(inner.startswith(f"{node_id}/") and said == err for inner, said in failures):
|
||
# 子工作流里那一步失败了，子工作流节点（「sub」）跟着失败、原话相同：同一件事，只报更具体的那一步（上面已报）
|
||
continue
|
||
failures.append((str(node_id), err))
|
||
add("node_failed", f"节点「{_step_name(str(node_id), titles)}」执行失败：{err}", BROKEN)
|
||
|
||
elif etype == EventType.RUN_FAILED:
|
||
err = str(data.get("error") or "").strip()[:200]
|
||
add("run_failed", f"运行失败：{err}", BROKEN)
|
||
|
||
elif etype == EventType.RETRIEVE_END:
|
||
if data.get("degraded") and "retrieve_degraded" not in kinds:
|
||
# knowledge.py 在这条事件**之前**已经把每条具体原因发成了 warn
|
||
#（"重排结果解析不出来"、"多少条向量对不上"）。两边都收就是同一件事
|
||
# 报两遍，一条含糊一条具体——这里只兜没有 warn 的情形
|
||
add("retrieve_degraded", "知识库检索能力降级，可能遗漏表述不同的内容", DEGRADED)
|
||
if not data.get("count"):
|
||
q = str(data.get("query") or "").strip()[:60]
|
||
add("retrieve_empty", f"知识库里没有检索到与「{q}」相关的内容", BROKEN)
|
||
|
||
elif etype == EventType.LOG and data.get("level") == "warn" and data.get("code") == REFUSED_CODE:
|
||
slots.setdefault(REFUSED_CODE, (len(signals), seq))
|
||
refused.append((str(node_id), data))
|
||
|
||
elif etype == EventType.LOG and data.get("level") == "warn":
|
||
message = str(data.get("message") or "").strip()
|
||
code = str(data.get("code") or "")
|
||
kind, severity = WARN_CODES.get(code, ("warn", DEGRADED))
|
||
if code in _SETTLED_BY_REPORT and _settled(code, finals.get(str(node_id))):
|
||
# 首稿没过、改过一次之后最终通过了（同一个报告节点最后一次 report.checked）：不是缺口，不进异常清单。
|
||
# 复核模型被别的信号叫起来时，在 notes() 里照实说「重写后已全部通过」
|
||
continue
|
||
reason = data.get("reason")
|
||
if message:
|
||
add(kind, message[:300], severity, reason if isinstance(reason, str) and reason else None)
|
||
|
||
# 先插靠后的那条（位置一样时先插后出现的），靠前那条的位置就不受影响
|
||
for kind in sorted(slots, key=lambda k: slots[k], reverse=True):
|
||
items = refused if kind == REFUSED_CODE else unfixed
|
||
signals.insert(slots[kind][0], Signal(kind, _analysis_signal(kind, items, titles), DEGRADED))
|
||
kinds.add(kind)
|
||
|
||
text = answer_text(output)
|
||
if not text:
|
||
add("empty_output", "这次运行没有产出任何成果内容", BROKEN)
|
||
elif any(mark in text for mark in _PLACEHOLDER_MARKS):
|
||
add("placeholder_output", "交付的是一句内部流程说明，而不是对问题的回答", BROKEN)
|
||
elif leaked_markup(text):
|
||
# 老运行、或者绕过了节点检查的路径：成果本身就是一段没执行的工具调用标记
|
||
add("markup_output", "交付的是一段未执行的工具调用标记，而不是对问题的回答", BROKEN)
|
||
|
||
return signals
|
||
|
||
|
||
def _step_name(node_id: str, titles: Mapping[str, str]) -> str:
|
||
"""失败的那一步给人看的名字。子工作流里的（「子工作流节点 id/…」）一层层接上 node.started 记的名字：「复核子流程 › 算一下」，
|
||
不写内部 id；没记名字的那一层照写 id 的那一段。"""
|
||
if "/" not in node_id:
|
||
return titles.get(node_id) or node_id
|
||
parts = node_id.split("/")
|
||
return " › ".join(titles.get("/".join(parts[:i + 1])) or parts[i] for i in range(len(parts)))
|
||
|
||
|
||
#: 分析工具名 → 算子（contract.TOOL_NAMES 反过来）：说「波动基线没有做成」，不写工具名
|
||
_TOOL_OPS = {name: op for op, name in TOOL_NAMES.items()}
|
||
|
||
|
||
def _analysis_who(node_id: str, data: Mapping[str, Any], titles: Mapping[str, str]) -> str | None:
|
||
"""没做成的那次分析是谁做的、哪一种：「「周报分析」的分析项「wow」（波动基线）」「「数据查询」的变化拆解」「「分析团队」的
|
||
成员「取数员」的趋势判断」。认不出是哪一种分析（老运行、事件里没带）为 None。"""
|
||
op = data.get("op") if data.get("op") in OP_LABELS else _TOOL_OPS.get(str(data.get("tool") or ""))
|
||
if op is None:
|
||
return None
|
||
where = f"「{_step_name(node_id, titles)}」" if node_id else ""
|
||
if data.get("item"):
|
||
return f"{where}的分析项「{data['item']}」（{OP_LABELS[op]}）"
|
||
member = f"的成员「{data['agent']}」" if data.get("agent") else ""
|
||
return f"{where}{member}的{OP_LABELS[op]}" if where or member else OP_LABELS[op]
|
||
|
||
|
||
def _analysis_why(kind: str, data: Mapping[str, Any]) -> str:
|
||
"""没做成的原因：算不了的照结论里的原因；调用写错的照交回 Agent 的那句（去掉「分析调用有误，没有记成结论：」的开头）。"""
|
||
if kind == REFUSED_CODE:
|
||
return str(data.get("reason") or "").strip() or "原因未知"
|
||
error = str(data.get("error") or "").strip()
|
||
return (error[len(CALL_ERROR_HEAD):] if error.startswith(CALL_ERROR_HEAD) else error).strip() or "原因未知"
|
||
|
||
|
||
def _analysis_signal(kind: str, items: list[tuple[str, Mapping[str, Any]]], titles: Mapping[str, str]) -> str:
|
||
"""一次运行里没做成的分析并成的那一句：一次时照实说哪一项、为什么；几次时先说「有 n 次分析没做成」，再点名前几次。
|
||
认不出是哪一种分析的（老运行的事件）照日志原话。复核模型、只写说明的复核都照这一句说。"""
|
||
told: list[tuple[str | None, str, str]] = []
|
||
for node_id, data in items:
|
||
told.append((_analysis_who(node_id, data, titles), _analysis_why(kind, data),
|
||
str(data.get("message") or data.get("error") or "").strip()))
|
||
verb = "没有做成" if kind == REFUSED_CODE else "的调用写错了、一直没改对，没有做成"
|
||
if len(told) == 1:
|
||
who, why, raw = told[0]
|
||
return (f"{who}{verb}：{why}" if who else raw or f"有 1 次分析没有做成：{why}")[:300]
|
||
head = f"有 {len(told)} 次分析没做成" if kind == REFUSED_CODE else f"有 {len(told)} 次分析调用写错了、一直没改对，没有做成"
|
||
named = "；".join(f"{who}：{why}" if who else raw or why for who, why, raw in told[:_ANALYSIS_SHOWN])
|
||
more = "；……" if len(told) > _ANALYSIS_SHOWN else ""
|
||
return f"{head}。{named}{more}"[:300]
|
||
|
||
|
||
#: 报告节点「改过一次」的信号：按同一个节点最后一次 report.checked 的结果定档（最终通过的不是缺口）
|
||
_SETTLED_BY_REPORT = ("report_repair", "report_rewrite")
|
||
#: report.checked 最多带几条违规（报告节点截到前 20 条）：带满了的，后面可能还有，不当作只剩标注
|
||
_CHECKED_CAP = 20
|
||
#: 证据不支持的几种判定（engine/judge 的 CONTRADICTED 加证据不足）：最终还剩这些，改写就没有改好
|
||
_UNSUPPORTED = ("contradicted", "unsupported", "insufficient")
|
||
|
||
|
||
def _final_checks(rows: list[Any]) -> dict[str, dict[str, Any]]:
|
||
"""每个报告节点最后一次 report.checked 的内容（节点 id → data）。"""
|
||
out: dict[str, dict[str, Any]] = {}
|
||
for e in rows:
|
||
if str(_event_field(e, "type", "")) == EventType.REPORT_CHECKED:
|
||
data = _event_field(e, "data") or {}
|
||
out[str(_event_field(e, "node_id") or "")] = data if isinstance(data, dict) else {}
|
||
return out
|
||
|
||
|
||
def _annotations(final: Mapping[str, Any]) -> list[Mapping[str, Any]] | None:
|
||
"""最后一次核对里剩下的违规都是只重写一次、留在报告里标注的那几类时，返回它们（没剩违规是空列表）；还有别的违规返回
|
||
None。标注的是：缺失断言（absence_claim，另发 report_absence_claim）、check 下的分析说法（engine/analysis_claims，另发
|
||
report_analysis_claims）——各有自己的信号说清楚，「改过一次」是不是改好了不看它们。strict 下的分析说法是出具缺口，
|
||
不算标注。"""
|
||
from app.engine.analysis_claims import CODES as ANALYSIS_CODES
|
||
|
||
left = [v for v in final.get("violations") or [] if isinstance(v, Mapping)]
|
||
marked = {"absence_claim", *(ANALYSIS_CODES if final.get("analysis_claims") == "check" else ())}
|
||
if final.get("ok") is True and not left:
|
||
return []
|
||
if not left or len(left) >= _CHECKED_CAP or any(v.get("code") not in marked for v in left):
|
||
return None
|
||
return left
|
||
|
||
|
||
def _settled(code: str, final: Mapping[str, Any] | None) -> bool:
|
||
"""「改过一次」之后最终是不是好了：report_repair 看最后一次核对有没有违规（ok、violations 为空），只剩标注
|
||
（_annotations）的也算好了；report_rewrite 看最后的裁判摘要里还有没有证据不支持的句子。找不到最终结果（老运行、
|
||
节点失败没写出 report.checked）的当没好。"""
|
||
if not final:
|
||
return False
|
||
if code == "report_repair":
|
||
return _annotations(final) is not None
|
||
counts = ((final.get("judge") or {}).get("counts") or {}) if isinstance(final.get("judge"), dict) else {}
|
||
return bool(counts) and not any(int(counts.get(k) or 0) for k in _UNSUPPORTED)
|
||
|
||
|
||
def notes(events: Iterable[Any]) -> list[str]:
|
||
"""不是异常、但复核模型被别的信号叫起来时该知道的事（review_prompt 的「已经处理好的」）：报告首稿没过核对、重写后
|
||
已全部通过（只剩标注的照实说剩下几处，那几处各有自己的信号）；裁判认为证据不支持的句子改写后都有了依据。只说处理好
|
||
了的，没好的照旧是信号（scan）。"""
|
||
rows = list(events)
|
||
finals = _final_checks(rows)
|
||
titles: dict[str, str] = {}
|
||
first: dict[tuple[str, str], dict[str, Any]] = {}
|
||
times: dict[tuple[str, str], int] = {}
|
||
for e in rows:
|
||
etype = str(_event_field(e, "type", ""))
|
||
data = _event_field(e, "data") or {}
|
||
node_id = str(_event_field(e, "node_id") or "")
|
||
if etype == EventType.NODE_STARTED and data.get("label"):
|
||
titles.setdefault(node_id, str(data["label"]))
|
||
if etype == EventType.LOG and data.get("code") in _SETTLED_BY_REPORT:
|
||
key = (node_id, str(data["code"]))
|
||
first.setdefault(key, data)
|
||
times[key] = times.get(key, 0) + 1
|
||
out: list[str] = []
|
||
for (node_id, code), data in first.items():
|
||
if not _settled(code, finals.get(node_id)):
|
||
continue
|
||
name = f"报告「{titles.get(node_id) or node_id}」"
|
||
count = data.get("count")
|
||
if code == "report_repair":
|
||
head = f"首稿有 {count} 处没过核对" if isinstance(count, int) and count > 0 else "首稿没过核对"
|
||
again = f"重写 {times[(node_id, code)]} 次后" if times[(node_id, code)] > 1 else "重写后"
|
||
left = _annotations(finals.get(node_id) or {}) or []
|
||
if not left:
|
||
out.append(f"{name}{head}，{again}已全部通过")
|
||
elif isinstance(count, int) and count > len(left):
|
||
said = {"absence_claim": "缺失断言"}
|
||
kinds = Counter(said.get(str(v.get("code")), "分析说法") for v in left)
|
||
rest = "、".join(f"{n} 处{kind}" for kind, n in kinds.items())
|
||
out.append(f"{name}{head}，{again}只剩已标出的 {rest}")
|
||
else:
|
||
head = f"有 {count} 句结论裁判认为证据不支持" if isinstance(count, int) and count > 0 \
|
||
else "有几句结论裁判认为证据不支持"
|
||
out.append(f"{name}{head}，改写后都有了依据")
|
||
return out
|
||
|
||
|
||
def worst(signals: Iterable[Signal]) -> str:
|
||
"""一组信号里最重的那档。没有信号时返回空串。"""
|
||
ranked = sorted(signals, key=lambda s: _SEVERITY_RANK.get(s.severity, 1), reverse=True)
|
||
return ranked[0].severity if ranked else ""
|
||
|
||
|
||
def should_retry(signals: Iterable[Signal]) -> bool:
|
||
"""这些信号里有没有「重跑一次大概率能好」的：严重的那几类（RETRYABLE），或者值得重建一次的缺口（REBUILD）。"""
|
||
return any((s.kind in RETRYABLE and s.severity == BROKEN) or s.kind in REBUILD for s in signals)
|
||
|
||
|
||
def check_numbers(original: str, revised: str) -> list[str]:
|
||
"""复核重写的答案里，有哪些数字是原答案里没有的。
|
||
|
||
复核模型手上只有原答案和一份异常清单，没有任何原始数据——它写出来的新数字
|
||
只能是凭空来的。这条是硬拦截，不是提示：叙述层无算术权限是这个项目的地基
|
||
（见 issuance.decide_tier）。
|
||
|
||
直接复用出具校验那套回指逻辑，把原答案的数字当成受控指标集。
|
||
"""
|
||
controlled = [
|
||
{"id": tok.raw, "value": tok.value} for tok in extract_numbers(original)
|
||
]
|
||
report = trace_numbers(revised, controlled)
|
||
return [item["token"] for item in report.unmatched]
|
||
|
||
|
||
REVIEW_SYSTEM = (
|
||
"你负责复核一次数据问答的结果。执行过程中检测到了异常，你的任务是让用户"
|
||
"在读这个答案之前就知道它哪里不可靠。\n\n"
|
||
"你会拿到：用户的问题、系统给出的原答案、这次运行的异常清单。\n\n"
|
||
"铁律：\n"
|
||
"1. 你看不到任何原始数据。**不许写出原答案里没有的数字**——一个都不行。"
|
||
"需要改写时只能重组原答案已有的内容。\n"
|
||
"2. 异常必须写进 note，不许淡化或略过。降级要明说，不能悄悄地降。\n"
|
||
"3. 如果异常说明这个答案根本不可信（比如取数工具就没跑通、检索什么都没查到），"
|
||
"就直说答不了、说清缺什么，不要拿一个看起来完整的答案去糊。\n"
|
||
"4. note 写给用户看，一到两句中文，说人话：出了什么事、对这个答案意味着什么。"
|
||
"不要复述事件名或代码。\n\n"
|
||
"输出 JSON：\n"
|
||
'- note：给用户的异常说明（必填）\n'
|
||
'- answer：重写后的答案。原答案本身没问题、只是需要补一句说明时，留空字符串\n'
|
||
'- retry：这次运行是不是「换个配置重跑一遍大概率就好了」（比如步数不够、'
|
||
"工具参数写错）。问题本身没问清楚导致的，填 false"
|
||
)
|
||
|
||
REVIEW_SCHEMA: dict[str, Any] = {
|
||
"title": "review",
|
||
"type": "object",
|
||
"properties": {
|
||
"note": {"type": "string", "description": "给用户的异常说明，一到两句中文"},
|
||
"answer": {"type": "string", "description": "重写后的答案；不需要改写就留空"},
|
||
"retry": {"type": "boolean", "description": "重跑一次是否大概率能好"},
|
||
},
|
||
"required": ["note"],
|
||
}
|
||
|
||
|
||
#: 答案带着逐段证据时追加给复核模型的话
|
||
LOCKED_NOTE = (
|
||
"\n\n这份答案带有逐段证据（报告里的每个数字都对应着核对过的出处），不能改写："
|
||
"answer 必须留空，只在 note 里说明异常。"
|
||
)
|
||
|
||
|
||
def review_prompt(question: str, answer: str, signals: list[Signal], *, locked: bool = False,
|
||
settled: Iterable[str] = ()) -> str:
|
||
"""拼给复核模型的请求。
|
||
|
||
只给信号清单，不给原始事件流：几百条事件塞进 prompt 既贵又没用，而信号
|
||
已经是规则层提炼过的结论。settled 是已经处理好的（notes()：首稿没过、重写后已全部通过……）：
|
||
照实告诉复核模型，免得它把「改过一次」写成「仍未全部通过、可能不准」。
|
||
"""
|
||
lines = "\n".join(
|
||
f"- [{'严重' if s.severity == BROKEN else '降级'}] {s.detail}" for s in signals
|
||
)
|
||
done = "\n".join(f"- {n}" for n in settled)
|
||
return (
|
||
f"用户的问题：\n{question or '（未记录）'}\n\n"
|
||
f"系统给出的原答案：\n{answer or '（空）'}\n\n"
|
||
f"检测到的异常：\n{lines}"
|
||
+ (f"\n\n已经处理好的（不是异常，不要写成缺口或「可能不准」）：\n{done}" if done else "")
|
||
+ (LOCKED_NOTE if locked else "")
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class ReviewResult:
|
||
"""复核结论。verdict 是对「系统实际做了什么」的陈述，不是模型的自我评价。"""
|
||
|
||
verdict: str # ok | annotated | rewritten | reverted | failed
|
||
note: str = ""
|
||
answer: str | None = None # None 表示沿用原答案
|
||
#: 改写之前的那份。改写是有损的，原件得跟着一起落库——否则刷新之后
|
||
#: 只剩一个被重组过的答案，想对照都没得对照
|
||
original: str | None = None
|
||
retry: bool = False
|
||
signals: list[Signal] = None # type: ignore[assignment]
|
||
severity: str = ""
|
||
|
||
def as_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"verdict": self.verdict,
|
||
"note": self.note,
|
||
"answer": self.answer,
|
||
"original": self.original,
|
||
"retry": self.retry,
|
||
"severity": self.severity,
|
||
"signals": [s.as_dict() for s in (self.signals or [])],
|
||
}
|
||
|
||
|
||
async def review(
|
||
model: Any, *, question: str, answer: str, signals: list[Signal], locked: bool = False,
|
||
settled: Iterable[str] = (),
|
||
) -> ReviewResult:
|
||
"""叫一次模型复核。调用方负责确保 signals 非空——空信号根本不该走到这里。
|
||
|
||
复核本身失败不能让整轮挂掉：答案已经在手上了，复核只是加一层说明。
|
||
所以这里把异常全兜住，最差也要把原答案原样交出去。
|
||
|
||
locked：成果带 _evidence（报告撰写节点产出、逐段对应证据文档）。这种答案只能
|
||
加说明、不能改写——改写之后界面上点开的出处，指向的是一段已经不存在的文字。
|
||
"""
|
||
severity = worst(signals)
|
||
rule_retry = should_retry(signals)
|
||
fallback_note = "；".join(s.detail for s in signals if s.severity == BROKEN) or "；".join(
|
||
s.detail for s in signals
|
||
)
|
||
if signals and all(s.kind in NOTE_ONLY for s in signals):
|
||
# 只写说明、不自动重来（NOTE_ONLY）：不叫模型，照信号原话说明
|
||
return ReviewResult(verdict="annotated", note=fallback_note, retry=False, signals=signals, severity=severity)
|
||
|
||
messages = [("system", REVIEW_SYSTEM),
|
||
("human", review_prompt(question, answer, signals, locked=locked, settled=settled))]
|
||
try:
|
||
raw = await model.with_structured_output(REVIEW_SCHEMA).ainvoke(messages)
|
||
if not isinstance(raw, dict):
|
||
raise TypeError("结构化输出没有返回对象")
|
||
except Exception: # noqa: BLE001 - 不支持结构化输出的模型退回文本解析
|
||
try:
|
||
from app.api.copilot import _extract_json
|
||
from app.engine.state import message_text
|
||
|
||
raw = _extract_json(message_text(await model.ainvoke(messages)))
|
||
except Exception: # noqa: BLE001
|
||
# 模型这条路彻底不通。规则层已经知道出了什么事，照样说得出话
|
||
return ReviewResult(
|
||
verdict="annotated", note=fallback_note, retry=rule_retry,
|
||
signals=signals, severity=severity,
|
||
)
|
||
|
||
note = str(raw.get("note") or "").strip() or fallback_note
|
||
revised = str(raw.get("answer") or "").strip()
|
||
# 规则层管**资格**，模型只有**否决权**。反过来（两边取并）的话，模型就能
|
||
# 越过规则强推重跑——而"检索什么都没查到"这类，重跑一遍还是查不到，
|
||
# 白烧一次建图加一次跑图。模型没表态时按规则走
|
||
retry = rule_retry and bool(raw.get("retry", True))
|
||
|
||
if locked or not revised or revised == answer.strip():
|
||
return ReviewResult(
|
||
verdict="annotated", note=note, retry=retry, signals=signals, severity=severity,
|
||
)
|
||
|
||
invented = check_numbers(answer, revised)
|
||
if invented:
|
||
# 重写里冒出了原答案没有的数字。整条退回——说明还是要给的，
|
||
# 但那份正文一个字都不能用
|
||
return ReviewResult(
|
||
verdict="reverted",
|
||
note=f"{note}\n\n（复核改写里出现了原答案中没有的数字"
|
||
f"（{'、'.join(invented[:5])}），已退回原答案。）",
|
||
retry=retry, signals=signals, severity=severity,
|
||
)
|
||
|
||
return ReviewResult(
|
||
verdict="rewritten", note=note, answer=revised, original=answer, retry=retry,
|
||
signals=signals, severity=severity,
|
||
)
|