From 888f746d31225174ee7ca892c3eeb121ee2b362f Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Tue, 6 Oct 2026 23:11:22 +0800 Subject: [PATCH] =?UTF-8?q?K3b=EF=BC=9A=E4=BA=A4=E7=BB=99=E6=A0=B8?= =?UTF-8?q?=E6=9F=A5=20agent=20=E7=9A=84=E8=AF=81=E6=8D=AE=E6=91=98?= =?UTF-8?q?=E5=BD=95=E5=85=88=E9=81=AE=EF=BC=9B=E5=8F=8D=E6=96=B9=E5=AE=A1?= =?UTF-8?q?=E7=A8=BF=E7=9A=84=E7=90=86=E7=94=B1=E8=BF=87=E6=9C=80=E5=90=8E?= =?UTF-8?q?=E4=B8=80=E9=81=93=E5=85=B3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 摘录由 judge.prepare 生成，以前 _hidden 只按列名遮：报告取数把遮罩列改了名字选出来（buyer_phone AS 手机号），质询、反方 agent 的任务里四个号码都在，模型照抄进理由。现在 prepare 收 kinds（{数据源名: 类型}，只有核查 agent 的摘录给），查询摘录、指标输入按 masking._derived_hidden 认出改了名字的遮罩列（和工具边界同一套规则，表结构按快照冻结的那份），认的过程出了意外整份的列都不给；结论句裁判自己的摘录不给 kinds，和以前逐字相同、复用键不变。反方审稿、审批质询再把摘录里别的列、SQL 里写着的遮罩值按值换成「已遮罩」（换了写法的号码也认），值同交给查询工具的那一份（报告快照里的，反方去掉这句结论原文里写着的）；解析 SQL、按值换放进线程。 - 反方审稿的理由按值遮之后，正文、理由文档（正文、原文、目录里的列名）里还认得出遮罩值的，这句记为无法判断、理由换成系统说明、不存理由文档（_kept，同审批质询的 _leaked），遮不住的「被推翻」不再让出具降档。 补测试：质询、反方各一条端到端（任务里的摘录没有号码，手机号列不给，SQL 里的号码、备注里换了写法的号码换掉），反方理由里 agent 自己查出的号码换着写法也换掉，最后一道关；分别去掉按 SQL 认列、两处按值换、agent 查到的值、最后一道关，对应的测试都变红。 --- backend/app/engine/inquiry.py | 35 ++++--- backend/app/engine/judge.py | 52 +++++++--- backend/app/engine/nodes/challenge.py | 82 ++++++++++++--- backend/tests/test_challenge_masking.py | 120 ++++++++++++++++++++++ backend/tests/test_inquiry_masking_e2e.py | 29 +++++- 5 files changed, 276 insertions(+), 42 deletions(-) create mode 100644 backend/tests/test_challenge_masking.py diff --git a/backend/app/engine/inquiry.py b/backend/app/engine/inquiry.py index f9796a0e..1556e6f5 100644 --- a/backend/app/engine/inquiry.py +++ b/backend/app/engine/inquiry.py @@ -7,8 +7,9 @@ 运行已经结束封存、核对一致、是探索运行——问答追加在封存之后（post_seal），是批注不是证据。正式运行封存之后不再追加 （同按需裁判）；设置里关了「审批质询 · 启用」一律不答。同一次运行同时只答一条：进程内登记（_ANSWERING），登记期间 这次运行的审批回复、继续运行也不受理（api/runs.py），免得运行往下走了、问答还在路上 -2. 回答（_answer）：交给 agent 被问的那句原文、它挂的证据摘录（照 judge.prepare 的办法取，数据源遮罩的列照旧不给，原报告 - 的编号标明「原报告」）、审批人选中的原文和问题；工具是这次运行用过的查询工具对应的数据源（同名重建的、没固定过版本的 +2. 回答（_answer）：交给 agent 被问的那句原文、它挂的证据摘录（照 judge.prepare 的办法取，原报告的编号标明「原报告」； + 摘录先遮：遮罩列不给，改了名字的遮罩列按 SQL 认出来也不给，别的列、SQL 里写着的遮罩值按值换掉，见 _excerpts）、 + 审批人选中的原文和问题；工具是这次运行用过的查询工具对应的数据源（同名重建的、没固定过版本的 上传表格不算，见 _usable_sources）；上限按设置（每次的查询次数、时长，全局每日金额）。回答先给结论，数字只能引用 agent 自己查询的格子，或者报告里已有的指标、运行输入；回答文档由 compose_doc 组装，裸数字被标出来（提问人自己写的数除外）。 封存之后的追问查的是数据源现在的数据，任务里给报告生成的时间，回答末尾注明；触顶、查询失败、引用解析不了也在末尾注明 @@ -125,9 +126,9 @@ 点、括号，含全角），带 +86 / 0086 / 86 前缀、全角数字：写成汉字（一三九……）、中间夹着别的字、拆成几处写的、跨行 写的认不出；和别的数字只隔着一个连字符或点连着的（13900000001-2、19914.74）不算号码；SQL 里不带引号的数不这样认 （199-1474-2785 在 SQL 里是三个数相减），字符串、带引号的名字、注释里照样认 -- 质询 agent 的任务里的证据摘录照 judge.prepare 按列名遮：报告取数把遮罩列改了名字选出来（buyer_phone AS 手机号）时， - 摘录里有原值，agent 看得到；它写进回答、SQL 的，按值认得出的照样换掉（报告快照里的值都收了），封存的是换过的。摘录 - 按 SQL 认改了名字的遮罩列要改 judge.py，另行处理 +- 质询 agent 的任务里的证据摘录先遮（_excerpts）：遮罩列、按 SQL 认出改了名字的遮罩列（和工具边界同一套规则）不给， + 别的列、SQL 里写着的遮罩值按值换（报告快照里的值），按值认的门槛同上：短的数（年龄、等级）写在 SQL 的条件里、别的列 + 里照原样。没指明是哪一句时任务里给的是报告正文，正文里写着的值照原样（读者本来就看得到） """ from __future__ import annotations @@ -582,9 +583,13 @@ def _table_text(block: dict[str, Any], markdown: str) -> str: def _excerpts(doc: dict[str, Any], catalog: dict[str, Any], cand: judging.Candidate, loader: Callable[[str], Any], - masked: dict[str, Any]) -> list[str]: - """这句挂的每件证据的摘录：和 judge.prepare 一样逐件摘（数据源遮罩的列不给，带期 4 标记的文档多来历行）， - 指向原报告编号的写法标上「原报告」——agent 自己的查询也编 Q1、Q2…，不分开的话会把它们混成一回事。""" + masked: dict[str, Any], kinds: dict[str, Any] | None = None, secrets: _Secrets | None = None + ) -> list[str]: + """这句挂的每件证据的摘录：和 judge.prepare 一样逐件摘（带期 4 标记的文档多来历行），指向原报告编号的写法标上 + 「原报告」——agent 自己的查询也编 Q1、Q2…，不分开的话会把它们混成一回事。 + 摘录先遮，模型从任务里也看不到遮罩值：数据源遮罩的列不给，改了名字的（buyer_phone AS 手机号）按 SQL 认出来也不给 + （kinds：{数据源名: 类型}，和工具边界同一套规则，judge._derived）；别的列、SQL 里写着的遮罩值按值换成「已遮罩」， + 换了写法的号码也认（secrets：报告快照里的遮罩值，同交给查询工具的那一份）。解析 SQL、按值换是纯计算，调用方放进线程。""" fetch = judging._Fetch(loader) provenance_on = doc.get("provenance") == DOC_PROVENANCE queries = {a for a, e in catalog.items() if isinstance(e, dict) and e.get("kind") == "query"} @@ -592,7 +597,9 @@ def _excerpts(doc: dict[str, Any], catalog: dict[str, Any], cand: judging.Candid for alias in cand.aliases: entry = catalog.get(alias) if isinstance(entry, dict) and (text := judging._excerpt(alias, entry, [cand], fetch, masked, catalog, - provenance_on=provenance_on)): + provenance_on=provenance_on, kinds=kinds)): + if secrets is not None and secrets.strong: + text = secrets.scrub(text, loose=True) out.append(mark_original(text, alias, queries)) return out @@ -1811,7 +1818,6 @@ async def _answer(op: _Opening, ask: Ask, target: _Target | None, settings: dict doc = op.doc catalog = doc.get("catalog") if isinstance(doc.get("catalog"), dict) else {} masked = await judging.source_masks(catalog, loader=op.loader) - excerpts = _excerpts(doc, catalog, target.cand, op.loader, masked) if target is not None else [] refs = _report_refs(catalog) wanted = _run_sources(op.events) usable = await _usable_sources(op.run_id, wanted, op.versions) @@ -1820,15 +1826,20 @@ async def _answer(op: _Opening, ask: Ask, target: _Target | None, settings: dict async with SessionLocal() as session: spec = await verify_model_spec(session, None, settings=settings) # 报告自己的查询快照：被问的那句可能直接写着遮罩列的值，agent 会照着复述；里面的遮罩值在派 agent 之前就交给它的 - # 查询工具（masked_values），agent 查到写着同一个号码的备注、别的表时，交给模型之前就换掉 + # 查询工具（masked_values），agent 查到写着同一个号码的备注、别的表时，交给模型之前就换掉；证据摘录也按它们先遮 given = [ask.question, ask.quote, *_ref_values(refs)] others = [snapshot for entry in catalog.values() if isinstance(entry, dict) and entry.get("kind") == "query" and _query_shaped(snapshot := _quiet_load(op.loader, entry.get("artifact")))] + seed = await snapshot_secrets(others, given) + excerpts: list[str] = [] + if target is not None: + kinds = await _source_kinds([*masked, *(str(s.get("source") or "") for s in others)]) + excerpts = await asyncio.to_thread(_excerpts, doc, catalog, target.cand, op.loader, masked, kinds, seed) outcome = await run_verify_agent( spec=spec, system=INQUIRY_SYSTEM, task=_task(op, ask, target, excerpts, refs, live, lost), sources=list(live), output_schema=INQUIRY_SCHEMA, max_queries=part.get("max_queries"), timeout_s=part.get("timeout_s"), daily_max_usd=settings.get("daily_max_usd"), run_id=op.run_id, data_versions=op.versions, - masked_values=await snapshot_secrets(others, given)) + masked_values=seed) # 遮罩：agent 自己的查询快照（查询时就遮好了，遮掉的值在 outcome.masked_values 里）、报告自己的查询快照 own: dict[str, dict[str, Any]] = {} diff --git a/backend/app/engine/judge.py b/backend/app/engine/judge.py index b5808583..f4f17ef2 100644 --- a/backend/app/engine/judge.py +++ b/backend/app/engine/judge.py @@ -342,6 +342,25 @@ def _hidden(snapshot: dict[str, Any], source: Any, masked: dict[str, Any] | None return names +def _derived(snapshot: dict[str, Any], hidden: set[str], fetch: _Fetch, kinds: dict[str, Any] | None) -> set[str]: + """核查 agent 的摘录（kinds 给了：{数据源名: 类型}）要遮的列：按列名的（hidden），加上按 SQL 认出改了名字的遮罩列 + （buyer_phone AS 手机号、套函数、经子查询转一道……），和核查 agent 的查询在工具边界遮的是同一套规则 + （masking._derived_hidden，表结构按快照冻结的那份、经 fetch 取）。认的过程出了意外，整份快照的列都算（宁可多遮）。 + kinds 为 None（结论句裁判自己的摘录）照旧只按列名，摘录、复用键都和以前一样。""" + if kinds is None or not hidden: + return hidden + from app.engine.masking import _derived_hidden, _schema_columns # masking 导入了别的引擎模块：用到时再导入 + + try: + schema = fetch(snapshot.get("schema_artifact")) if snapshot.get("schema_artifact") else None + tables = _schema_columns(schema) if isinstance(schema, dict) else None + kind = kinds.get(str(snapshot.get("source") or "")) + return set(hidden) | _derived_hidden(dict(snapshot), set(hidden), kind, tables)[0] + except Exception: # noqa: BLE001 - 认不准就整份遮：摘录少几列，不让遮罩值交给核查 agent + logger.warning("judge: 按 SQL 认摘录里改了名字的遮罩列时出错，整份快照的列都按遮罩处理", exc_info=True) + return set(hidden) | {str(c).lower() for c in snapshot.get("columns") or []} + + def _shown_columns(columns: list[str], cited: Iterable[str]) -> list[str]: """摘录给哪几列：被引用的列一律给（超过上限也给），再按原来的顺序用别的列补满到 EXCERPT_COLS 列。 @@ -354,11 +373,13 @@ def _shown_columns(columns: list[str], cited: Iterable[str]) -> list[str]: def _query_excerpt(alias: str, entry: dict[str, Any], rows: set[int], cols: Iterable[str], whole: bool, - fetch: _Fetch, masked: dict[str, Any] | None, *, provenance_on: bool = False) -> str | None: + fetch: _Fetch, masked: dict[str, Any] | None, *, provenance_on: bool = False, + kinds: dict[str, Any] | None = None) -> str | None: """一条查询的摘录：头、SQL、（期 4）来历行、列、被引用的行。 provenance_on：文档带期 4 标记时为真（prepare 算）。这时上传表格的查询快照（有 data_version）在 SQL 之后、「列：」 之前多几行来历（_provenance_lines）；其余情况一个字都不多，期 4 之前的文档、手工源的摘录和以前逐字相同。 + kinds：给核查 agent 的摘录才给，改了名字的遮罩列也不给（_derived）。 """ snapshot = fetch(entry.get("artifact")) if not isinstance(snapshot, dict) or not isinstance(snapshot.get("rows"), list): @@ -372,7 +393,7 @@ def _query_excerpt(alias: str, entry: dict[str, Any], rows: set[int], cols: Iter lines = [head, f"SQL：{sql[:SQL_CHARS]}{'…' if len(sql) > SQL_CHARS else ''}"] if snapshot.get("source") == MERGE_SOURCE: lines.append(_merge_line(snapshot)) - hidden = _hidden(snapshot, source, masked) + hidden = _derived(snapshot, _hidden(snapshot, source, masked), fetch, kinds) if provenance_on and snapshot.get("data_version"): lines += _provenance_lines(snapshot, fetch, hidden) columns = [str(c) for c in snapshot.get("columns") or [] if str(c).lower() not in hidden] @@ -483,7 +504,7 @@ def _cell_words(locator: dict[str, Any], hidden: set[str]) -> str: def _metric_inputs(metric: dict[str, Any], fetch: _Fetch, catalog: dict[str, Any], - masked: dict[str, Any] | None) -> list[str]: + masked: dict[str, Any] | None, kinds: dict[str, Any] | None = None) -> list[str]: """指标的输入取自哪次查询、哪一格，再附上那次查询的 SQL（截到 SQL_CHARS，最多 ENTITY_QUERIES 次）。 「活跃用户按会员号去重」这种讲口径的句子，证据在取数的 SQL 里；只给原式，裁判只能判证据不足。 @@ -504,8 +525,8 @@ def _metric_inputs(metric: dict[str, Any], fetch: _Fetch, catalog: dict[str, Any snapshot = fetch(artifact) snap = snapshot if isinstance(snapshot, dict) else {} source = (catalog.get(alias) or {}).get("source") if alias else None - hidden = _source_hidden(source or snap.get("source"), catalog, fetch, masked) \ - | _mask_names(snap.get("mask_columns")) + hidden = _derived(snap, _source_hidden(source or snap.get("source"), catalog, fetch, masked) + | _mask_names(snap.get("mask_columns")), fetch, kinds) label = f"查询 {alias} " if alias else "一次查询（不在这份报告的证据目录中）" lines.append(f"输入 {path}：取自{label}的 {_cell_words(item.get('locator') or {}, hidden)}") sql = _clean(str(snap.get("sql") or "")) @@ -518,7 +539,7 @@ def _metric_inputs(metric: dict[str, Any], fetch: _Fetch, catalog: dict[str, Any def _metric_excerpt(alias: str, entry: dict[str, Any], fetch: _Fetch, catalog: dict[str, Any], - masked: dict[str, Any] | None) -> str: + masked: dict[str, Any] | None, kinds: dict[str, Any] | None = None) -> str: name = entry.get("name") or (entry.get("locator") or {}).get("metric") or alias rendered = entry.get("rendered") where = " ".join(v for v in (f"口径「{entry['caliber']}」" if entry.get("caliber") else "", @@ -533,7 +554,7 @@ def _metric_excerpt(alias: str, entry: dict[str, Any], fetch: _Fetch, catalog: d if metric is not None: if metric.get("expression"): lines.append(f"原式：{_clean(str(metric['expression']))}") - lines.extend(_metric_inputs(metric, fetch, catalog, masked)) + lines.extend(_metric_inputs(metric, fetch, catalog, masked, kinds)) return "\n".join(lines) @@ -631,12 +652,13 @@ def _retrieval_excerpt(alias: str, entry: dict[str, Any], picked: list[int], who def _excerpt(alias: str, entry: dict[str, Any], cands: list[Candidate], fetch: _Fetch, - masked: dict[str, Any] | None, catalog: dict[str, Any], *, provenance_on: bool = False) -> str | None: + masked: dict[str, Any] | None, catalog: dict[str, Any], *, provenance_on: bool = False, + kinds: dict[str, Any] | None = None) -> str | None: """一件证据的摘录。provenance_on 只交给查询摘录（_query_excerpt）：指标摘录里「输入取自查询 Qn」的那几行不加 - 来历（P4-SPEC 调整 1），表名、字段名的摘录照旧。""" + 来历（P4-SPEC 调整 1），表名、字段名的摘录照旧。kinds 只给核查 agent 的摘录（见 _derived）。""" kind = entry.get("kind") if kind == "metric": - return _metric_excerpt(alias, entry, fetch, catalog, masked) + return _metric_excerpt(alias, entry, fetch, catalog, masked, kinds) if kind == "input": field_name = (entry.get("locator") or {}).get("field") or alias return f"【{alias}】运行输入：{field_name} = {entry.get('rendered')}" @@ -644,7 +666,7 @@ def _excerpt(alias: str, entry: dict[str, Any], cands: list[Candidate], fetch: _ rows = {r for c in cands for a, r in c.rows if a == alias} cols = [col for c in cands for a, col in c.cols if a == alias] return _query_excerpt(alias, entry, rows, cols, any(alias in c.whole for c in cands), fetch, masked, - provenance_on=provenance_on) + provenance_on=provenance_on, kinds=kinds) if kind == "retrieval": picked = [h for c in cands for a, h in c.hits if a == alias] return _retrieval_excerpt(alias, entry, picked, any(alias in c.whole for c in cands), fetch) @@ -739,10 +761,14 @@ class JudgeRequest: def prepare(doc: dict[str, Any], catalog: dict[str, Any] | None = None, *, units: Iterable[str] | None = None, - loader: Callable[[str], Any] | None = None, masked: dict[str, Any] | None = None) -> JudgeRequest: + loader: Callable[[str], Any] | None = None, masked: dict[str, Any] | None = None, + kinds: dict[str, Any] | None = None) -> JudgeRequest: """裁判的输入：候选句和证据摘录。纯函数（快照经 loader 取，缺省读工件库），不调模型。 catalog 缺省用文档自己存的目录；masked = {数据源名: 要遮的列}，和快照里记下的 mask_columns 合在一起。 + kinds = {数据源名: 类型} 只在给核查 agent（反方审稿、审批质询）准备摘录时给：查询摘录、指标输入里改了名字的遮罩列 + （buyer_phone AS 手机号）按 SQL 认出来也不给，和核查 agent 的查询在工具边界遮的是同一套规则（_derived）。结论句裁判 + 自己的摘录不给它，和以前逐字相同。 摘录按文档标记分版本（P4-SPEC 1.4、3.3）：带 `provenance: DOC_PROVENANCE` 的文档，上传表格的查询摘录多几行 来历；没有标记的（期 4 之前组装的）和以前逐字相同，复用键、断点指纹都不变，已判过的句子照旧复用。 @@ -755,7 +781,7 @@ def prepare(doc: dict[str, Any], catalog: dict[str, Any] | None = None, *, units for alias in dict.fromkeys(a for c in cands for a in c.aliases): entry = catalog.get(alias) if isinstance(entry, dict) and (text := _excerpt(alias, entry, cands, fetch, masked, catalog, - provenance_on=provenance_on)): + provenance_on=provenance_on, kinds=kinds)): excerpts[alias] = text picked = {c.unit for c in cands} skipped: dict[str, str] = {} diff --git a/backend/app/engine/nodes/challenge.py b/backend/app/engine/nodes/challenge.py index 00f0c22d..f9ddfff3 100644 --- a/backend/app/engine/nodes/challenge.py +++ b/backend/app/engine/nodes/challenge.py @@ -11,9 +11,10 @@ 的 Q1 分开），要它先想清楚「如果这句不成立，数据里会看到什么」，写只读查询去找，最后交结论（经受住 / 被削弱 / 被推翻 / 无法判断）和理由。理由里的数字只能引用它自己查询的格子，裸数字由 compose_doc 标出来。 遮罩：agent 的查询结果在工具边界就遮好了；这份报告引用的查询快照里的遮罩值（按 SQL 认出改了名字的遮罩列，同审批 - 质询）反查之前就交给查询工具，agent 查到写着同一个号码的别的表时也看不到。摘录按列名遮，改了名字的遮罩列原值 - agent 看得到，会照着写进理由：理由正文和理由文档按值换成「已遮罩」（换了写法的号码也认），这句结论原文里写着的 - 不换（_masked_rationale） + 质询）反查之前就交给查询工具，agent 查到写着同一个号码的别的表时也看不到。任务里的证据摘录先遮：遮罩列、按 SQL + 认出改了名字的遮罩列（buyer_phone AS 手机号）不给，别的列、SQL 里写着的遮罩值按值换（_task）。理由正文和理由文档 + 仍按值换成「已遮罩」（换了写法的号码也认，agent 查询时遮掉的值也算），这句结论原文里写着的不换（_masked_rationale）； + 换完还认得出的，这句记为无法判断、理由不记（_kept，同审批质询的最后一道关） 3. 出处：一条查询都没执行成功的，判什么都记为无法判断；判「被推翻」的，理由里还得引用到自己的查询结果——它要让 正式出具降档，审稿记录里得查得到凭什么。不采信时系统写明原因，反方原本的结论和理由照录在后面（纯文字：解析 不了的引用写成「（引用无效）」，裸数字注明「（无出处）」）。这句该查的数据源都不存在或已停用、核查模型本身用 @@ -49,6 +50,8 @@ from __future__ import annotations import asyncio +import json +import logging import re from collections.abc import Callable, Collection, Iterable from dataclasses import replace @@ -76,8 +79,12 @@ from app.engine.state import GraphState from app.engine.verify_agent import RATIONALE, VerifyOutcome, run_verify_agent from app.tools.datasource import run_versions +logger = logging.getLogger(__name__) + #: 四种结论：经受住 / 被削弱 / 被推翻 / 无法判断。counts 的键照这个顺序 VERDICTS = ("stands", "weakened", "refuted", "inconclusive") +#: 理由按值遮之后还认得出遮罩值时，这句的理由换成这段系统说明（_kept） +LEAKED = "反方的理由里有设置了遮罩的值，无法遮住，这句的结论不予采信，记为无法判断" #: 节点产出、审批卡里最多列几句（审稿记录里一句不少） ITEMS_MAX = 10 #: 出具缺口、报错里最多点几句的原文，每句最多几个字 @@ -378,6 +385,12 @@ def _claim_of(cand: judging.Candidate, outcome: VerifyOutcome) -> tuple[dict[str return _claim(cand, verdict, rationale, doc=outcome.rationale, queries=outcome.queries, usage=outcome.usage), None +def _scrubbed_excerpt(text: str, known: Any) -> str: + """证据摘录里的遮罩值换成「已遮罩」：别的列、SQL 里写着的（换了写法的号码也认），交给核查 agent 之前就换掉。值是 + 报告快照里的遮罩值去掉这句结论原文里写着的（_spared），同交给查询工具的那一份。""" + return known.scrub(text, loose=True) if known is not None and known.strong else text + + def _spared(seed: Any, claim: str) -> Any: """报告快照里的遮罩值（seed）去掉这句结论原文里写着的，交给这一句的反查。seed 本身不动（下一句另算）。""" known = seed.copy() @@ -391,7 +404,7 @@ def _masked_rationale(outcome: VerifyOutcome, secrets: Any, *, given: Iterable[s 原文里写着的（given）已经摆在读者眼前，不换。反方 agent 的查询结果在工具边界就遮好了，可证据摘录按列名遮，改了名字 的遮罩列（buyer_phone AS 手机号）的原值它看得到，会写进理由。 换的办法同审批质询的回答（_scrubbed_source）：原文里的值先换掉，再照原来的目录重新组装理由文档；引用到的格子按 - 遮好的快照渲染（文字格里夹着的遮罩值换掉那一截）。理由里认不出遮罩值的原样交回。""" + 遮好的快照渲染（文字格里夹着的遮罩值换掉那一截）。理由里认不出遮罩值的原样交回。换完还认得出的过最后一道关（_kept）。""" from app.engine.evidence import compose_doc from app.engine.masking import _masked_view, _scrubbed_source, _Secrets @@ -406,7 +419,7 @@ def _masked_rationale(outcome: VerifyOutcome, secrets: Any, *, given: Iterable[s pool.prepare() source = str(doc.get("source") or "") if not (pool.spans(str(doc.get("markdown") or ""), prose=True) or pool.spans(source, prose=True)): - return outcome + return _kept(outcome, pool) views: dict[str, Any] = {} fetch = _fetcher() for query in outcome.queries: @@ -422,8 +435,23 @@ def _masked_rationale(outcome: VerifyOutcome, secrets: Any, *, given: Iterable[s except Exception: # noqa: BLE001 - 重新组装出了意外，值也不能留着 # 正文按值换掉，理由文档不要了（界面照正文显示；判「被推翻」的因此查不到出处，按无法判断记） said = pool.scrub(str(doc.get("markdown") or ""), prose=True) - return replace(outcome, rationale=None, result={**result, RATIONALE: said}) - return replace(outcome, rationale=again, result={**result, RATIONALE: again["markdown"]}) + return _kept(replace(outcome, rationale=None, result={**result, RATIONALE: said}), pool) + return _kept(replace(outcome, rationale=again, result={**result, RATIONALE: again["markdown"]}), pool) + + +def _kept(outcome: VerifyOutcome, pool: Any) -> VerifyOutcome: + """最后一道关（同审批质询的 _leaked）：按值遮之后，理由正文、理由文档（正文、原文、目录里的列名）里还认得出遮罩值的 + （换了写法的号码也算），这句的结论不采信、理由不记：记为无法判断，理由换成系统的说明（LEAKED）。遮不住的「被推翻」 + 不能让出具降档，也不能把号码存进审稿记录、随运行封存。""" + doc = outcome.rationale if isinstance(outcome.rationale, dict) else {} + texts = [str((outcome.result or {}).get(RATIONALE) or ""), str(doc.get("markdown") or ""), + str(doc.get("source") or ""), + json.dumps(doc.get("catalog") or {}, ensure_ascii=False, default=str)] + if not any(pool.spans(text, prose=True) for text in texts): + return outcome + logger.warning("反方审稿的理由里有遮罩值遮不住，这句记为无法判断（运行 %s）", doc.get("run_id")) + return replace(outcome, rationale=None, result={**(outcome.result or {}), "verdict": "inconclusive", + RATIONALE: LEAKED}) def _cites(doc: dict[str, Any] | None) -> bool: @@ -573,6 +601,8 @@ class _Review: max_claims, max_queries, timeout_s, daily_max_usd self.spec, self.masks, self.writer_model = spec, masks, writer_model self.fetch = _fetcher() + #: 报告用到的数据源的类型（摘录按 SQL 认改了名字的遮罩列时按它的方言解析）。run() 里查好 + self.kinds: dict[str, Any] = {} #: 这一段执行的总时长（engine/deadline.py）：有句子的时长按剩余时间收紧过、或者不再开始时记下，说明和日志里写 self.run_seconds: float | None = None #: 这份报告目录里的查询编号：摘录里指向它们的写法标上「原报告」（mark_original） @@ -597,13 +627,20 @@ class _Review: """当天的核查金额用完、这句一次模型都没调用时的理由：派出去就撞上的和后面不再派出的，是同一回事。""" 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) + def _task(self, cand: judging.Candidate, sources: list[str], lost: list[str], *, own: bool, + known: Any = None) -> str: + """交给核查 agent 的任务：这句结论、它挂的证据摘录（原报告的编号标明，见 mark_original）、可以查的数据源；该查的 + 源有停用了的，也告诉它查不了哪几个，免得它把查不到当成没有反例。own：源是这句自己引用的，不是没挂依据、按整份 + 报告用到的源查的。 + 摘录先遮，模型从任务里也看不到遮罩值：遮罩列不给，改了名字的（buyer_phone AS 手机号）按 SQL 认出来也不给 + （judge.prepare 的 kinds，和工具边界同一套规则）；别的列、SQL 里写着的遮罩值按值换成「已遮罩」，换了写法的号码 + 也认（known：报告快照里的遮罩值去掉这句结论原文里写着的，同交给查询工具的那一份）。解析 SQL、按值换是纯计算， + 在线程里做。""" + request = judging.prepare(self.doc, self.catalog, units=[cand.unit], loader=self.fetch, masked=self.masks, + kinds=self.kinds) 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) + evidence = "\n\n".join(mark_original(_scrubbed_excerpt(request.excerpts[a], known), a, self.queries) + for a in aliases) gone = f"（{'这句还引用了' if own else '这份报告还用到了'}{_quoted(lost)}，已不存在或已停用，无法查询）" \ if lost else "" return CHALLENGE_TASK_PROMPT.format( @@ -620,6 +657,7 @@ class _Review: # 这句没挂依据时在整份报告引用的数据源里查：写作者看得到的就是这几个库 everywhere = claim_sources(self.catalog, self.catalog, self.fetch) usable = await _usable_sources(everywhere) + self.kinds = await self._kinds(everywhere) # 报告引用的查询快照里遮罩列的值：反查之前交给查询工具（agent 查到写着同一个号码的别的表时看不到），反查之后 # 据此换掉理由里的遮罩值（_masked_rationale） seed = await self._seed() @@ -653,12 +691,15 @@ class _Review: claims.append(_claim(cand, "inconclusive", late[0])) overdue += 1 continue + # 这句结论原文里写着的值已经摆在读者眼前，不当遮罩值。上万个值过一遍要几十毫秒，放进线程 + known = await asyncio.to_thread(_spared, seed, cand.text) 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 "" + task_text = await asyncio.to_thread(self._task, cand, live, lost, own=bool(own), known=known) \ + if live else "" except Exception as e: # noqa: BLE001 - 一句的摘录出了意外（坏形状的工件），只让这一句没审成 why = f"整理证据摘录时出错（{describe_exception(e)}）" unreached(cand, f"反查未完成：{why}", why) @@ -686,8 +727,6 @@ class _Review: if allot.tightened: self.run_seconds = allot.seconds # 收紧过的：核查 agent 到点自己发的 verify_limit 改说运行时长上限将至（deadline.relabel），不写收紧后的秒数 - # 这句结论原文里写着的值已经摆在读者眼前，不当遮罩值。上万个值过一遍要几十毫秒，放进线程 - known = await asyncio.to_thread(_spared, seed, cand.text) 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=allot.timeout_s, @@ -777,6 +816,17 @@ class _Review: return await snapshot_secrets(self.fetch(entry.get("artifact")) for entry in self.catalog.values() if isinstance(entry, dict) and entry.get("kind") == "query") + async def _kinds(self, names: Iterable[str]) -> dict[str, Any]: + """这几个数据源的类型（摘录按 SQL 认改了名字的遮罩列时用）。快照记的数据源也算上（旧台账条目没记数据源）。""" + from app.engine.inquiry import _source_kinds # inquiry 导入了这个模块，用到时再导入 + + wanted = set(names) + for entry in self.catalog.values(): + snapshot = self.fetch(entry.get("artifact")) if isinstance(entry, dict) else None + if isinstance(snapshot, dict) and snapshot.get("source"): + wanted.add(str(snapshot["source"])) + return await _source_kinds(wanted) + 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, diff --git a/backend/tests/test_challenge_masking.py b/backend/tests/test_challenge_masking.py new file mode 100644 index 00000000..b102023c --- /dev/null +++ b/backend/tests/test_challenge_masking.py @@ -0,0 +1,120 @@ +"""反方审稿的遮罩收口（engine/nodes/challenge.py）： + +- 交给反方 agent 的证据摘录先遮：报告取数把遮罩列改了名字选出来（buyer_phone AS 手机号），摘录里不给这一列（按 SQL + 认，和工具边界同一套规则）；别的列（备注里换了写法的号码）、SQL 里写着的遮罩值换成「已遮罩」。以前摘录只按列名遮， + 四个号码原样交给模型，模型照抄进理由 +- 理由正文按值遮（报告快照里的值、agent 查询时遮掉的值、换了写法的号码），端到端；组装之后还认得出遮罩值的（第二道 + 防线漏了），整句降为无法判断，理由不记，同审批质询的最后一道关 + +模型一律是 MockChatModel 剧本，数据源是现造的 SQLite 文件，区域、金额、手机号都是虚构的。 +""" +from __future__ import annotations + +import json +import sqlite3 +import uuid + +import pytest + +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 DataSource +from app.engine import masking +from app.engine.nodes import challenge as challenge_mod +from tests.test_challenge_node import Models, Play, engine_up, finish, graph, output_of # noqa: F401 - engine_up 是自动用上的夹具 + +#: 四笔订单，买家手机号设了遮罩；备注没设遮罩，其中一条写着换了写法的号码 +SCRIPT = ("CREATE TABLE sales (id INTEGER PRIMARY KEY, region TEXT, buyer_phone TEXT, amount INTEGER, refunded INTEGER," + " note TEXT);" + "INSERT INTO sales VALUES (1, '华东', '13900000001', 1200, 0, '客户 139-0000-0002 代下单')," + " (2, '华东', '13900000002', 300, 1, ''), (3, '华南', '13900000003', 900, 0, '例行回访')," + " (4, '华北', '13900000004', 400, 0, '');") +SECRETS = ("1390000000", "0000-0002", "139 0000") + + +@pytest.fixture +async def masked_shop(tmp_path): + path = tmp_path / f"{uuid.uuid4().hex[:8]}.db" + db = sqlite3.connect(path) + db.executescript(SCRIPT) + 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={"mask_columns": ["buyer_phone"]}) + session.add(row) + await session.flush() + row.schema_cache = await introspect.introspect(row) + await session.commit() + source_id = row.id + yield name + await engines.invalidate(source_id) + + +def _clean(text: str) -> bool: + return not any(s in text for s in SECRETS) + + +#: 报告取数：手机号改了名字选出来，备注照选，按号码排序时把号码写进了 SQL +REPORT_SQL = ("SELECT buyer_phone AS 手机号, note AS 备注, amount AS 金额 FROM sales " + "ORDER BY buyer_phone = '13900000001' DESC, amount DESC") +WRITER = ("金额最高的一笔 [[v:Q1.r0.金额]]。[[see:Q1]]",) +CLAIM = "金额最高的一笔" + + +async def test_the_excerpt_in_the_task_is_masked_before_the_agent_sees_it(monkeypatch, masked_shop): + """反方 agent 收到的任务里：改了名字的遮罩列（手机号）不给，备注里换了写法的号码、SQL 里写着的号码换成「已遮罩」， + 金额照常。以前「列：金额 | 手机号」「r0：1,200 | 13900000001」…，四个号码都在。""" + name = masked_shop + plays = {CLAIM: Play(["SELECT COUNT(*) AS 笔数 FROM sales"], "stands", "共 [[v:Q1.r0.笔数]] 笔。[[see:Q1]]")} + models = Models(monkeypatch, name, writer=WRITER, plays=plays) + row = await finish(graph(name, sql=REPORT_SQL), run_class="exploratory") + assert row.status == "succeeded", row.error + task = models.tasks()[CLAIM] + assert "金额最高的一笔 1,200。" in task and "【原报告 Q1】" in task + excerpt = task[task.index("【原报告 Q1】"):] + assert _clean(excerpt), excerpt + columns = next(line for line in excerpt.splitlines() if line.startswith("列：")) + assert "手机号" not in columns and "金额" in columns and "备注" in columns + assert "客户 已遮罩 代下单" in excerpt and "buyer_phone = '已遮罩'" in excerpt + assert "（有的列按数据源的设置遮罩了，没有给出）" in excerpt + + +async def test_masked_values_the_agent_found_itself_are_masked_in_the_rationale(monkeypatch, masked_shop): + """端到端：反方 agent 自己查出的号码（查询时在工具边界遮掉、只在内存里）被它换着写法写进理由：审稿记录里的理由正文、 + 理由文档都换成「已遮罩」。报告没有选出号码，值只来自 agent 查询时遮掉的那一份。""" + name = masked_shop + top = "SELECT buyer_phone AS 电话, amount AS 金额 FROM sales WHERE region = '华南'" + said = "华南那笔来自 +86 139 0000 0003，金额 [[v:Q1.r0.金额]]。[[see:Q1]]" + plays = {CLAIM: Play([top], "stands", said)} + Models(monkeypatch, name, writer=WRITER, plays=plays) + row = await finish(graph(name, sql="SELECT region AS 区域, amount AS 金额 FROM sales ORDER BY amount DESC"), + run_class="exploratory") + assert row.status == "succeeded", row.error + record = artifact_store.load((await output_of(row.id, "chal"))["challenge_artifact"]) + [claim] = [c for c in record["claims"] if CLAIM in c["text"]] + assert claim["verdict"] == "stands" and claim["rationale"] == "华南那笔来自 已遮罩，金额 900。" + assert claim["rationale_doc"]["markdown"] == claim["rationale"] + assert "0000 0003" not in json.dumps(record, ensure_ascii=False) + + +async def test_a_rationale_still_showing_a_masked_value_is_not_kept(monkeypatch, masked_shop): + """最后一道关：按值遮之后组装出来的理由里还认得出遮罩值（这里让按值换失手），这句降为无法判断，理由不记、不存， + 原值不进审稿记录。同审批质询的回答。""" + name = masked_shop + monkeypatch.setattr(masking, "_scrubbed_source", lambda text, secrets: text) + said = "号码 13900000001 的那笔属实，金额 [[v:Q1.r0.笔数]]。[[see:Q1]]" + plays = {CLAIM: Play(["SELECT COUNT(*) AS 笔数 FROM sales"], "refuted", said)} + Models(monkeypatch, name, writer=WRITER, plays=plays) + row = await finish(graph(name, sql=REPORT_SQL), run_class="exploratory") + assert row.status == "succeeded", row.error + record = artifact_store.load((await output_of(row.id, "chal"))["challenge_artifact"]) + [claim] = [c for c in record["claims"] if CLAIM in c["text"]] + assert claim["verdict"] == "inconclusive" and claim["rationale_doc"] is None + assert claim["rationale"] == challenge_mod.LEAKED + assert _clean(json.dumps(record, ensure_ascii=False)) + out = await output_of(row.id, "chal") + assert out["counts"]["refuted"] == 0, "遮不住的理由不能让出具降档" diff --git a/backend/tests/test_inquiry_masking_e2e.py b/backend/tests/test_inquiry_masking_e2e.py index f791f120..ffae833f 100644 --- a/backend/tests/test_inquiry_masking_e2e.py +++ b/backend/tests/test_inquiry_masking_e2e.py @@ -14,7 +14,7 @@ import json import pytest from app.core import artifact_store -from tests.test_inquiry_api import ( # noqa: F401 - engine_up 是自动用上的夹具，client、big_shop 是夹具 +from tests.test_inquiry_api import ( # noqa: F401 - engine_up 是自动用上的夹具，client、big_shop、masked_shop 是夹具 AMOUNTS, Models, Play, @@ -24,8 +24,11 @@ from tests.test_inquiry_api import ( # noqa: F401 - engine_up 是自动用上 engine_up, events, fetching, + masked_shop, pause, pending, + report_doc, + unit_of, ) BAND = "CASE WHEN amount >= 20000 THEN '大单' ELSE '小单' END" @@ -61,3 +64,27 @@ async def test_a_band_grouped_outside_its_derived_source_reads_as_written(monkey got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{query['artifact']}") assert got.status_code == 200, got.text assert (got.json()["content"]["rows"], got.json()["masked"]) == ([["大单", 3], ["小单", 2]], []) + + +#: 报告取数：手机号改了名字选出来，按号码排序时把号码写进了 SQL +RENAMED = "SELECT buyer_phone AS 手机号, amount AS 金额 FROM sales ORDER BY buyer_phone = '13900000001' DESC, amount DESC" + + +async def test_the_excerpt_in_the_task_is_masked_before_the_agent_sees_it(monkeypatch, masked_shop, client): + """针对「金额最高的一笔 1,200。」发起质询：任务里这句挂的证据摘录不给改了名字的遮罩列（手机号），SQL 里写着的号码 + 换成「已遮罩」，金额照常。以前「列：金额 | 手机号」「r0：1,200 | 13900000001」…，四个号码都在。""" + models = Models(monkeypatch, masked_shop, writer="金额最高的一笔 [[v:Q1.r0.金额]]。[[see:Q1]]", + plays={"这笔是谁买的": Play(["SELECT COUNT(*) AS 笔数 FROM sales"], "共 [[v:Q1.r0.笔数]] 笔。[[see:Q1]]")}) + paused = await pause(masked_shop, graph=fetching(masked_shop, RENAMED)) + [approval] = await pending(paused.id) + doc = await report_doc(paused.id) + r = await ask(client, paused.id, report="write", unit=unit_of(doc, "金额最高的一笔 1,200。"), + question="这笔是谁买的？", approval_id=approval.id) + assert r.status_code == 200, r.text + [task] = models.tasks + excerpt = task[task.index("【原报告 Q1】"):] + assert "1390000000" not in task, excerpt + columns = next(line for line in excerpt.splitlines() if line.startswith("列：")) + assert "手机号" not in columns and "金额" in columns + assert "buyer_phone = '已遮罩'" in excerpt and "（有的列按数据源的设置遮罩了，没有给出）" in excerpt + assert r.json()["inquiry"]["answer"]["ok"] is True