diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index 32994055..33ea5d50 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -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 diff --git a/backend/app/engine/nodes/challenge.py b/backend/app/engine/nodes/challenge.py index a2348aad..f6195010 100644 --- a/backend/app/engine/nodes/challenge.py +++ b/backend/app/engine/nodes/challenge.py @@ -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"] diff --git a/backend/app/engine/nodes/human.py b/backend/app/engine/nodes/human.py index f314011b..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: @@ -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 diff --git a/backend/app/engine/schema.py b/backend/app/engine/schema.py index caf383cb..10c945d3 100644 --- a/backend/app/engine/schema.py +++ b/backend/app/engine/schema.py @@ -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): diff --git a/backend/tests/test_challenge_node.py b/backend/tests/test_challenge_node.py new file mode 100644 index 00000000..41bc16ac --- /dev/null +++ b/backend/tests/test_challenge_node.py @@ -0,0 +1,1238 @@ +"""反方审稿节点（engine/nodes/challenge.py）：报告写完后挑几句关键结论，各派一个只读 agent 站在反方找反例。 + +- 选句照结论句裁判的预筛（judge.candidates / claim_priority）：挂了依据的、带比较或因果的排前面，取前「最多审几句」句 +- 每句一个只读 agent（engine/verify_agent.py）：工具只给这句引用的查询所在的数据源，任务里给这句结论和它挂的证据摘录 + （judge.prepare，遮罩照旧）；理由里的数字只能引用它自己的查询格子，裸数字被标出来 +- 结论四选一：经受住 / 被削弱 / 被推翻 / 无法判断；完整记录写进 challenge_artifact，随运行封存；台账只记 kind challenge， + 它的查询不进任何报告的写作目录 +- 结论要有出处：零查询的判什么都记为无法判断，推翻还得引用到自己的查询结果，照录反方的理由时不露出内部标记、 + 裸数字注明无出处；没有可查的源、模型用不了时不派 agent，只停用了一部分源的照常反查，理由和时间线写明哪几个没查 +- 被推翻的：on_refuted 为 degrade（默认）时写 issuance_gaps，出具降档；fail 判节点失败；flag 只记录。产出和审批卡里 + 的 on_refuted 写实际生效的那一种 +- 上限：句数、每句查询数、每句时长、全局每日金额（都按发起时的设置快照）；全局关闭、报告没有可审的结论句时直通 +- 整个反查放在一个 @task 里：接着跑时不重复执行、不重复记账；改了审稿本身的配置才重审；节点重试不重审 +- 人工审批审这份报告时（含「报告 → 反方审稿 → 人工审批」的接法），审批卡附上反方审稿的概况 +- 画布校验：要审的报告必须是上游的报告撰写节点；「报错并停止运行」不能配「出错时：记录错误并继续」；不在成果节点 + 上游的「出具降档」给警告 + +模型一律是 MockChatModel 剧本，数据源是现造的 SQLite 文件，区域、金额、电话都是虚构的。 +""" +from __future__ import annotations + +import asyncio +import json +import re +import sqlite3 +import uuid +from dataclasses import dataclass, field + +import pytest +from langchain_core.messages import AIMessage, ToolMessage +from sqlalchemy import delete, select + +from app.core import artifact_store +from app.data import introspect +from app.data.engine import engines +from app.db.base import SessionLocal +from app.db.models import Approval, DataSource, Run, RunEvent, Setting +from app.engine import spend, verify_agent +from app.engine.context import NodeContext, NodeError, RunContext +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, ValidationResult, validate_graph +from app.providers.factory import ProviderNotConfigured +from app.providers.mock_model import MockChatModel + +PRICED = "claude-sonnet-5" +#: 各区域销售额（含退款）：华东 1500、华南 900、华北 400 +QUERY = "SELECT region AS 区域, SUM(amount) AS 销售额 FROM sales GROUP BY region ORDER BY 销售额 DESC" +#: 一句有依据的结论、一句预埋的错误结论（华南其实低于华东） +TEXT = "华东销售额 [[v:Q1.r0.销售额]]，居各区域之首。[[see:Q1]]华南销售额超过华东。[[see:Q1]]" +EAST, SOUTH = "华东销售额 1,500，居各区域之首。", "华南销售额超过华东。" +NET = "SELECT region AS 区域, SUM(amount) AS 净销售额 FROM sales WHERE refunded = 0 GROUP BY region ORDER BY 净销售额 DESC" +REFUNDS = "SELECT COUNT(*) AS 退款笔数 FROM sales WHERE refunded = 1" + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + async with SessionLocal() as session: + await session.execute(delete(Setting).where(Setting.key.in_( + [spend.SPEND_KEY, JUDGE_SPEND_KEY, "verify", "judge", "copilot"]))) + await session.commit() + yield + await run_manager.shutdown() + + +async def _source(tmp_path, *, options=None) -> tuple[str, int]: + path = tmp_path / f"{uuid.uuid4().hex[:8]}.db" + db = sqlite3.connect(path) + db.executescript( + "CREATE TABLE sales (id INTEGER PRIMARY KEY, region TEXT, buyer_phone TEXT, amount INTEGER, refunded INTEGER);" + "INSERT INTO sales VALUES (1, '华东', '13900000001', 1200, 0), (2, '华东', '13900000002', 300, 1)," + " (3, '华南', '13900000003', 900, 0), (4, '华北', '13900000004', 400, 0);") + db.commit() + db.close() + name = f"shop_{uuid.uuid4().hex[:8]}" + async with SessionLocal() as session: + row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True, + options=options or {}) + session.add(row) + await session.flush() + row.schema_cache = await introspect.introspect(row) + await session.commit() + return name, row.id + + +@pytest.fixture +async def shop(tmp_path): + """只读的手工源：四笔订单，华东两笔（其中一笔 300 已退款）、华南一笔 900、华北一笔 400。返回源名。""" + name, source_id = await _source(tmp_path) + yield name + await engines.invalidate(source_id) + + +def node(nid, ntype, label=None, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label or nid, "config": config}} + + +def chain(*nodes, extra_edges=()): + edges = [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])] + return {"nodes": list(nodes), "edges": [*edges, *extra_edges]} + + +def graph(source: str, *, sql: str = QUERY, contract: bool = True, **challenge) -> dict: + """取数 → 报告 → 反方审稿 → 成果。成果按报告文档复核（引用模式的出具契约），数字都有出处：不加反方审稿是完整出具。""" + challenge.setdefault("report", "write") + return chain( + node("start", "input"), + node("fetch", "tool", tool=f"db_query__{source}", args={"sql": sql}), + node("write", "report", "写周报", instructions="写周报"), + node("chal", "challenge", "审稿", **challenge), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}], + **({"contract": {"report_from": "write"}} if contract else {})), + ) + + +@dataclass +class Play: + """反查 agent 对某一句的剧本：依次执行的查询，最后交的结论和理由。""" + + queries: list[str] = field(default_factory=list) + verdict: str = "stands" + 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]]"), + "居各区域之首": Play([REFUNDS], "weakened", "华东含 [[v:Q1.r0.退款笔数]] 笔退款，扣除后仍居首，但销售额被高估。"), +} + + +class Models: + """写作者和反查 agent 两份剧本。写作者依次回 writer 里的几段；反查 agent 按任务里的结论认出是哪一句，照 plays 查询、 + 交结果。calls 记下反查 agent 的每次调用：(step / final, 认出的句子片段, 收到的消息)。""" + + def __init__(self, monkeypatch, source: str, *, writer=(TEXT,), plays=None, model_id=PRICED, latency=0.0, + broken: Exception | None = None): + self.calls: list[tuple[str, str | None, list]] = [] + 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: + task = str(messages[1].content) if len(messages) > 1 else "" + return next((k for k in plays if k in task), None) + + def decide(model, messages): + system = str(messages[0].content) + if challenge_mod.CHALLENGE_SYSTEM in system: + key = which(messages) + play = plays.get(key) or Play() + if model.response_format: + self.calls.append(("final", key, list(messages))) + return AIMessage(content=json.dumps({"verdict": play.verdict, "rationale": play.rationale}, + ensure_ascii=False)) + self.calls.append(("step", key, list(messages))) + self.tools.setdefault(key, set()).update(t["name"] for t in model.tools) + done = sum(1 for m in messages if isinstance(m, ToolMessage)) + if done < len(play.queries): + call = {"name": f"db_query__{source}", "args": {"sql": play.queries[done]}, + "id": f"{(key or 'x')[:4]}-{done}"} + return AIMessage(content="查一下。", tool_calls=[call]) + return AIMessage(content="查完了。") + if report_mod.ROLE in system: + self.writes.append(list(messages)) + return AIMessage(content=writer[min(len(self.writes), len(writer)) - 1]) + raise AssertionError(f"没想到的模型调用：{system[:60]}") + + 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 + + monkeypatch.setattr(verify_agent, "get_chat_model", verify_model) + + def tasks(self) -> dict[str, str]: + """反查 agent 收到的任务：{句子片段: 任务原文}。""" + return {key: str(messages[1].content) for kind, key, messages in self.calls if kind == "step" and key} + + def reviewed(self) -> list[str | None]: + return list(dict.fromkeys(key for _, key, _ in self.calls)) + + +async def finish(spec: dict, *, run_class="formal", options=None, statuses=("succeeded", "failed")) -> Run: + run = await run_manager.start(graph=spec, input_payload={}, run_class=run_class, options=options) + return await wait(run.id, statuses) + + +async def wait(run_id, statuses=("succeeded", "failed")) -> Run: + for _ in range(600): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row.status in statuses and (row.status == "interrupted" or row.manifest_seq is not None): + await run_manager.wait_idle(run_id) + return row + raise AssertionError(f"运行 {run_id} 没跑完：{row.status}") + + +async def events(run_id: str, etype: str, node_id: str | None = None) -> list[RunEvent]: + async with SessionLocal() as session: + q = select(RunEvent).where(RunEvent.run_id == run_id, RunEvent.type == etype) + if node_id: + q = q.where(RunEvent.node_id == node_id) + return list((await session.execute(q.order_by(RunEvent.seq))).scalars()) + + +async def logs(run_id: str, code: str) -> list[dict]: + return [e.data for e in await events(run_id, "log") if e.data.get("code") == code] + + +async def output_of(run_id: str, node_id: str) -> dict: + finished = await events(run_id, "node.finished", node_id) + return artifact_store.load(finished[-1].data["artifact"]) + + +async def report_doc(run_id: str) -> dict: + [checked] = await events(run_id, "report.checked", "write") + return artifact_store.load(checked.data["doc_artifact"]) + + +def unit_of(doc: dict, text: str) -> str: + md = doc["markdown"] + return next(u["id"] for _, u in iter_units(doc) if md[u["span"][0]:u["span"][1]].strip() == text) + + +COUNT_KEYS = ("stands", "weakened", "refuted", "inconclusive") + + +# -------------------------------------------------------------------------- +# 被推翻 → 出具降档 +# -------------------------------------------------------------------------- + + +async def test_a_refuted_conclusion_degrades_the_issuance(monkeypatch, shop): + models = Models(monkeypatch, shop) + row = await finish(graph(shop)) + assert row.status == "succeeded", row.error + assert models.reviewed() == ["居各区域之首", "华南销售额超过华东"], "按优先级、再按文档先后逐句反查" + + issuance = row.output["_issuance"] + assert issuance["tier"] == "degraded" + assert issuance["gaps"] == [f"反方审稿「审稿」：反方审稿推翻了 1 句结论：「{SOUTH}」"], "缺口不带节点名，前缀由出具契约加" + + doc = await report_doc(row.id) + east, south = unit_of(doc, EAST), unit_of(doc, SOUTH) + out = await output_of(row.id, "chal") + assert out["counts"] == {"stands": 0, "weakened": 1, "refuted": 1, "inconclusive": 0} + assert set(out["counts"]) == set(COUNT_KEYS) + assert out["items"] == [{"unit": east, "text": EAST, "verdict": "weakened"}, + {"unit": south, "text": SOUTH, "verdict": "refuted"}] + assert out["on_refuted"] == "degrade" + assert out["issuance_gaps"] == [f"反方审稿推翻了 1 句结论：「{SOUTH}」"] + + record = artifact_store.load(out["challenge_artifact"]) + assert set(record) == {"report", "doc_artifact", "claims", "model"} + assert record["report"] == "write" and record["model"] == PRICED + [checked] = await events(row.id, "report.checked", "write") + assert record["doc_artifact"] == checked.data["doc_artifact"] + first, second = record["claims"] + assert set(second) == {"unit", "text", "verdict", "rationale", "rationale_doc", "queries", "usage"} + assert (second["unit"], second["text"], second["verdict"]) == (south, SOUTH, "refuted") + assert second["rationale"] == "扣除退款后华南为 900，仍低于华东的 1,200。" + assert second["rationale_doc"]["violations"] == [] + [query] = second["queries"] + assert set(query) == {"alias", "artifact", "sql", "row_count", "truncated", "source"} + assert query["alias"] == "Q1" and query["source"] == shop and "refunded = 0" in query["sql"] + assert artifact_store.load(query["artifact"])["rows"] == [["华东", 1200], ["华南", 900], ["华北", 400]] + assert query["row_count"] == 3 and query["truncated"] is False + assert second["usage"]["calls"] >= 2 and second["usage"]["input_tokens"] > 0 + assert first["verdict"] == "weakened" and first["rationale"] == "华东含 1 笔退款，扣除后仍居首，但销售额被高估。" + + # 时间线：完成、推翻各一条，按约定带结构化字段 + [done] = await logs(row.id, "challenge_done") + assert done["counts"] == out["counts"] and done["artifact"] == out["challenge_artifact"] + [refuted] = await logs(row.id, "challenge_refuted") + assert refuted["items"] == [{"unit": south, "text": SOUTH}] + # 反查的查询发 tool.end（role: verify，带查询快照），可下钻、随运行封存 + ends = [e.data for e in await events(row.id, "tool.end", "chal")] + assert len(ends) == 2 and all(d["role"] == "verify" and d.get("query_artifact") for d in ends) + assert {d["query_artifact"] for d in ends} == {first["queries"][0]["artifact"], query["artifact"]} + llm_ends = [e.data for e in await events(row.id, "llm.end", "chal")] + assert llm_ends and all(d["role"] == "verify" for d in llm_ends) + # 台账只记 kind challenge，绝不记 query：核查的查询不能被当成报告可以引用的证据 + [finished] = await events(row.id, "node.finished", "chal") + assert finished.data["evidence"] == [{"kind": "challenge", "node_id": "chal", "exec": 1, + "artifact": out["challenge_artifact"]}] + # 节点用量记进运行用量 + assert row.usage["calls"] >= sum(c["usage"]["calls"] for c in record["claims"]) + verdict = await verify_manifest(row.id) + assert verdict["ok"] is True + assert finished.seq <= row.manifest_seq + + +async def test_flag_only_records_and_fail_stops_the_run(monkeypatch, shop): + Models(monkeypatch, shop) + row = await finish(graph(shop, on_refuted="flag")) + assert row.status == "succeeded", row.error + assert row.output["_issuance"]["tier"] == "formal" and row.output["_issuance"]["gaps"] == [] + out = await output_of(row.id, "chal") + assert out["on_refuted"] == "flag" and "issuance_gaps" not in out and out["counts"]["refuted"] == 1 + assert len(await logs(row.id, "challenge_refuted")) == 1, "flag 也照实记一条" + + Models(monkeypatch, shop) + failed = await finish(graph(shop, on_refuted="fail")) + assert failed.status == "failed" + assert failed.error_node_id == "chal" + assert "反方审稿推翻了 1 句结论" in failed.error and SOUTH in failed.error + assert "请核对这句结论" in failed.error, "只推翻了一句，不说「这几句」" + assert "「结论被推翻时」改为「只标注」" in failed.error, "报错里点的选项名和检查器下拉里的一致" + # 判失败之前审稿记录已经存下、时间线照常有完成那一条 + [done] = await logs(failed.id, "challenge_done") + record = artifact_store.load(done["artifact"]) + assert [c["verdict"] for c in record["claims"]] == ["weakened", "refuted"] + + +async def test_nothing_refuted_means_no_gap(monkeypatch, shop): + plays = {"华南销售额超过华东": Play([NET], "inconclusive", "查不到判断所需的数据。"), + "居各区域之首": 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} + assert "issuance_gaps" not in out + assert await logs(row.id, "challenge_refuted") == [] + + +async def test_the_verify_queries_never_reach_a_writer_downstream(monkeypatch, shop): + """反方审稿之后再写一份报告：它的证据目录里只有原来的查询（Q1），没有反查 agent 的查询。""" + Models(monkeypatch, shop, writer=(TEXT, "华东销售额 [[v:Q1.r0.销售额]]。[[see:Q1]]")) + spec = chain( + node("start", "input"), + node("fetch", "tool", tool=f"db_query__{shop}", args={"sql": QUERY}), + node("write", "report", "写周报", instructions="写周报"), + node("chal", "challenge", "审稿", report="write"), + node("again", "report", "复写", instructions="再写一遍"), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.again.text }}"}]), + ) + row = await finish(spec, run_class="exploratory") + assert row.status == "succeeded", row.error + [checked] = await events(row.id, "report.checked", "again") + doc = artifact_store.load(checked.data["doc_artifact"]) + queries = [a for a, e in doc["catalog"].items() if e.get("kind") == "query"] + 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") == [], "查过了，不算没能反查" + + +@pytest.mark.parametrize("queries, said, quoted", [ + ([NET], "扣除退款后华南为 [[v:Q1.r99.净销售额]]。", "扣除退款后华南为（引用无效）。"), + ([NET], "扣除退款后华南为 [[v:Q2.r1.净销售额]] 元。", "扣除退款后华南为（引用无效）元。"), + ([], "华南为 [[v:Q1.r1.净销售额]]，不成立。", "华南为（引用无效），不成立。"), + ([NET], "扣除退款后华南为 900，低于华东 1200。", "扣除退款后华南为 900（无出处），低于华东 1200（无出处）。"), +]) +async def test_an_unfounded_rationale_is_quoted_in_plain_words(monkeypatch, shop, queries, said, quoted): + """不采信时照录反方的理由，界面按纯文字显示（没有理由文档）：解析不了的引用不能把内部的占位标记露出来，换成 + 「（引用无效）」；裸数字逐个注明「（无出处）」——文档里它们会被标出来，照录成文字时也得标。""" + plays = {"华南销售额超过华东": Play(queries, "refuted", said), + "居各区域之首": Play([REFUNDS], "stands", "没有找到反例。")} + Models(monkeypatch, shop, plays=plays) + row = await finish(graph(shop), run_class="exploratory") + assert row.status == "succeeded", row.error + south = artifact_store.load((await output_of(row.id, "chal"))["challenge_artifact"])["claims"][1] + assert south["verdict"] == "inconclusive" and south["rationale_doc"] is None + assert south["rationale"].endswith(f"「被推翻」不予采信，记为无法判断。反方的理由：{quoted}"), south["rationale"] + assert "⟦" not in south["rationale"] and "[[" not in south["rationale"] + + +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"] + + +async def test_a_claim_with_part_of_its_sources_gone_says_so(monkeypatch, tmp_path, shop): + """两个数据源的报告，审稿开始前停用了其中一个：引用两个源的那句、没挂依据的那句都只能在剩下的源里查。判断照常 + 采信，但理由末尾由系统写明哪个源没查（理由文档里也有：界面按文档画），时间线上另记一条——不然「经受住」看着像 + 两个源都查过了。""" + other, other_id = await _source(tmp_path) + try: + text = ("华东销售额 [[v:Q1.r0.销售额]]，退款只有 [[v:Q2.r0.退款笔数]] 笔。[[see:Q1]][[see:Q2]]" + "增长主要来自华东。") + plays = {"退款只有": Play([NET], "stands", "没有找到反例。"), + "增长主要来自": Play([NET], "weakened", "华东净销售额为 [[v:Q1.r0.净销售额]]，增长说过了头。")} + real = report_mod._store + + async def store_then_disable(doc, ctx): + out = await real(doc, ctx) + async with SessionLocal() as session: + row = await session.get(DataSource, other_id) + row.enabled = False + await session.commit() + return out + + monkeypatch.setattr(report_mod, "_store", store_then_disable) + models = Models(monkeypatch, shop, writer=(text,), plays=plays) + spec = chain( + node("start", "input"), + node("fetch", "tool", tool=f"db_query__{shop}", args={"sql": QUERY}), + node("fetch2", "tool", tool=f"db_query__{other}", args={"sql": REFUNDS}), + 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 + assert models.reviewed() == ["退款只有", "增长主要来自"] + assert models.tools["退款只有"] == {f"db_schema__{shop}", f"db_query__{shop}"}, "停用的源不给工具" + task = models.tasks()["退款只有"] + assert f"「{other}」" in task and "已不存在或已停用" in task, "给 agent 的任务里也说了，免得它把查不到当成没有反例" + assert "这份报告还用到了" in models.tasks()["增长主要来自"], "没挂依据的那句不说「这句引用了」" + + record = artifact_store.load((await output_of(row.id, "chal"))["challenge_artifact"]) + both, grow = record["claims"] + assert (both["verdict"], grow["verdict"]) == ("stands", "weakened"), "查过、有出处的判断照常采信" + assert both["rationale"] == (f"没有找到反例。\n\n注：这句引用的数据源「{other}」已不存在或已停用，未能查询，" + "以上判断不含其中的数据。") + assert grow["rationale"] == (f"华东净销售额为 1,200，增长说过了头。\n\n注：这份报告用到的数据源「{other}」" + "已不存在或已停用，未能查询，以上判断不含其中的数据。") + for claim in (both, grow): + doc = claim["rationale_doc"] + assert doc["markdown"] == claim["rationale"], "界面按文档画：说明要在文档里" + segments = [s for _, u in iter_units(doc) for s in u["segments"]] + assert "".join(s["text"] for s in segments) == doc["markdown"] + assert [s["id"] for s in segments] == [f"s{i}" for i in range(len(segments))] + *_, note = iter_units(doc) + assert note[1]["kind"] == "connective" and note[1]["cites"] == [], "说明不是反方的结论，也不挂出处" + cited = [s for _, u in iter_units(grow["rationale_doc"]) for s in u["segments"] if s.get("cite")] + assert [s["text"] for s in cited] == ["1,200"], "反方自己的引用照旧解析" + + [lost] = await logs(row.id, "challenge_sources_lost") + assert lost["level"] == "warn" and lost["claims"] == 2 and lost["sources"] == [other] + assert f"「{other}」" in lost["message"] and "2 句" in lost["message"] + assert await logs(row.id, "challenge_unreviewed") == [], "两句都查过了，不算没能反查" + finally: + await engines.invalidate(other_id) + + +def test_the_note_reads_like_one_more_paragraph_of_the_rationale(): + """理由末尾的系统说明照 compose_doc 的样子接上：和把说明写进原文再组装出来的块、句、片段、统计一模一样（片段的 + 偏移连着、编号接着往下数），只是反方的原文（source）不动、传进来的那份文档不改。""" + from app.engine.evidence import compose_doc + + said = "华南只有 900，低于华东。\n\n- 华北为 [[v:Q1.r2.销售额]]" + note = "注：这句引用的数据源「仓库」已不存在或已停用，未能查询，以上判断不含其中的数据。" + doc = compose_doc(said, {}, node_id="chal") + before = json.dumps(doc, ensure_ascii=False, sort_keys=True) + noted = challenge_mod.with_note(doc, note) + again = compose_doc(f"{said}\n\n{note}", {}, node_id="chal") + assert noted["markdown"] == again["markdown"] and noted["blocks"] == again["blocks"] + assert noted["stats"] == again["stats"] + assert noted["source"] == doc["source"] == said + assert json.dumps(doc, ensure_ascii=False, sort_keys=True) == before + + +# -------------------------------------------------------------------------- +# 理由里的裸数字 +# -------------------------------------------------------------------------- + + +async def test_bare_numbers_in_the_rationale_are_flagged(monkeypatch, shop): + plays = {"华南销售额超过华东": Play([NET], "refuted", "华南只有 900，低于华东的 [[v:Q1.r0.净销售额]]。"), + "居各区域之首": Play([REFUNDS], "stands", "没有找到反例。")} + Models(monkeypatch, shop, plays=plays) + row = await finish(graph(shop), run_class="exploratory") + assert row.status == "succeeded", row.error + record = artifact_store.load((await output_of(row.id, "chal"))["challenge_artifact"]) + south = record["claims"][1] + assert south["rationale"] == "华南只有 900，低于华东的 1,200。" + bare = [v for v in south["rationale_doc"]["violations"] if v["code"] == "uncited_number"] + assert [v["text"] for v in bare] == ["900"], "自己写的数字不采信，标成没有出处" + assert south["rationale_doc"]["stats"]["uncited_numbers"] == 1 + + +# -------------------------------------------------------------------------- +# 上限 +# -------------------------------------------------------------------------- + + +async def test_the_claim_limit_reviews_the_first_claims_only(monkeypatch, shop): + models = Models(monkeypatch, shop) + row = await finish(graph(shop, max_claims=1)) + assert row.status == "succeeded", row.error + assert models.reviewed() == ["居各区域之首"] + out = await output_of(row.id, "chal") + assert [i["text"] for i in out["items"]] == [EAST] and sum(out["counts"].values()) == 1 + assert row.output["_issuance"]["tier"] == "formal", "没审到那句错误结论：不能凭空降档" + [limit] = await logs(row.id, "challenge_limit") + assert limit["limit"] == "max_claims" and limit["claims_skipped"] == 1 + assert "1 句" in limit["message"] + + +async def test_the_settings_snapshot_sets_the_limits_and_the_node_overrides_it(monkeypatch, shop): + """上限按发起时快照的核查设置（run.options["verify"]）；节点上写了的盖过它。""" + models = Models(monkeypatch, shop) + row = await finish(graph(shop), options={"verify": {"challenge": {"max_claims": 1}}}) + assert row.status == "succeeded", row.error + assert models.reviewed() == ["居各区域之首"] + + models = Models(monkeypatch, shop) + row = await finish(graph(shop, max_claims=2), options={"verify": {"challenge": {"max_claims": 1}}}) + assert models.reviewed() == ["居各区域之首", "华南销售额超过华东"] + 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}})) + await session.commit() + models = Models(monkeypatch, shop) + row = await finish(graph(shop)) + assert row.status == "succeeded", row.error + assert models.reviewed() == ["居各区域之首"] + + +async def test_the_query_limit_applies_to_each_claim(monkeypatch, shop): + plays = {"华南销售额超过华东": Play([NET, REFUNDS], "refuted", "华南为 [[v:Q1.r1.净销售额]]。"), + "居各区域之首": Play([REFUNDS], "stands", "没有找到反例。")} + Models(monkeypatch, shop, plays=plays) + row = await finish(graph(shop, max_queries=1), run_class="exploratory") + assert row.status == "succeeded", row.error + record = artifact_store.load((await output_of(row.id, "chal"))["challenge_artifact"]) + south = record["claims"][1] + assert [q["alias"] for q in south["queries"]] == ["Q1"], "第二条查询超过每句的查询次数上限，没有执行" + assert south["verdict"] == "refuted" and south["rationale"] == "华南为 900。", "触顶照样按查到的结果收尾" + [limit] = await logs(row.id, "challenge_limit") + assert limit["limit"] == "max_queries" and limit["claims_skipped"] == 0 + + +async def test_the_time_limit_applies_to_each_claim_and_still_settles(monkeypatch, shop): + monkeypatch.setattr(verify_agent, "SETTLE_GRACE_S", 2.0) + endless = [f"SELECT COUNT(*) AS n FROM sales WHERE id > {i}" for i in range(40)] + plays = {"华南销售额超过华东": Play(endless, "inconclusive", "时间到了，没能查完。"), + "居各区域之首": Play([], "stands", "没有找到反例。")} + Models(monkeypatch, shop, plays=plays, latency=0.1) + row = await finish(graph(shop, timeout_s=0.4), run_class="exploratory") + assert row.status == "succeeded", row.error + record = artifact_store.load((await output_of(row.id, "chal"))["challenge_artifact"]) + south = record["claims"][1] + assert south["verdict"] == "inconclusive" and 1 <= len(south["queries"]) < 20 + limits = await logs(row.id, "challenge_limit") + assert [(lm["limit"], lm["claims_skipped"]) for lm in limits] == [("timeout_s", 0)] + + +async def test_the_daily_cap_stops_the_remaining_claims(monkeypatch, shop): + """当天的核查金额已经用完：一句都不反查，照实记成无法判断，不降档。""" + async with SessionLocal() as session: + session.add(Setting(key=spend.SPEND_KEY, value={"date": spend._today(), "usd": 1.9999, "calls": 3})) + await session.commit() + models = Models(monkeypatch, shop) + row = await finish(graph(shop), options={"verify": {"daily_max_usd": 2.0}}) + assert row.status == "succeeded", row.error + assert [kind for kind, _, _ in models.calls] == [], "没钱就不调用模型" + out = await output_of(row.id, "chal") + assert out["counts"] == {"stands": 0, "weakened": 0, "refuted": 0, "inconclusive": 2} + record = artifact_store.load(out["challenge_artifact"]) + words = "已达上限（全局每日核查金额上限 $2.00），这句未反查" + assert [c["rationale"] for c in record["claims"]] == [words, words], \ + "第一句派出去就撞上上限、一次模型都没调用，和后面不再派出的句子是同一回事，同一种说法" + [limit] = await logs(row.id, "challenge_limit") + assert limit["limit"] == "daily_max_usd" and limit["claims_skipped"] == 2 + assert len(await logs(row.id, "verify_limit")) == 1, "第一句撞上每日上限之后，后面的句子不再派 agent 去试" + assert row.output["_issuance"]["tier"] == "formal" + + +# -------------------------------------------------------------------------- +# 审批卡 +# -------------------------------------------------------------------------- + + +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=message), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]), + ] + spec = chain(*nodes[:-1]) + spec["nodes"].append(nodes[-1]) + spec["edges"].append({"source": "gate", "target": "out", "sourceHandle": "approved"}) + return spec + + +async def paused(spec: dict, *, options=None) -> tuple[Run, Approval]: + row = await finish(spec, run_class="exploratory", options=options, statuses=("interrupted", "failed", "succeeded")) + assert row.status == "interrupted", row.error + async with SessionLocal() as session: + approval = (await session.execute(select(Approval).where(Approval.run_id == row.id))).scalar_one() + return row, approval + + +async def test_the_approval_card_carries_the_challenge_summary(monkeypatch, shop): + Models(monkeypatch, shop) + row, approval = await paused(gated(shop)) + [report] = approval.payload["reports"] + out = await output_of(row.id, "chal") + assert report["node_id"] == "write" + assert report["challenge"] == {"counts": out["counts"], "items": out["items"], + "artifact": out["challenge_artifact"], "on_refuted": "degrade"} + assert report["challenge"]["counts"]["refuted"] == 1 + # human.requested 事件里也带着：回放时界面一样读得到 + [requested] = await events(row.id, "human.requested", "gate") + assert requested.data["reports"] == approval.payload["reports"] + + +async def test_the_approval_card_shows_what_flag_means(monkeypatch, shop): + Models(monkeypatch, shop) + _, approval = await paused(gated(shop, on_refuted="flag")) + assert approval.payload["reports"][0]["challenge"]["on_refuted"] == "flag" + + +async def test_a_skipped_challenge_adds_nothing_to_the_card(monkeypatch, shop): + Models(monkeypatch, shop) + _, approval = await paused(gated(shop, enabled=False)) + [report] = approval.payload["reports"] + 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" + claims[13]["verdict"] = "weakened" + items = challenge_mod.pick_items(claims) + assert len(items) == 10 + assert [i["unit"] for i in items][-2:] == ["u12", "u13"], "最多 10 条：被推翻、被削弱的不能被挤掉，顺序照原来的" + assert [i["unit"] for i in items][:8] == [f"u{i}" for i in range(8)] + + +# -------------------------------------------------------------------------- +# 重放 +# -------------------------------------------------------------------------- + + +async def test_continuing_after_a_failure_does_not_run_the_agents_again(monkeypatch, shop): + models = Models(monkeypatch, shop) + real = challenge_mod._finish + broken = {"on": True} + + def flaky(*args, **kwargs): + if broken["on"]: + raise RuntimeError("落盘时断了") + return real(*args, **kwargs) + + monkeypatch.setattr(challenge_mod, "_finish", flaky) + run = await run_manager.start(graph=graph(shop), input_payload={}, run_class="formal") + row = await wait(run.id) + assert row.status == "failed" and "落盘时断了" in (row.error or ""), row.error + calls = len(models.calls) + assert calls > 0 + before = await spend.daily_spend() + verify_ends = len(await events(run.id, "llm.end", "chal")) + tool_ends = len(await events(run.id, "tool.end", "chal")) + + broken["on"] = False + await run_manager.continue_failed(run.id) + row = await wait(run.id) + assert row.status == "succeeded", row.error + assert len(models.calls) == calls, "接着跑时反查 agent 又执行了一遍：反查没进 checkpoint" + assert await spend.daily_spend() == before, "接着跑时核查的每日账又记了一次" + assert len(await events(run.id, "llm.end", "chal")) == verify_ends + assert len(await events(run.id, "tool.end", "chal")) == tool_ends + assert len(await logs(row.id, "challenge_done")) == 1 + 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, "两次审稿各记各的" + + +# -------------------------------------------------------------------------- +# 直通 +# -------------------------------------------------------------------------- + + +async def test_switched_off_globally_it_passes_through(monkeypatch, shop): + models = Models(monkeypatch, shop) + row = await finish(graph(shop), options={"verify": {"challenge": {"enabled": False}}}) + assert row.status == "succeeded", row.error + assert models.calls == [] + [skipped] = await logs(row.id, "challenge_skipped") + assert skipped["reason"] == "disabled" and skipped["level"] == "info" + out = await output_of(row.id, "chal") + assert out == {"skipped": True, "reason": "disabled"} + assert row.output["_issuance"]["tier"] == "formal" + assert await events(row.id, "llm.end", "chal") == [] + + +async def test_the_node_can_turn_it_on_or_off(monkeypatch, shop): + models = Models(monkeypatch, shop) + row = await finish(graph(shop, enabled=True), options={"verify": {"challenge": {"enabled": False}}}) + assert models.reviewed() == ["居各区域之首", "华南销售额超过华东"], "节点写了 enabled: true，全局关闭也照审" + assert row.output["_issuance"]["tier"] == "degraded" + + models = Models(monkeypatch, shop) + row = await finish(graph(shop, enabled=False)) + assert models.calls == [] and (await logs(row.id, "challenge_skipped"))[0]["reason"] == "disabled" + + +async def test_a_report_without_conclusions_passes_through(monkeypatch, shop): + models = Models(monkeypatch, shop, writer=("下面看细节。",)) + row = await finish(graph(shop, contract=False), run_class="exploratory") + assert row.status == "succeeded", row.error + assert models.calls == [] + [skipped] = await logs(row.id, "challenge_skipped") + assert skipped["reason"] == "no_claims" + assert await output_of(row.id, "chal") == {"skipped": True, "reason": "no_claims"} + + +async def test_claims_without_any_query_behind_them_are_not_sent_out(monkeypatch): + """报告的证据里没有一次数据库查询（数字来自整形节点）：没有数据源可查，不派 agent，照实记无法判断。""" + metrics = [{"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, "expression": "vars.kpi.gmv"}] + models = Models(monkeypatch, "unused", writer=("本周销售额 [[m:gmv]]，比上周有所增长。[[see:m:gmv]]",)) + spec = chain( + node("start", "input"), + node("fetch", "transform", mode="expression", expression="{'gmv': 45678.5}", assign_to="kpi"), + 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 + assert models.calls == [] + out = await output_of(row.id, "chal") + assert out["counts"] == {"stands": 0, "weakened": 0, "refuted": 0, "inconclusive": 1} + [claim] = artifact_store.load(out["challenge_artifact"])["claims"] + assert claim["verdict"] == "inconclusive" and "数据库查询" in claim["rationale"] + assert claim["queries"] == [] and claim["rationale_doc"] is None + + +async def test_an_unavailable_model_leaves_the_claims_inconclusive(monkeypatch, shop): + models = Models(monkeypatch, shop, broken=ProviderNotConfigured("没有配置可用的模型接入")) + row = await finish(graph(shop)) + assert row.status == "succeeded", row.error + out = await output_of(row.id, "chal") + assert out["counts"]["inconclusive"] == 2 + 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"] + + +# -------------------------------------------------------------------------- +# 同模型提示、默认的报告、摘录遮罩 +# -------------------------------------------------------------------------- + + +async def test_reviewing_with_the_writers_own_model_is_called_out(monkeypatch, shop): + Models(monkeypatch, shop, model_id="mock-fast") + row = await finish(graph(shop), run_class="exploratory") + assert row.status == "succeeded", row.error + [warn] = await logs(row.id, "challenge_same_model") + assert warn["model"] == "mock-fast" and warn["level"] == "warn" + assert "mock-fast" in warn["message"] + + Models(monkeypatch, shop) + other = await finish(graph(shop), run_class="exploratory") + 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) + spec["nodes"][3]["data"]["config"].pop("report") + row = await finish(spec, run_class="exploratory") + assert row.status == "succeeded", row.error + assert models.reviewed() == ["居各区域之首", "华南销售额超过华东"] + assert artifact_store.load((await output_of(row.id, "chal"))["challenge_artifact"])["report"] == "write" + + +@pytest.mark.parametrize("when", ["at_query", "after_query"]) +async def test_the_task_gives_the_claim_its_evidence_and_only_its_sources(monkeypatch, tmp_path, when): + """任务里给这句结论和它挂的证据摘录：遮罩的列照旧不给——查询当时设的（记在查询快照里），和数据源现在设的 + （运行中途才设上遮罩，同证据面板的口径）都算；工具只给这句引用的查询所在的数据源。""" + masked = {"mask_columns": ["buyer_phone"]} + name, source_id = await _source(tmp_path, options=masked if when == "at_query" else None) + try: + if when == "after_query": + real = report_mod._store + + async def store_then_mask(doc, ctx): + # 报告写完、反方审稿开始之前，管理员给数据源设上了遮罩 + out = await real(doc, ctx) + async with SessionLocal() as session: + row = await session.get(DataSource, source_id) + row.options = masked + await session.commit() + return out + + monkeypatch.setattr(report_mod, "_store", store_then_mask) + sql = "SELECT region AS 区域, buyer_phone, amount AS 金额 FROM sales ORDER BY id" + plays = {"华东有一笔": Play(["SELECT COUNT(*) AS 笔数 FROM sales WHERE region = '华东'"], "stands", "没有找到反例。")} + models = Models(monkeypatch, name, writer=("华东有一笔 [[v:Q1.r0.金额]] 的订单。[[see:Q1]]",), plays=plays) + row = await finish(graph(name, sql=sql), run_class="exploratory") + assert row.status == "succeeded", row.error + task = models.tasks()["华东有一笔"] + assert "华东有一笔 1,200 的订单。" in task + assert "【原报告 Q1】" in task and "SELECT region AS 区域" in task, "摘录里有这句挂的查询，编号标明属于原报告" + assert "13900000001" not in task, "遮罩的列不给" + assert "buyer_phone" not in task.split("列：")[1].split("\n")[0] + assert "遮罩" in task + assert models.tools["华东有一笔"] == {f"db_schema__{name}", f"db_query__{name}"} + record = artifact_store.load((await output_of(row.id, "chal"))["challenge_artifact"]) + assert [q["source"] for q in record["claims"][0]["queries"]] == [name] + finally: + 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"(? GraphSpec: + if edges is None: + edges = [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])] + return GraphSpec.model_validate({"nodes": list(nodes), "edges": edges}) + + +def problems(spec: GraphSpec, nid: str = "chal") -> list[tuple[str | None, str, str | None]]: + return [(i.field, i.message, i.code) for i in validate_graph(spec).issues if i.level == "error" and i.node_id == nid] + + +BASE = [node("start", "input"), node("fetch", "tool", tool="db_query__shop", args={"sql": "SELECT 1"}), + node("write", "report", "写周报", instructions="写周报")] + + +def test_the_report_must_be_an_upstream_report_writer(): + ok = spec_of(*BASE, node("chal", "challenge", report="write")) + assert problems(ok) == [] + assert problems(spec_of(*BASE, node("chal", "challenge"))) == [], "不写就取上游唯一的报告" + + [(fld, message, code)] = problems(spec_of(*BASE, node("chal", "challenge", report="fetch"))) + assert fld == "report" and code == "challenge.report_invalid" + assert "「fetch」" in message and "调用工具" in message and "报告撰写" in message + + [(_, message, _)] = problems(spec_of(*BASE, node("chal", "challenge", report="ghost"))) + assert "「ghost」" in message and "不存在" in message + + later = spec_of(*BASE, node("chal", "challenge", report="late"), node("late", "report", "后写的报告")) + [(_, message, _)] = problems(later) + assert "后写的报告" in message and "上游" in message + + [(fld, message, _)] = problems(spec_of(node("start", "input"), node("chal", "challenge"))) + assert fld == "report" and "没有" in message and "报告撰写" in message + + two = spec_of(node("start", "input"), node("a", "report", "周报甲"), node("b", "report", "周报乙"), + node("chal", "challenge"), + edges=[{"source": "start", "target": "a"}, {"source": "start", "target": "b"}, + {"source": "a", "target": "chal"}, {"source": "b", "target": "chal"}]) + [(fld, message, _)] = problems(two) + assert fld == "report" and "周报甲" in message and "周报乙" in message and "选择" in message + + # 两份报告前后相接：直接连着的那一份就是要审的 + stacked = spec_of(node("start", "input"), node("a", "report", "初稿"), node("b", "report", "终稿"), + node("chal", "challenge")) + assert problems(stacked) == [] + from app.engine.schema import challenge_report + + assert challenge_report(stacked, stacked.node_map()["chal"]) == ("b", None) + + +@pytest.mark.parametrize("config, field_name", [ + ({"max_claims": 0}, "max_claims"), ({"max_claims": 1.5}, "max_claims"), ({"max_claims": True}, "max_claims"), + ({"max_queries": -1}, "max_queries"), ({"timeout_s": 0}, "timeout_s"), ({"timeout_s": "30"}, "timeout_s"), + ({"on_refuted": "withhold"}, "on_refuted"), ({"enabled": "yes"}, "enabled"), ({"model": 3}, "model"), + ({"provider": ["a"]}, "provider"), +]) +def test_bad_settings_are_named_in_chinese(config, field_name): + [(fld, message, code)] = problems(spec_of(*BASE, node("chal", "challenge", report="write", **config))) + assert fld == field_name and code == "challenge.config_invalid" + assert field_name not in message, "写界面上的叫法，不写键名" + assert message.startswith("「反方审稿」的「") + if field_name == "on_refuted": + assert "只能选「出具降档」、「只标注」或「报错并停止运行」" in message, "选项名和检查器下拉里的一致" + if field_name == "enabled": + assert "开启、关闭或跟随设置" in message + + +def test_good_settings_pass(): + spec = spec_of(*BASE, node("chal", "challenge", report="write", max_claims=3, max_queries=2, timeout_s=45.5, + on_refuted="flag", enabled=None, provider=None, model="claude-sonnet-5")) + assert problems(spec) == [] + + +def test_fail_with_continue_on_error_is_refused(): + """「报错并停止运行」配「出错时：记录错误并继续」：报的错会被当成普通错误跳过，被推翻的结论照常完整出具——比默认的 + 降档还松。挡住运行，说清怎么改。""" + spec = spec_of(*BASE, node("chal", "challenge", report="write", on_refuted="fail", on_error="continue")) + [(fld, message, code)] = problems(spec) + assert fld == "on_refuted" and code == "challenge.config_invalid" + assert "「报错并停止运行」" in message and "「记录错误并继续」" in message and "完整出具" in message + assert "on_error" not in message and "continue" not in message + for config in ({"on_refuted": "degrade", "on_error": "continue"}, {"on_refuted": "fail", "retries": 2}): + assert problems(spec_of(*BASE, node("chal", "challenge", report="write", **config))) == [], config + + +def test_a_challenge_off_the_issuance_path_is_called_out(): + """反方审稿接在侧枝上（报告 → 成果、报告 → 反方审稿）：它写的出具缺口没有成果节点收，按默认的「出具降档」也不会 + 降档。画布上说破；接在成果之前、没有出具契约、不靠降档的处置都不提醒。""" + def warned(nodes, edges=None): + return [i for i in validate_graph(spec_of(*nodes, edges=edges)).issues if i.code == "challenge.no_issuance"] + + out = node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}], + contract={"report_from": "write"}) + side = [{"source": "start", "target": "fetch"}, {"source": "fetch", "target": "write"}, + {"source": "write", "target": "chal"}, {"source": "write", "target": "out"}] + [issue] = warned([*BASE, node("chal", "challenge", "审稿", report="write"), out], side) + assert issue.level == "warning" and issue.node_id == "chal" + assert "成果" in issue.message and "降档" in issue.message + assert warned([*BASE, node("chal", "challenge", "审稿", report="write"), out]) == [], "接在成果之前" + loose = node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]) + assert warned([*BASE, node("chal", "challenge", "审稿", report="write"), loose], side) == [], "没有出具契约" + for disposition in ("flag", "fail"): + assert warned([*BASE, node("chal", "challenge", "审稿", report="write", on_refuted=disposition), out], + side) == [], disposition + + +def test_reviewing_with_the_writer_model_is_a_warning(): + spec = spec_of(node("start", "input"), node("write", "report", "写周报", model="claude-sonnet-5"), + node("chal", "challenge", model="claude-sonnet-5")) + warnings = [i for i in validate_graph(spec).issues if i.level == "warning" and i.node_id == "chal"] + assert [(i.field, i.code) for i in warnings] == [("model", "challenge.same_model")] + assert "自己审自己" in warnings[0].message or "难以发现" in warnings[0].message + + +async def test_the_node_refuses_the_same_bad_settings_at_run_time(): + """跳过画布校验直接执行（旧版本发布的图、改坏了的配置）：节点自己报同一句话。""" + spec = spec_of(*BASE, node("chal", "challenge", report="write", on_refuted="withhold")) + [(_, message, _)] = problems(spec) + ctx = NodeContext(node=spec.node_map()["chal"], run=RunContext(run_id="r", thread_id="r", spec=spec)) + with pytest.raises(NodeError) as caught: + await challenge_mod.run_challenge({"nodes": {}}, ctx) + assert str(caught.value) == message + + wrong = spec_of(*BASE, node("chal", "challenge", report="fetch")) + [(_, message, _)] = problems(wrong) + ctx = NodeContext(node=wrong.node_map()["chal"], run=RunContext(run_id="r", thread_id="r", spec=wrong)) + with pytest.raises(NodeError) as caught: + await challenge_mod.run_challenge({"nodes": {}}, ctx) + assert str(caught.value) == message diff --git a/backend/tests/test_copilot_challenge_guide.py b/backend/tests/test_copilot_challenge_guide.py new file mode 100644 index 00000000..3812c7c9 --- /dev/null +++ b/backend/tests/test_copilot_challenge_guide.py @@ -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 diff --git a/backend/tests/test_issuance_upstream_gaps.py b/backend/tests/test_issuance_upstream_gaps.py index cc6929c8..92fbfc12 100644 --- a/backend/tests/test_issuance_upstream_gaps.py +++ b/backend/tests/test_issuance_upstream_gaps.py @@ -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}')