审批质询(后端):运行停在审批时,审批人针对报告里的一句话提问,只读 agent 现查现答,问答写进运行、随运行封存

- engine/inquiry.py:开放条件(停在审批且问的是待处理的那张卡;或已封存、核对一致的探索运行,追加在封存之后;
  正式运行封存后、设置关闭、还在执行一律 409);同一运行同时只答一条(进程内登记);回答交给 run_verify_agent
  (不传 emit,数据版本取运行记录,上限和模型按核查设置),任务里给被问的那句、照 judge.prepare 取的证据摘录
  (遮罩照旧、原报告编号标明)、选中的原文和问题,数据源取这次运行用过的查询工具;回答由 compose_doc 组装成文档
  落工件,数字可以引用自己的查询格子或报告已有的指标、运行输入,裸数字被标出;引用到遮罩列的格子渲染成「已遮罩」,
  回答正文和查询清单里出现的遮罩列的值换成「已遮罩」;写入走 append_events,运行在回答期间变了就不写、照实说明,
  触顶和失败照样记一条
- api/inquiry.py:POST 提问、GET 列表(按报告、句子筛选)、GET 质询工件(只认质询引用过的,遮罩照证据面板);
  输入不合法回 {detail, field}
- api/runs.py:质询正在回答时,审批回复(批准 / 驳回)和继续运行回 409 inquiry_busy,拦在记签批人之前
This commit is contained in:
ThearchyHelios
2026-10-05 18:51:51 +08:00
parent 14dcbe19bb
commit 7df224106d
4 changed files with 1649 additions and 3 deletions
+55 -3
View File
@@ -1,9 +1,61 @@
"""审批质询:审批人在批准前就待审的报告提出质询。
"""审批质询:审批人针对待审报告里的一句话提问,只读 agent 现查现答(规则和做法见 engine/inquiry.py)。
目前只登记了路由前缀,接口由审批质询的工作包补上。路径挂在 /api/runs/{run_id}/… 下面。
- POST /api/runs/{run_id}/inquiries:{report, unit, quote?, question, approval_id?} → {inquiry, recorded, note}
- GET /api/runs/{run_id}/inquiries?report=&unit=:{items: [Inquiry]},封存前后的都算,按先后
- GET /api/runs/{run_id}/inquiries/artifacts/{artifact_id}:质询引用过的工件,查询快照的遮罩列照证据面板遮好
不受理:409 {detail, code}(inquiry_not_open / inquiry_busy / inquiry_disabled);输入不合法 422 {detail, field},
field 指明是哪一项(前后端约定的形状,不是 FastAPI 自己那种校验错误数组,所以请求体自己解析)。
"""
from __future__ import annotations
from fastapi import APIRouter
from typing import Any
from fastapi import APIRouter, Header, HTTPException, Request
from fastapi.responses import JSONResponse
from app.api.coded import CodedHTTPException
from app.api.runs import actor_of
from app.engine import inquiry
from app.engine.inquiry import InquiryRefused
router = APIRouter(prefix="/api/runs", tags=["inquiry"])
def _refused(e: InquiryRefused) -> Any:
"""不受理 → HTTP 响应。422 照约定只回 {detail, field};其余带机读码的回 {detail, code}。"""
if e.status == 422:
return JSONResponse(status_code=422, content={"detail": e.message, "field": e.field})
if e.code:
raise CodedHTTPException(e.status, e.message, e.code)
raise HTTPException(e.status, e.message)
@router.post("/{run_id}/inquiries", response_model=None)
async def ask_inquiry(run_id: str, request: Request, x_actor: str | None = Header(default=None)) -> Any:
"""问一个问题。回答要等 agent 查完库(最长约为设置里的时长上限再加半分钟收尾),接口一直等到回答完才返回。"""
try:
body = await request.json()
except (ValueError, UnicodeDecodeError):
body = None
try:
return await inquiry.ask(run_id, inquiry.parse_ask(body), actor=actor_of(x_actor))
except InquiryRefused as e:
return _refused(e)
@router.get("/{run_id}/inquiries", response_model=None)
async def list_inquiries(run_id: str, report: str | None = None, unit: str | None = None) -> Any:
try:
return {"items": await inquiry.list_inquiries(run_id, report=report or None, unit=unit or None)}
except InquiryRefused as e:
return _refused(e)
@router.get("/{run_id}/inquiries/artifacts/{artifact_id}", response_model=None)
async def inquiry_artifact(run_id: str, artifact_id: str) -> Any:
try:
return await inquiry.inquiry_artifact(run_id, artifact_id)
except InquiryRefused as e:
return _refused(e)
+12
View File
@@ -15,6 +15,7 @@ from app.api.coded import CodedHTTPException
from app.core.bus import bus
from app.db.base import SessionLocal, get_session
from app.db.models import Approval, Artifact, Run, RunEvent
from app.engine.inquiry import BUSY_CONTINUE, BUSY_DECIDE, INQUIRY_BUSY, inquiry_busy
# 发起规则和发起前检查绑定的工具搬到了 engine/launch.py。这几个名字以前从这里导入,照旧可用
from app.engine.launch import ( # noqa: F401
TOOL_MISSING, LaunchRefused, launch_run, missing_tools, node_setting, tool_bindings,
@@ -261,10 +262,18 @@ def _missing(e: KeyError) -> HTTPException:
return HTTPException(404, str(e.args[0]) if e.args else "运行记录不存在")
def _not_while_inquiring(run_id: str, detail: str) -> None:
"""这次运行正在回答审批质询(engine/inquiry.py 的进程内登记)时,审批回复和继续运行一律 409:问答要写进
运行停着的这一段,运行先往下走了,审批人看着回答再批的那份依据就落了空。"""
if inquiry_busy(run_id):
raise CodedHTTPException(409, detail, INQUIRY_BUSY)
@router.post("/{run_id}/resume", response_model=RunOut)
async def resume_run(
run_id: str, payload: ResumeIn, x_actor: str | None = Header(default=None)
) -> Run:
_not_while_inquiring(run_id, BUSY_DECIDE)
try:
# approval_id 以前定义了却从没传下去——一次运行里有多条待审批时,
# 引擎无从知道这次回复的是哪一条
@@ -296,6 +305,7 @@ async def continue_run(
一行状态判定,而 checkpoint 一直都在——改个模型 id 就得把前面两分钟的取数
重跑一遍,纯属白等。
"""
_not_while_inquiring(run_id, BUSY_CONTINUE)
try:
return await run_manager.continue_failed(
run_id, graph=payload.graph, actor=actor_of(x_actor),
@@ -593,6 +603,8 @@ async def decide(
raise HTTPException(404, "审批记录不存在,所属的运行可能已被删除")
if approval.status != "pending":
raise HTTPException(409, "这条审批已处理过,请刷新页面查看最新状态")
# 在落签批人之前拦:拒掉的批准不该留下「谁试图批的」
_not_while_inquiring(approval.run_id, BUSY_DECIDE)
# 责任归属先落库再恢复执行——恢复失败也要留下"谁试图批的"
approval.resolved_by = actor_of(x_actor)
+814
View File
@@ -0,0 +1,814 @@
"""审批质询:审批人针对待审报告里的一句话提问,一个只读 agent 当场查库作答,问答写进这次运行。
审批卡上只有报告正文和核对概况。审批人对某句话有疑问(「排除测试账号会怎样」「这个数含不含退款」),以前只能驳回重跑,
或者自己去查库。质询让一个只读 agent(engine/verify_agent.py)当场作答,报告本身一个字不改:
1. 开放条件(_open):运行停在审批上、问的是那张待处理的审批卡——问答落在以后的封存范围里,随运行结束一起封存;或者
运行已经结束封存、核对一致、是探索运行——问答追加在封存之后(post_seal),是批注不是证据。正式运行封存之后不再追加
(同按需裁判);设置里关了「审批质询 · 启用」一律不答。同一次运行同时只答一条:进程内登记(_ANSWERING),登记期间
这次运行的审批回复、继续运行也不受理(api/runs.py),免得运行往下走了、问答还在路上
2. 回答(_answer):交给 agent 被问的那句原文、它挂的证据摘录(照 judge.prepare 的办法取,数据源遮罩的列照旧不给,原报告
的编号标明「原报告」)、审批人选中的原文和问题;工具是这次运行用过的查询工具对应的数据源;上限按设置(每次的查询次数、
时长,全局每日金额)。回答先给结论,数字只能引用 agent 自己查询的格子,或者报告里已有的指标、运行输入;回答文档由
compose_doc 组装,裸数字被标出来
3. 遮罩:回答里引用到遮罩列的格子渲染成「已遮罩」(按遮罩过的快照组装,_masking_loader);回答正文、查询清单里出现的
遮罩列的值换成「已遮罩」(_Secrets)。哪些列要遮照证据面板的规则:查询当时记下的,加上数据源现在设的
4. 写入(_record):一条 approval.inquiry 事件,node_id 为空,用量写在数据的 usage 里(计进运行用量,见 runner 的
_usage_of_events)。质询的查询快照只放在 inquiry_artifacts / queries 里,绝不放在 query_artifact 下:证据接口按那个键
收集封存的查询,会把质询的查询当成报告的证据。回答期间运行状态变了(被批准、被放弃、继续运行之后重新封存),
append_events 不写,回答照样交回给提问的人,说明没有写进运行记录。触顶、失败照样记一条
5. 读取:列表读 approval.inquiry 事件(封存前后的都算,按先后);质询工件只认质询事件引用过的,遮罩照证据面板
不发 tool.start、llm.start 这类事件(run_verify_agent 不传 emit):界面从接口的返回取回答;runner 的 _stale_replay
只看 tool.start,质询不影响恢复后的重放。
"""
from __future__ import annotations
import contextlib
import json
import logging
import re
import time
from collections.abc import Callable, Iterable, Iterator
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any
from sqlalchemy import select
from app.core import artifact_store
from app.core.errors import describe_exception
from app.core.events import EventType
from app.data.provenance_types import DOC_PROVENANCE
from app.db.base import SessionLocal
from app.db.models import Approval, DataSource, Run, RunEvent
from app.engine import judge as judging
from app.engine.evidence import MARKER_RE, RenderError, compose_doc, iter_units, render_cell
from app.engine.expressions import numeric_text
from app.engine.merge_query import MERGE_SOURCE
from app.engine.nodes.challenge import mark_original, with_note
from app.engine.run_append import append_events
from app.engine.toolcalls import QUERY_PREFIX
from app.engine.verify_agent import RATIONALE, VerifyOutcome, run_verify_agent
logger = logging.getLogger(__name__)
#: 不受理时的机读码(前后端约定)。审批回复、继续运行在质询期间回的也是 INQUIRY_BUSY(api/runs.py)
INQUIRY_NOT_OPEN = "inquiry_not_open"
INQUIRY_BUSY = "inquiry_busy"
INQUIRY_DISABLED = "inquiry_disabled"
RUN_NOT_FOUND = "run_not_found"
ARTIFACT_NOT_FOUND = "inquiry_artifact_not_found"
ARTIFACT_TAMPERED = "snapshot_tampered"
#: 问题、选中的原文最长多少字
QUESTION_MAX = 1000
QUOTE_MAX = 500
#: 回答期间运行状态变了、问答没写进运行记录时给提问的人的说明
NOT_RECORDED = "运行状态已经变化,这次质询的问答没有写入运行记录"
BUSY = "这次运行另有一条质询正在回答,回答完再提问"
#: 质询正在回答时,这次运行的审批回复(批准 / 驳回)、继续运行回的话(api/runs.py)
BUSY_DECIDE = "正在回答质询,回答完再批准或驳回"
BUSY_CONTINUE = "正在回答质询,回答完再继续运行"
#: 触顶原因:前后端约定的四种。verify_agent 另有 max_steps(模型想了太多轮),轮数是按查询次数推出来的,算作查询次数到顶
LIMITS = ("max_queries", "timeout_s", "max_cost_usd", "daily_max_usd")
#: 事件、接口里的用量只给这几项(runner 计进运行用量的也是这几项,_INQUIRY_KEYS)
USAGE_KEYS = ("input_tokens", "output_tokens", "cost_usd", "calls", "unpriced_calls")
#: 查询清单每条给的几项:没有行,查询结果只在快照里(取的时候按遮罩遮好)
QUERY_KEYS = ("alias", "artifact", "sql", "row_count", "truncated", "source")
#: 回答文档存成工件时的种类
ANSWER_KIND = "inquiry_answer"
#: 没指明句子时交给 agent 的报告正文最多多少字
REPORT_CHARS = 4000
#: 列给 agent 的报告指标、运行输入最多几条
REFS_MAX = 40
#: 被问的是表格里的一格时,交给 agent 的表格最多几行(不含表头)
TABLE_ROWS = 20
#: 遮罩列的值短于这么多字就不在正文里换:一个字的值(「是」「男」)换掉,会把不相干的词也换得面目全非
SECRET_MIN = 2
#: 遮罩列的值最多收这么多个(一份快照最多上千行):再多就按已经收到的换
SECRETS_MAX = 20_000
#: 运行的终态。封存之后的追问只在这几种状态上
_TERMINAL = ("succeeded", "failed", "cancelled")
#: 按词边界认值的字符:值的头尾是这样的字时,前后不能紧挨着同类的字(13900000001 不换掉 139000000012 的一截)
_WORD = re.compile(r"[0-9A-Za-z_]")
INQUIRY_SYSTEM = (
"你在帮审批人核对一份待批准的报告:审批人针对报告里的一句话提了问题,你查库作答。你不写报告,也不改报告。\n"
"- 问题是「如果……会怎样」(排除测试账号、换个时间范围、只看某个区域)时,写只读查询把它算出来,不要凭印象推测\n"
"- 回答直截了当:第一句直接给结论(是还是不是、是多少、差多少),再用一两句说依据;不要复述问题,不要客套\n"
"- 回答里的每个数字都写成引用:你自己查询结果里的一格写 [[v:Qn.rX.列名]],报告里已有的指标、运行输入照任务里列出的"
"写法引用;不要直接写数字,没有出处的数字会被标出来\n"
"- 查不到、数据里没有判断所需的信息,就照实说查不到,不要编造"
)
INQUIRY_TASK_PROMPT = (
"报告「{report}」{where}\n\n"
"{evidence}"
"{refs}"
"可以查询的数据源:{sources}\n\n"
"审批人的问题:\n{question}\n\n"
"按上面的做法作答,回答写在 rationale 里。"
)
#: 交给核查 agent 的结果结构:只有回答一项(verify_agent 的理由字段,沿用这里写的说明)
INQUIRY_SCHEMA: dict[str, Any] = {
"type": "object",
"properties": {RATIONALE: {
"type": "string",
"description": "回答:第一句直接给结论,再用一两句说依据;数字一律写成引用([[v:Qn.rX.列名]],"
"或任务里列出的报告指标、运行输入)",
}},
"required": [RATIONALE],
"additionalProperties": False,
}
# 不受理的说法
DISABLED_WORDS = "审批质询已关闭(设置中的「{label}」),如需质询,请在设置的「核查」一节开启"
STILL_RUNNING = "运行还在执行,停在人工审批时、或者运行结束封存之后才能质询"
NO_CARD = "运行没有停在人工审批上(可能是服务重启时中断的),没有可以质询的审批"
CARD_GONE = "这张审批卡已处理,或不是这次运行正在等待的审批,请刷新页面查看最新状态"
CARD_DONE = "这张审批卡已处理,运行已经结束,请刷新页面查看最新状态"
FORMAL_SEALED = "正式运行已结束并封存,问答不能再追加进运行记录;封存之后只有探索运行可以继续质询"
UNSEALED = "运行已结束,但还没有封存,暂时不能质询"
BROKEN_SEAL = "封存核对未通过:封存后有事件被修改、删除或插入,报告和证据已不可信,不能再质询"
NEED_CARD = "运行停在人工审批上,需要指明质询的是哪张审批卡"
BAD_BODY = ("请求格式有误:需要一个对象,写明报告(report,报告撰写节点的 ID)和问题(question),"
"可选句子编号(unit)、选中的原文(quote)、审批卡(approval_id)")
DOC_TAMPERED = "报告文档与哈希不一致,疑似被修改,不能据此质询"
DOC_MISSING = "无法读取报告文档"
NO_ANSWER = "模型没有给出回答内容"
class InquiryRefused(Exception):
"""质询不受理:status 是 HTTP 状态码,message 是给人看的话;409 带机读码 code,422 带出错的那一项 field。"""
def __init__(self, status: int, message: str, *, code: str | None = None, field: str | None = None) -> None:
super().__init__(message)
self.status, self.message, self.code, self.field = status, message, code, field
def _invalid(message: str, name: str | None) -> InquiryRefused:
return InquiryRefused(422, message, field=name)
def _not_open(message: str) -> InquiryRefused:
return InquiryRefused(409, message, code=INQUIRY_NOT_OPEN)
# --------------------------------------------------------------------------
# 同一次运行同时只答一条
# --------------------------------------------------------------------------
class _Answering:
"""正在回答质询的运行(进程内登记)。
只在本进程里有效:另一个进程里的批准拦不住,那时追加按条件更新核对(append_events),写不进就照实说没写进。"""
def __init__(self) -> None:
self._runs: set[str] = set()
def busy(self, run_id: str) -> bool:
return run_id in self._runs
@contextlib.contextmanager
def hold(self, run_id: str) -> Iterator[None]:
# 查和登记之间没有 await:同一个事件循环里的两条质询不会都看到「没人在答」
if run_id in self._runs:
raise InquiryRefused(409, BUSY, code=INQUIRY_BUSY)
self._runs.add(run_id)
try:
yield
finally:
self._runs.discard(run_id)
_ANSWERING = _Answering()
def inquiry_busy(run_id: str) -> bool:
"""这次运行是不是正在回答质询(api/runs.py 据此让审批回复、继续运行回 409)。"""
return _ANSWERING.busy(run_id)
# --------------------------------------------------------------------------
# 请求
# --------------------------------------------------------------------------
@dataclass(frozen=True)
class Ask:
"""一条质询:哪份报告(报告撰写节点的 id)的哪一句(句子单元 id,可以不指明)、选中的原文、问题、问的是哪张审批卡。"""
report: str
unit: str | None
quote: str | None
question: str
approval_id: str | None
def parse_ask(body: Any) -> Ask:
"""请求体 → Ask。不合法时抛 422,field 指明是哪一项。"""
if not isinstance(body, dict):
raise _invalid(BAD_BODY, None)
report = body.get("report")
if not isinstance(report, str) or not report.strip():
raise _invalid("请指明质询的是哪份报告(报告撰写节点的 ID)", "report")
unit = body.get("unit")
if unit is not None and (not isinstance(unit, str) or not unit.strip()):
raise _invalid("句子编号需要是文字,例如「u12」;不指明句子时留空", "unit")
quote = body.get("quote")
if quote is not None and not isinstance(quote, str):
raise _invalid("选中的原文需要是文字", "quote")
quote = (quote or "").strip() or None
if quote is not None and len(quote) > QUOTE_MAX:
raise _invalid(f"选中的原文最长 {QUOTE_MAX} 字,当前为 {len(quote)} 字,请缩短选区", "quote")
question = body.get("question")
if not isinstance(question, str) or not question.strip():
raise _invalid("请填写要质询的问题", "question")
question = question.strip()
if len(question) > QUESTION_MAX:
raise _invalid(f"问题最长 {QUESTION_MAX} 字,当前为 {len(question)} 字", "question")
approval_id = body.get("approval_id")
if approval_id is not None and (not isinstance(approval_id, str) or not approval_id.strip()):
raise _invalid("审批卡的 ID 需要是文字;不是在审批卡上提问时留空", "approval_id")
return Ask(report=report.strip(), unit=unit.strip() if unit else None, quote=quote, question=question,
approval_id=approval_id.strip() if approval_id else None)
# --------------------------------------------------------------------------
# 开放条件
# --------------------------------------------------------------------------
Events = list[tuple[int, str, str | None, Any]]
@dataclass
class _Opening:
"""开放了的这一次质询:往哪种运行里写(mode)、写的时候要核对什么(approval_id / seal),以及被问的报告文档。
loader 是取证据的函数:封存之后只认封存范围内的工件(同按需裁判,api/evidence._sealed_loader),停在审批时和
报告节点里的裁判一样直接读工件库。events 是这次运行的事件(封存之后只取封存范围内的)。"""
run_id: str
mode: str
title: str
versions: Any
events: Events
doc: dict[str, Any]
loader: Callable[[str], Any]
approval_id: str | None = None
seal: tuple[int | None, str | None] | None = None
async def _open(run_id: str, ask: Ask) -> _Opening:
async with SessionLocal() as session:
run = await session.get(Run, run_id)
if run is None:
raise InquiryRefused(404, "运行记录不存在", code=RUN_NOT_FOUND)
status, run_class, graph, versions = run.status, run.run_class, run.graph or {}, run.data_versions
cards = [str(a) for a in (await session.execute(
select(Approval.id).where(Approval.run_id == run_id, Approval.status == "pending"))).scalars()] \
if status == "interrupted" else []
rows: Events = [tuple(r) for r in await session.execute(
select(RunEvent.seq, RunEvent.type, RunEvent.node_id, RunEvent.data)
.where(RunEvent.run_id == run_id).order_by(RunEvent.seq))] if status == "interrupted" else []
title = _node_title(graph, ask.report)
if status == "interrupted":
if not cards:
raise _not_open(NO_CARD)
if ask.approval_id is None:
raise _invalid(NEED_CARD, "approval_id")
if ask.approval_id not in cards:
raise _not_open(CARD_GONE)
doc = _waiting_doc(rows, run_id, ask.report, title)
return _Opening(run_id=run_id, mode="interrupted", title=title, versions=versions, events=rows, doc=doc,
loader=artifact_store.load, approval_id=ask.approval_id)
if status not in _TERMINAL:
raise _not_open(STILL_RUNNING)
if ask.approval_id is not None:
raise _not_open(CARD_DONE)
if run_class == "formal":
raise _not_open(FORMAL_SEALED)
from app.api.evidence import _reports, _sealed, _sealed_loader # api.evidence 导入了 engine 里的模块
sealed = await _sealed(run_id)
if not sealed.sealed:
raise _not_open(UNSEALED)
if not sealed.trusted:
raise _not_open(BROKEN_SEAL)
report = next((r for r in _reports(sealed) if r.node_id == ask.report), None)
if report is None:
raise _invalid(_no_report(title), "report")
if report.hash_ok is False:
raise _invalid(DOC_TAMPERED, "report")
if report.doc is None:
raise _invalid(DOC_MISSING, "report")
# 带期 4 标记的文档才让摘录取得到哈希链上的清单,和按需裁判同一个口径
loader = _sealed_loader(sealed, chain=report.doc.get("provenance") == DOC_PROVENANCE)
return _Opening(run_id=run_id, mode="sealed", title=title, versions=versions, events=list(sealed.events),
doc=report.doc, loader=loader, seal=(sealed.run_manifest_seq, sealed.manifest_hash))
def _node_title(graph: dict[str, Any], node_id: str) -> str:
for n in graph.get("nodes") or []:
if isinstance(n, dict) and n.get("id") == node_id:
return str((n.get("data") or {}).get("label") or node_id)
return node_id
def _no_report(title: str) -> str:
return f"这次运行里没有报告「{title}」的报告文档,请刷新页面后重新选择"
def _waiting_doc(events: Events, run_id: str, report: str, title: str) -> dict[str, Any]:
"""停在审批时被问的报告文档:这个报告节点最后一次 report.checked 里的那份(同证据接口的 _reports)。文档要真是
这次运行里这个节点写的——内容寻址只保证没被改,不保证没被张冠李戴。"""
artifact = None
for _, etype, node_id, data in events:
if etype == EventType.REPORT_CHECKED and node_id == report and isinstance(data, dict) and data.get("doc_artifact"):
artifact = str(data["doc_artifact"])
if artifact is None:
raise _invalid(_no_report(title), "report")
try:
doc = artifact_store.load(artifact)
except ValueError as e:
raise _invalid(DOC_TAMPERED, "report") from e
if not isinstance(doc, dict):
raise _invalid(DOC_MISSING, "report")
if doc.get("node_id") != report or doc.get("run_id") != run_id:
raise _invalid(DOC_TAMPERED, "report")
return doc
# --------------------------------------------------------------------------
# 被问的那句
# --------------------------------------------------------------------------
@dataclass
class _Target:
"""被问的那句:按裁判的候选句建(judge._candidate,带它挂的依据),表格里的一格另带整张表。"""
cand: judging.Candidate
table: str | None = None
def _target(doc: dict[str, Any], unit: str) -> _Target:
"""文档里的这一句。按 judge.candidates 的走法记小节和先后;不是结论句的(表格里的一格、标题)也照样建,
审批人问的不一定是结论句。没有这一句抛 422。"""
markdown = str(doc.get("markdown") or "")
section = ""
for order, (block, item) in enumerate(iter_units(doc)):
text = judging._unit_text(item, markdown)
if block.get("type") == "heading":
section = text
if item.get("id") != unit:
continue
priority = judging.claim_priority(item, block, markdown) or 3
cand = judging._candidate(item, priority, order, text, section)
return _Target(cand, _table_text(block, markdown) if block.get("type") == "table" else None)
raise _invalid(f"报告里没有这句({unit}),请刷新页面后重新选择", "unit")
def _table_text(block: dict[str, Any], markdown: str) -> str:
"""表格整张压成几行(表头在前),一格的值离了行列说不清是什么。"""
rows: dict[int, dict[int, str]] = {}
for unit in block.get("units") or []:
loc = unit.get("loc") if isinstance(unit, dict) else None
if isinstance(loc, dict) and isinstance(loc.get("row"), int) and isinstance(loc.get("col"), int):
rows.setdefault(loc["row"], {})[loc["col"]] = judging._unit_text(unit, markdown)
lines = [" | ".join(cells[c] for c in sorted(cells)) for _, cells in sorted(rows.items())]
kept = lines[:TABLE_ROWS + 1]
return "\n".join(kept) + ("\n…(后面的行省略了)" if len(lines) > len(kept) else "")
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…,不分开的话会把它们混成一回事。"""
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"}
out: list[str] = []
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)):
out.append(mark_original(text, alias, queries))
return out
def _report_refs(catalog: dict[str, Any]) -> dict[str, dict[str, Any]]:
"""报告目录里回答可以直接引用的出处:口径卡指标(m:)和运行输入(i:)。它们的编号和 agent 自己的 Q1… 不会撞;
报告的查询格子不行——agent 的查询也编 Q1,引用 Q1 说不清是哪一份。"""
return {a: e for a, e in catalog.items() if isinstance(e, dict) and e.get("kind") in ("metric", "input")}
def _run_sources(events: Events) -> list[str]:
"""这次运行用过的查询工具对应的数据源,按第一次用到的先后。"""
out: list[str] = []
for _, etype, _, data in events:
if etype not in (EventType.TOOL_START, EventType.TOOL_END) or not isinstance(data, dict):
continue
tool = str(data.get("tool") or "")
name = tool[len(QUERY_PREFIX):] if tool.startswith(QUERY_PREFIX) else ""
if name and name not in out:
out.append(name)
return out
async def _usable_sources(names: Iterable[str]) -> set[str]:
"""现在能查的:存在、没有停用(核查建工具时也只认这样的)。"""
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 _quoted(names: Iterable[str]) -> str:
return "、".join(f"「{n}」" for n in names)
def _task(op: _Opening, ask: Ask, target: _Target | None, excerpts: list[str], refs: dict[str, dict[str, Any]],
live: list[str], lost: list[str]) -> str:
if target is not None:
cand = target.cand
where = f"中的这句话{f'(所在小节「{cand.section}」)' if cand.section else ''}:\n「{cand.text}」"
if target.table:
where += f"\n\n这句话在一张表格里,表格是(每行一条,单元格用 | 隔开):\n{target.table}"
else:
where = ",审批人没有指明是哪一句"
if ask.quote:
where += f"\n\n审批人选中的原文:\n「{ask.quote}」"
if target is None:
markdown = str(op.doc.get("markdown") or "")
cut = markdown[:REPORT_CHARS] + ("\n…(后面的省略了)" if len(markdown) > REPORT_CHARS else "")
where += f"\n\n报告正文:\n{cut}"
evidence = ""
if target is not None:
evidence = ("这句话挂的证据摘录(编号属于原报告;你的查询另起编号,回答里只能引用你自己的查询):\n"
+ "\n\n".join(excerpts) + "\n\n") if excerpts else "这句话没有挂依据。\n\n"
lines = [f"[[{alias}]] {entry.get('label') or alias}" for alias, entry in list(refs.items())[:REFS_MAX]]
more = f"\n…另有 {len(refs) - REFS_MAX} 项" if len(refs) > REFS_MAX else ""
refs_text = ("报告里已有的指标和运行输入(回答里可以照这个写法直接引用):\n" + "\n".join(lines) + more + "\n\n"
if lines else "")
sources = "、".join(live) if live else "(没有可以查询的数据源,只能依据上面的报告内容作答)"
if lost:
sources += f"(这次运行还用到了{_quoted(lost)},已不存在或已停用,无法查询)"
return INQUIRY_TASK_PROMPT.format(report=op.title, where=where, evidence=evidence, refs=refs_text,
sources=sources, question=ask.question)
# --------------------------------------------------------------------------
# 遮罩
# --------------------------------------------------------------------------
def _query_shaped(content: Any) -> bool:
return isinstance(content, dict) and isinstance(content.get("rows"), list)
async def _hidden_columns(snapshot: dict[str, Any], masks: Any) -> tuple[set[str], list[str]]:
"""一份查询快照要遮的列(小写)和已经不在的数据源,照证据面板的规则(api/evidence.py 的 _query_step):查询当时
记下的,加上数据源现在设的;合并查询按它合并的那几个数据源算。"""
from app.api.evidence import _merge_masks, _recorded_masks
if snapshot.get("source") == MERGE_SOURCE:
return await _merge_masks(snapshot, masks)
cols, found = await masks.of(snapshot.get("source"))
return cols | _recorded_masks(snapshot), ([] if found or not snapshot.get("source") else [str(snapshot["source"])])
def _masked_word() -> str:
from app.api.evidence import MASKED
return MASKED
def masked_snapshot(snapshot: dict[str, Any], hidden: set[str]) -> dict[str, Any]:
"""遮罩列的每一格换成「已遮罩」,这几列的类型记成文本(渲染时不当数)。没有要遮的列原样返回。"""
columns = [str(c) for c in snapshot.get("columns") or []]
index = {i for i, c in enumerate(columns) if c.lower() in hidden}
if not index:
return snapshot
word = _masked_word()
def row_of(row: Any) -> Any:
if isinstance(row, list):
return [word if i in index else v for i, v in enumerate(row)]
if isinstance(row, dict):
return {k: word if str(k).lower() in hidden else v for k, v in row.items()}
return row
types = dict(snapshot["column_types"]) if isinstance(snapshot.get("column_types"), dict) else {}
types.update({columns[i]: "text" for i in index})
return {**snapshot, "rows": [row_of(r) for r in snapshot["rows"]], "column_types": types}
def _masking_loader(hidden: dict[str, set[str]]) -> Callable[[str], Any]:
"""组装回答文档用的 loader:agent 自己的查询快照按遮罩遮好再交出去,引用到遮罩列的格子就渲染成「已遮罩」。
别的查询快照不交(回答的目录里只有 agent 自己的查询),表结构快照这类没有行的原样。哈希照旧复验。"""
def load(artifact: str) -> Any:
content = artifact_store.load(artifact)
if not _query_shaped(content):
return content
return masked_snapshot(content, hidden[artifact]) if artifact in hidden else None
return load
def _forms(value: Any) -> set[str]:
"""一格的值在正文里可能的写法:原样(压成一行),是数的再加上千分位的写法(13,900,000,001)。"""
text = json.dumps(value, ensure_ascii=False, default=str) if isinstance(value, (dict, list)) else str(value)
text = re.sub(r"\s+", " ", text).strip()
forms = {text}
if isinstance(value, (int, float)) or numeric_text(value) is not None:
with contextlib.suppress(RenderError, ValueError, ArithmeticError):
forms.add(render_cell(value))
return {f for f in forms if len(f) >= SECRET_MIN}
def _pattern(value: str) -> str:
"""一个值的正则:头尾是字母数字的按词边界认,数字还不能是更长的数的一截(1,390 里的 390、12.5 里的 2.5)。"""
head = tail = ""
if _WORD.match(value[0]):
head = r"(?<![0-9A-Za-z_])" + (r"(?<!\d[.,])" if value[0].isdigit() else "")
if _WORD.match(value[-1]):
tail = r"(?![0-9A-Za-z_])" + (r"(?![.,]\d)" if value[-1].isdigit() else "")
return head + re.escape(value) + tail
@dataclass
class _Secrets:
"""遮罩列里出现过的值。模型看得到原始的查询结果(工具交回的是原样),回答里照着复述、或者把值写进查询条件,
都会把它带进回答正文和查询清单;按值找出来换成「已遮罩」。这只减少暴露,不是安全边界(同证据面板的遮罩)。"""
values: set[str] = field(default_factory=set)
def add(self, snapshot: Any, hidden: set[str]) -> None:
if not _query_shaped(snapshot) or not hidden:
return
columns = [str(c) for c in snapshot.get("columns") or []]
index = [i for i, c in enumerate(columns) if c.lower() in hidden]
for row in snapshot["rows"]:
if len(self.values) >= SECRETS_MAX:
return
if isinstance(row, list):
cells = [row[i] for i in index if i < len(row)]
elif isinstance(row, dict):
cells = [v for k, v in row.items() if str(k).lower() in hidden]
else:
continue
for value in cells:
if value is not None and not isinstance(value, bool):
self.values |= _forms(value)
def scrubber(self) -> Callable[[str], str]:
if not self.values:
return lambda text: text
# 长的在前:13900000001 先于它的一截被认出来
pattern = re.compile("|".join(_pattern(v) for v in sorted(self.values, key=lambda v: (-len(v), v))))
word = _masked_word()
return lambda text: pattern.sub(word, text) if text else text
def _outside_markers(text: str, scrub: Callable[[str], str]) -> str:
"""只换引用标记之外的文字:标记里是编号、行号、列名,换了会把一个好好的引用弄坏。"""
out: list[str] = []
last = 0
for m in MARKER_RE.finditer(text):
out += [scrub(text[last:m.start()]), m.group(0)]
last = m.end()
out.append(scrub(text[last:]))
return "".join(out)
# --------------------------------------------------------------------------
# 回答
# --------------------------------------------------------------------------
def _limit(outcome: VerifyOutcome) -> str | None:
if outcome.limit in LIMITS:
return outcome.limit
return "max_queries" if outcome.limit == "max_steps" else None
async def _answer(op: _Opening, ask: Ask, target: _Target | None, settings: dict[str, Any]) -> dict[str, Any]:
"""问 agent,组装回答文档,交回事件数据里回答的那几项。永远不抛:模型用不了、触顶都落在 ok / limit / error 里。"""
from app.api.evidence import _Masks
from app.api.settings import verify_model_spec
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(wanted)
live, lost = [s for s in wanted if s in usable], [s for s in wanted if s not in usable]
part = settings.get("inquiry") if isinstance(settings.get("inquiry"), dict) else {}
async with SessionLocal() as session:
spec = await verify_model_spec(session, None, settings=settings)
outcome = await run_verify_agent(
spec=spec, system=INQUIRY_SYSTEM, task=_task(op, ask, target, excerpts, refs, live, lost), sources=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)
# 遮罩:agent 自己的查询快照、报告自己的查询快照(被问的那句可能直接写着遮罩列的值,agent 会照着复述)
masks = _Masks()
hidden: dict[str, set[str]] = {}
secrets = _Secrets()
for query in outcome.queries:
snapshot = _quiet_load(artifact_store.load, query.get("artifact"))
if _query_shaped(snapshot):
hidden[query["artifact"]] = (await _hidden_columns(snapshot, masks))[0]
secrets.add(snapshot, hidden[query["artifact"]])
for entry in catalog.values():
if isinstance(entry, dict) and entry.get("kind") == "query":
snapshot = _quiet_load(op.loader, entry.get("artifact"))
if _query_shaped(snapshot):
secrets.add(snapshot, (await _hidden_columns(snapshot, masks))[0])
scrub = secrets.scrubber()
text, doc_id = "", None
source = (outcome.rationale or {}).get("source") if outcome.result is not None else None
if isinstance(source, str) and source.strip():
answer = _compose(source, {**refs, **((outcome.rationale or {}).get("catalog") or {})}, op.run_id, hidden,
scrub)
if lost:
answer = with_note(answer, f"注:这次运行还用到了数据源{_quoted(lost)},已不存在或已停用,未能查询,"
"以上回答不含其中的数据。")
text = str(answer.get("markdown") or "")
doc_id = await _store(answer, op, ask)
ok = bool(text.strip())
queries = [{**{k: q.get(k) for k in QUERY_KEYS}, "sql": scrub(str(q.get("sql") or ""))} for q in outcome.queries]
return {
"answer_text": text, "answer_doc": doc_id, "ok": ok, "limit": _limit(outcome),
"error": None if ok else (outcome.error or NO_ANSWER),
"inquiry_artifacts": [q["artifact"] for q in queries if q.get("artifact")], "queries": queries,
"model": outcome.model, "usage": {k: outcome.usage.get(k, 0) for k in USAGE_KEYS},
}
def _quiet_load(loader: Callable[[str], Any], artifact: Any) -> Any:
"""取一份快照;取不到、哈希不符都当没有(这里只是为了找遮罩列的值,取不到的那份本来也渲染不出来)。"""
if not artifact:
return None
try:
return loader(str(artifact))
except Exception: # noqa: BLE001
return None
def _compose(source: str, catalog: dict[str, Any], run_id: str, hidden: dict[str, set[str]],
scrub: Callable[[str], str]) -> dict[str, Any]:
"""agent 写的回答原文(带引用标记)→ 回答文档:目录是 agent 自己的查询,加上报告里的指标和运行输入;快照按遮罩
遮好再渲染。遮罩列的值先在标记之外的文字里换掉;换完正文里还认得出遮罩列的值(标记里写了值,解析不了,原样
出现在正文里),就连标记一起换了再组装一遍——宁可弄坏一个引用,也不让值露出来。"""
loader = _masking_loader(hidden)
doc = compose_doc(_outside_markers(source, scrub), catalog, run_id=run_id, loader=loader)
markdown = str(doc.get("markdown") or "")
if scrub(markdown) != markdown:
doc = compose_doc(scrub(source), catalog, run_id=run_id, loader=loader)
return doc
async def _store(doc: dict[str, Any], op: _Opening, ask: Ask) -> str | None:
try:
return await artifact_store.put_json(doc, kind=ANSWER_KIND, run_id=op.run_id,
meta={"report": ask.report, "unit": ask.unit})
except Exception: # noqa: BLE001 - 存不下不影响回答本身:正文照常交回,只是数字点不开
logger.warning("质询的回答文档保存失败(运行 %s)", op.run_id, exc_info=True)
return None
# --------------------------------------------------------------------------
# 写入与读取
# --------------------------------------------------------------------------
async def _record(op: _Opening, data: dict[str, Any]) -> tuple[Any, str | None]:
"""记一条 approval.inquiry。返回 (写入的事件, 没写进时的说明)。
停在审批时:那张卡仍待审批、运行仍停着才写(批过了,运行可能已经停到下一张卡上)。封存之后:仍是开始回答时
的那份封存才写。append_events 返回 None 就是运行在回答期间变了。"""
items = [(EventType.APPROVAL_INQUIRY, None, data)]
try:
if op.mode == "interrupted":
written = await append_events(op.run_id, items, require="interrupted", approval_id=op.approval_id)
else:
written = await append_events(op.run_id, items, require="sealed", seal=op.seal)
except Exception as e: # noqa: BLE001 - 写不进运行记录也要把回答交回给提问的人
logger.warning("质询的问答写入运行记录失败(运行 %s)", op.run_id, exc_info=True)
return None, f"这次质询的问答没能写入运行记录({describe_exception(e)})"
return (written[0], None) if written else (None, NOT_RECORDED)
def _iso(ts: Any) -> str | None:
if isinstance(ts, bool) or not isinstance(ts, (int, float)):
return None
return datetime.fromtimestamp(float(ts), timezone.utc).isoformat()
def inquiry_of(data: Any, *, seq: int | None, ts: Any) -> dict[str, Any]:
"""事件数据 → 接口上的 Inquiry 对象(前后端约定的形状)。POST 的返回和 GET 的列表都经这里,两边一样。"""
data = data if isinstance(data, dict) else {}
usage = data.get("usage") if isinstance(data.get("usage"), dict) else {}
return {
"seq": seq, "at": _iso(ts), "actor": data.get("actor"), "report": data.get("report"), "unit": data.get("unit"),
"quote": data.get("quote"), "question": str(data.get("question") or ""),
"answer": {"text": str(data.get("answer_text") or ""), "doc_artifact": data.get("answer_doc"),
"ok": data.get("ok") is True, "limit": data.get("limit"), "error": data.get("error")},
"queries": [{k: q.get(k) for k in QUERY_KEYS} for q in data.get("queries") or [] if isinstance(q, dict)],
"model": data.get("model"),
"usage": {k: usage.get(k, 0) for k in USAGE_KEYS},
"post_seal": bool(data.get("post_seal")), "approval_id": data.get("approval_id"),
}
async def ask(run_id: str, request: Ask, *, actor: str | None = None) -> dict[str, Any]:
"""答一条质询:{inquiry, recorded, note}。不受理时抛 InquiryRefused。
设置按现在的(不按发起时快照进运行选项的那份):质询不在图里执行、不参与重放,关掉「审批质询」应当马上生效。"""
from app.api.settings import VERIFY_SETTING_LABEL, verify_settings
async with SessionLocal() as session:
settings = await verify_settings(session)
part = settings.get("inquiry") if isinstance(settings.get("inquiry"), dict) else {}
if not part.get("enabled", True):
raise InquiryRefused(409, DISABLED_WORDS.format(label=VERIFY_SETTING_LABEL["inquiry.enabled"]),
code=INQUIRY_DISABLED)
with _ANSWERING.hold(run_id):
opening = await _open(run_id, request)
target = _target(opening.doc, request.unit) if request.unit else None
data = {"question": request.question, "quote": request.quote, "report": request.report, "unit": request.unit,
"approval_id": opening.approval_id, "actor": actor,
**await _answer(opening, request, target, settings)}
event, note = await _record(opening, data)
if event is not None:
return {"inquiry": inquiry_of(event.data, seq=event.seq, ts=event.ts), "recorded": True, "note": None}
shown = {**data, "post_seal": True} if opening.mode == "sealed" else data
return {"inquiry": inquiry_of(shown, seq=None, ts=time.time()), "recorded": False, "note": note}
async def _inquiry_events(run_id: str) -> list[tuple[int, float, dict[str, Any]]]:
async with SessionLocal() as session:
if await session.get(Run, run_id) is None:
raise InquiryRefused(404, "运行记录不存在", code=RUN_NOT_FOUND)
rows = await session.execute(
select(RunEvent.seq, RunEvent.ts, RunEvent.data)
.where(RunEvent.run_id == run_id, RunEvent.type == EventType.APPROVAL_INQUIRY).order_by(RunEvent.seq))
return [(seq, ts, data if isinstance(data, dict) else {}) for seq, ts, data in rows]
async def list_inquiries(run_id: str, *, report: str | None = None, unit: str | None = None) -> list[dict[str, Any]]:
"""这次运行的质询(封存前后的都算),按先后;给了 report / unit 只要问那份报告、那一句的。"""
return [inquiry_of(data, seq=seq, ts=ts) for seq, ts, data in await _inquiry_events(run_id)
if (not report or data.get("report") == report) and (not unit or data.get("unit") == unit)]
async def inquiry_artifact(run_id: str, artifact_id: str) -> dict[str, Any]:
"""质询引用过的工件:查询快照(遮罩列照证据面板遮好,SQL 里写着的遮罩列的值同样换掉)或回答文档(组装时就遮好了)。
{id, content, masked[, mask_note]}。不是这次运行的质询引用过的一律 404:工件 id 不能拿来取任意快照。"""
from app.api.evidence import SOURCE_GONE, _Masks
referenced = [data for _, _, data in await _inquiry_events(run_id)
if artifact_id in (data.get("inquiry_artifacts") or []) or data.get("answer_doc") == artifact_id]
if not referenced:
raise InquiryRefused(404, "这次运行的质询没有引用这件工件", code=ARTIFACT_NOT_FOUND)
try:
content = artifact_store.load(artifact_id)
except ValueError as e:
raise InquiryRefused(409, "工件与哈希不一致,疑似被修改,不予展示", code=ARTIFACT_TAMPERED) from e
if content is None:
raise InquiryRefused(404, "无法读取这件工件,可能已被删除", code=ARTIFACT_NOT_FOUND)
if not _query_shaped(content):
return {"id": artifact_id, "content": content, "masked": []}
masks = _Masks()
hidden, gone = await _hidden_columns(content, masks)
# 遮罩列的值按引用它的那几条质询各自查到的全部快照找:查询条件里写着的值,不一定在这一份快照的结果里
secrets = _Secrets()
for data in referenced:
for other in dict.fromkeys(str(a) for a in data.get("inquiry_artifacts") or []):
snapshot = content if other == artifact_id else _quiet_load(artifact_store.load, other)
if _query_shaped(snapshot):
secrets.add(snapshot, hidden if other == artifact_id else (await _hidden_columns(snapshot, masks))[0])
shown = masked_snapshot(content, hidden)
if isinstance(shown.get("sql"), str):
shown = {**shown, "sql": secrets.scrubber()(shown["sql"])}
columns = [str(c) for c in content.get("columns") or []]
out: dict[str, Any] = {"id": artifact_id, "content": shown, "masked": [c for c in columns if c.lower() in hidden]}
if gone:
out["mask_note"] = SOURCE_GONE.format(source="、".join(gone))
return out
__all__ = [
"BUSY_CONTINUE", "BUSY_DECIDE", "INQUIRY_BUSY", "INQUIRY_DISABLED", "INQUIRY_NOT_OPEN", "INQUIRY_SCHEMA",
"INQUIRY_SYSTEM", "NOT_RECORDED", "QUESTION_MAX", "QUOTE_MAX", "Ask", "InquiryRefused", "ask", "inquiry_artifact",
"inquiry_busy", "inquiry_of", "list_inquiries", "masked_snapshot", "parse_ask",
]
+768
View File
@@ -0,0 +1,768 @@
"""审批质询(engine/inquiry.py、api/inquiry.py):运行停在人工审批时,审批人针对报告里的一句话提问,一个只读 agent
现查现答;问答写进这次运行(approval.inquiry 事件),随运行结束一起封存,报告本身一个字不改。
- 停在审批时质询:事件 node_id 为空、数据按前后端约定的形状;质询的查询工件只在 inquiry_artifacts / queries 里,
不在 query_artifact 下(证据接口不把它们当成报告的证据);批准、运行结束封存之后 verify 一致,封存范围包含这条事件;
用量计入运行用量
- 回答期间运行被批准(竞态):回答照样交回,recorded 为假、note 说明,事件不落库
- 同一运行同时只答一条:第二条质询、审批回复(批准 / 驳回)、继续运行都回 409 inquiry_busy,拒掉的批准不留痕迹
- 封存之后:探索运行可以追问(post_seal,落在封存之后,不计运行用量),正式运行 409;设置里关了 409;
没停在那张审批卡上、还在执行、封存核对不通过都 409;输入不合法 422,指明是哪一项
- 回答:先给结论,数字只能引用自己查询的格子或报告已有的指标,裸数字被标出;遮罩列的格子渲染成「已遮罩」,
回答正文、查询清单、事件里都不出现遮罩列的值
- 触顶、失败照样记一条事件
- 读取:列表按报告、句子筛选,按时间排;质询工件只认质询引用过的,遮罩照证据面板
- 恢复后的重放不受影响:前面的节点不重复执行
模型一律是 MockChatModel 剧本,数据源是现造的 SQLite 文件,区域、金额、手机号都是虚构的。
"""
from __future__ import annotations
import asyncio
import json
import sqlite3
import uuid
from dataclasses import dataclass, field
import pytest
from httpx import ASGITransport, AsyncClient
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 inquiry as inquiry_mod
from app.engine import spend, verify_agent
from app.engine.evidence import iter_segments, iter_units
from app.engine.judge import SPEND_KEY as JUDGE_SPEND_KEY
from app.engine.nodes import report as report_mod
from app.engine.runner import run_manager, verify_manifest
from app.main import app
from app.providers.factory import ProviderNotConfigured
from app.providers.mock_model import MockChatModel
PRICED = "claude-sonnet-5"
#: 各区域销售额:华东 1500(其中 300 是测试账号下的单)、华南 900、华北 400
QUERY = "SELECT region AS 区域, SUM(amount) AS 销售额 FROM sales GROUP BY region ORDER BY 销售额 DESC"
TEXT = "华东销售额 [[v:Q1.r0.销售额]],居各区域之首。[[see:Q1]]华南销售额 [[v:Q1.r1.销售额]]。[[see:Q1]]"
EAST, SOUTH = "华东销售额 1,500,居各区域之首。", "华南销售额 900。"
NO_TEST = ("SELECT region AS 区域, SUM(amount) AS 销售额 FROM sales WHERE is_test = 0 GROUP BY region "
"ORDER BY 销售额 DESC")
NO_TEST_ANSWER = "仍居各区域之首:排除测试账号后华东为 [[v:Q1.r0.销售额]],华南为 [[v:Q1.r1.销售额]]。[[see:Q1]]"
NO_TEST_TEXT = "仍居各区域之首:排除测试账号后华东为 1,200,华南为 900。"
QUESTION = "排除测试账号会怎样?"
USAGE_KEYS = {"input_tokens", "output_tokens", "cost_usd", "calls", "unpriced_calls"}
ITEM_KEYS = {"seq", "at", "actor", "report", "unit", "quote", "question", "answer", "queries", "model", "usage",
"post_seal", "approval_id"}
EVENT_KEYS = {"question", "quote", "report", "unit", "approval_id", "actor", "answer_text", "answer_doc", "ok",
"limit", "error", "inquiry_artifacts", "queries", "model", "usage"}
QUERY_KEYS = {"alias", "artifact", "sql", "row_count", "truncated", "source"}
NOT_RECORDED = "运行状态已经变化,这次质询的问答没有写入运行记录"
BUSY_DECIDE = {"detail": "正在回答质询,回答完再批准或驳回", "code": "inquiry_busy"}
@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()
@pytest.fixture
async def client():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
yield c
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, is_test 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)
@pytest.fixture
async def masked_shop(tmp_path):
"""同样的四笔订单,买家手机号这一列设了遮罩。"""
name, source_id = await _source(tmp_path, options={"mask_columns": "buyer_phone"})
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):
return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
def gated(source: str, *, gate: bool = True) -> dict:
"""取数 → 报告 → 人工审批 → 成果。gate 为假时没有审批,一口气跑完、封存。"""
return chain(
node("start", "input"),
node("fetch", "tool", tool=f"db_query__{source}", args={"sql": QUERY}),
node("write", "report", "写周报", instructions="写周报"),
*([node("gate", "human", mode="approve", title="周报可以出具吗")] if gate else []),
node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]),
)
@dataclass
class Play:
"""质询 agent 对某个问题的剧本:依次执行的查询,最后交的回答。"""
queries: list[str] = field(default_factory=list)
answer: str = "查不到判断所需的数据。"
PLAYS = {"排除测试账号": Play([NO_TEST], NO_TEST_ANSWER)}
class Models:
"""写作者和质询 agent 两份剧本。写作者回 writer;质询 agent 按任务里的问题认出剧本,照着查询、交回答。
before:质询 agent 每次调用模型之前先 await 它(测试在这里卡住回答、或者在回答途中批掉审批)。
tasks 记下质询 agent 收到的任务原文,finals 记下交回答那次调用收到的消息,specs 记下每次要的模型。"""
def __init__(self, monkeypatch, source: str | None = None, *, writer: str = TEXT, plays=None,
model_id: str = PRICED, broken: Exception | None = None, before=None):
self.tasks: list[str] = []
self.finals: list[list] = []
self.writes = 0
self.specs: list = []
plays = PLAYS if plays is None else plays
def which(task: str) -> Play:
return next((p for k, p in plays.items() if k in task), Play())
def decide(model, messages):
system = str(messages[0].content)
if inquiry_mod.INQUIRY_SYSTEM in system:
task = str(messages[1].content) if len(messages) > 1 else ""
play = which(task)
if model.response_format:
self.finals.append(list(messages))
return AIMessage(content=json.dumps({"rationale": play.answer}, ensure_ascii=False))
if not any(isinstance(m, ToolMessage) for m in messages):
self.tasks.append(task)
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"iq-{done}"}
return AIMessage(content="查一下。", tool_calls=[call])
return AIMessage(content="查完了。")
if report_mod.ROLE in system:
self.writes += 1
return AIMessage(content=writer)
raise AssertionError(f"没想到的模型调用:{system[:60]}")
monkeypatch.setattr(MockChatModel, "_decide", decide)
original = MockChatModel._agenerate
async def held(chat, messages, stop=None, run_manager=None, **kwargs):
if before is not None and inquiry_mod.INQUIRY_SYSTEM in str(messages[0].content):
await before()
return await original(chat, messages, stop, run_manager, **kwargs)
monkeypatch.setattr(MockChatModel, "_agenerate", held)
async def verify_model(session, spec):
self.specs.append(spec)
if broken is not None:
raise broken
return MockChatModel(model_name=model_id), model_id
monkeypatch.setattr(verify_agent, "get_chat_model", verify_model)
async def _row(run_id: str) -> Run:
async with SessionLocal() as session:
return await session.get(Run, run_id)
async def wait(run_id: str, statuses=("succeeded", "failed")) -> Run:
for _ in range(600):
row = await _row(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 await _row(run_id)
await asyncio.sleep(0.03)
raise AssertionError(f"运行 {run_id} 停在 {row.status}")
async def pause(source: str, *, run_class: str = "exploratory", graph: dict | None = None) -> Run:
run = await run_manager.start(graph=graph or gated(source), input_payload={}, run_class=run_class)
return await wait(run.id, ("interrupted",))
async def finish(source: str, *, run_class: str = "exploratory", graph: dict | None = None) -> Run:
run = await run_manager.start(graph=graph or gated(source, gate=False), input_payload={}, run_class=run_class)
return await wait(run.id)
async def pending(run_id: str) -> list[Approval]:
async with SessionLocal() as session:
return list((await session.execute(
select(Approval).where(Approval.run_id == run_id, Approval.status == "pending"))).scalars())
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 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)
async def ask(client, run_id: str, **body):
return await client.post(f"/api/runs/{run_id}/inquiries", json=body)
async def set_verify(value: dict) -> None:
async with SessionLocal() as session:
session.add(Setting(key="verify", value=value))
await session.commit()
# --------------------------------------------------------------------------
# 停在审批时质询 → 写进事件 → 批准 → 封存
# --------------------------------------------------------------------------
async def test_asking_while_waiting_is_recorded_and_sealed_with_the_run(monkeypatch, shop, client):
models = Models(monkeypatch, shop)
paused = await pause(shop)
[approval] = await pending(paused.id)
doc = await report_doc(paused.id)
east = unit_of(doc, EAST)
r = await ask(client, paused.id, report="write", unit=east, quote="居各区域之首", question=QUESTION,
approval_id=approval.id)
assert r.status_code == 200, r.text
body = r.json()
assert body.keys() == {"inquiry", "recorded", "note"}
assert body["recorded"] is True and body["note"] is None
item = body["inquiry"]
assert item.keys() == ITEM_KEYS
assert (item["report"], item["unit"], item["quote"], item["question"], item["approval_id"]) == (
"write", east, "居各区域之首", QUESTION, approval.id)
assert item["post_seal"] is False and item["actor"] is None and isinstance(item["at"], str)
answer = item["answer"]
assert answer.keys() == {"text", "doc_artifact", "ok", "limit", "error"}
assert (answer["ok"], answer["limit"], answer["error"]) == (True, None, None)
assert answer["text"] == NO_TEST_TEXT, "回答先给结论,数字按查询结果换好"
answer_doc = artifact_store.load(answer["doc_artifact"])
assert answer_doc["markdown"] == answer["text"] and answer_doc["violations"] == []
cited = [s["cite"] for s in iter_segments(answer_doc) if s.get("cite")]
assert [(c["alias"], c["locator"]["row"], c["value"]) for c in cited] == [("Q1", 0, 1200), ("Q1", 1, 900)]
[query] = item["queries"]
assert query.keys() == QUERY_KEYS
assert (query["alias"], query["source"], query["sql"], query["row_count"], query["truncated"]) == (
"Q1", shop, NO_TEST, 3, False)
assert artifact_store.load(query["artifact"])["rows"] == [["华东", 1200], ["华南", 900], ["华北", 400]]
assert item["model"] == PRICED
assert item["usage"].keys() == USAGE_KEYS and item["usage"]["calls"] >= 2 and item["usage"]["cost_usd"] > 0
# 交给质询 agent 的:被问的那句、它挂的证据摘录(原报告的编号标明)、选中的原文、问题、可以查的数据源
[task] = models.tasks
assert EAST in task and "【原报告 Q1】" in task and "居各区域之首" in task and QUESTION in task and shop in task
# 事件:node_id 为空,数据按约定;查询工件不在 query_artifact 下
[event] = await events(paused.id, "approval.inquiry")
assert event.node_id is None and event.seq == item["seq"]
assert event.data.keys() == EVENT_KEYS
assert event.data["inquiry_artifacts"] == [query["artifact"]]
assert event.data["queries"] == item["queries"]
assert (event.data["answer_text"], event.data["answer_doc"]) == (answer["text"], answer["doc_artifact"])
assert "query_artifact" not in event.data
assert (await _row(paused.id)).status == "interrupted"
# 列表里是同一条
listed = await client.get(f"/api/runs/{paused.id}/inquiries")
assert listed.status_code == 200 and listed.json() == {"items": [item]}
# 批准 → 运行结束封存:这条事件在封存范围里,verify 一致
r = await client.post(f"/api/approvals/{approval.id}/decide", json={"approved": True})
assert r.status_code == 200, r.text
final = await wait(paused.id)
assert final.status == "succeeded", final.error
assert final.manifest_seq > event.seq
assert (await verify_manifest(paused.id))["ok"] is True
# 用量计进运行用量,明细在 inquiry 下
assert final.usage["inquiry"]["calls"] == item["usage"]["calls"]
assert final.usage["inquiry"]["input_tokens"] == item["usage"]["input_tokens"]
# 证据接口不把质询的查询当成报告的证据
from app.api.evidence import _sealed
assert query["artifact"] not in (await _sealed(paused.id)).queries
# 报告本身一个字没改
assert (await report_doc(paused.id)) == doc
# 改它一个字,封存核对就对不上:它确实在封存范围里
async with SessionLocal() as session:
stored = await session.get(RunEvent, event.id)
stored.data = {**stored.data, "answer_text": "改过的回答"}
await session.commit()
assert (await verify_manifest(paused.id))["ok"] is False
async def test_an_approval_that_lands_while_answering_leaves_no_record(monkeypatch, shop, client):
"""回答途中运行被批准了(另一个进程、或者抢在登记之前的批准):回答照样交回给提问的人,但不写进运行记录。"""
seen: dict = {}
async def approve_meanwhile():
if not seen.get("done"):
seen["done"] = True
await run_manager.resume(seen["run"], {"approved": True}, approval_id=seen["approval"])
Models(monkeypatch, shop, before=approve_meanwhile)
paused = await pause(shop)
[approval] = await pending(paused.id)
seen.update(run=paused.id, approval=approval.id)
r = await ask(client, paused.id, report="write", unit=None, question=QUESTION, approval_id=approval.id)
assert r.status_code == 200, r.text
body = r.json()
assert body["recorded"] is False and body["note"] == NOT_RECORDED
assert body["inquiry"]["seq"] is None and body["inquiry"]["answer"]["ok"] is True
assert body["inquiry"]["answer"]["text"] == NO_TEST_TEXT
final = await wait(paused.id)
assert final.status == "succeeded", final.error
assert await events(paused.id, "approval.inquiry") == []
assert (await verify_manifest(paused.id))["ok"] is True
assert "inquiry" not in final.usage
assert (await client.get(f"/api/runs/{paused.id}/inquiries")).json() == {"items": []}
async def test_while_answering_approving_resuming_and_continuing_are_refused(monkeypatch, shop, client):
entered, release = asyncio.Event(), asyncio.Event()
async def hold():
entered.set()
await release.wait()
Models(monkeypatch, shop, before=hold)
paused = await pause(shop)
[approval] = await pending(paused.id)
asking = asyncio.create_task(ask(client, paused.id, report="write", unit=None, question=QUESTION,
approval_id=approval.id))
try:
await asyncio.wait_for(entered.wait(), 10)
refused = await client.post(f"/api/approvals/{approval.id}/decide", json={"approved": False, "note": "先别批"},
headers={"X-Actor": "reviewer"})
assert refused.status_code == 409 and refused.json() == BUSY_DECIDE
refused = await client.post(f"/api/runs/{paused.id}/resume",
json={"response": {"approved": True}, "approval_id": approval.id})
assert refused.status_code == 409 and refused.json() == BUSY_DECIDE
refused = await client.post(f"/api/runs/{paused.id}/continue", json={})
assert refused.status_code == 409 and refused.json()["code"] == "inquiry_busy"
second = await ask(client, paused.id, report="write", unit=None, question="华南为什么低?",
approval_id=approval.id)
assert second.status_code == 409 and second.json()["code"] == "inquiry_busy"
# 拒掉的批准不留痕迹:卡仍待审批,没记下签批人
async with SessionLocal() as session:
card = await session.get(Approval, approval.id)
assert card.status == "pending" and card.resolved_by is None
assert (await _row(paused.id)).status == "interrupted"
finally:
release.set()
answered = await asyncio.wait_for(asking, 30)
assert answered.status_code == 200 and answered.json()["recorded"] is True
# 回答完了,照常批准
r = await client.post(f"/api/approvals/{approval.id}/decide", json={"approved": True})
assert r.status_code == 200, r.text
assert (await wait(paused.id)).status == "succeeded"
# --------------------------------------------------------------------------
# 封存之后:探索运行可以追问,正式运行不行
# --------------------------------------------------------------------------
async def test_after_the_seal_only_exploratory_runs_take_inquiries(monkeypatch, shop, client):
Models(monkeypatch, shop)
done = await finish(shop)
assert done.status == "succeeded", done.error
east = unit_of(await report_doc(done.id), EAST)
r = await ask(client, done.id, report="write", unit=east, question=QUESTION)
assert r.status_code == 200, r.text
body = r.json()
assert body["recorded"] is True and body["note"] is None
item = body["inquiry"]
assert item["post_seal"] is True and item["approval_id"] is None and item["answer"]["text"] == NO_TEST_TEXT
[event] = await events(done.id, "approval.inquiry")
assert event.data["post_seal"] is True and event.seq > done.manifest_seq
after = await _row(done.id)
assert (after.manifest_seq, after.manifest_hash) == (done.manifest_seq, done.manifest_hash)
assert (await verify_manifest(done.id))["ok"] is True
assert "inquiry" not in after.usage, "封存之后的问答是批注,不计进运行用量"
assert (await client.get(f"/api/runs/{done.id}/inquiries")).json()["items"] == [item]
# 带着审批卡来问:那张卡已经不在等了
r = await ask(client, done.id, report="write", unit=east, question=QUESTION, approval_id="no-such-card")
assert r.status_code == 409 and r.json()["code"] == "inquiry_not_open"
formal = await finish(shop, run_class="formal")
assert formal.status == "succeeded", formal.error
r = await ask(client, formal.id, report="write", unit=unit_of(await report_doc(formal.id), EAST),
question=QUESTION)
assert r.status_code == 409 and r.json()["code"] == "inquiry_not_open"
assert "正式运行" in r.json()["detail"]
assert await events(formal.id, "approval.inquiry") == []
async def test_a_formal_run_waiting_for_approval_takes_inquiries(monkeypatch, shop, client):
"""正式运行停在审批时可以质询:问答随运行一起封存,正式运行同样可以复核。"""
Models(monkeypatch, shop)
paused = await pause(shop, run_class="formal")
[approval] = await pending(paused.id)
r = await ask(client, paused.id, report="write", unit=None, question=QUESTION, approval_id=approval.id)
assert r.status_code == 200 and r.json()["recorded"] is True
await client.post(f"/api/approvals/{approval.id}/decide", json={"approved": True})
final = await wait(paused.id)
[event] = await events(paused.id, "approval.inquiry")
assert final.manifest_seq > event.seq and (await verify_manifest(paused.id))["ok"] is True
async def test_switched_off_in_settings_it_is_refused(monkeypatch, shop, client):
await set_verify({"inquiry": {"enabled": False}})
models = Models(monkeypatch, shop)
paused = await pause(shop)
[approval] = await pending(paused.id)
r = await ask(client, paused.id, report="write", unit=None, question=QUESTION, approval_id=approval.id)
assert r.status_code == 409 and r.json()["code"] == "inquiry_disabled"
assert "「审批质询 · 启用」" in r.json()["detail"]
assert models.tasks == [] and models.specs == []
assert await events(paused.id, "approval.inquiry") == []
async def test_the_run_must_be_waiting_on_that_approval(monkeypatch, shop, client):
models = Models(monkeypatch, shop)
paused = await pause(shop)
[approval] = await pending(paused.id)
r = await ask(client, paused.id, report="write", unit=None, question=QUESTION)
assert r.status_code == 422 and r.json() == {"detail": r.json()["detail"], "field": "approval_id"}
r = await ask(client, paused.id, report="write", unit=None, question=QUESTION, approval_id="no-such-card")
assert r.status_code == 409 and r.json()["code"] == "inquiry_not_open"
# 还在执行的运行
async with SessionLocal() as session:
running = Run(workflow_name="还在执行", status="running", graph={}, input={})
session.add(running)
await session.commit()
running_id = running.id
r = await ask(client, running_id, report="write", unit=None, question=QUESTION)
assert r.status_code == 409 and r.json()["code"] == "inquiry_not_open"
r = await ask(client, "no-such-run", report="write", unit=None, question=QUESTION)
assert r.status_code == 404
# 封存核对不通过的探索运行:报告和证据已不可信,不再质询
done = await finish(shop)
async with SessionLocal() as session:
first = (await session.execute(select(RunEvent).where(RunEvent.run_id == done.id)
.order_by(RunEvent.seq).limit(1))).scalar_one()
first.data = {**(first.data or {}), "tampered": True}
await session.commit()
r = await ask(client, done.id, report="write", unit=None, question=QUESTION)
assert r.status_code == 409 and r.json()["code"] == "inquiry_not_open"
assert models.tasks == [], "开放条件不满足时不问模型"
@pytest.mark.parametrize("body, field_name", [
({"unit": None, "question": QUESTION}, "report"),
({"report": 3, "question": QUESTION}, "report"),
({"report": "write", "unit": None}, "question"),
({"report": "write", "unit": None, "question": " "}, "question"),
({"report": "write", "unit": None, "question": "问" * 1001}, "question"),
({"report": "write", "unit": None, "question": QUESTION, "quote": "引" * 501}, "quote"),
({"report": "write", "unit": None, "question": QUESTION, "quote": 5}, "quote"),
({"report": "write", "unit": 12, "question": QUESTION}, "unit"),
({"report": "write", "unit": None, "question": QUESTION, "approval_id": 7}, "approval_id"),
])
async def test_bad_requests_name_the_field(monkeypatch, shop, client, body, field_name):
Models(monkeypatch, shop)
paused = await pause(shop)
[approval] = await pending(paused.id)
r = await ask(client, paused.id, **{"approval_id": approval.id, **body})
assert r.status_code == 422, r.text
assert r.json().keys() == {"detail", "field"} and r.json()["field"] == field_name
assert r.json()["detail"] and "[" not in r.json()["detail"]
async def test_the_report_and_the_sentence_must_exist(monkeypatch, shop, client):
models = Models(monkeypatch, shop)
paused = await pause(shop)
[approval] = await pending(paused.id)
r = await ask(client, paused.id, report="no-such-report", unit=None, question=QUESTION, approval_id=approval.id)
assert r.status_code == 422 and r.json()["field"] == "report"
r = await ask(client, paused.id, report="write", unit="u999", question=QUESTION, approval_id=approval.id)
assert r.status_code == 422 and r.json()["field"] == "unit"
r = await client.post(f"/api/runs/{paused.id}/inquiries", content=b"[1, 2]",
headers={"content-type": "application/json"})
assert r.status_code == 422 and r.json().keys() == {"detail", "field"}
r = await client.post(f"/api/runs/{paused.id}/inquiries", content=b"{not json",
headers={"content-type": "application/json"})
assert r.status_code == 422 and r.json().keys() == {"detail", "field"}
assert models.tasks == []
# --------------------------------------------------------------------------
# 回答:出处、裸数字、遮罩
# --------------------------------------------------------------------------
TOP_BUYER = "SELECT buyer_phone, amount AS 金额 FROM sales ORDER BY amount DESC LIMIT 1"
BUYER_ORDERS = "SELECT COUNT(*) AS 笔数 FROM sales WHERE buyer_phone = '13900000001'"
async def test_bare_numbers_are_flagged_and_masked_values_never_show(monkeypatch, masked_shop, client):
plays = {"买得最多": Play([TOP_BUYER, BUYER_ORDERS],
"金额最高的一笔来自 [[v:Q1.r0.buyer_phone]](手机号 13900000001),金额 [[v:Q1.r0.金额]];"
"这位客户共 [[v:Q2.r0.笔数]] 笔订单,另有 3 笔小额订单。")}
Models(monkeypatch, masked_shop, plays=plays)
paused = await pause(masked_shop)
[approval] = await pending(paused.id)
r = await ask(client, paused.id, report="write", unit=None, question="买得最多的是哪位客户?",
approval_id=approval.id)
assert r.status_code == 200, r.text
assert "1390000000" not in r.text, "回答、查询清单里都不能出现遮罩列的值"
item = r.json()["inquiry"]
answer = item["answer"]
assert answer["ok"] is True
assert answer["text"] == "金额最高的一笔来自 已遮罩(手机号 已遮罩),金额 1,200;这位客户共 1 笔订单,另有 3 笔小额订单。"
doc = artifact_store.load(answer["doc_artifact"])
assert "1390000000" not in json.dumps(doc, ensure_ascii=False)
# 裸数字被标出来:界面在它下面画「无出处」
assert [s["text"] for s in iter_segments(doc) if s.get("issue") == "uncited_number"] == ["3"]
# 遮罩列的格子:出处照记,值是「已遮罩」
masked = [s["cite"] for s in iter_segments(doc) if (s.get("cite") or {}).get("locator", {}).get("column")
== "buyer_phone"]
assert [(c["alias"], c["status"], c["value"]) for c in masked] == [("Q1", "resolved", "已遮罩")]
first, second = item["queries"]
assert first["sql"] == TOP_BUYER
assert second["sql"] == "SELECT COUNT(*) AS 笔数 FROM sales WHERE buyer_phone = '已遮罩'"
[event] = await events(paused.id, "approval.inquiry")
assert "1390000000" not in json.dumps(event.data, ensure_ascii=False)
# 质询工件:遮罩照证据面板,查询里写着的遮罩列的值同样换掉
got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{first['artifact']}")
assert got.status_code == 200, got.text
assert got.json()["content"]["rows"] == [["已遮罩", 1200]] and got.json()["masked"] == ["buyer_phone"]
got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{second['artifact']}")
assert got.status_code == 200 and "1390000000" not in got.text
assert got.json()["content"]["rows"] == [[1]]
async def test_the_answer_may_cite_the_reports_own_metrics(monkeypatch, client):
"""报告里已有的出处(口径卡指标)回答里可以直接引用;这次运行没用到查询工具,agent 只能依据证据作答。"""
metrics = [{"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, "expression": "vars.kpi.gmv"}]
graph = chain(
node("start", "input"),
node("calc", "transform", mode="expression", expression="{'gmv': 45678.5}", assign_to="kpi"),
node("caliber", "metrics", caliber="周报口径", caliber_version="v2", metrics=metrics),
node("write", "report", instructions="写周报"),
node("gate", "human", mode="approve", title="确认"),
node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]),
)
plays = {"销售额": Play([], "按口径卡,本周销售额为 [[m:gmv]],这是本周的口径值。[[see:m:gmv]]")}
models = Models(monkeypatch, None, writer="本周销售额 [[m:gmv]]。[[see:m:gmv]]", plays=plays)
paused = await pause("", graph=graph)
[approval] = await pending(paused.id)
doc = await report_doc(paused.id)
r = await ask(client, paused.id, report="write", unit=unit_of(doc, "本周销售额 45,678.5元。"),
question="销售额是怎么算的?", approval_id=approval.id)
assert r.status_code == 200, r.text
answer = r.json()["inquiry"]["answer"]
assert answer["text"] == "按口径卡,本周销售额为 45,678.5元,这是本周的口径值。"
assert artifact_store.load(answer["doc_artifact"])["violations"] == []
[task] = models.tasks
assert "[[m:gmv]]" in task and "销售额 = 45,678.5元" in task
assert r.json()["inquiry"]["queries"] == []
async def test_a_cell_in_a_table_comes_with_the_whole_table(monkeypatch, shop, client):
"""问的是表格里的一格:一格的值离了行列说不清是什么,任务里另带整张表(表头在前)和这一格的出处摘录。"""
models = Models(monkeypatch, shop, writer="各区域销售额如下。\n\n[[table:Q1]]")
paused = await pause(shop)
[approval] = await pending(paused.id)
doc = await report_doc(paused.id)
cell = unit_of(doc, "1,500")
r = await ask(client, paused.id, report="write", unit=cell, question=QUESTION, approval_id=approval.id)
assert r.status_code == 200, r.text
[task] = models.tasks
assert "「1,500」" in task and "区域 | 销售额\n华东 | 1,500\n华南 | 900\n华北 | 400" in task
# 摘录照裁判的办法:被引用的列排在前面,只给被引用的那一行
assert "【原报告 Q1】" in task and "列:销售额 | 区域\nr0:1,500 | 华东\n" in task
def test_masked_values_are_matched_as_whole_values():
"""遮罩列的值按整值认:更长的数里的一截、千分位写法里的一截不换;一个字的值不换(换了会把不相干的词弄坏);
数按原样和千分位两种写法认。"""
secrets = inquiry_mod._Secrets()
snapshot = {"columns": ["phone", "name", "gender", "code", "salary", "region"],
"rows": [["13900000001", "张伟", "男", "390", 12000, "华东"]]}
secrets.add(snapshot, {"phone", "name", "gender", "code", "salary"})
scrub = secrets.scrubber()
assert scrub("客户 13900000001、139000000012、1,390 元、390 笔、张伟买了男装,月薪 12,000,另 12000.5") == (
"客户 已遮罩、139000000012、1,390 元、已遮罩 笔、已遮罩买了男装,月薪 已遮罩,另 12000.5")
assert scrub("华东") == "华东", "没遮的列不换"
assert inquiry_mod._Secrets().scrubber()("13900000001") == "13900000001"
# --------------------------------------------------------------------------
# 触顶与失败
# --------------------------------------------------------------------------
async def test_hitting_the_query_limit_is_recorded(monkeypatch, shop, client):
await set_verify({"inquiry": {"max_queries": 1}})
plays = {"排除测试账号": Play([NO_TEST, QUERY], NO_TEST_ANSWER)}
models = Models(monkeypatch, shop, plays=plays)
paused = await pause(shop)
[approval] = await pending(paused.id)
r = await ask(client, paused.id, report="write", unit=None, question=QUESTION, approval_id=approval.id)
assert r.status_code == 200, r.text
item = r.json()["inquiry"]
assert (item["answer"]["ok"], item["answer"]["limit"]) == (True, "max_queries"), "触顶照样按查到的结果收尾"
# 没指明句子:任务里给报告正文,不给哪一句的摘录
[task] = models.tasks
assert "审批人没有指明是哪一句" in task and f"报告正文:\n{EAST}{SOUTH}" in task and "原报告" not in task
assert [q["sql"] for q in item["queries"]] == [NO_TEST]
[event] = await events(paused.id, "approval.inquiry")
assert (event.data["ok"], event.data["limit"]) == (True, "max_queries")
async def test_an_unavailable_model_is_recorded_as_a_failure(monkeypatch, shop, client):
Models(monkeypatch, shop, broken=ProviderNotConfigured("没有配置模型接入"))
# 写作者照常用演示模型:只有质询 agent 的模型用不了
paused = await pause(shop)
[approval] = await pending(paused.id)
r = await ask(client, paused.id, report="write", unit=None, question=QUESTION, approval_id=approval.id)
assert r.status_code == 200, r.text
body = r.json()
assert body["recorded"] is True
answer = body["inquiry"]["answer"]
assert answer["ok"] is False and answer["text"] == "" and answer["doc_artifact"] is None
assert answer["error"] and "模型" in answer["error"]
assert body["inquiry"]["queries"] == [] and body["inquiry"]["usage"]["calls"] == 0
[event] = await events(paused.id, "approval.inquiry")
assert event.data["ok"] is False and event.data["error"] == answer["error"]
async def test_the_daily_cap_is_recorded_as_a_limit(monkeypatch, shop, client):
async with SessionLocal() as session:
session.add(Setting(key=spend.SPEND_KEY, value={"date": spend._today(), "usd": 5.0, "calls": 3}))
await session.commit()
models = Models(monkeypatch, shop)
paused = await pause(shop)
[approval] = await pending(paused.id)
r = await ask(client, paused.id, report="write", unit=None, question=QUESTION, approval_id=approval.id)
assert r.status_code == 200, r.text
answer = r.json()["inquiry"]["answer"]
assert (answer["ok"], answer["limit"]) == (False, "daily_max_usd") and answer["error"]
assert models.finals == []
[event] = await events(paused.id, "approval.inquiry")
assert event.data["limit"] == "daily_max_usd"
# --------------------------------------------------------------------------
# 读取
# --------------------------------------------------------------------------
async def test_listing_filters_by_report_and_sentence_and_artifacts_are_whitelisted(monkeypatch, shop, client):
plays = {"排除测试账号": Play([NO_TEST], NO_TEST_ANSWER), "华南为什么": Play([], "查不到判断所需的数据。")}
Models(monkeypatch, shop, plays=plays)
paused = await pause(shop)
[approval] = await pending(paused.id)
doc = await report_doc(paused.id)
east, south = unit_of(doc, EAST), unit_of(doc, SOUTH)
first = (await ask(client, paused.id, report="write", unit=east, question=QUESTION,
approval_id=approval.id)).json()["inquiry"]
second = (await ask(client, paused.id, report="write", unit=south, question="华南为什么低?",
approval_id=approval.id)).json()["inquiry"]
assert first["seq"] < second["seq"]
listed = await client.get(f"/api/runs/{paused.id}/inquiries")
assert listed.json() == {"items": [first, second]}
assert (await client.get(f"/api/runs/{paused.id}/inquiries", params={"report": "write", "unit": south})
).json() == {"items": [second]}
assert (await client.get(f"/api/runs/{paused.id}/inquiries", params={"report": "other"})).json() == {"items": []}
assert (await client.get("/api/runs/no-such-run/inquiries")).status_code == 404
[query] = first["queries"]
got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{query['artifact']}")
assert got.status_code == 200, got.text
assert got.json()["id"] == query["artifact"] and got.json()["masked"] == []
assert got.json()["content"]["rows"] == [["华东", 1200], ["华南", 900], ["华北", 400]]
# 回答文档也是质询引用过的工件
got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{first['answer']['doc_artifact']}")
assert got.status_code == 200 and got.json()["content"]["markdown"] == NO_TEST_TEXT
# 报告自己的查询快照不是质询引用过的
[fetched] = [e for e in await events(paused.id, "tool.end", "fetch")]
got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{fetched.data['query_artifact']}")
assert got.status_code == 404
assert (await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/nothing")).status_code == 404
# --------------------------------------------------------------------------
# 重放
# --------------------------------------------------------------------------
async def test_resuming_after_an_inquiry_does_not_repeat_the_nodes(monkeypatch, shop, client):
"""质询不发 tool.start(_stale_replay 只看它),批准后从断点继续:取数、报告都不再执行一遍。"""
models = Models(monkeypatch, shop)
paused = await pause(shop)
[approval] = await pending(paused.id)
r = await ask(client, paused.id, report="write", unit=None, question=QUESTION, approval_id=approval.id)
assert r.json()["recorded"] is True
assert [e for e in await events(paused.id, "tool.start") if e.node_id is None] == []
await client.post(f"/api/approvals/{approval.id}/decide", json={"approved": True})
final = await wait(paused.id)
assert final.status == "succeeded", final.error
for node_id in ("fetch", "write"):
assert len(await events(paused.id, "node.started", node_id)) == 1, f"{node_id} 又执行了一遍"
assert len(await events(paused.id, "tool.start", "fetch")) == 1
assert models.writes == 1