合入期 3 反方审稿(后端):报告里的关键结论各派只读 agent 主动找反例,给出经受住、被削弱、被推翻、无法判断和有出处的理由;被推翻的计入出具缺口;审批卡附反方审稿概况

# Conflicts:
#	backend/tests/test_issuance_upstream_gaps.py
This commit is contained in:
ThearchyHelios
2026-10-05 18:08:39 +08:00
7 changed files with 2142 additions and 33 deletions
+12
View File
@@ -270,6 +270,18 @@ NODE_REFERENCE = """\
写 null 表示不限,预算要用户定,拿不准就问;只有受管级别要这样,普通草稿里默认开的 judge 不写预算、不为预算问人);契约用 report_from(受管级别不收 narrative),成果字段只取报告节点的正文;给口径卡
供数的 agent 配 output_schema 和 cite_fields: true,code 节点喂口径卡的必须是真取数(evidence_role: "source"),
否则把计算写进口径卡;llm、agent、协作团队写的文字不能绕过报告节点直接进出口,llm 也不能喂口径卡
- challenge:反方审稿。报告写完后挑出几句关键结论(挂了依据的、带比较或因果的优先),每句派一个只读 agent 站在反方
自己写查询找反例,逐句判经受住、被削弱、被推翻或无法判断,理由引用它自己查到的数据;不改报告。
config: {report, max_claims, max_queries, timeout_s, provider, model, on_refuted, enabled}
**多个来源汇总成的分析报告、要正式出具的报告**加这个节点,串在 report 和 output 之间:report → challenge → output
(有审批时 report → challenge → human → output)。challenge 要在 output 上游:只从 report 拉出一条侧枝接 challenge、
output 不经过它时,降档不会发生。
report 写要审的报告节点 id(上游只有一个 report 时可以不写);on_refuted 是结论被推翻时怎么办:degrade(默认,计入
出具缺口、出具降档)/ flag(只标注,不影响出具)/ fail(报错并停止运行,不要再配 on_error: continue)。max_claims
(每份报告最多审几句)、max_queries(每句最多查询次数)、timeout_s(每句时长上限,秒)、enabled(不写就跟随设置里
「核查」一节的开关)一般都不写,取设置里的默认;provider、model 指定反方模型,不写用设置里的核查模型(建议和写作
模型不同,同一个模型等于自己审自己)。challenge 后面直接接 human 时,审批卡自动附上这份报告的核对和审稿结果;
human 的 message 里引用 {{ nodes.报告节点id.text }},审批人能直接读到正文。只回答一个问题、一次查询的不必加
搭图规则(有数字结论时必须遵守):
1. 取数(tool 节点查库;或者 agent 查库并配 output_schema + cite_fields)→ 口径卡 → report → output
+704 -11
View File
@@ -1,21 +1,714 @@
"""反方审稿节点:让另一个模型站在反方,专挑上游报告的毛病,逐条提出反驳意见。
"""反方审稿节点:报告写完后,挑几句关键结论,各派一个只读 agent 站在反方主动找反例。
目前只是占位:图里放了这个节点也能运行,执行器原样跳过、记一条日志,不改任何状态。
真正的审稿由后续工作包补上;没能回应的反驳通过节点产出里的 issuance_gaps 交给成果节点的
出具契约降档(见 nodes/io.py 的 _upstream_issuance_gaps)。
结论句裁判(engine/judge.py)只看写作者挂的证据,判「说法有没有被证据支持」;可证据本身就可能挑过——只查了含
退款的销售额,只看了一个月。反方审稿换个方向:不看证据够不够,自己写查询去证伪。
1. 选句:报告文档里的结论句按裁判的预筛排好(judge.candidates / claim_priority:挂了依据的、带比较或因果的在前),
取前「最多审几句」句;裁判判成「不是结论」的不审。同一份文档永远选出同样的句子、同样的顺序
2. 反查:每句一个只读 agent(verify_agent.run_verify_agent)。工具只给这句引用的查询所在的数据源(指标按它的输入
取自的查询算,合并查询按它合并的那几个源算);这句没挂依据,就在这份报告引用的数据源里查。任务里给这句结论
和它挂的证据摘录(judge.prepare,数据源遮罩的列照旧不给;摘录里原报告的编号都标成「原报告 Qn」,和 agent 自己
的 Q1 分开),要它先想清楚「如果这句不成立,数据里会看到什么」,写只读查询去找,最后交结论(经受住 / 被削弱 /
被推翻 / 无法判断)和理由。理由里的数字只能引用它自己查询的格子,裸数字由 compose_doc 标出来
3. 出处:一条查询都没执行成功的,判什么都记为无法判断;判「被推翻」的,理由里还得引用到自己的查询结果——它要让
正式出具降档,审稿记录里得查得到凭什么。不采信时系统写明原因,反方原本的结论和理由照录在后面(纯文字:解析
不了的引用写成「(引用无效)」,裸数字注明「(无出处)」)。这句该查的数据源都不存在或已停用、核查模型本身用
不了时不派 agent(模型用不了的,后面的句子照同一个原因记,不再逐句去建模型);只停用了一部分的照常派,理由
末尾由系统注明哪几个没查(写进理由文档,见 with_note)
4. 记录:每句的结论、理由(渲染后的正文和组装好的文档)、查询快照、用量写进审稿记录(challenge_artifact),随运行
封存。查询照常发 tool.end(role: verify,带查询快照,可下钻);台账只记 kind challenge,绝不记 query——核查的
查询进了写作目录,下游的写作者会把它们当成可以引用的证据。有句子没能反查的另记一条 challenge_unreviewed、
只查了部分数据源的另记一条 challenge_sources_lost,不然时间线上只有一句「审了 N 句」,看不出根本没审成、
没审全(这两个代码还不在期 3 约定的日志代码表里,界面认不得时照原话显示)
5. 处置:被推翻的按 on_refuted 处置。degrade(默认)在产出里写 issuance_gaps,成果节点的出具契约照单收进缺口、
出具降档(nodes/io.py 的 _upstream_issuance_gaps,所以反方审稿要接在成果节点上游,画布校验会提醒);fail 判节点
失败;flag 只记录。产出和审批卡里的 on_refuted 是实际生效的那一种(只有 degrade / flag,见 _finish)。审批卡
(human.py 的 _reviewed_reports)附上审稿概况
上限:句数、每句查询次数、每句时长、全局每日金额(核查和裁判各记各的账,engine/spend.py)。都按发起时快照的核查
设置(run.options["verify"]),节点上写了的盖过它:审批停下期间改了设置,恢复、接着跑仍按发起时的那份。没有快照
的运行按设置的当前值。「启用」节点上写了就按节点(写开启时全局关闭也照审),没写跟随设置;不启用时直通,报告没有
可审的结论句、报告文档读不到时也直通(都记一条 challenge_skipped)。
逐句串行反查:并行的话几句的工具调用在时间线上交错,看不出哪条查询在审哪一句;每日余额按先后扣,结果也才确定。
整个反查连同存审稿记录、发日志都在一个 @task 里:接着跑时取 checkpoint 里的结果,不重复查库、不重复记账,时间线
上也不再发一遍。结果带着输入指纹,接着跑时改了审稿本身的配置就重审;同一次执行里的重试沿用已经审完的(见 _reviewed)。
"""
from __future__ import annotations
import re
from collections.abc import Callable, Collection, Iterable
from typing import Any
from app.core.events import EventType
from app.engine.context import NodeContext
from app.engine.state import GraphState
from langgraph.func import task
from sqlalchemy import select
#: 占位期间的跳过原因:节点产出和日志里都写这一句
PENDING_REASON = "反方审稿功能开发中,本节点暂未生效,已跳过"
from app.core import artifact_store
from app.core.artifact_store import canonical_json, content_hash
from app.core.errors import describe_exception
from app.core.events import EventType
from app.db.base import SessionLocal
from app.db.models import DataSource
from app.engine import judge as judging
from app.engine.context import NodeContext, NodeError
from app.engine.evidence import ENTITY_KINDS, iter_units, ledger_enabled, next_exec
from app.engine.judge import _usd
from app.engine.merge_query import MERGE_SOURCE
from app.engine.replay import _in_graph
from app.engine.schema import CHALLENGE_LABEL, CHALLENGE_ON_REFUTED_LABEL, GraphSpec, NodeType, challenge_problems, \
challenge_report
from app.engine.state import GraphState
from app.engine.verify_agent import RATIONALE, VerifyOutcome, run_verify_agent
from app.tools.datasource import run_versions
#: 四种结论:经受住 / 被削弱 / 被推翻 / 无法判断。counts 的键照这个顺序
VERDICTS = ("stands", "weakened", "refuted", "inconclusive")
#: 节点产出、审批卡里最多列几句(审稿记录里一句不少)
ITEMS_MAX = 10
#: 出具缺口、报错里最多点几句的原文,每句最多几个字
_NAMED = 3
_QUOTE_CHARS = 40
#: 审稿记录里每次查询记的几项
_QUERY_KEYS = ("alias", "artifact", "sql", "row_count", "truncated", "source")
#: 每句反查都会遇到的上限:触顶时发 challenge_limit。步数上限(verify_agent 的 max_steps)由核查自己的
#: verify_limit 说明,不在这里另记
_CLAIM_LIMITS = ("max_queries", "timeout_s", "daily_max_usd")
#: 列「审批卡上的几句」时谁先留:被推翻、被削弱、没能判断的比经受住的更要让审批人看到
_RANK = {"refuted": 0, "weakened": 1, "inconclusive": 2, "stands": 3}
#: 系统改判时照录反方原本的结论用的叫法(界面的叫法见 lib/terms.ts 的 CHALLENGE_TEXT.verdict)
_VERDICT_LABEL = {"stands": "经受住", "weakened": "被削弱", "refuted": "被推翻", "inconclusive": "无法判断"}
#: 照录反方的理由时:解析不了的引用换成的字、裸数字后面加的注(见 _said)
_INVALID_REF = "(引用无效)"
_UNSOURCED = "(无出处)"
#: 解析不了的引用在渲染后的正文里的占位(evidence.placeholder),连同两边隔开它的一个空格
_PLACEHOLDER = re.compile(r" ?⟦\?[^⟧]*⟧ ?")
#: 接着跑时 checkpoint 里缓存的审稿最多翻几条去找这一份配置的(每条都是按先前某一份配置审的,对不上就往后翻)
_STALE_LIMIT = 16
#: 同一次执行里审完的结果放在 RunContext.extra 的这个键下:{(节点, 第几次执行, 输入指纹): 审稿结果}。重试沿用它
_REVIEWS_KEY = "challenge_reviews"
CHALLENGE_SYSTEM = (
"你是反方审稿人:报告已经写完,你站在反方,专门找其中一句结论的反例。你不写报告,也不改报告。\n"
"- 先想清楚:如果这句结论不成立,数据里会看到什么;再写只读查询去找这样的反例\n"
"- 换一个角度核对:按别的维度拆开、换时间范围、核对口径和过滤条件(退款、作废、重复记录)、找被忽略的数据,"
"不要只是重复写作者的查询\n"
"- verdict 四选一:stands 认真找过、没有找到反例;weakened 大体成立,但有重要的例外或前提,或者把范围、程度、"
"原因说过了头;refuted 查到的数据和这句结论相矛盾;inconclusive 查不到判断所需的数据\n"
"- 没有找到反例不等于结论被推翻:只有数据和结论冲突时才判 refuted\n"
"- 结论要有出处:一条查询都没执行成功时,判什么都不会被采信;判 refuted 时,理由里至少要引用一处你自己查询的"
"结果([[v:Qn.rX.列名]] 或 [[see:Qn]]),否则同样不会被采信"
)
CHALLENGE_TASK_PROMPT = (
"要审的结论(报告「{report}」中的一句{section}):\n「{claim}」\n\n"
"写作者给这句挂的证据摘录(编号属于原报告;你的查询另起编号,理由里只能引用你自己的查询):\n{evidence}\n\n"
"可以查询的数据源:{sources}\n\n"
"按上面的做法反查,最后交结论(verdict)和理由(rationale):理由说清查了什么、看到了什么,里面的数字一律写成"
"对你自己查询结果的引用,不要直接写数字。"
)
#: 交给核查 agent 的结果结构:结论一项;理由由 verify_agent 自动加上(rationale)
CHALLENGE_SCHEMA: dict[str, Any] = {
"type": "object",
"properties": {"verdict": {
"type": "string", "enum": list(VERDICTS),
"description": "stands 经受住、weakened 被削弱、refuted 被推翻、inconclusive 无法判断",
}},
"required": ["verdict"],
}
# --------------------------------------------------------------------------
# 执行器
# --------------------------------------------------------------------------
async def run_challenge(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
ctx.emit(EventType.LOG, level="info", code="node_pending", message=PENDING_REASON)
return {"nodes": {ctx.node.id: {"skipped": True, "reason": PENDING_REASON}}}
spec, node = ctx.run.spec, ctx.node
# 节点自己的配置直接读 node.config,不走 ctx.cfg:图级 defaults 里的 provider / model 是写作模型的,
# 拿来当反方模型就成了自己审自己
config = node.config
# 画布校验也拦这几项;执行时再守一道,报的是同一句话(跳过校验执行的旧图、改坏了的配置)
if problems := challenge_problems(config):
raise NodeError(node.id, ";".join(message for _, message in problems))
report_id, problem = challenge_report(spec, node)
if report_id is None:
raise NodeError(node.id, problem or "反方审稿找不到要审的报告")
title = spec.node_map()[report_id].title
settings = await _verify_settings(ctx)
part = settings.get("challenge") if isinstance(settings.get("challenge"), dict) else {}
enabled = config.get("enabled")
if not (part.get("enabled", True) if enabled is None else enabled):
from app.api.settings import VERIFY_SETTING_LABEL
return _skip(ctx, "disabled", f"反方审稿已关闭(设置中的「{VERIFY_SETTING_LABEL['challenge.enabled']}」,"
f"或节点的「{CHALLENGE_LABEL['enabled']}」),本节点已跳过")
written = (state.get("nodes") or {}).get(report_id)
written = written if isinstance(written, dict) else {}
doc_artifact = written.get("doc_artifact") if isinstance(written.get("doc_artifact"), str) else None
doc = _load_doc(doc_artifact)
if doc is None or doc_artifact is None:
return _skip(ctx, "no_report", f"报告「{title}」的报告文档无法读取(没有写出,或保存失败),本节点已跳过")
claims = pick_claims(doc)
if not claims:
return _skip(ctx, "no_claims", f"报告「{title}」中没有可审的结论句,本节点已跳过")
max_claims = _limit(config, part, "max_claims")
max_queries = _limit(config, part, "max_queries")
timeout_s = _limit(config, part, "timeout_s")
catalog = doc.get("catalog") if isinstance(doc.get("catalog"), dict) else {}
from app.api.settings import verify_model_spec # 在这里导入:api.settings 导入了 engine 里的模块
async with SessionLocal() as session:
model_spec = await verify_model_spec(session, config, settings=settings)
review = _Review(
ctx, report_id=report_id, title=title, doc_artifact=doc_artifact, doc=doc, catalog=catalog, claims=claims,
max_claims=None if max_claims is None else int(max_claims),
max_queries=None if max_queries is None else int(max_queries),
timeout_s=None if timeout_s is None else float(timeout_s), daily_max_usd=settings.get("daily_max_usd"),
spec=model_spec, masks=await judging.source_masks(catalog),
writer_model=written.get("model") if isinstance(written.get("model"), str) else None,
)
outcome = await _reviewed(ctx, state, review)
return _finish(ctx, state, outcome, on_refuted=str(config.get("on_refuted") or "degrade"))
async def _reviewed(ctx: NodeContext, state: GraphState, review: _Review) -> dict[str, Any]:
"""这一次执行的审稿结果。
- 进 checkpoint:整个反查在一个 @task 里,接着跑时直接取这里的结果,不再查库、不再记账(每日账也记在这里面);
事件都在这里面发,取缓存时不会再发一遍
- 带输入指纹(_Review.key):接着跑时可以带着改过配置的图(runner.continue_failed)。只改了「结论被推翻时」的,
审稿沿用(判失败的报错就是这么建议的);改了要审的报告、反方模型、上限的,缓存里那份是按旧配置审的,对不上就
往后翻、重新审——不然换了报告,审批卡会把对另一份报告的审稿挂上来(做法同 report.py 的裁判)
- 同一次执行里的重试(节点的「失败重试次数」)沿用已经审完的:重试把执行器整个再调一遍,task 成了这次执行里的
又一次调用,不命中缓存。再审一遍既多花一份钱,模型换个说法还会把「被推翻」重审掉,最严的处置反倒放行
"""
reviews = ctx.run.extra.setdefault(_REVIEWS_KEY, {})
slot = (ctx.node.id, next_exec(state, ctx.node.id), review.key)
if slot not in reviews:
reviews[slot] = await _review_once(review)
return reviews[slot]
async def _review_once(review: _Review) -> dict[str, Any]:
if _in_graph():
@task
async def challenge_step(key: str) -> dict[str, Any]:
return {**await review.run(), "key": key}
for _ in range(_STALE_LIMIT):
outcome = await challenge_step(review.key)
if outcome.get("key") == review.key:
return outcome
return await review.run()
def _skip(ctx: NodeContext, reason: str, message: str) -> dict[str, Any]:
"""直通:不改任何状态,记一条 challenge_skipped。审批卡认不出审稿结果(没有 counts),不会附上概况。"""
ctx.emit(EventType.LOG, level="info", code="challenge_skipped", reason=reason, message=message)
return {"nodes": {ctx.node.id: {"skipped": True, "reason": reason}}}
async def _verify_settings(ctx: NodeContext) -> dict[str, Any]:
"""核查设置:发起时快照进运行选项的那份(run.options["verify"]),补齐默认、去掉坏值;没有快照的运行按设置的
当前值。"""
from app.api.settings import verify_settings, verify_values
from app.engine.runner import RUN_OPTIONS_KEY
options = ctx.run.extra.get(RUN_OPTIONS_KEY)
snapshot = options.get("verify") if isinstance(options, dict) else None
if isinstance(snapshot, dict):
return verify_values(snapshot)
async with SessionLocal() as session:
return await verify_settings(session)
def _limit(config: dict[str, Any], part: dict[str, Any], key: str) -> Any:
"""节点上写了的上限盖过设置;留空(不写、写 null)沿用设置,设置里的 null 才是不限。"""
value = config.get(key)
return part.get(key) if value in (None, "") else value
def _load_doc(artifact: str | None) -> dict[str, Any] | None:
if not artifact:
return None
try:
doc = artifact_store.load(artifact)
except ValueError: # 和哈希对不上:被改过的文档不审
return None
return doc if isinstance(doc, dict) else None
def pick_claims(doc: dict[str, Any]) -> list[judging.Candidate]:
"""要审的结论句:裁判预筛排好的候选(judge.candidates),去掉裁判已经判成「不是结论」的。"""
units = {u.get("id"): u for _, u in iter_units(doc)}
out: list[judging.Candidate] = []
for cand in judging.candidates(doc):
verdict = (units.get(cand.unit) or {}).get("verdict")
if isinstance(verdict, dict) and verdict.get("status") == "not_a_claim":
continue
out.append(cand)
return out
# --------------------------------------------------------------------------
# 数据源:这句引用的证据背后是哪几个库
# --------------------------------------------------------------------------
def _fetcher() -> Callable[[Any], Any]:
"""按工件 id 取工件(复验哈希),同一份只读一次;取不到、被改过的当没有。"""
memo: dict[str, Any] = {}
def fetch(artifact: Any) -> Any:
key = str(artifact or "")
if key not in memo:
try:
memo[key] = artifact_store.load(key) if key else None
except Exception: # noqa: BLE001 - 取不回来就当没有:少一个源,不让整个审稿失败
memo[key] = None
return memo[key]
return fetch
def _snapshot_sources(snapshot: Any) -> list[str]:
"""一份查询快照背后的数据源:合并查询是它合并的那几个源,其余是快照记的源。"""
if not isinstance(snapshot, dict):
return []
if snapshot.get("source") == MERGE_SOURCE:
return [str(s) for s in snapshot.get("sources") or [] if s]
return [str(snapshot["source"])] if snapshot.get("source") else []
def claim_sources(aliases: Iterable[str], catalog: dict[str, Any], fetch: Callable[[Any], Any]) -> list[str]:
"""这几件证据背后的数据源(去重保序):查询的、口径卡指标的输入取自的查询的、表和字段的。知识库检索、运行输入
背后没有数据源。认法同裁判找遮罩(judge.source_masks)。"""
out: list[str] = []
for alias in aliases:
entry = catalog.get(alias)
if not isinstance(entry, dict):
continue
kind = entry.get("kind")
found: list[str] = []
if kind == "query":
source = entry.get("source")
found = [str(source)] if source and source != MERGE_SOURCE else _snapshot_sources(fetch(entry.get("artifact")))
elif kind == "metric":
card = fetch(entry.get("artifact"))
wanted = (entry.get("locator") or {}).get("metric")
for metric in (card.get("metrics") if isinstance(card, dict) else None) or []:
if isinstance(metric, dict) and metric.get("id") == wanted:
for item in metric.get("inputs") or []:
if isinstance(item, dict) and item.get("artifact"):
found += _snapshot_sources(fetch(item["artifact"]))
elif kind in ENTITY_KINDS:
source = entry.get("source")
found = [str(source)] if source else next(
(_snapshot_sources(fetch(o.get("artifact"))) for o in entry.get("sources") or []
if isinstance(o, dict) and o.get("kind") == "schema"), [])
out.extend(s for s in found if s not in out)
return out
# --------------------------------------------------------------------------
# 逐句反查
# --------------------------------------------------------------------------
def _zero_usage() -> dict[str, Any]:
return {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0, "cost_usd": 0.0, "calls": 0, "unpriced_calls": 0}
def _add_usage(total: dict[str, Any], more: dict[str, Any]) -> dict[str, Any]:
out = dict(total)
for key, value in (more or {}).items():
if isinstance(value, (int, float)) and not isinstance(value, bool) and key in out:
out[key] += value
out["cost_usd"] = round(out["cost_usd"], 6)
return out
def _claim(cand: judging.Candidate, verdict: str, rationale: str, *, doc: dict[str, Any] | None = None,
queries: Iterable[dict[str, Any]] = (), usage: dict[str, Any] | None = None) -> dict[str, Any]:
"""审稿记录里的一句。rationale 是渲染后的正文;系统替它说的(没有可查的源、没能反查)rationale_doc 为 None。"""
return {"unit": cand.unit, "text": cand.text, "verdict": verdict, "rationale": rationale, "rationale_doc": doc,
"queries": [{k: q.get(k) for k in _QUERY_KEYS} for q in queries], "usage": usage or _zero_usage()}
def _claim_of(cand: judging.Candidate, outcome: VerifyOutcome) -> tuple[dict[str, Any], str | None]:
"""反查 agent 交回的 → (审稿记录里的一句, 这句没能反查的原因)。反查过了的原因是 None。
反方的结论要有出处才采信:一条查询都没执行成功的(没有可用的工具、查询全部失败、根本没查),判什么都记为无法
判断——零查询的「被推翻」能让正式出具降档,零查询的「经受住」会在审批卡上给人假安心。判「被推翻」的,理由里还得
引用到自己的查询结果(resolved 的 [[v:]] / [[see:]]),审稿记录里才查得到凭什么推翻。"""
if outcome.result is None:
# 没交成结果(模型不可用、调用失败、触顶时连收尾都没赶上):查过的照记,结论只能是无法判断
why = outcome.error or "原因不明"
return _claim(cand, "inconclusive", f"反查未完成:{why}", queries=outcome.queries, usage=outcome.usage), why
verdict = str(outcome.result.get("verdict") or "").strip().lower()
verdict = verdict if verdict in VERDICTS else "inconclusive"
rationale = str(outcome.result.get(RATIONALE) or "")
if not outcome.queries:
why = "反查没有执行成功任何查询"
return _unfounded(cand, outcome, verdict, why), why
if verdict == "refuted" and not _cites(outcome.rationale):
return _unfounded(cand, outcome, verdict, "理由没有引用到任何查询结果"), None
return _claim(cand, verdict, rationale, doc=outcome.rationale, queries=outcome.queries, usage=outcome.usage), None
def _cites(doc: dict[str, Any] | None) -> bool:
"""理由文档里有没有一处引用解析到了 agent 自己的查询结果(compose_doc 只把解析成功的引用记进 unit.cites)。"""
return isinstance(doc, dict) and any(unit.get("cites") for _, unit in iter_units(doc))
def _unfounded(cand: judging.Candidate, outcome: VerifyOutcome, verdict: str, why: str) -> dict[str, Any]:
"""没有出处的结论:记为无法判断,系统写明原因,反方原本的结论和理由照录在后面(照录的写法见 _said)。
rationale_doc 为 None:这段话是系统说的,界面照 rationale 显示,不拿反方那份文档冒充理由。查过的照记。"""
taken = f"反方给出的「{_VERDICT_LABEL[verdict]}」不予采信,记为无法判断" if verdict != "inconclusive" else "记为无法判断"
rationale = _said(outcome.rationale, str((outcome.result or {}).get(RATIONALE) or "")).strip()
said = f"反方的理由:{rationale}" if rationale else "反方没有给出理由"
return _claim(cand, "inconclusive", f"{why},{taken}。{said}", queries=outcome.queries, usage=outcome.usage)
def _said(doc: dict[str, Any] | None, rendered: str) -> str:
"""反方的理由照录成纯文字。界面没有文档可画、按文字显示,文档里标出来的东西得写成字:解析不了的引用渲染成内部
的占位(⟦?v:Q1.r9.列⟧),换成「(引用无效)」,连同两边隔开它的空格;裸数字后面注明「(无出处)」。按片段在正文
里的偏移改(片段互不重叠,从后往前改偏移不乱);文档和正文对不上时只换占位。"""
if not isinstance(doc, dict) or doc.get("markdown") != rendered:
return _PLACEHOLDER.sub(_INVALID_REF, rendered)
edits: list[tuple[int, int, str]] = []
for _, unit in iter_units(doc):
for seg in unit.get("segments") or []:
span = seg.get("span")
if not (isinstance(span, list) and len(span) == 2 and all(isinstance(x, int) for x in span)):
continue
if seg.get("issue") == "uncited_number":
edits.append((span[1], span[1], _UNSOURCED))
elif seg.get("issue") == "unresolved_ref" and str(seg.get("text") or "").startswith("⟦?"):
edits.append((span[0], span[1], _INVALID_REF))
text = rendered
for start, end, new in sorted(edits, reverse=True):
text = text[:start] + new + text[end:]
return text.replace(f" {_INVALID_REF}", _INVALID_REF).replace(f"{_INVALID_REF} ", _INVALID_REF)
def with_note(doc: dict[str, Any], note: str) -> dict[str, Any]:
"""理由文档末尾另起一段,写系统的说明(界面按文档画理由,说明只写进 rationale 正文的话看不到)。
照 compose_doc 的样子接:段与段之间的空行是这一句开头的结构片段,说明是一句连接性的话,块、句、片段的编号接着
往下数,统计跟着加——和把说明写进原文再组装出来的一模一样。只是说明不挂出处、不查裸数字(是系统的话,数据源的
名字里也可能有数字),source(反方的原文)不动。传进来的文档不改,返回新的一份。"""
markdown = str(doc.get("markdown") or "")
blocks = list(doc.get("blocks") or [])
units = [u for b in blocks if isinstance(b, dict) for u in b.get("units") or []]
first = sum(len(u.get("segments") or []) for u in units if isinstance(u, dict))
gap = "\n\n" if markdown else ""
start = len(markdown) + len(gap)
pieces = [("structural", gap, len(markdown))] if gap else []
pieces.append(("text", note, start))
segments = [{"kind": kind, "text": text, "state": "neutral", "span": [at, at + len(text)], "id": f"s{first + i}"}
for i, (kind, text, at) in enumerate(pieces)]
unit = {"id": f"u{len(units)}", "kind": "connective", "span": [start, start + len(note)], "cites": [], "see": [],
"segments": segments}
out = {**doc, "markdown": markdown + gap + note,
"blocks": [*blocks, {"type": "paragraph", "units": [unit], "id": f"b{len(blocks)}"}]}
if isinstance(doc.get("stats"), dict):
out["stats"] = stats = dict(doc["stats"])
for key, more in (("units", 1), ("segments", len(segments)), ("connective", 1)):
if isinstance(stats.get(key), int):
stats[key] += more
return out
def _noted(claim: dict[str, Any], note: str) -> dict[str, Any]:
"""审稿记录里的一句,理由末尾加上系统的说明:有理由文档的写进文档(with_note),正文照文档;没有的接在正文后面。"""
doc = claim.get("rationale_doc")
if isinstance(doc, dict):
doc = with_note(doc, note)
return {**claim, "rationale": doc["markdown"], "rationale_doc": doc}
said = str(claim.get("rationale") or "").rstrip()
return {**claim, "rationale": f"{said}\n\n{note}" if said else note}
async def _usable_sources(names: Iterable[str]) -> set[str]:
"""这几个数据源里现在能查的:存在、没有停用(核查建工具时也只认这样的,tools/datasource.build_datasource_tools)。
报告写完到审稿之间源被停用了,这句就没有可查的源:不派 agent,免得白建一次模型、白花一次钱。"""
wanted = list(dict.fromkeys(names))
if not wanted:
return set()
async with SessionLocal() as session:
rows = await session.execute(
select(DataSource.name).where(DataSource.name.in_(wanted), DataSource.enabled.is_(True)))
return {str(name) for name in rows.scalars()}
def mark_original(excerpt: str, alias: str, queries: Collection[str]) -> str:
"""证据摘录里指向原报告编号的写法都标上「原报告」:开头的【编号】,和正文里的「查询 Qn」(指标的输入取自哪次
查询、那次查询的 SQL,表名、字段名出现在哪次查询)。核查 agent 自己的查询也编 Q1、Q2…,不分开的话,它在理由里
引用的 Q1 会被当成原报告那份快照的格子。只认摘录自己的这两种写法、只认这份报告目录里的查询编号:数据里的值
(季度列里的 Q1)不动。"""
text = excerpt.replace(f"【{alias}】", f"【原报告 {alias}】", 1)
if not queries:
return text
names = "|".join(re.escape(q) for q in sorted(queries, key=len, reverse=True))
return re.sub(rf"查询 ({names})(?!\w)", r"原报告 \1", text)
def _quoted(names: Iterable[str]) -> str:
"""「a」、「b」:点名几个数据源。"""
return "、".join(f"「{n}」" for n in names)
def _named(claims: list[dict[str, Any]]) -> str:
"""「「…」、「…」」:点名最多 _NAMED 句,每句截到 _QUOTE_CHARS 个字。"""
names = "、".join(f"「{judging.clip_text(c['text'], _QUOTE_CHARS)}」" for c in claims[:_NAMED])
return names + ("等" if len(claims) > _NAMED else "")
def _refuted_words(claims: list[dict[str, Any]]) -> str:
"""出具缺口、时间线、报错里的那句:「反方审稿推翻了 N 句结论:「…」」。不带节点名:出具契约统一加前缀。"""
return f"反方审稿推翻了 {len(claims)} 句结论:{_named(claims)}"
def pick_items(claims: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""节点产出、审批卡里列的几句 {unit, text, verdict}:最多 ITEMS_MAX 句,照原来的顺序。多出来的时候先留被推翻、
被削弱、没能判断的,再按先后——被推翻的那句不能因为排得靠后就不给审批人看。"""
keep = range(len(claims)) if len(claims) <= ITEMS_MAX else \
sorted(sorted(range(len(claims)), key=lambda i: (_RANK.get(claims[i]["verdict"], len(_RANK)), i))[:ITEMS_MAX])
return [{k: claims[i][k] for k in ("unit", "text", "verdict")} for i in keep]
class _Review:
"""一次审稿:逐句派核查 agent,写审稿记录,发日志。run() 的返回值进 checkpoint,只放节点产出用得到的。
key 是这次审稿的输入指纹:审哪份报告的哪一稿、反方模型、四个上限、数据源现在的遮罩(摘录按它遮)。「结论被推翻时」
不在里面:它只管审完之后怎么处置,改了它接着跑不必重审。"""
def __init__(self, ctx: NodeContext, *, report_id: str, title: str, doc_artifact: str, doc: dict[str, Any],
catalog: dict[str, Any], claims: list[judging.Candidate], max_claims: int | None,
max_queries: int | None, timeout_s: float | None, daily_max_usd: float | None, spec: Any,
masks: dict[str, Any], writer_model: str | None) -> None:
self.ctx = ctx
self.report_id, self.title, self.doc_artifact, self.doc, self.catalog = \
report_id, title, doc_artifact, doc, catalog
self.total = len(claims)
self.picked = claims if max_claims is None else claims[:max_claims]
self.max_claims, self.max_queries, self.timeout_s, self.daily_max_usd = \
max_claims, max_queries, timeout_s, daily_max_usd
self.spec, self.masks, self.writer_model = spec, masks, writer_model
self.fetch = _fetcher()
#: 这份报告目录里的查询编号:摘录里指向它们的写法标上「原报告」(mark_original)
self.queries = {a for a, e in catalog.items() if isinstance(e, dict) and e.get("kind") == "query"}
self.key = content_hash(canonical_json({
"report": report_id, "doc": doc_artifact, "model": [spec.provider, spec.model],
"limits": [max_claims, max_queries, timeout_s, daily_max_usd], "masks": masks,
}))
def _limit_words(self, limit: str) -> str:
if limit == "max_claims":
return f"每份报告最多审 {self.max_claims} 句"
if limit == "max_queries":
return f"每句最多查询 {self.max_queries} 次"
if limit == "timeout_s":
return f"每句时长上限 {self.timeout_s:g} 秒"
return f"全局每日核查金额上限 {_usd(self.daily_max_usd)}"
def _capped(self) -> str:
"""当天的核查金额用完、这句一次模型都没调用时的理由:派出去就撞上的和后面不再派出的,是同一回事。"""
return f"已达上限({self._limit_words('daily_max_usd')}),这句未反查"
def _task(self, cand: judging.Candidate, sources: list[str], lost: list[str], *, own: bool) -> str:
"""交给核查 agent 的任务:这句结论、它挂的证据摘录(遮罩照旧,原报告的编号标明,见 mark_original)、可以查的
数据源;该查的源有停用了的,也告诉它查不了哪几个,免得它把查不到当成没有反例。own:源是这句自己引用的,
不是没挂依据、按整份报告用到的源查的。"""
request = judging.prepare(self.doc, self.catalog, units=[cand.unit], loader=self.fetch, masked=self.masks)
aliases = dict.fromkeys(a for c in request.cands for a in c.aliases if a in request.excerpts)
evidence = "\n\n".join(mark_original(request.excerpts[a], a, self.queries) for a in aliases)
gone = f"({'这句还引用了' if own else '这份报告还用到了'}{_quoted(lost)},已不存在或已停用,无法查询)" \
if lost else ""
return CHALLENGE_TASK_PROMPT.format(
report=self.title, section=f",所在小节「{cand.section}」" if cand.section else "", claim=cand.text,
evidence=evidence or "(这句没有挂依据)", sources="、".join(sources) + gone)
async def run(self) -> dict[str, Any]:
ctx = self.ctx
if len(self.picked) < self.total:
skipped = self.total - len(self.picked)
ctx.emit(EventType.LOG, level="warn", code="challenge_limit", limit="max_claims", claims_skipped=skipped,
message=f"报告「{self.title}」有 {self.total} 句可审的结论,已达上限({self._limit_words('max_claims')}),"
f"{skipped} 句未审")
# 这句没挂依据时在整份报告引用的数据源里查:写作者看得到的就是这几个库
everywhere = claim_sources(self.catalog, self.catalog, self.fetch)
usable = await _usable_sources(everywhere)
claims: list[dict[str, Any]] = []
usage = _zero_usage()
model: str | None = None
warned = False
hit: dict[str, list[int]] = {} # 上限 → [触顶的句数, 因此没能给出结论的句数]
broke = False # 全局每日金额用完了:后面的句子不再派 agent
broken: str | None = None # 核查模型本身用不了:后面的句子照这个原因记,不再逐句去建模型
unreviewed: list[str] = [] # 没能反查的句子各自的原因(触顶的由 challenge_limit 说,不在这里)
partial: list[list[str]] = [] # 反查了、但该查的源有停用了的句子:各自没查成的源
def unreached(cand: judging.Candidate, rationale: str, why: str) -> None:
claims.append(_claim(cand, "inconclusive", rationale))
unreviewed.append(why)
for cand in self.picked:
if broke:
hit["daily_max_usd"][1] += 1
claims.append(_claim(cand, "inconclusive", self._capped()))
continue
if broken is not None:
unreached(cand, f"反查未完成:{broken}", broken)
continue
try:
own = claim_sources(cand.aliases, self.catalog, self.fetch)
sources = own or everywhere
live = [s for s in sources if s in usable]
lost = [s for s in sources if s not in usable]
task_text = self._task(cand, live, lost, own=bool(own)) if live else ""
except Exception as e: # noqa: BLE001 - 一句的摘录出了意外(坏形状的工件),只让这一句没审成
why = f"整理证据摘录时出错({describe_exception(e)})"
unreached(cand, f"反查未完成:{why}", why)
continue
# 没挂依据的句子查的是整份报告用到的源,不是「这句引用的」
where = "这句引用的" if own else "这份报告用到的"
if not sources:
why = "这份报告引用的证据中没有数据库查询,没有可以反查的数据源"
unreached(cand, f"{why},这句未反查", why)
continue
if not live:
why = f"数据源{_quoted(sources)}已不存在或已停用"
unreached(cand, f"{where}{why},没有可以反查的数据源,这句未反查", why)
continue
outcome = await run_verify_agent(
spec=self.spec, system=CHALLENGE_SYSTEM, task=task_text, sources=live,
output_schema=CHALLENGE_SCHEMA, max_queries=self.max_queries, timeout_s=self.timeout_s,
daily_max_usd=self.daily_max_usd, emit=ctx.emit, run_id=ctx.run.run_id, node_id=ctx.node.id,
data_versions=run_versions(ctx.run))
if outcome.error and not outcome.model:
broken = outcome.error # 连模型都没建成(没配好、接入出错):换一句也一样
model = model or outcome.model
if not warned and outcome.model and outcome.model == self.writer_model:
warned = True
ctx.emit(EventType.LOG, level="warn", code="challenge_same_model", model=outcome.model,
message=f"反方模型与报告的写作模型相同(「{outcome.model}」),难以发现写作模型自身的错误。"
f"请在设置的「核查」一节选择另一个模型,或在节点的「{CHALLENGE_LABEL['model']}」中指定")
claim, why = _claim_of(cand, outcome)
if outcome.limit == "daily_max_usd" and outcome.result is None and not outcome.usage.get("calls"):
# 派出去就撞上每日金额上限、一次模型都没调用:和后面不再派出的句子是同一回事,同一种说法
claim = _claim(cand, "inconclusive", self._capped(), queries=outcome.queries, usage=outcome.usage)
elif lost and why is None:
# 反查了,可该查的源有停用了的:判断照常采信,理由末尾写明哪几个没查(只写在任务里的话,审稿记录和
# 审批卡上的「经受住」看着就像每个源都查过了)
claim = _noted(claim, f"注:{where}数据源{_quoted(lost)}已不存在或已停用,未能查询,以上判断不含其中的数据。")
partial.append(lost)
claims.append(claim)
usage = _add_usage(usage, outcome.usage)
if outcome.limit in _CLAIM_LIMITS:
counted = hit.setdefault(outcome.limit, [0, 0])
counted[0] += 1
counted[1] += why is not None
broke = outcome.limit == "daily_max_usd"
elif why is not None:
unreviewed.append(why)
counts = {v: sum(1 for c in claims if c["verdict"] == v) for v in VERDICTS}
artifact = await self._store({"report": self.report_id, "doc_artifact": self.doc_artifact, "claims": claims,
"model": model})
for limit, (touched, skipped) in hit.items():
words = self._limit_words(limit)
message = f"已达上限({words}),{skipped} 句未反查" if limit == "daily_max_usd" else \
f"有 {touched} 句反查已达上限({words}),已按查到的结果收尾" + \
(f",其中 {skipped} 句未能给出结论" if skipped else "")
ctx.emit(EventType.LOG, level="warn", code="challenge_limit", limit=limit, claims_skipped=skipped,
message=message)
if unreviewed:
more = "等" if len(set(unreviewed)) > 1 else ""
ctx.emit(EventType.LOG, level="warn", code="challenge_unreviewed", claims_skipped=len(unreviewed),
reason=unreviewed[0],
message=f"反方审稿有 {len(unreviewed)} 句结论未能反查,已记为无法判断:{unreviewed[0]}{more}")
if partial:
names = list(dict.fromkeys(s for gone in partial for s in gone))
ctx.emit(EventType.LOG, level="warn", code="challenge_sources_lost", claims=len(partial), sources=names,
message=f"反方审稿有 {len(partial)} 句结论只查了部分数据源:{_quoted(names)}已不存在或已停用,"
f"未能查询,{'这句' if len(partial) == 1 else '这几句'}的判断不含其中的数据")
# 真正查过的和没查成的分开说:「无法判断 2 句」可能是数据不够,也可能是一条查询都没执行成功
idle = sum(1 for c in claims if not c["queries"])
ctx.emit(EventType.LOG, level="info", code="challenge_done", counts=counts, artifact=artifact,
message=f"反方审稿完成:审了 {len(claims)} 句结论,经受住 {counts['stands']} 句、被削弱 "
f"{counts['weakened']} 句、被推翻 {counts['refuted']} 句、无法判断 {counts['inconclusive']} 句"
+ (f",其中 {idle} 句未能反查(没有执行成功任何查询)" if idle else ""))
if refuted := [c for c in claims if c["verdict"] == "refuted"]:
ctx.emit(EventType.LOG, level="warn", code="challenge_refuted",
items=[{"unit": c["unit"], "text": c["text"]} for c in refuted[:ITEMS_MAX]],
message=_refuted_words(refuted))
return {"artifact": artifact, "counts": counts, "usage": usage,
"claims": [{k: c[k] for k in ("unit", "text", "verdict")} for c in claims]}
async def _store(self, record: dict[str, Any]) -> str | None:
try:
return await artifact_store.put_json(record, kind="challenge", run_id=self.ctx.run.run_id,
node_id=self.ctx.node.id, meta={"report": self.report_id})
except Exception as e: # noqa: BLE001 - 记录写失败不毁掉运行:结论照常处置,只是理由和查询点不开
self.ctx.emit(EventType.LOG, level="warn", code="evidence_store_failed",
message=f"反方审稿记录保存失败({describe_exception(e)}),审稿的理由和查询将无法查看")
return None
def _finish(ctx: NodeContext, state: GraphState, outcome: dict[str, Any], *, on_refuted: str) -> dict[str, Any]:
"""反查的结果 → 节点产出;被推翻的按 on_refuted 处置。重放时 outcome 取自 checkpoint,算出来的一样。
产出里的 on_refuted 写实际生效的处置,只有 degrade / flag 两种(期 3 的约定:审批卡据此决定写不写「出具会降档」)。
配的是「报错并停止运行」却走到了这里,只有两种情形:没有句子被推翻,什么都没处置,记成只标注;有被推翻的却又配了
「出错时:记录错误并继续」(画布校验拦这个组合,跳过校验的旧图才会走到)——这时报错会被当成普通错误吞掉,被推翻
的结论照常完整出具,比默认的降档还松,所以改按出具降档处置。"""
claims = outcome.get("claims") or []
refuted = [c for c in claims if c.get("verdict") == "refuted"]
if on_refuted == "fail":
if refuted and ctx.node.config.get("on_error") != "continue":
which = "这句" if len(refuted) == 1 else "这几句"
raise NodeError(ctx.node.id, f"{_refuted_words(refuted)}。请核对{which}结论;如只需记录,可将"
f"「{CHALLENGE_LABEL['on_refuted']}」改为「{CHALLENGE_ON_REFUTED_LABEL['flag']}」")
on_refuted = "degrade" if refuted else "flag"
result: dict[str, Any] = {"challenge_artifact": outcome.get("artifact"), "counts": outcome.get("counts") or {},
"items": pick_items(claims), "on_refuted": on_refuted}
if refuted and on_refuted == "degrade":
result["issuance_gaps"] = [_refuted_words(refuted)]
updates: dict[str, Any] = {"nodes": {ctx.node.id: result}}
spent = outcome.get("usage") or {}
if spent.get("calls"):
updates["usage"] = {k: spent.get(k, 0) for k in ("input_tokens", "output_tokens", "total_tokens", "cost_usd",
"calls")}
if outcome.get("artifact") and ledger_enabled(ctx.run):
# 台账只记审稿记录本身(kind challenge):它的查询不进写作目录,报告按台账编 Q1… 时看不到它们
updates["evidence"] = [{"kind": "challenge", "node_id": ctx.node.id, "exec": next_exec(state, ctx.node.id),
"artifact": outcome["artifact"]}]
return updates
# --------------------------------------------------------------------------
# 审批卡
# --------------------------------------------------------------------------
def challenge_summary(spec: GraphSpec, outputs: dict[str, Any], report_id: str) -> dict[str, Any] | None:
"""审批卡上这份报告的反方审稿概况 {counts, items, artifact, on_refuted}:审它的、已经审完的反方审稿节点给的。
没有(没加反方审稿、还没审到、直通了)返回 None,审批卡不加这个键。几个反方审稿节点审同一份报告时取画布上
靠后的那个。只看图和状态:重放时算出来的一样。"""
found: dict[str, Any] | None = None
for node in spec.nodes:
result = outputs.get(node.id)
if node.type != NodeType.CHALLENGE or not isinstance(result, dict) or not isinstance(result.get("counts"), dict):
continue
if challenge_report(spec, node)[0] != report_id:
continue
found = {"counts": dict(result["counts"]), "items": list(result.get("items") or [])[:ITEMS_MAX],
"artifact": result.get("challenge_artifact"), "on_refuted": result.get("on_refuted") or "degrade"}
return found
__all__ = ["CHALLENGE_SYSTEM", "ITEMS_MAX", "VERDICTS", "challenge_summary", "claim_sources", "mark_original",
"pick_claims", "pick_items", "run_challenge", "with_note"]
+20 -6
View File
@@ -81,12 +81,16 @@ _JUDGE_KEYS = ("mode", "counts", "complete", "unjudged", "limits_hit")
def reviewed_report_ids(spec: Any, human: Any) -> list[str]:
"""一个人工审批节点审的是哪几份报告:上游直接连着的报告撰写节点,加上标题、说明里引用了正文的
({{ nodes.<id>.text }},或者报告的 assign_to 变量)。只看图,不看运行状态。
"""一个人工审批节点审的是哪几份报告:上游直接连着的报告撰写节点,直接连着的反方审稿节点审的那份报告,加上
标题、说明里引用了正文的({{ nodes.<id>.text }},或者报告的 assign_to 变量)。只看图,不看运行状态。
「报告 → 反方审稿 → 人工审批」是反方审稿推荐的接法:审批节点的直接上游是反方审稿,只认直接连着的报告撰写节点的话,
审批卡上一份报告都没有,被推翻的结论审批人批准时完全不知道。反方审稿审哪份照 schema.challenge_report 认,和它
执行时审的是同一份。
审批卡(_reviewed_reports)和报告节点(reviewed_by_human:要交给人审的报告,探索运行也在节点里判)
用的是同一套认法,两边说的「这份报告要人审」不会对不上。"""
from app.engine.schema import NodeType
from app.engine.schema import NodeType, challenge_report
config = human.config or {}
text = " ".join(json.dumps(config.get(key), ensure_ascii=False, default=str)
@@ -95,9 +99,15 @@ def reviewed_report_ids(spec: Any, human: Any) -> list[str]:
by_var = {str(n.config.get("assign_to")): n.id for n in spec.nodes
if n.type == NodeType.REPORT and n.config.get("assign_to")}
nodes = spec.node_map()
wanted = ([e.source for e in spec.edges if e.target == human.id] + _NODE_REF.findall(text)
def upstream(source: str) -> str | None:
node = nodes.get(source)
return challenge_report(spec, node)[0] if node is not None and node.type == NodeType.CHALLENGE else source
wanted = ([upstream(e.source) for e in spec.edges if e.target == human.id] + _NODE_REF.findall(text)
+ [by_var[v] for v in _VAR_REF.findall(text) if v in by_var])
return [rid for rid in dict.fromkeys(wanted) if (n := nodes.get(rid)) is not None and n.type == NodeType.REPORT]
return [rid for rid in dict.fromkeys(wanted)
if rid and (n := nodes.get(rid)) is not None and n.type == NodeType.REPORT]
def reviewed_by_human(spec: Any, report_id: str) -> bool:
@@ -112,7 +122,9 @@ def _reviewed_reports(state: GraphState, ctx: NodeContext) -> list[dict[str, Any
真实运行里报告有一句结论和证据相矛盾,审批人批准时完全不知道:审批卡只有正文。这里带上报告节点产出里的
stats(同 report.checked)、结论句策略、裁判摘要和报告文档的工件 id,审批卡据此写核对概况、给打开证据的入口。
只认已经跑完、带着核对统计的报告;重放时 state 一样,算出来的也一样。"""
审这份报告的反方审稿已经审完的,再带上它的概况(challenge:几句被推翻、被削弱,打开审稿记录的入口);没有就
不加这个键。只认已经跑完、带着核对统计的报告;重放时 state 一样,算出来的也一样。"""
from app.engine.nodes.challenge import challenge_summary
from app.engine.schema import NodeType
spec = ctx.run.spec
@@ -134,6 +146,8 @@ def _reviewed_reports(state: GraphState, ctx: NodeContext) -> list[dict[str, Any
item["doc_artifact"] = result["doc_artifact"]
if isinstance(result.get("judge"), dict):
item["judge"] = {k: result["judge"][k] for k in _JUDGE_KEYS if k in result["judge"]}
if (challenge := challenge_summary(spec, outputs, rid)) is not None:
item["challenge"] = challenge
out.append(item)
return out
+133
View File
@@ -717,6 +717,137 @@ def _check_report_from(node: GraphNode, report_from: Any, spec: GraphSpec,
node_id=node.id, field=field, code=code)
# --------------------------------------------------------------------------
# 反方审稿节点(engine/nodes/challenge.py):审哪份报告、各项配置
# --------------------------------------------------------------------------
#: 反方审稿节点各项配置在界面上的叫法:报错写叫法,不写键名
CHALLENGE_LABEL = {
"report": "要审的报告", "max_claims": "最多审几句", "max_queries": "每句最多查询次数",
"timeout_s": "每句时长上限(秒)", "provider": "反方模型 · 接入", "model": "反方模型",
"on_refuted": "结论被推翻时", "enabled": "启用",
}
#: 结论被推翻时怎么办:degrade 计入出具缺口、出具降档(默认)/ flag 只标注 / fail 判节点失败、运行停止
CHALLENGE_ON_REFUTED = ("degrade", "flag", "fail")
#: 选项的叫法和检查器下拉一致(lib/terms.ts 的 CHALLENGE_TEXT.onRefuted),只去掉「(默认)」:报错让人去改某一项时,
#: 人在下拉里找得到这个名字
CHALLENGE_ON_REFUTED_LABEL = {"degrade": "出具降档", "flag": "只标注", "fail": "报错并停止运行"}
def _challenge_field(key: str) -> str:
return f"「{type_label(NodeType.CHALLENGE)}」的「{CHALLENGE_LABEL[key]}」"
def challenge_problems(config: dict[str, Any]) -> list[tuple[str, str]]:
"""反方审稿各项配置有什么问题(要审的报告另见 challenge_report):[(字段, 人话)],没问题返回 []。
校验和节点执行时报的是同一句话。
三个上限留空(不写或写 null)就是沿用设置里「核查」一节的值,和「启用」留空一个意思。节点上没有「不限」的写法:
要放开上限去设置里改,免得一个节点悄悄把核查的花费放开了。
"""
out: list[tuple[str, str]] = []
for key, words in (("max_claims", "每份报告最多反查的句数"), ("max_queries", "每句最多执行的查询次数")):
value = config.get(key)
if value not in (None, "") and not _positive(value, integer=True):
out.append((key, f"{_challenge_field(key)}需要填写正整数({words}),留空表示沿用设置,当前为「{value}」"))
value = config.get("timeout_s")
if value not in (None, "") and not _positive(value):
out.append(("timeout_s", f"{_challenge_field('timeout_s')}需要填写大于 0 的秒数,留空表示沿用设置,"
f"当前为「{value}」"))
for key, what in (("provider", "模型接入的名称"), ("model", "模型 ID")):
value = config.get(key)
if value is not None and not isinstance(value, str):
out.append((key, f"{_challenge_field(key)}需要填写{what},留空表示使用设置中的核查模型,当前为「{value}」"))
value = config.get("on_refuted")
if value not in (None, "") and value not in CHALLENGE_ON_REFUTED:
names = [f"「{CHALLENGE_ON_REFUTED_LABEL[v]}」" for v in CHALLENGE_ON_REFUTED]
out.append(("on_refuted", f"{_challenge_field('on_refuted')}只能选{'、'.join(names[:-1])}或{names[-1]},"
f"当前为「{value}」"))
value = config.get("enabled")
if value is not None and not isinstance(value, bool):
out.append(("enabled", f"{_challenge_field('enabled')}只能是开启、关闭或跟随设置,当前为「{value}」"))
return out
def challenge_report(spec: GraphSpec, node: GraphNode) -> tuple[str | None, str | None]:
"""反方审稿审的是哪份报告:(报告撰写节点 id, None);定不下来时 (None, 为什么)。校验、执行器、审批卡都按它认。
写了「要审的报告」就按写的,必须是它上游的报告撰写节点。没写就从上游找:直接连着的报告撰写节点只有一个,
就是它(「初稿 → 终稿 → 反方审稿」审终稿);否则整个上游里只有一个报告撰写节点才取它。两份以上说不准审哪一份,
宁可让人选,也不能悄悄审错一份。
"""
who, label = f"「{type_label(NodeType.CHALLENGE)}」", CHALLENGE_LABEL["report"]
nodes = spec.node_map()
upstream = _ancestors(spec, node.id)
chosen = node.config.get("report")
if chosen in (None, ""):
direct = list(dict.fromkeys(e.source for e in spec.incoming(node.id)
if (n := nodes.get(e.source)) is not None and n.type == NodeType.REPORT))
if len(direct) == 1:
return direct[0], None
found = [n for n in spec.nodes if n.id in upstream and n.type == NodeType.REPORT]
if len(found) == 1:
return found[0].id, None
if not found:
return None, (f"{who}的上游没有「报告撰写」节点:反方审稿只审报告撰写节点写出的报告。"
f"请把它接在报告撰写节点之后,或在「{label}」中选择一个")
names = "、".join(f"「{n.title}」" for n in found[:5]) + ("等" if len(found) > 5 else "")
return None, f"{who}的上游有多个「报告撰写」节点({names}),请在「{label}」中选择要审的那一份"
if not isinstance(chosen, str):
return None, f"{who}的「{label}」需要选择一个「报告撰写」节点,当前为「{chosen}」"
target = nodes.get(chosen)
if target is None:
return None, f"{who}的「{label}」选择了「{chosen}」,但工作流中不存在这个节点"
if target.type != NodeType.REPORT:
return None, (f"{who}的「{label}」选择了「{target.title}」,它是「{type_label(target.type)}」节点,"
"不是「报告撰写」节点:反方审稿只审报告撰写节点写出的报告")
if chosen not in upstream:
return None, (f"{who}的「{label}」选择的「{target.title}」不在它的上游:审稿时这份报告还没有写出。"
"请把报告撰写节点连到反方审稿之前")
return chosen, None
def _check_challenge(node: GraphNode, spec: GraphSpec, result: ValidationResult) -> None:
"""反方审稿的配置写错了挡住运行:执行时节点照样会报这一句,与其取数、写完报告才死在审稿这一步,不如在画图时说破。
反方模型和要审的报告的写作模型写成同一个的给警告:等于自己审自己。
两种「看着配了、其实不生效」的接法也在这里说破:
- 「报错并停止运行」配「出错时:记录错误并继续」:节点报的错被当成普通错误跳过,被推翻的结论照常完整出具,比默认的
降档还松,挡住运行。不放进 challenge_problems:执行时再报一遍的错照样会被这个设置吞掉,执行时改按降档处置
- 「出具降档」靠成果节点的出具契约收上游的出具缺口(nodes/io.py 只认成果节点的上游):反方审稿接在侧枝上、或者在
成果之后,降档不会发生,给警告。图里没有带出具契约的成果节点时无所谓出具,不提醒
"""
who = f"「{type_label(NodeType.CHALLENGE)}」"
for field, message in challenge_problems(node.config):
result.add(message, node_id=node.id, field=field, code="challenge.config_invalid")
on_refuted = node.config.get("on_refuted") or "degrade"
if on_refuted == "fail" and node.config.get("on_error") == "continue":
result.add(f"{_challenge_field('on_refuted')}选了「{CHALLENGE_ON_REFUTED_LABEL['fail']}」,"
f"「{field_label('on_error')}」却是「{option_label('on_error', 'continue')}」:结论被推翻时节点报的错"
"会被当作普通错误跳过,被推翻的结论照常完整出具。"
f"请把「{CHALLENGE_LABEL['on_refuted']}」改为「{CHALLENGE_ON_REFUTED_LABEL['degrade']}」,"
f"或把「{field_label('on_error')}」改回「中断整个运行」",
node_id=node.id, field="on_refuted", code="challenge.config_invalid")
if on_refuted == "degrade":
issued = [n for n in spec.nodes if n.type == NodeType.OUTPUT and n.config.get("contract")]
if issued and not any(node.id in _ancestors(spec, n.id) for n in issued):
report_kind, output_kind = type_label(NodeType.REPORT), type_label(NodeType.OUTPUT)
result.add(f"{who}不在带出具契约的「{output_kind}」节点上游:结论被推翻时,「{output_kind}」节点收不到它记下的"
f"出具缺口,出具不会降档。请改为「{report_kind} → {type_label(NodeType.CHALLENGE)} → {output_kind}」"
f"的接法;只需记录的,把「{CHALLENGE_LABEL['on_refuted']}」改为「{CHALLENGE_ON_REFUTED_LABEL['flag']}」",
level="warning", node_id=node.id, code="challenge.no_issuance")
report, problem = challenge_report(spec, node)
if report is None:
result.add(problem or f"{who}找不到要审的报告", node_id=node.id, field="report", code="challenge.report_invalid")
return
writer = spec.node_map()[report]
model = node.config.get("model")
if isinstance(model, str) and model and model == (writer.config.get("model") or spec.defaults.get("model")):
result.add(f"反方模型与报告「{writer.title}」的写作模型相同(「{model}」),等于自己审自己,难以发现写作模型自身"
f"的错误。请在「{CHALLENGE_LABEL['model']}」中选择另一个模型", level="warning", node_id=node.id,
field="model", code="challenge.same_model")
def validate_graph(spec: GraphSpec) -> ValidationResult:
"""编译前的静态检查。错误会挡住运行,警告只在画布上提示。"""
result = ValidationResult()
@@ -785,6 +916,8 @@ def validate_graph(spec: GraphSpec) -> ValidationResult:
_check_report_claims(node, spec, result)
elif node.type == NodeType.MERGE:
_check_merge(node, spec, result)
elif node.type == NodeType.CHALLENGE:
_check_challenge(node, spec, result)
elif node.type == NodeType.OUTPUT:
contract = cfg.get("contract")
if contract and not isinstance(contract, dict):
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,35 @@
"""助手的节点手册要教反方审稿节点:什么时候加(多个来源汇总的分析报告、要正式出具的报告),接在哪里,各配置项怎么写。
手册里没写的节点,助手就不会用;写错的键名、取值,画布校验会一直把图打回来。所以这里不只查关键词,还拿校验认的那一份
配置项和取值来对。
"""
from __future__ import annotations
from app.api.copilot import NODE_REFERENCE
from app.engine.schema import CHALLENGE_LABEL, CHALLENGE_ON_REFUTED
def _challenge_entry() -> str:
start = NODE_REFERENCE.index("- challenge:反方审稿")
return NODE_REFERENCE[start:NODE_REFERENCE.index("\n\n", start)]
def test_the_guide_names_every_setting_the_validator_knows():
entry = _challenge_entry()
assert all(key in entry for key in CHALLENGE_LABEL), [k for k in CHALLENGE_LABEL if k not in entry]
assert all(value in entry for value in CHALLENGE_ON_REFUTED)
def test_the_guide_says_when_to_add_it_and_where():
entry = _challenge_entry()
# 什么时候加:多个来源汇总的分析报告、要正式出具的报告;串在报告和成果之间(降档靠成果节点收上游的缺口,
# 接在侧枝上不生效),审批节点在后面时引用报告正文
assert "多个来源" in entry and "正式出具" in entry and "report → challenge → output" in entry
assert "侧枝" in entry
assert "{{ nodes.报告节点id.text }}" in entry
# 「报错并停止运行」配「出错时:记录错误并继续」画布校验会挡:助手别这么配
assert "on_error: continue" in entry
# 反方模型建议和写作模型不同:同一个模型等于自己审自己
assert "自己审自己" in entry
# 不写上限就取设置里的默认:助手不要替人把核查的花费写死
assert "取设置里的默认" in entry
@@ -85,22 +85,6 @@ async def test_upstream_gap_degrades_issuance_and_names_the_node():
assert (await issuance_event(row.id))["gaps"] == issuance["gaps"]
async def test_stub_gate_nodes_pass_through_without_touching_issuance():
"""反方审稿目前是占位:放在图里照样能运行,出具不受影响。双盲复算已经实现,没有可复算的指标时直接放行、
不碰出具,见 test_dual_check_node.py"""
for kind in ("challenge",):
row = await finish(graph(node("gate", kind)))
issuance = row.output["_issuance"]
assert issuance["tier"] == "formal", (kind, issuance)
assert issuance["gaps"] == []
async with SessionLocal() as session:
logs = (await session.execute(
select(RunEvent).where(RunEvent.run_id == row.id, RunEvent.type == "log", RunEvent.node_id == "gate")
)).scalars().all()
assert [e.data.get("level") for e in logs] == ["info"]
assert "开发中" in logs[0].data["message"]
async def test_without_issuance_gaps_nothing_changes():
"""没写 issuance_gaps 的节点,和别的支路上写了的节点,都不影响这份成果"""
plain = node("recheck", "transform", "复核销售额", mode="json", template='{"ok": true}')