From b54545b5d450b464a96a52822535d3d8574c4ee7 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Tue, 6 Oct 2026 13:01:20 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8F=8C=E7=9B=B2=E5=A4=8D=E7=AE=97=E5=9C=A8?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E6=97=B6=E9=95=BF=E4=B8=8A=E9=99=90=E5=86=85?= =?UTF-8?q?=E6=94=B6=E5=B0=BE=EF=BC=9A=E6=AF=8F=E9=A1=B9=E7=9B=B2=E7=AE=97?= =?UTF-8?q?=E5=BC=80=E5=A7=8B=E4=B9=8B=E5=89=8D=E7=9C=8B=E8=BF=99=E4=B8=80?= =?UTF-8?q?=E6=AE=B5=E6=89=A7=E8=A1=8C=E8=BF=98=E5=89=A9=E5=A4=9A=E5=B0=91?= =?UTF-8?q?=E6=97=B6=E9=97=B4=EF=BC=8C=E4=B8=8D=E5=A4=9F=E5=B0=B1=E4=B8=8D?= =?UTF-8?q?=E5=86=8D=E5=BC=80=E5=A7=8B=E3=80=81=E6=8C=89=E6=97=A0=E6=B3=95?= =?UTF-8?q?=E5=A4=8D=E7=AE=97=E8=AE=B0=E7=BC=BA=E5=8F=A3=EF=BC=8C=E5=A4=9F?= =?UTF-8?q?=E7=9A=84=E8=AF=9D=E6=97=B6=E9=95=BF=E4=B8=8A=E9=99=90=E6=8C=89?= =?UTF-8?q?=E5=89=A9=E4=BD=99=E6=97=B6=E9=97=B4=E6=94=B6=E7=B4=A7=EF=BC=9B?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E4=B8=8D=E8=A2=AB=E4=B8=AD=E6=AD=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为什么：盲算最多 3 路并行、每项 120 秒再加 30 秒收尾宽限，指标多了一样能耗光一段执行的时长上限，整段盲算没进 checkpoint，「继续运行」全部重做、再花一次钱。 - 只动安排盲算的 _blind：开始一项之前问 deadline.allot。不够时交回一个没有模型的结果（原因「本次运行即将达到 600 秒的时长上限，为保证运行按时完成，未进行盲算」），走现有的路径：_compare 记无法复算、dual_check_unchecked 点名、出具缺口照记；第一项就不够时和核查模型用不了一样，后面的不再一个个去试 - 时长收紧过的盲算到了点：照样按无法复算记（同触了每项时长上限），原因说运行时长上限将至、盲算时长已按剩余时间收紧；不留 limit，不发 dual_check_limit——它叫人调高每项的上限，调高了只会更快撞线 - 决定都在 dual_check_blind 这个 @task 里：审批恢复后重放取 checkpoint 里的结果，不再盲算，哪几项没来得及照旧 - 时间充裕时时长上限照传原值，和以前一样；_plan_metric 等别处没动 测试：假的钟钉住五个指标只够盲算 2 项、第一项都来不及、收紧的那项到点怎么说、审批恢复后不重做；真的钟用 5 秒的时长上限、一项接一项的慢盲算，以前被中止，现在运行成功。去掉判断后这几条都变红。 --- backend/app/engine/nodes/dual_check.py | 32 ++++- backend/tests/test_verify_deadline.py | 184 ++++++++++++++++++++++++- 2 files changed, 208 insertions(+), 8 deletions(-) diff --git a/backend/app/engine/nodes/dual_check.py b/backend/app/engine/nodes/dual_check.py index 94a32f0f..02b95481 100644 --- a/backend/app/engine/nodes/dual_check.py +++ b/backend/app/engine/nodes/dual_check.py @@ -23,7 +23,8 @@ SQL，就跟着差异说明进了审批卡、对照工件和目录日志。完 选了会怎样要在卡上看得出：payload 在约定的几项之外带上实际生效的 on_reject，摘要里每个指标带 catalog——选 a、选 b 按发起审批时的目录各会往目录里写什么、不写的为什么（dual_check_facts.preview，和真写同一套判断）；几个指标落到目录 同一项上时每种选择组合合起来会怎样写（preview_joint），放在对照工件的 catalog_joint 里。 -盲算触顶、出错、交不出数，或者输入本身没法单独复算的指标按「无法复算」记缺口，发 dual_check_unchecked。 +盲算触顶、出错、交不出数，或者输入本身没法单独复算的指标按「无法复算」记缺口，发 dual_check_unchecked。运行这一段的 +时长上限将至时不再开始新的盲算、开始的按剩余时间收紧时长（engine/deadline.py），没来得及的照此记，运行不被中止。 **产出的状态。** 全部一致是 agreed；没有不一致、但有无法复算的：一个都没核成是 unchecked，核成了一部分是 partial； 有不一致、人裁决过的是 resolved。产出和台账都带 counts（一致、不一致、无法复算各几个）。 @@ -55,7 +56,7 @@ from app.core.errors import describe_exception from app.core.events import EventType from app.db.base import SessionLocal from app.db.models import DataSource -from app.engine import recompute +from app.engine import deadline, recompute from app.engine.approval import read_decision from app.engine.context import NodeContext, NodeError from app.engine.dual_check_facts import preview, preview_joint, record_choices @@ -117,6 +118,9 @@ GAP_BLIND = "指标「{name}」两路结果不一致，审核人选择了盲算 GAP_NEITHER = "指标「{name}」两路结果不一致，审核人判断两边都不对" GAP_UNANSWERED = "指标「{name}」两路结果不一致，审核人没有作出选择" GAP_UNCHECKED = "指标「{name}」无法复算：{reason}" +#: 运行时长上限将至（engine/deadline.py）：没来得及开始的盲算、时长按剩余时间收紧后到了点的盲算，都按无法复算记 +LATE = "本次运行即将达到 {seconds:g} 秒的时长上限，为保证运行按时完成，未进行盲算" +SQUEEZED = "本次运行即将达到 {seconds:g} 秒的时长上限，盲算时长已按剩余时间收紧，到时未能完成" BLIND_SYSTEM = ( "你在做双盲复算：另一条流程已经算出了口径卡上的一个指标，你要在看不到它的 SQL 和结果的情况下，按业务定义独立算出" @@ -706,10 +710,12 @@ async def _blind(ctx: NodeContext, jobs: dict[str, dict[str, Any]], options: _Op 先单独做第一个：核查模型本身用不了（没配好、接入出错）时，后面的一样用不了，照这一次的原因记，不再一个个去试。 其余的同时进行，最多 BLIND_CONCURRENCY 个：串行时几个慢的输入（每个最多 timeout_s 加收尾的宽限）就能把运行这一段 - 的时长上限耗光，整段盲算没有 checkpoint，接着运行时全部重做、再花一次钱。结果按格子写回，日志按任务的先后发， - 和完成的先后无关。""" + 的时长上限耗光，整段盲算没有 checkpoint，接着运行时全部重做、再花一次钱。指标多了并行也一样耗得光，所以每项开始 + 之前还要看这一段还剩多少时间（engine/deadline.py）：不够就不再开始、按无法复算记，够的话时长上限按剩余时间收紧。 + 结果按格子写回，日志按任务的先后发，和完成的先后无关。""" from app.api.settings import verify_model_spec - from app.engine.verify_agent import run_verify_agent + from app.engine import verify_agent + from app.engine.verify_agent import VerifyOutcome, run_verify_agent from app.tools.datasource import run_versions usage = {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0, "cost_usd": 0.0, "calls": 0} @@ -723,11 +729,21 @@ async def _blind(ctx: NodeContext, jobs: dict[str, dict[str, Any]], options: _Op if job["source"] not in catalogs: catalogs[job["source"]] = await _catalog_text(sources, job["source"]) outcomes: dict[str, Any] = {} + squeezed: dict[str, float] = {} # 时长按剩余时间收紧过的盲算 → 这一段执行的总时长（到了点时说明里写） async def one(key: str, job: dict[str, Any]) -> None: + # 给下游留出余量后不够再做一项：不再开始，交回一个没有模型的结果，按无法复算记——第一项就不够时，和核查模型 + # 用不了一样，后面的不再一个个去试。在 @task 里定，重放时不再重新定 + allot = deadline.allot(options.timeout_s, grace=verify_agent.SETTLE_GRACE_S) + if not allot.start: + outcomes[key] = VerifyOutcome(result=None, cells=[], queries=[], rationale=None, usage={}, + error=LATE.format(seconds=allot.seconds)) + return + if allot.tightened: + squeezed[key] = allot.seconds outcomes[key] = await run_verify_agent( spec=spec, system=BLIND_SYSTEM, task=_task_text(job, catalogs[job["source"]]), sources=[job["source"]], - output_schema=BLIND_SCHEMA, max_queries=options.max_queries, timeout_s=options.timeout_s, + output_schema=BLIND_SCHEMA, max_queries=options.max_queries, timeout_s=allot.timeout_s, daily_max_usd=options.daily_max_usd, emit=ctx.emit, run_id=ctx.run.run_id, node_id=ctx.node.id, data_versions=run_versions(ctx.run)) @@ -751,6 +767,10 @@ async def _blind(ctx: NodeContext, jobs: dict[str, dict[str, Any]], options: _Op results[key] = dict(results[keys[0]]) continue results[key] = _slim(outcome) + if key in squeezed and results[key]["limit"] == "timeout_s": + # 收紧过的时长到了点：照样按无法复算记（同触了每项时长上限），原因说运行时长上限将至。不留 limit，免得发 + # dual_check_limit——它叫人调高每项的上限，调高了只会更快撞线 + results[key].update(value=None, limit=None, reason=SQUEEZED.format(seconds=squeezed[key])) for k in usage: usage[k] += outcome.usage.get(k) or 0 if outcome.model and model is None: diff --git a/backend/tests/test_verify_deadline.py b/backend/tests/test_verify_deadline.py index 5084aa6b..cfa205f3 100644 --- a/backend/tests/test_verify_deadline.py +++ b/backend/tests/test_verify_deadline.py @@ -1,11 +1,13 @@ -"""核查节点在运行时长上限内收尾（engine/deadline.py）：反方审稿逐句开始之前，看这一段执行还剩多少时间。 +"""核查节点在运行时长上限内收尾（engine/deadline.py）：反方审稿逐句、双盲复算逐项开始之前，看这一段执行还剩多少时间。 - 反方审稿：5 句、截止时刻只够审 2 句 → 审 2 句，其余 3 句记为未能反查（理由、时间线写明运行时长上限将至），运行 成功、不被中止，日志和产出的 unreviewed 对得上；剩下的时间不够整段时这句的时长收紧，到点按查到的结果收尾，说的是 运行时长上限将至，不说「每句时长上限 90 秒」；接着运行时取 checkpoint 里的审稿，不重审 +- 双盲复算：同理，没来得及开始的盲算按无法复算记（照现有的路径记缺口）；收紧过的时长到点也按无法复算记，不发提醒 + 调高上限的 dual_check_limit；审批恢复后重放，盲算不再执行、哪几项没来得及照旧 - 时间充裕时（默认 600 秒的一段）时长上限照传原值：和没有这道检查时一样 -两种钟：精确的计数用假的钟（每派出一句就把 allot() 看到的钟往前拨若干秒，真实只过去几毫秒，runner 的 asyncio.timeout +两种钟：精确的计数用假的钟（每派出一项就把 allot() 看到的钟往前拨若干秒，真实只过去几毫秒，runner 的 asyncio.timeout 远远不到）；「运行不被中止」用真的钟、真的短时长上限和真的慢 agent，只断言不随机器快慢变的事。模型都是 MockChatModel 剧本，数据源是现造的 SQLite 文件，区域、金额都是虚构的。 """ @@ -14,11 +16,22 @@ from __future__ import annotations import asyncio from collections.abc import Callable +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import select + from app.core import artifact_store from app.core.config import settings +from app.db.base import SessionLocal +from app.db.models import Approval, Run from app.engine import deadline, verify_agent from app.engine.nodes import challenge as challenge_mod +from app.engine.nodes import dual_check as dual_check_mod from app.engine.runner import run_manager, verify_manifest +from app.main import app +from app.providers.mock_model import MockChatModel +from tests.fixtures.dual_check_shop import WEEK, Blind, add_source, build +from tests.fixtures.sources import drop_source from tests.test_challenge_node import ( # noqa: F401 - engine_up、shop 是夹具，按名字引入 NET, REFUNDS, @@ -203,3 +216,170 @@ async def test_a_slow_review_no_longer_aborts_the_run(monkeypatch, shop): [warn] = await logs(row.id, "challenge_unreviewed") assert warn["claims_skipped"] == len(late) and f"剩下 {len(late)} 句未审" in warn["reason"] assert "本次运行即将达到 7 秒的时长上限" in warn["reason"] + + +# -------------------------------------------------------------------------- +# 双盲复算 +# -------------------------------------------------------------------------- + +#: 一条查询算出五个指标（按支付日、排除退款）：销售额 330.0、订单数 3、付款客户数 2、最大单笔 200.0、最小单笔 30.0 +FIVE_SQL = (f"SELECT SUM(amount) AS gmv, COUNT(*) AS orders, COUNT(DISTINCT customer_id) AS buyers, " + f"MAX(amount) AS top, MIN(amount) AS low FROM orders WHERE status <> 'refunded' AND paid_date {WEEK}") +NAMES = {"gmv": "销售额", "orders": "订单数", "buyers": "付款客户数", "top": "最大单笔", "low": "最小单笔"} +AGGS = {"gmv": "SUM(amount)", "orders": "COUNT(*)", "buyers": "COUNT(DISTINCT customer_id)", "top": "MAX(amount)", + "low": "MIN(amount)"} +METRICS = [{"id": mid, "name": name, "expression": f"cell(nodes.fetch, 0, '{mid}')"} for mid, name in NAMES.items()] +#: 盲算各自写一条只算这一个数的 SQL：和原流程的结果一致 +PLANS = {name: (f"SELECT {AGGS[mid]} AS blind_v FROM orders WHERE status <> 'refunded' AND paid_date {WEEK}", "blind_v") + for mid, name in NAMES.items()} +#: 没来得及开始的盲算（默认 600 秒的一段）：指标的原因、出具缺口 +LATE_BLIND = "本次运行即将达到 600 秒的时长上限，为保证运行按时完成，未进行盲算" + + +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 dual_graph(source: str, **dual) -> dict: + """取数 → 口径卡（五个指标）→ 双盲复算 → 成果。成果的叙述里没有数字：没有缺口时是完整出具。""" + nodes = [ + node("start", "input"), + node("fetch", "tool", tool=f"db_query__{source}", args={"sql": FIVE_SQL}), + node("card", "metrics", caliber="销售口径", caliber_version="v1", metrics=METRICS), + node("dual", "dual_check", "复核五个指标", **dual), + node("out", "output", fields=[{"name": "结论", "value": "本周指标见口径卡"}], + contract={"metrics_from": "card", "narrative": "本周指标见口径卡"}), + ] + return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +@pytest.fixture +async def orders(tmp_path): + name, source_id = await add_source(build(tmp_path / "orders.db")) + yield name + await drop_source(source_id) + + +async def dual_run(spec: dict, **kwargs) -> Run: + run = await run_manager.start(graph=spec, input_payload={}, run_class="formal", **kwargs) + return await wait(run.id, ("succeeded", "failed", "interrupted")) + + +def unchecked_reasons(out: dict) -> dict[str, str]: + return {m["metric_id"]: m["reason"] for m in out["metrics"] if m["status"] == "unchecked"} + + +async def test_five_metrics_with_time_for_two_check_two(monkeypatch, orders): + """截止时刻只够盲算 2 项（每项像花了 260 秒）：前 2 项照常盲算、一致，后 3 项不再开始，按无法复算记缺口；运行成功、 + 出具降档。时间线上 dual_check_unchecked 点名这 3 项和原因，不发提醒调高上限的 dual_check_limit。""" + blind = Blind(monkeypatch, orders, PLANS) + ticker = Ticker(monkeypatch, verify_agent, step=260) + row = await dual_run(dual_graph(orders)) + assert row.status == "succeeded", row.error + assert blind.runs == 2 and ticker.timeouts == [120, 120], "时间充裕的两项照传每项时长上限，不收紧" + + out = await output_of(row.id, "dual") + assert out["status"] == "partial" and out["counts"] == {"agreed": 2, "disagreed": 0, "unchecked": 3} + assert [m["status"] for m in out["metrics"]] == ["agreed", "agreed", "unchecked", "unchecked", "unchecked"] + assert unchecked_reasons(out) == {"buyers": LATE_BLIND, "top": LATE_BLIND, "low": LATE_BLIND} + gaps = [f"指标「{NAMES[m]}」无法复算：{LATE_BLIND}" for m in ("buyers", "top", "low")] + assert out["issuance_gaps"] == gaps + assert row.output["_issuance"]["tier"] == "degraded" + assert row.output["_issuance"]["gaps"] == [f"双盲复算「复核五个指标」：{gap}" for gap in gaps] + + [unchecked] = await logs(row.id, "dual_check_unchecked") + assert unchecked["metrics"] == [{"id": m, "name": NAMES[m], "reason": LATE_BLIND} for m in ("buyers", "top", "low")] + assert await logs(row.id, "dual_check_limit") == [] + assert (await verify_manifest(row.id))["ok"] is True + + +async def test_no_time_for_even_the_first_blind_check(monkeypatch, orders): + """一开始就只剩 40 秒：第一项都不开始（后面的也就不再一个个去试），五项都按无法复算记，节点状态是一个都没核成。""" + blind = Blind(monkeypatch, orders, PLANS) + ticker = Ticker(monkeypatch, verify_agent, step=0, start=560) + row = await dual_run(dual_graph(orders)) + assert row.status == "succeeded", row.error + assert blind.calls == [] and ticker.timeouts == [] + out = await output_of(row.id, "dual") + assert out["status"] == "unchecked" and out["counts"] == {"agreed": 0, "disagreed": 0, "unchecked": 5} + assert set(unchecked_reasons(out).values()) == {LATE_BLIND} + assert len(out["issuance_gaps"]) == 5 + + +async def test_a_blind_check_squeezed_by_the_deadline_says_so(monkeypatch, orders): + """第 2 项开始前只剩约 150 秒：时长收紧到 60 秒以内；它到点时按无法复算记，原因说运行时长上限将至、时长已收紧， + 不说「已达每项时长上限（120 秒）」，也不发叫人调大上限的 dual_check_limit。""" + blind = Blind(monkeypatch, orders, PLANS) + ticker = Ticker(monkeypatch, verify_agent, step=450, hit=lambda timeout: timeout < 120) + row = await dual_run(dual_graph(orders)) + assert row.status == "succeeded", row.error + assert blind.runs == 2 + first, second = ticker.timeouts + assert first == 120 and 50 < second < 60, ticker.timeouts + out = await output_of(row.id, "dual") + reasons = unchecked_reasons(out) + assert set(reasons) == {"orders", "buyers", "top", "low"} + assert "本次运行即将达到 600 秒的时长上限" in reasons["orders"] and "收紧" in reasons["orders"] + assert "每项时长上限" not in reasons["orders"] + assert {reasons[m] for m in ("buyers", "top", "low")} == {LATE_BLIND} + assert await logs(row.id, "dual_check_limit") == [] + + +async def test_resuming_after_the_review_replays_the_blind_checks(monkeypatch, orders): + """销售额两路不一致、挂起等审核人裁决；后 3 项没来得及。审批恢复后节点从头重放：盲算取 checkpoint 里的，不再执行， + 没来得及的 3 项照旧——不看恢复后这一段还剩多少时间。""" + plans = dict(PLANS) + plans["销售额"] = (f"SELECT SUM(amount) AS blind_v FROM orders WHERE paid_date {WEEK}", "blind_v") # 没排除退款：380.0 + blind = Blind(monkeypatch, orders, plans) + ticker = Ticker(monkeypatch, verify_agent, step=260) + row = await dual_run(dual_graph(orders)) + assert row.status == "interrupted", row.error + calls, timeouts = list(blind.calls), list(ticker.timeouts) + assert blind.runs == 2 + + ticker.offset = 0 # 恢复后的一段：时间又充裕了，重做的话五项都会盲算 + async with SessionLocal() as session: + approval = (await session.execute(select(Approval).where( + Approval.run_id == row.id, Approval.status == "pending"))).scalar_one() + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + res = await client.post(f"/api/approvals/{approval.id}/decide", + json={"approved": True, "value": {"choices": {"gmv": "a"}, "note": ""}}) + assert res.status_code == 200, res.text + row = await wait(row.id) + assert row.status == "succeeded", row.error + assert blind.calls == calls and ticker.timeouts == timeouts, "恢复后盲算又执行了：没取 checkpoint" + out = await output_of(row.id, "dual") + assert out["status"] == "resolved" and out["choices"] == {"gmv": "a"} + assert [m["status"] for m in out["metrics"]] == ["disagreed", "agreed", "unchecked", "unchecked", "unchecked"] + assert set(unchecked_reasons(out).values()) == {LATE_BLIND} + assert len(await logs(row.id, "dual_check_unchecked")) == 1, "盲算那一步的日志只发一次" + + +async def test_slow_blind_checks_no_longer_abort_the_run(monkeypatch, orders): + """真的钟：这一段只有 5 秒，五项盲算一项接一项（并行数 1），每项约 1 秒（模型每次调用 0.3 秒，每项 3 次），合起来 + 超过 5 秒——以前整次运行在盲算到一半时被中止。现在时间不够就不再开始、开始的按剩余时间收紧，没来得及的按无法复算 + 记，运行成功。每项时长上限 3 秒，一项自己用不完：无法复算的只会是因为运行时长上限将至。几项来得及随机器快慢变， + 只断言不变的。""" + monkeypatch.setattr(settings, "max_run_seconds", 5) + monkeypatch.setattr(deadline, "RESERVE_S", 1.2) + monkeypatch.setattr(verify_agent, "SETTLE_GRACE_S", 0.4) + monkeypatch.setattr(dual_check_mod, "BLIND_CONCURRENCY", 1) + + async def slow_model(session, spec): + return MockChatModel(model_name="slow-verify", latency=0.3), "slow-verify" + + monkeypatch.setattr(verify_agent, "get_chat_model", slow_model) + Blind(monkeypatch, orders, PLANS) + row = await dual_run(dual_graph(orders), options={"verify": {"dual_check": {"timeout_s": 3}}}) + assert row.status == "succeeded", row.error + assert "时长上限，已中止" not in (row.error or "") + + out = await output_of(row.id, "dual") + reasons = unchecked_reasons(out) + assert out["counts"]["agreed"] >= 1 and reasons, out["metrics"] + assert out["counts"]["agreed"] + len(reasons) == 5 and out["counts"]["disagreed"] == 0 + assert all("本次运行即将达到 5 秒的时长上限" in why for why in reasons.values()), reasons + assert [m["status"] for m in out["metrics"]][:out["counts"]["agreed"]] == ["agreed"] * out["counts"]["agreed"], \ + "一项接一项：来得及的在前" + [unchecked] = await logs(row.id, "dual_check_unchecked") + assert {m["id"] for m in unchecked["metrics"]} == set(reasons)