From 4ceac6b74ca7ac9025f94ad3d725677fa52d5876 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 5 Oct 2026 15:08:53 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8F=8D=E6=96=B9=E5=AE=A1=E7=A8=BF=E5=A4=8D?= =?UTF-8?q?=E6=A0=B8=E4=BF=AE=E6=AD=A3=EF=BC=9A=E6=B2=A1=E6=9C=89=E5=87=BA?= =?UTF-8?q?=E5=A4=84=E7=9A=84=E7=BB=93=E8=AE=BA=E4=B8=8D=E9=87=87=E4=BF=A1?= =?UTF-8?q?=EF=BC=8C=E6=8E=A8=E8=8D=90=E6=8E=A5=E6=B3=95=E4=B8=8B=E5=AE=A1?= =?UTF-8?q?=E6=89=B9=E5=8D=A1=E6=9C=89=E6=8A=A5=E5=91=8A=EF=BC=8C=E9=87=8D?= =?UTF-8?q?=E8=AF=95=E5=92=8C=E6=8E=A5=E7=9D=80=E8=B7=91=E4=B8=8D=E9=87=8D?= =?UTF-8?q?=E5=AE=A1=E3=80=81=E4=B8=8D=E6=8B=BF=E6=97=A7=E5=AE=A1=E7=A8=BF?= =?UTF-8?q?=E5=86=92=E5=85=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 出处：一条查询都没执行成功的判什么都记为无法判断，判「被推翻」的理由里还得引用到自己的查询结果——零查询的推翻能让 正式出具降档，零查询的「经受住」会给审批人假安心。系统写明原因，反方原本的结论和理由照录。这句引用的数据源都已 停用时不派 agent；核查模型本身用不了时后面的句子照同一个原因记，不再逐句去建模型 - 时间线：有句子没能反查时另记一条 challenge_unreviewed（带第一条原因），challenge_done 的说法分开写查过的和没查成的 - 审批卡：「报告 → 反方审稿 → 人工审批」时审批说明不引用正文也认得出这份报告（reviewed_report_ids 穿过直接连着的 反方审稿，reviewed_by_human 同步），审稿概况附得上 - 重试与接着跑：同一次执行里的重试沿用已经审完的结果，不再多花一份钱、也不会把「被推翻」重审掉；审稿结果带输入指纹 （报告、文档、反方模型、上限、遮罩），接着跑时只改「结论被推翻时」沿用，改了审稿本身就重审 - 处置：产出和审批卡里的 on_refuted 写实际生效的 degrade / flag；「报错并停止运行」配「出错时：记录错误并继续」画布 校验挡住，跳过校验的旧图执行时改按出具降档；反方审稿不在带出具契约的成果节点上游时画布给警告（降档不会发生） - 摘录里指向原报告查询的写法（指标取自查询 Qn、查询 Qn 的 SQL、出现在查询 Qn）都标成「原报告 Qn」，和反查 agent 自己的 Q1 分开 - 选项叫法和检查器下拉一致：出具降档 / 只标注 / 报错并停止运行，「启用」写跟随设置；助手手册同步 - 补测试：快照里的反方模型和每日上限、裁判判成不是结论的句子不审、审批卡只给被审的那份报告挂概况 --- backend/app/api/copilot.py | 14 +- backend/app/engine/nodes/challenge.py | 228 +++++++++--- backend/app/engine/nodes/human.py | 20 +- backend/app/engine/schema.py | 37 +- backend/tests/test_challenge_node.py | 330 +++++++++++++++++- backend/tests/test_copilot_challenge_guide.py | 8 +- 6 files changed, 565 insertions(+), 72 deletions(-) diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index e78db38d..3158b74a 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -261,13 +261,15 @@ NODE_REFERENCE = """\ - challenge：反方审稿。报告写完后挑出几句关键结论（挂了依据的、带比较或因果的优先），每句派一个只读 agent 站在反方 自己写查询找反例，逐句判经受住、被削弱、被推翻或无法判断，理由引用它自己查到的数据；不改报告。 config: {report, max_claims, max_queries, timeout_s, provider, model, on_refuted, enabled} - **多个来源汇总成的分析报告、要正式出具的报告**加这个节点，接在 report 之后：report → challenge → human 或 output。 + **多个来源汇总成的分析报告、要正式出具的报告**加这个节点，串在 report 和 output 之间：report → challenge → output + （有审批时 report → challenge → human → output）。challenge 要在 output 上游：只从 report 拉出一条侧枝接 challenge、 + output 不经过它时，降档不会发生。 report 写要审的报告节点 id（上游只有一个 report 时可以不写）；on_refuted 是结论被推翻时怎么办：degrade（默认，计入 - 出具缺口、出具降档）/ flag（只记录）/ fail（判节点失败）。max_claims（每份报告最多审几句）、max_queries（每句最多 - 查询次数）、timeout_s（每句时长上限，秒）、enabled（不写就跟随设置里「核查」一节的开关）一般都不写，取设置里的默认； - provider、model 指定反方模型，不写用设置里的核查模型（建议和写作模型不同，同一个模型等于自己审自己）。后面接 human - 时，human 的 message 里引用 {{ nodes.报告节点id.text }}，审批卡会附上这份报告的审稿结果。只回答一个问题、一次查询的 - 不必加 + 出具缺口、出具降档）/ 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 diff --git a/backend/app/engine/nodes/challenge.py b/backend/app/engine/nodes/challenge.py index b7aa24dd..836c3ef3 100644 --- a/backend/app/engine/nodes/challenge.py +++ b/backend/app/engine/nodes/challenge.py @@ -7,15 +7,20 @@ 取前「最多审几句」句；裁判判成「不是结论」的不审。同一份文档永远选出同样的句子、同样的顺序 2. 反查：每句一个只读 agent（verify_agent.run_verify_agent）。工具只给这句引用的查询所在的数据源（指标按它的输入 取自的查询算，合并查询按它合并的那几个源算）；这句没挂依据，就在这份报告引用的数据源里查。任务里给这句结论 - 和它挂的证据摘录（judge.prepare，数据源遮罩的列照旧不给），要它先想清楚「如果这句不成立，数据里会看到什么」， - 写只读查询去找，最后交结论（经受住 / 被削弱 / 被推翻 / 无法判断）和理由。理由里的数字只能引用它自己查询的 - 格子，裸数字由 compose_doc 标出来 -3. 记录：每句的结论、理由（渲染后的正文和组装好的文档）、查询快照、用量写进审稿记录（challenge_artifact），随运行 + 和它挂的证据摘录（judge.prepare，数据源遮罩的列照旧不给；摘录里原报告的编号都标成「原报告 Qn」，和 agent 自己 + 的 Q1 分开），要它先想清楚「如果这句不成立，数据里会看到什么」，写只读查询去找，最后交结论（经受住 / 被削弱 / + 被推翻 / 无法判断）和理由。理由里的数字只能引用它自己查询的格子，裸数字由 compose_doc 标出来 +3. 出处：一条查询都没执行成功的，判什么都记为无法判断；判「被推翻」的，理由里还得引用到自己的查询结果——它要让 + 正式出具降档，审稿记录里得查得到凭什么。不采信时系统写明原因，反方原本的结论和理由照录在后面。这句引用的数据源 + 都不存在或已停用、核查模型本身用不了时不派 agent（模型用不了的，后面的句子照同一个原因记，不再逐句去建模型） +4. 记录：每句的结论、理由（渲染后的正文和组装好的文档）、查询快照、用量写进审稿记录（challenge_artifact），随运行 封存。查询照常发 tool.end（role: verify，带查询快照，可下钻）；台账只记 kind challenge，绝不记 query——核查的 - 查询进了写作目录，下游的写作者会把它们当成可以引用的证据 -4. 处置：被推翻的按 on_refuted 处置。degrade（默认）在产出里写 issuance_gaps，成果节点的出具契约照单收进缺口、 - 出具降档（nodes/io.py 的 _upstream_issuance_gaps）；fail 判节点失败；flag 只记录。审批卡（human.py 的 - _reviewed_reports）附上审稿概况 + 查询进了写作目录，下游的写作者会把它们当成可以引用的证据。有句子没能反查的另记一条 challenge_unreviewed， + 不然时间线上只有一句「审了 N 句」，看不出根本没审成 +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"]），节点上写了的盖过它：审批停下期间改了设置，恢复、接着跑仍按发起时的那份。没有快照 @@ -24,19 +29,23 @@ 逐句串行反查：并行的话几句的工具调用在时间线上交错，看不出哪条查询在审哪一句；每日余额按先后扣，结果也才确定。 整个反查连同存审稿记录、发日志都在一个 @task 里：接着跑时取 checkpoint 里的结果，不重复查库、不重复记账，时间线 -上也不再发一遍。 +上也不再发一遍。结果带着输入指纹，接着跑时改了审稿本身的配置就重审；同一次执行里的重试沿用已经审完的（见 _reviewed）。 """ from __future__ import annotations -from collections.abc import Callable, Iterable +import re +from collections.abc import Callable, Collection, Iterable from typing import Any from langgraph.func import task +from sqlalchemy import select 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 @@ -63,6 +72,12 @@ _QUERY_KEYS = ("alias", "artifact", "sql", "row_count", "truncated", "source") _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": "无法判断"} +#: 接着跑时 checkpoint 里缓存的审稿最多翻几条去找这一份配置的（每条都是按先前某一份配置审的，对不上就往后翻） +_STALE_LIMIT = 16 +#: 同一次执行里审完的结果放在 RunContext.extra 的这个键下：{(节点, 第几次执行, 输入指纹): 审稿结果}。重试沿用它 +_REVIEWS_KEY = "challenge_reviews" CHALLENGE_SYSTEM = ( "你是反方审稿人：报告已经写完，你站在反方，专门找其中一句结论的反例。你不写报告，也不改报告。\n" @@ -71,7 +86,9 @@ CHALLENGE_SYSTEM = ( "不要只是重复写作者的查询\n" "- verdict 四选一：stands 认真找过、没有找到反例；weakened 大体成立，但有重要的例外或前提，或者把范围、程度、" "原因说过了头；refuted 查到的数据和这句结论相矛盾；inconclusive 查不到判断所需的数据\n" - "- 没有找到反例不等于结论被推翻：只有数据和结论冲突时才判 refuted" + "- 没有找到反例不等于结论被推翻：只有数据和结论冲突时才判 refuted\n" + "- 结论要有出处：一条查询都没执行成功时，判什么都不会被采信；判 refuted 时，理由里至少要引用一处你自己查询的" + "结果（[[v:Qn.rX.列名]] 或 [[see:Qn]]），否则同样不会被采信" ) CHALLENGE_TASK_PROMPT = ( "要审的结论（报告「{report}」中的一句{section}）：\n「{claim}」\n\n" @@ -143,17 +160,39 @@ async def run_challenge(state: GraphState, ctx: NodeContext) -> dict[str, Any]: 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() -> dict[str, Any]: - # 反查进 checkpoint：接着跑时直接取这里的结果，不再查库、不再记账（每日账也记在这里面）； - # 事件都在这里面发，取缓存时不会再发一遍 - return await review.run() + async def challenge_step(key: str) -> dict[str, Any]: + return {**await review.run(), "key": key} - outcome = await challenge_step() - else: - outcome = await review.run() - return _finish(ctx, state, outcome, on_refuted=str(config.get("on_refuted") or "degrade")) + 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]: @@ -289,14 +328,68 @@ def _claim(cand: judging.Candidate, verdict: str, rationale: str, *, doc: dict[s "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) -> dict[str, Any]: +def _claim_of(cand: judging.Candidate, outcome: VerifyOutcome) -> tuple[dict[str, Any], str | None]: + """反查 agent 交回的 → (审稿记录里的一句, 这句没能反查的原因)。反查过了的原因是 None。 + + 反方的结论要有出处才采信：一条查询都没执行成功的（没有可用的工具、查询全部失败、根本没查），判什么都记为无法 + 判断——零查询的「被推翻」能让正式出具降档，零查询的「经受住」会在审批卡上给人假安心。判「被推翻」的，理由里还得 + 引用到自己的查询结果（resolved 的 [[v:]] / [[see:]]），审稿记录里才查得到凭什么推翻。""" if outcome.result is None: # 没交成结果（模型不可用、调用失败、触顶时连收尾都没赶上）：查过的照记，结论只能是无法判断 - return _claim(cand, "inconclusive", f"反查未完成：{outcome.error or '原因不明'}", queries=outcome.queries, - usage=outcome.usage) + 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() - return _claim(cand, verdict if verdict in VERDICTS else "inconclusive", str(outcome.result.get(RATIONALE) or ""), - doc=outcome.rationale, queries=outcome.queries, usage=outcome.usage) + 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, rationale, why), why + if verdict == "refuted" and not _cites(outcome.rationale): + return _unfounded(cand, outcome, verdict, rationale, "理由没有引用任何查询结果"), 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, rationale: str, + why: str) -> dict[str, Any]: + """没有出处的结论：记为无法判断，系统写明原因，反方原本的结论和理由照录在后面。rationale_doc 为 None：这段话是 + 系统说的，界面照 rationale 显示，不拿反方那份文档冒充理由。查过的照记。""" + taken = f"反方给出的「{_VERDICT_LABEL[verdict]}」不予采信，记为无法判断" if verdict != "inconclusive" else "记为无法判断" + said = f"反方的理由：{rationale.strip()}" if rationale.strip() else "反方没有给出理由" + return _claim(cand, "inconclusive", f"{why}，{taken}。{said}", queries=outcome.queries, usage=outcome.usage) + + +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: @@ -319,7 +412,10 @@ def pick_items(claims: list[dict[str, Any]]) -> list[dict[str, Any]]: class _Review: - """一次审稿：逐句派核查 agent，写审稿记录，发日志。run() 的返回值进 checkpoint，只放节点产出用得到的。""" + """一次审稿：逐句派核查 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, @@ -334,6 +430,12 @@ class _Review: 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": @@ -344,15 +446,16 @@ class _Review: return f"每句时长上限 {self.timeout_s:g} 秒" return f"全局每日核查金额上限 {_usd(self.daily_max_usd)}" - def _task(self, cand: judging.Candidate, sources: list[str]) -> str: - """交给核查 agent 的任务：这句结论、它挂的证据摘录（遮罩照旧）、可以查的数据源。摘录开头的编号改成 - 「原报告 Qn」：agent 自己的查询也编 Q1、Q2，不分开的话理由里引用的会是另一份快照的格子。""" + def _task(self, cand: judging.Candidate, sources: list[str], lost: list[str]) -> str: + """交给核查 agent 的任务：这句结论、它挂的证据摘录（遮罩照旧，原报告的编号标明，见 mark_original）、可以查的 + 数据源；这句引用的源有停用了的，也告诉它查不了哪几个，免得它把查不到当成没有反例。""" 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(request.excerpts[a].replace(f"【{a}】", f"【原报告 {a}】", 1) for a in aliases) + evidence = "\n\n".join(mark_original(request.excerpts[a], a, self.queries) for a in aliases) + gone = f"（这句还引用了{_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)) + evidence=evidence or "（这句没有挂依据）", sources="、".join(sources) + gone) async def run(self) -> dict[str, Any]: ctx = self.ctx @@ -363,45 +466,67 @@ class _Review: 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 说，不在这里） + + 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", f"已达上限（{self._limit_words('daily_max_usd')}），这句未反查")) continue + if broken is not None: + unreached(cand, f"反查未完成：{broken}", broken) + continue try: sources = claim_sources(cand.aliases, self.catalog, self.fetch) or everywhere - task_text = self._task(cand, sources) if sources else "" + live = [s for s in sources if s in usable] + task_text = self._task(cand, live, [s for s in sources if s not in usable]) if live else "" except Exception as e: # noqa: BLE001 - 一句的摘录出了意外（坏形状的工件），只让这一句没审成 - claims.append(_claim(cand, "inconclusive", f"反查未完成：{describe_exception(e)}")) + why = f"整理证据摘录时出错（{describe_exception(e)}）" + unreached(cand, f"反查未完成：{why}", why) continue if not sources: - claims.append(_claim(cand, "inconclusive", - "这份报告引用的证据中没有数据库查询，没有可以反查的数据源，这句未反查")) + why = "这份报告引用的证据中没有数据库查询，没有可以反查的数据源" + unreached(cand, f"{why}，这句未反查", why) + continue + if not live: + why = f"数据源{_quoted(sources)}不存在或已停用" + unreached(cand, f"这句引用的{why}，没有可以反查的数据源，这句未反查", why) continue outcome = await run_verify_agent( - spec=self.spec, system=CHALLENGE_SYSTEM, task=task_text, sources=sources, + 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']}」中指定") - claims.append(_claim_of(cand, outcome)) + claim, why = _claim_of(cand, outcome) + 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] += outcome.result is None + 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, @@ -413,9 +538,17 @@ class _Review: (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}") + # 真正查过的和没查成的分开说：「无法判断 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"{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]], @@ -434,12 +567,19 @@ class _Review: def _finish(ctx: NodeContext, state: GraphState, outcome: dict[str, Any], *, on_refuted: str) -> dict[str, Any]: - """反查的结果 → 节点产出；被推翻的按 on_refuted 处置。重放时 outcome 取自 checkpoint，算出来的一样。""" + """反查的结果 → 节点产出；被推翻的按 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 refuted and on_refuted == "fail": - raise NodeError(ctx.node.id, f"{_refuted_words(refuted)}。请核对这几句结论；如只需记录，可将" - f"「{CHALLENGE_LABEL['on_refuted']}」改为「{CHALLENGE_ON_REFUTED_LABEL['flag']}」") + if on_refuted == "fail": + if refuted and ctx.node.config.get("on_error") != "continue": + raise NodeError(ctx.node.id, f"{_refuted_words(refuted)}。请核对这几句结论；如只需记录，可将" + 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": @@ -477,5 +617,5 @@ def challenge_summary(spec: GraphSpec, outputs: dict[str, Any], report_id: str) return found -__all__ = ["CHALLENGE_SYSTEM", "ITEMS_MAX", "VERDICTS", "challenge_summary", "claim_sources", "pick_claims", - "pick_items", "run_challenge"] +__all__ = ["CHALLENGE_SYSTEM", "ITEMS_MAX", "VERDICTS", "challenge_summary", "claim_sources", "mark_original", + "pick_claims", "pick_items", "run_challenge"] diff --git a/backend/app/engine/nodes/human.py b/backend/app/engine/nodes/human.py index 8d55ea36..a84725d6 100644 --- a/backend/app/engine/nodes/human.py +++ b/backend/app/engine/nodes/human.py @@ -81,12 +81,16 @@ _JUDGE_KEYS = ("mode", "counts", "complete", "unjudged", "limits_hit") def reviewed_report_ids(spec: Any, human: Any) -> list[str]: - """一个人工审批节点审的是哪几份报告：上游直接连着的报告撰写节点，加上标题、说明里引用了正文的 - （{{ nodes..text }}，或者报告的 assign_to 变量）。只看图，不看运行状态。 + """一个人工审批节点审的是哪几份报告：上游直接连着的报告撰写节点，直接连着的反方审稿节点审的那份报告，加上 + 标题、说明里引用了正文的（{{ nodes..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: diff --git a/backend/app/engine/schema.py b/backend/app/engine/schema.py index db1e5692..2ddfa513 100644 --- a/backend/app/engine/schema.py +++ b/backend/app/engine/schema.py @@ -623,9 +623,11 @@ CHALLENGE_LABEL = { "timeout_s": "每句时长上限（秒）", "provider": "反方模型 · 接入", "model": "反方模型", "on_refuted": "结论被推翻时", "enabled": "启用", } -#: 结论被推翻时怎么办：degrade 计入出具缺口、出具降档（默认）/ flag 只记录 / fail 判节点失败 +#: 结论被推翻时怎么办：degrade 计入出具缺口、出具降档（默认）/ flag 只标注 / fail 判节点失败、运行停止 CHALLENGE_ON_REFUTED = ("degrade", "flag", "fail") -CHALLENGE_ON_REFUTED_LABEL = {"degrade": "出具降档", "flag": "仅记录", "fail": "判为失败"} +#: 选项的叫法和检查器下拉一致（lib/terms.ts 的 CHALLENGE_TEXT.onRefuted），只去掉「（默认）」：报错让人去改某一项时， +#: 人在下拉里找得到这个名字 +CHALLENGE_ON_REFUTED_LABEL = {"degrade": "出具降档", "flag": "只标注", "fail": "报错并停止运行"} def _challenge_field(key: str) -> str: @@ -659,7 +661,7 @@ def challenge_problems(config: dict[str, Any]) -> list[tuple[str, str]]: f"当前为「{value}」")) value = config.get("enabled") if value is not None and not isinstance(value, bool): - out.append(("enabled", f"{_challenge_field('enabled')}只能是开启、关闭或留空（沿用设置），当前为「{value}」")) + out.append(("enabled", f"{_challenge_field('enabled')}只能是开启、关闭或跟随设置，当前为「{value}」")) return out @@ -703,13 +705,36 @@ def challenge_report(spec: GraphSpec, node: GraphNode) -> tuple[str | None, str 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"「{type_label(NodeType.CHALLENGE)}」找不到要审的报告", node_id=node.id, field="report", - code="challenge.report_invalid") + 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") diff --git a/backend/tests/test_challenge_node.py b/backend/tests/test_challenge_node.py index ab0ccc71..145d1ec7 100644 --- a/backend/tests/test_challenge_node.py +++ b/backend/tests/test_challenge_node.py @@ -5,11 +5,14 @@ （judge.prepare，遮罩照旧）；理由里的数字只能引用它自己的查询格子，裸数字被标出来 - 结论四选一：经受住 / 被削弱 / 被推翻 / 无法判断；完整记录写进 challenge_artifact，随运行封存；台账只记 kind challenge， 它的查询不进任何报告的写作目录 -- 被推翻的：on_refuted 为 degrade（默认）时写 issuance_gaps，出具降档；fail 判节点失败；flag 只记录 -- 上限：句数、每句查询数、每句时长、全局每日金额；全局关闭、报告没有可审的结论句时直通 -- 整个反查放在一个 @task 里：接着跑时不重复执行、不重复记账 -- 人工审批审这份报告时，审批卡附上反方审稿的概况 -- 画布校验：要审的报告必须是上游的报告撰写节点 +- 结论要有出处：零查询的判什么都记为无法判断，推翻还得引用到自己的查询结果；没有可查的源、模型用不了时不派 agent +- 被推翻的：on_refuted 为 degrade（默认）时写 issuance_gaps，出具降档；fail 判节点失败；flag 只记录。产出和审批卡里 + 的 on_refuted 写实际生效的那一种 +- 上限：句数、每句查询数、每句时长、全局每日金额（都按发起时的设置快照）；全局关闭、报告没有可审的结论句时直通 +- 整个反查放在一个 @task 里：接着跑时不重复执行、不重复记账；改了审稿本身的配置才重审；节点重试不重审 +- 人工审批审这份报告时（含「报告 → 反方审稿 → 人工审批」的接法），审批卡附上反方审稿的概况 +- 画布校验：要审的报告必须是上游的报告撰写节点；「报错并停止运行」不能配「出错时：记录错误并继续」；不在成果节点 + 上游的「出具降档」给警告 模型一律是 MockChatModel 剧本，数据源是现造的 SQLite 文件，区域、金额、电话都是虚构的。 """ @@ -17,6 +20,7 @@ from __future__ import annotations import asyncio import json +import re import sqlite3 import uuid from dataclasses import dataclass, field @@ -36,8 +40,9 @@ from app.engine.evidence import iter_units from app.engine.judge import SPEND_KEY as JUDGE_SPEND_KEY from app.engine.nodes import challenge as challenge_mod from app.engine.nodes import report as report_mod +from app.engine.nodes.human import reviewed_by_human, reviewed_report_ids from app.engine.runner import run_manager, verify_manifest -from app.engine.schema import GraphSpec, validate_graph +from app.engine.schema import GraphSpec, ValidationResult, validate_graph from app.providers.factory import ProviderNotConfigured from app.providers.mock_model import MockChatModel @@ -121,6 +126,20 @@ class Play: rationale: str = "没有找到反例。" +@dataclass +class Fickle: + """第一次交结果判「被推翻」，之后都判「经受住」：同一句重审一遍，模型就换了说法。""" + + queries: list[str] + rationale: str + asked: int = 0 + + @property + def verdict(self) -> str: + self.asked += 1 + return "refuted" if self.asked == 1 else "stands" + + #: 默认剧本：错误结论被推翻（扣除退款后华南仍低于华东），首句被削弱（华东含退款） PLAYS = { "华南销售额超过华东": Play([NET], "refuted", "扣除退款后华南为 [[v:Q1.r1.净销售额]]，仍低于华东的 [[v:Q1.r0.净销售额]]。[[see:Q1]]"), @@ -138,6 +157,8 @@ class Models: self.writes: list[list] = [] #: 反查 agent 每句拿到的工具：{句子片段: 工具名集合} self.tools: dict[str | None, set[str]] = {} + #: 核查每次建模型时要的是哪个模型（ModelSpec）：反查一句要建两次（查库、交结果） + self.specs: list = [] plays = PLAYS if plays is None else plays def which(messages) -> str | None: @@ -169,6 +190,7 @@ class Models: monkeypatch.setattr(MockChatModel, "_decide", decide) async def verify_model(session, spec): + self.specs.append(spec) if broken is not None: raise broken return MockChatModel(model_name=model_id, latency=latency), model_id @@ -308,6 +330,7 @@ async def test_flag_only_records_and_fail_stops_the_run(monkeypatch, shop): assert failed.status == "failed" assert failed.error_node_id == "chal" assert "反方审稿推翻了 1 句结论" in failed.error and SOUTH in failed.error + assert "「结论被推翻时」改为「只标注」" in failed.error, "报错里点的选项名和检查器下拉里的一致" # 判失败之前审稿记录已经存下、时间线照常有完成那一条 [done] = await logs(failed.id, "challenge_done") record = artifact_store.load(done["artifact"]) @@ -346,6 +369,74 @@ async def test_the_verify_queries_never_reach_a_writer_downstream(monkeypatch, s assert queries == ["Q1"] and doc["catalog"]["Q1"]["node_id"] == "fetch" +# -------------------------------------------------------------------------- +# 没有出处的结论不采信 +# -------------------------------------------------------------------------- + + +async def test_a_verdict_without_any_query_is_not_taken(monkeypatch, shop): + """一条查询都没执行成功就判「被推翻」「经受住」：没有出处，记为无法判断——不让正式出具降档，也不给审批人「经受住」 + 的假安心。反方原本给的结论和理由照录在理由里，审稿记录看得出发生了什么。""" + plays = {"华南销售额超过华东": Play([], "refuted", "我认为这句不成立。"), + "居各区域之首": Play([], "stands", "没有找到反例。")} + Models(monkeypatch, shop, plays=plays) + row = await finish(graph(shop)) + assert row.status == "succeeded", row.error + assert row.output["_issuance"]["tier"] == "formal", "零查询的推翻没有出处，不能让正式出具降档" + out = await output_of(row.id, "chal") + assert out["counts"] == {"stands": 0, "weakened": 0, "refuted": 0, "inconclusive": 2} + assert "issuance_gaps" not in out and await logs(row.id, "challenge_refuted") == [] + east, south = artifact_store.load(out["challenge_artifact"])["claims"] + assert south["verdict"] == "inconclusive" and south["queries"] == [] and south["rationale_doc"] is None + assert "没有执行成功任何查询" in south["rationale"] + assert "「被推翻」" in south["rationale"] and "我认为这句不成立。" in south["rationale"] + assert east["verdict"] == "inconclusive" and "「经受住」" in east["rationale"] + [warn] = await logs(row.id, "challenge_unreviewed") + assert warn["claims_skipped"] == 2 and "没有执行成功任何查询" in warn["reason"] + + +async def test_a_refutation_that_cites_none_of_its_queries_is_not_taken(monkeypatch, shop): + """查过了，可理由里一处查询结果都没引用就判「被推翻」：推翻没有出处，记为无法判断、不降档；查过的照录。""" + plays = {"华南销售额超过华东": Play([NET], "refuted", "扣除退款后华南仍低于华东。"), + "居各区域之首": Play([REFUNDS], "stands", "没有找到反例。")} + Models(monkeypatch, shop, plays=plays) + row = await finish(graph(shop)) + assert row.status == "succeeded", row.error + assert row.output["_issuance"]["tier"] == "formal" + out = await output_of(row.id, "chal") + assert out["counts"] == {"stands": 1, "weakened": 0, "refuted": 0, "inconclusive": 1} + east, south = artifact_store.load(out["challenge_artifact"])["claims"] + assert south["verdict"] == "inconclusive" and len(south["queries"]) == 1 and south["rationale_doc"] is None + assert "没有引用" in south["rationale"] and "扣除退款后华南仍低于华东。" in south["rationale"] + assert east["verdict"] == "stands", "「经受住」不要求引用：它不改变出具" + assert await logs(row.id, "challenge_unreviewed") == [], "查过了，不算没能反查" + + +async def test_claims_whose_sources_are_gone_are_not_sent_out(monkeypatch, shop): + """报告写完、审稿开始之前数据源被停用：这句没有可查的源，不建模型、不花钱，照实记无法判断并点出是哪个源。""" + real = report_mod._store + + async def store_then_disable(doc, ctx): + out = await real(doc, ctx) + async with SessionLocal() as session: + row = (await session.execute(select(DataSource).where(DataSource.name == shop))).scalar_one() + row.enabled = False + await session.commit() + return out + + monkeypatch.setattr(report_mod, "_store", store_then_disable) + models = Models(monkeypatch, shop) + row = await finish(graph(shop)) + assert row.status == "succeeded", row.error + assert models.calls == [] and models.specs == [], "没有可查的源就不建模型、不调用" + out = await output_of(row.id, "chal") + assert out["counts"]["inconclusive"] == 2 and row.output["_issuance"]["tier"] == "formal" + claims = artifact_store.load(out["challenge_artifact"])["claims"] + assert all(f"「{shop}」" in c["rationale"] and "不存在或已停用" in c["rationale"] for c in claims) + [warn] = await logs(row.id, "challenge_unreviewed") + assert warn["claims_skipped"] == 2 and f"「{shop}」" in warn["reason"] + + # -------------------------------------------------------------------------- # 理由里的裸数字 # -------------------------------------------------------------------------- @@ -396,6 +487,31 @@ async def test_the_settings_snapshot_sets_the_limits_and_the_node_overrides_it(m assert await logs(row.id, "challenge_limit") == [] +async def test_the_snapshot_decides_the_model(monkeypatch, shop): + """反方模型也按发起时快照的核查设置：设置里现在写的是另一个模型，照样用快照里的。""" + async with SessionLocal() as session: + session.add(Setting(key="verify", value={"model": "stored-model"})) + await session.commit() + models = Models(monkeypatch, shop) + row = await finish(graph(shop), run_class="exploratory", options={"verify": {"model": "snap-model"}}) + assert row.status == "succeeded", row.error + assert models.specs and {s.model for s in models.specs} == {"snap-model"} + + +async def test_the_snapshot_decides_the_daily_cap(monkeypatch, shop): + """每日金额上限也按快照：设置现在放宽到 $5，发起时的 $0.50 当天已经用完，照样一句不反查。""" + async with SessionLocal() as session: + session.add(Setting(key="verify", value={"daily_max_usd": 5.0})) + session.add(Setting(key=spend.SPEND_KEY, value={"date": spend._today(), "usd": 1.0, "calls": 3})) + await session.commit() + models = Models(monkeypatch, shop) + row = await finish(graph(shop), options={"verify": {"daily_max_usd": 0.5}}) + assert row.status == "succeeded", row.error + assert [kind for kind, _, _ in models.calls] == [] + [limit] = await logs(row.id, "challenge_limit") + assert limit["limit"] == "daily_max_usd" and "$0.50" in limit["message"] + + async def test_without_a_snapshot_the_current_settings_apply(monkeypatch, shop): async with SessionLocal() as session: session.add(Setting(key="verify", value={"challenge": {"max_claims": 1}})) @@ -459,14 +575,15 @@ async def test_the_daily_cap_stops_the_remaining_claims(monkeypatch, shop): # -------------------------------------------------------------------------- -def gated(source: str, **challenge) -> dict: +def gated(source: str, *, message: str = "{{ nodes.write.text }}", **challenge) -> dict: + """取数 → 报告 → 反方审稿 → 人工审批 → 成果。message 是审批说明，缺省引用报告正文。""" challenge.setdefault("report", "write") nodes = [ node("start", "input"), node("fetch", "tool", tool=f"db_query__{source}", args={"sql": QUERY}), node("write", "report", "写周报", instructions="写周报"), node("chal", "challenge", "审稿", **challenge), - node("gate", "human", "主管审批", mode="approve", title="审核周报", message="{{ nodes.write.text }}"), + node("gate", "human", "主管审批", mode="approve", title="审核周报", message=message), node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]), ] spec = chain(*nodes[:-1]) @@ -510,6 +627,49 @@ async def test_a_skipped_challenge_adds_nothing_to_the_card(monkeypatch, shop): assert "challenge" not in report, "没审就不加这个键：老行为逐字不变" +async def test_report_then_challenge_then_human_shows_the_report_on_the_card(monkeypatch, shop): + """按推荐的接法「报告 → 反方审稿 → 人工审批」、审批说明里不引用正文：审批卡上照样有这份报告和它的审稿概况， + 审批人批准时知道有一句被推翻了。报告节点也认它要交给人审（两边是同一个认法，不会对不上）。""" + Models(monkeypatch, shop) + spec = gated(shop, message="请审批本周周报") + row, approval = await paused(spec) + [report] = approval.payload["reports"] + out = await output_of(row.id, "chal") + assert report["node_id"] == "write" and report["challenge"]["counts"] == out["counts"] + assert report["challenge"]["counts"]["refuted"] == 1 + parsed = GraphSpec.model_validate(spec) + assert reviewed_report_ids(parsed, parsed.node_map()["gate"]) == ["write"] + assert reviewed_by_human(parsed, "write") + + +async def test_fail_with_nothing_refuted_reports_what_took_effect(monkeypatch, shop): + """「报错并停止运行」又没有句子被推翻：走到了审批，节点产出和审批卡里的 on_refuted 写实际生效的处置（只标注）， + 不写约定之外的 fail。""" + plays = {"华南销售额超过华东": Play([NET], "stands", "没有找到反例。"), + "居各区域之首": Play([REFUNDS], "stands", "没有找到反例。")} + Models(monkeypatch, shop, plays=plays) + row, approval = await paused(gated(shop, on_refuted="fail")) + assert approval.payload["reports"][0]["challenge"]["on_refuted"] == "flag" + assert (await output_of(row.id, "chal"))["on_refuted"] == "flag" + + +def test_the_card_only_carries_the_review_of_that_report(): + """一个审批节点审两份报告，反方审稿只审了其中一份：审稿概况只挂在被审的那份上。""" + spec = spec_of(node("start", "input"), node("a", "report", "周报甲"), node("b", "report", "周报乙"), + node("chal", "challenge", report="a"), node("gate", "human"), + edges=[{"source": "start", "target": "a"}, {"source": "start", "target": "b"}, + {"source": "a", "target": "chal"}, {"source": "chal", "target": "gate"}, + {"source": "b", "target": "gate"}]) + counts = {"stands": 1, "weakened": 0, "refuted": 0, "inconclusive": 0} + items = [{"unit": "u0", "text": "周报甲的一句结论。", "verdict": "stands"}] + outputs = {"chal": {"challenge_artifact": "a1", "counts": counts, "items": items, "on_refuted": "degrade"}} + assert challenge_mod.challenge_summary(spec, outputs, "a") == { + "counts": counts, "items": items, "artifact": "a1", "on_refuted": "degrade"} + assert challenge_mod.challenge_summary(spec, outputs, "b") is None + # 直接连着的报告，和直接连着的反方审稿审的那份，都算这个审批节点审的 + assert reviewed_report_ids(spec, spec.node_map()["gate"]) == ["a", "b"] + + def test_items_keep_the_refuted_ones_when_there_are_many(): claims = [{"unit": f"u{i}", "text": f"第 {i} 句", "verdict": "stands"} for i in range(14)] claims[12]["verdict"] = "refuted" @@ -557,6 +717,67 @@ async def test_continuing_after_a_failure_does_not_run_the_agents_again(monkeypa assert row.output["_issuance"]["tier"] == "degraded" +async def test_retrying_after_a_refutation_does_not_review_again(monkeypatch, shop): + """「报错并停止运行」配了失败重试：重试沿用同一次审稿，不再派 agent——再审一遍既多花一份钱，模型换个说法还会把 + 「被推翻」重审掉，最严的处置反倒放行。""" + plays = {"华南销售额超过华东": Fickle([NET], "扣除退款后华南为 [[v:Q1.r1.净销售额]]。[[see:Q1]]"), + "居各区域之首": Play([REFUNDS], "stands", "没有找到反例。")} + models = Models(monkeypatch, shop, plays=plays) + row = await finish(graph(shop, on_refuted="fail", retries=1, retry_backoff=0.01)) + assert row.status == "failed" and "反方审稿推翻了 1 句结论" in (row.error or ""), row.error + assert [key for kind, key, _ in models.calls if kind == "final"] == ["居各区域之首", "华南销售额超过华东"] + assert len(await logs(row.id, "challenge_done")) == 1 + assert len(await logs(row.id, "node_retry")) == 1, "重试照常发生，只是不再审一遍" + + +async def test_a_fail_that_would_be_swallowed_degrades_instead(monkeypatch, shop): + """「报错并停止运行」配「出错时：记录错误并继续」（画布校验拦这个组合，这里模拟跳过了校验的旧图）：报的错会被吞掉， + 被推翻的结论照常完整出具。节点改按出具降档处置。""" + monkeypatch.setattr("app.engine.runner.validate_graph", lambda spec: ValidationResult()) + Models(monkeypatch, shop) + row = await finish(graph(shop, on_refuted="fail", on_error="continue")) + assert row.status == "succeeded", row.error + out = await output_of(row.id, "chal") + assert out["on_refuted"] == "degrade" and out["issuance_gaps"] == [f"反方审稿推翻了 1 句结论：「{SOUTH}」"] + assert row.output["_issuance"]["tier"] == "degraded" + + +async def test_continuing_with_only_the_disposition_changed_keeps_the_review(monkeypatch, shop): + """判失败后只把「结论被推翻时」改成「只标注」再接着跑（报错里就是这么建议的）：审稿沿用 checkpoint 里的，不重审。""" + models = Models(monkeypatch, shop) + run = await run_manager.start(graph=graph(shop, on_refuted="fail"), input_payload={}, run_class="formal") + row = await wait(run.id) + assert row.status == "failed", row.error + calls = len(models.calls) + await run_manager.continue_failed(run.id, graph=graph(shop, on_refuted="flag")) + row = await wait(run.id) + assert row.status == "succeeded", row.error + assert len(models.calls) == calls + out = await output_of(row.id, "chal") + assert out["on_refuted"] == "flag" and out["counts"]["refuted"] == 1 + assert row.output["_issuance"]["tier"] == "formal" + + +async def test_continuing_with_a_revised_review_reviews_again(monkeypatch, shop): + """接着跑时改了审稿本身（句数、反方模型）：checkpoint 里那份是按旧配置审的，对不上就重审，不拿旧结果冒充。""" + models = Models(monkeypatch, shop) + run = await run_manager.start(graph=graph(shop, on_refuted="fail"), input_payload={}, run_class="formal") + row = await wait(run.id) + assert row.status == "failed", row.error + assert models.reviewed() == ["居各区域之首", "华南销售额超过华东"] + models.calls.clear() + models.specs.clear() + revised = graph(shop, on_refuted="flag", max_claims=1, model="another-model") + await run_manager.continue_failed(run.id, graph=revised) + row = await wait(run.id) + assert row.status == "succeeded", row.error + assert models.reviewed() == ["居各区域之首"], "按新的句数重审" + assert models.specs and {s.model for s in models.specs} == {"another-model"}, "按新的反方模型重审" + out = await output_of(row.id, "chal") + assert [i["text"] for i in out["items"]] == [EAST] and sum(out["counts"].values()) == 1 + assert len(await logs(row.id, "challenge_done")) == 2, "两次审稿各记各的" + + # -------------------------------------------------------------------------- # 直通 # -------------------------------------------------------------------------- @@ -619,7 +840,7 @@ async def test_claims_without_any_query_behind_them_are_not_sent_out(monkeypatch async def test_an_unavailable_model_leaves_the_claims_inconclusive(monkeypatch, shop): - Models(monkeypatch, shop, broken=ProviderNotConfigured("没有配置可用的模型接入")) + models = Models(monkeypatch, shop, broken=ProviderNotConfigured("没有配置可用的模型接入")) row = await finish(graph(shop)) assert row.status == "succeeded", row.error out = await output_of(row.id, "chal") @@ -627,6 +848,12 @@ async def test_an_unavailable_model_leaves_the_claims_inconclusive(monkeypatch, record = artifact_store.load(out["challenge_artifact"]) assert all(c["rationale"].startswith("反查未完成：核查模型不可用") for c in record["claims"]) assert row.output["_issuance"]["tier"] == "formal" + assert len(models.specs) == 1, "核查模型本身用不了：后面的句子照这一次的原因记，不再逐句去建模型" + # 时间线上看得出根本没审成：不是只有一条「审了 2 句」 + [warn] = await logs(row.id, "challenge_unreviewed") + assert warn["level"] == "warn" and warn["claims_skipped"] == 2 and "核查模型不可用" in warn["reason"] + [done] = await logs(row.id, "challenge_done") + assert "2 句未能反查" in done["message"] # -------------------------------------------------------------------------- @@ -647,6 +874,20 @@ async def test_reviewing_with_the_writers_own_model_is_called_out(monkeypatch, s assert await logs(other.id, "challenge_same_model") == [] +async def test_sentences_the_judge_ruled_out_are_not_reviewed(monkeypatch, shop): + """结论句裁判已经判成「不是结论」的句子不审：同一份文档，标上之后就不在要审的句子里。""" + Models(monkeypatch, shop) + row = await finish(graph(shop), run_class="exploratory") + assert row.status == "succeeded", row.error + doc = await report_doc(row.id) + east, south = unit_of(doc, EAST), unit_of(doc, SOUTH) + assert [c.unit for c in challenge_mod.pick_claims(doc)] == [east, south] + for _, unit in iter_units(doc): + if unit["id"] == south: + unit["verdict"] = {"status": "not_a_claim"} + assert [c.unit for c in challenge_mod.pick_claims(doc)] == [east] + + async def test_the_upstream_report_is_taken_when_none_is_chosen(monkeypatch, shop): models = Models(monkeypatch, shop) spec = graph(shop) @@ -695,6 +936,40 @@ async def test_the_task_gives_the_claim_its_evidence_and_only_its_sources(monkey await engines.invalidate(source_id) +async def test_a_metric_excerpt_marks_the_report_query_numbers_too(monkeypatch, shop): + """句子挂的是口径卡指标（正式报告的主路径「取数 → 口径卡 → 报告」）：摘录正文里「取自查询 Q1」「查询 Q1 的 SQL」 + 也标成原报告的编号，和反查 agent 自己的 Q1 分开；工具按指标的输入取自的查询找源。""" + metrics = [{"id": "gmv", "name": "华东销售额", "unit": "元", "decimals": 0, + "expression": "cell(nodes.fetch, 0, '销售额')"}] + plays = {"本周华东销售额": Play([NET], "stands", "没有找到反例。")} + models = Models(monkeypatch, shop, writer=("本周华东销售额 [[m:gmv]]，居各区域之首。[[see:m:gmv]]",), plays=plays) + spec = chain( + node("start", "input"), + node("fetch", "tool", tool=f"db_query__{shop}", args={"sql": QUERY}), + node("caliber", "metrics", caliber="周报口径", caliber_version="v1", metrics=metrics), + node("write", "report", "写周报", instructions="写周报"), + node("chal", "challenge", "审稿"), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]), + ) + row = await finish(spec, run_class="exploratory") + assert row.status == "succeeded", row.error + task = models.tasks()["本周华东销售额"] + evidence = task.split("写作者给这句挂的证据摘录")[1].split("可以查询的数据源")[0] + assert "【原报告 m:gmv】" in evidence and "取自原报告 Q1 的" in evidence and "原报告 Q1 的 SQL" in evidence + assert re.findall(r"(?