From a552f4feab2e582f65027f482881ee0451ee2d38 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Tue, 6 Oct 2026 12:58:37 +0800 Subject: [PATCH] =?UTF-8?q?=E6=A0=B8=E6=9F=A5=E6=A8=A1=E5=9E=8B=E7=9A=84?= =?UTF-8?q?=E8=BE=93=E5=87=BA=E9=A2=9D=E5=BA=A6=E6=94=B9=E4=B8=BA=2016384?= =?UTF-8?q?=E3=80=81=E6=9C=8D=E5=8A=A1=E6=96=B9=E5=AB=8C=E5=A4=A7=E6=97=B6?= =?UTF-8?q?=E9=80=80=E5=9B=9E=204096=EF=BC=8C=E6=88=AA=E6=96=AD=E7=85=A7?= =?UTF-8?q?=E5=AE=9E=E8=AF=B4=EF=BC=9A=E5=B7=A5=E5=85=B7=E5=BE=AA=E7=8E=AF?= =?UTF-8?q?=E3=80=81=E6=94=B6=E5=B0=BE=E7=9A=84=E6=AF=8F=E6=AC=A1=E8=B0=83?= =?UTF-8?q?=E7=94=A8=E9=83=BD=E7=9C=8B=E6=9C=89=E6=B2=A1=E6=9C=89=E5=88=B0?= =?UTF-8?q?=E4=BA=86=E4=B8=8A=E9=99=90=E8=A2=AB=E6=88=AA=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 额度：judge.MAX_TOKENS 从 4096 改为 16384（核查的 verify_model_spec 沿用它）。经 OpenAI 兼容接入的推理模型收不到 thinking=off，思考算在额度里，4096 常常想完就没额度发查询、交结果；16384 是常见网关接受的输出上限里偏大的一档，cite_fields 的抽取也是这个数。服务方嫌大（400/422 且报错提到 max_tokens 一类）时退回 judge.FALLBACK_MAX_TOKENS（4096）再试一次，工具循环和收尾共用一个开关，之后都用小的；小额度的模型要退回时才建。 截断：工具循环里没有工具调用、finish_reason=length（或 stop_reason=max_tokens）不再当成「模型说完了」，一条查询都没成功时不再收尾（收尾只能交「查不到」，双盲复算就会记成「盲算没有查到这个数」）；查到过东西的照常收尾，取不到要的数时同样交回截断的原因，取得到的照常交、notes 里注明。收尾被截断（Anthropic 强制工具调用写到一半、openai 结构化输出抛 LengthFinishReasonError、普通调用只有思考）时交回截断的原因，不再退回普通调用多花一次额度；截断那次的用量从异常里取，照记账。给人看的原因写明「核查模型的输出达到上限（N token）被截断，核查未能完成。通常是推理占满了输出额度，可在「设置 → 偏好设置 → 核查」中换用不带推理的模型」，经 error 交给双盲复算的无法复算原因、反方审稿的「反查未完成」、质询的报错；节点里调用时另记一条 verify_truncated，llm.end 标出 truncated 和额度。没有截断时原有说法不变。 测试：新增 test_verify_truncation（假模型返回 finish_reason=length、只有思考、LengthFinishReasonError，覆盖工具循环、收尾结构化和普通调用、退回 4096、别的 400 不重试、三条渠道）；三处钉着 4096 的断言改成 16384。 --- backend/app/api/settings.py | 4 +- backend/app/engine/judge.py | 14 +- backend/app/engine/verify_agent.py | 110 ++++++-- backend/tests/test_evidence_judge.py | 4 +- backend/tests/test_evidence_judge_api.py | 2 +- backend/tests/test_judge_settings.py | 3 +- backend/tests/test_verify_truncation.py | 334 +++++++++++++++++++++++ 7 files changed, 447 insertions(+), 24 deletions(-) create mode 100644 backend/tests/test_verify_truncation.py diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index 566fae74..5c98ba86 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -549,7 +549,9 @@ async def verify_model_spec(session: AsyncSession, node_config: dict[str, Any] | """核查用哪个模型：节点配置里的 provider / model > 设置里的核查模型 > 证据裁判模型的取值链（裁判设置 > 助手模型 > 默认接入，见 judge.judge_model_spec）。哪一级写了就整组用哪一级，不跨级拼接入和模型。 - 和裁判一样关掉 thinking、max_tokens 4096：核查最后要按结构交结果（强制工具调用），和思考模式不兼容。 + 和裁判一样关掉 thinking：核查最后要按结构交结果（强制工具调用），和思考模式不兼容。额度也和裁判一样（judge.MAX_TOKENS， + 16384）：经 OpenAI 兼容接入的推理模型收不到 thinking=off，思考算在额度里，4096 常常想完就没额度查库、交结果了； + 服务方嫌大时核查 agent 退回 judge.FALLBACK_MAX_TOKENS（engine/verify_agent.py）。 建议和被核查的模型不同，否则等于自己核自己；比较、提醒由调用方做（它知道被核查的是哪个模型）。 """ node = node_config if isinstance(node_config, dict) else {} diff --git a/backend/app/engine/judge.py b/backend/app/engine/judge.py index 1dc0e67d..27ba26c0 100644 --- a/backend/app/engine/judge.py +++ b/backend/app/engine/judge.py @@ -12,7 +12,8 @@ （按需裁判传只认封存工件的那个）。规则里写明数字已经由系统核对过。带期 4 文档标记的报告，上传表格的 查询在 SQL 之后另有来历行（来源、口径、已接受、区域外文字，_provenance_lines） 3. 调用（judge_doc）：一份报告合并成一次调用，超过每批的句数就分批；结构化输出失败退回 JSON 解析， - 再失败就记未裁判。thinking 关掉，max_tokens 4096 + 再失败就记未裁判。thinking 关掉，输出额度 MAX_TOKENS（推理模型的思考也算在里面），服务方嫌大时退回 + FALLBACK_MAX_TOKENS；输出到了上限被截断的记未裁判（truncated），照实说是额度不够、不再退回纯文本多花一次 4. 预算：每份报告的句数、金额、时长和全局每日金额，每一项都能是 None（不限）。调用前按目录价估算， 超了按优先级截断；触顶就停，已判的保留，没判的记未裁判并说明是哪个上限——不报错 5. 每日计数落在设置表（SPEND_KEY），按本地日期滚动。目录里没有价格的模型估不出金额，金额上限对它 @@ -104,7 +105,13 @@ BATCH_SIZE = 8 #: 一波最多几批同时问：几批共用剩下的时长，句数上限内的句子一两波就判完。第一批单独问：接入调不通（鉴权、 #: 网关挂了）时只撞这一次，不让一整波一起失败 PARALLEL_BATCHES = 4 -MAX_TOKENS = 4096 +#: 裁判（和核查：api/settings.py 的 verify_model_spec、engine/verify_agent.py 沿用这一档）的输出额度。thinking 关掉只对 +#: Anthropic 有用：经 OpenAI 兼容接入的推理模型（DeepSeek、GLM、Kimi 这一类）收不到 thinking=off，思考算在额度里，原来的 +#: 4096 常常想完就没额度写判定了（cite_fields 的抽取在 2026-10 撞过同一堵墙，那边也改成了 16384）。16384 是常见网关、 +#: 模型接受的输出上限里偏大的一档（GPT-4o 的上限就是它）；更小的服务方会拒收，那时退回 FALLBACK_MAX_TOKENS +MAX_TOKENS = 16384 +#: 服务方嫌 MAX_TOKENS 太大（400 / 422，报错里有 max_tokens 一类字样）时退回这个：原来的额度，各家都收。之后的调用都用它 +FALLBACK_MAX_TOKENS = 4096 #: 证据摘录：SQL 前几个字、每次查询最多几行几列、每段知识库片段最多几个字、整份引用的检索给几段。 #: 列数上限只管没被引用的列：句子引用的格子所在的列一律给（遮罩的除外），再用别的列补满到这么多列 SQL_CHARS = 300 @@ -897,7 +904,8 @@ def _text(value: Any) -> str | None: async def judge_model_spec(session: Any, node_judge: dict[str, Any] | None = None, *, settings: dict[str, Any] | None = None) -> ModelSpec: """裁判用哪个模型：节点上的 judge.model（连同 provider）> 设置里的证据裁判模型 > Copilot 模型 > - 默认 provider。哪一级写了就整组用哪一级，不跨级拼 provider 和模型。thinking 关、max_tokens 4096。 + 默认 provider。哪一级写了就整组用哪一级，不跨级拼 provider 和模型。thinking 关、额度 MAX_TOKENS（推理模型关不掉 + 思考，见 MAX_TOKENS 的说明；服务方嫌大时 run_request 退回 FALLBACK_MAX_TOKENS）。 建议和写作模型不同，否则等于自己审自己（报告节点发现相同会发 judge_same_model 警告）。 """ diff --git a/backend/app/engine/verify_agent.py b/backend/app/engine/verify_agent.py index ad0a39d2..ad5b59a7 100644 --- a/backend/app/engine/verify_agent.py +++ b/backend/app/engine/verify_agent.py @@ -25,6 +25,12 @@ query_artifact 和 role: verify——随运行封存、证据面板能下钻。 每日上限不传就按设置里的「核查每日金额上限」，明确传 None 才是不限。模型不在价格目录里时估不出金额，金额上限 对它无效，照裁判的说法记一条说明。每次调用都记进核查的每日账（engine/spend.py）。 +**输出额度。** 经 OpenAI 兼容接入的推理模型收不到 thinking=off，思考也算在输出额度里：额度按 spec（缺省 judge.MAX_TOKENS）， +服务方嫌大时退回 judge.FALLBACK_MAX_TOKENS 再试一次，之后工具循环、收尾都用小的（providers/budget.py）。每次调用都看 +有没有到了上限被截断：工具循环里被截断（话没说完、也没发查询）不算模型自己停下，一条查询都没成功时不再收尾；收尾被截断 +时不再退回普通调用多花一次额度。因为截断没交成结果的，error 照实说是输出额度（TRUNCATED_ERROR），不说成「没查到」 +「没按结构交」；节点里调用时另记一条 verify_truncated。 + 永远不抛：模型没配好、调用失败、格式不对、触顶、库写不进去，都落在返回值里（error / limit）。模型调用都经 get_chat_model，测试换成剧本。 """ @@ -49,11 +55,12 @@ from app.db.base import SessionLocal from app.engine import spend from app.engine.evidence import build_catalog, cell_eid, compose_doc, query_entry_fields, schema_ledger_entry from app.engine.expressions import CellError, column_kind, locate_cell, numeric_text -from app.engine.judge import _json_in, _tokens, _usd +from app.engine.judge import FALLBACK_MAX_TOKENS, MAX_TOKENS, _json_in, _tokens, _usd from app.engine.nodes.llm import _DIGEST_BUDGET, _digest, _split_structured, _typed, explain_model_error from app.engine.state import message_text from app.engine.toolcalls import ToolTimeout, limit_fields, limit_of, run_bounded from app.providers import catalog as pricing +from app.providers.budget import BudgetFallback, cut_off, cut_words, error_usage, truncated_error from app.providers.factory import ModelSpec, ProviderNotConfigured, bind_tools_safely, get_chat_model from app.tools.datasource import QUERY_PREFIX, SCHEMA_PREFIX, build_datasource_tools from app.tools.registry import ToolArgsError, ToolContext, Versions, args_model_of, prepare_args @@ -79,6 +86,12 @@ _REF_DEPTH = 8 RATIONALE = "rationale" #: 触顶原因 LIMITS = ("max_queries", "timeout_s", "max_cost_usd", "daily_max_usd", "max_steps") +#: 核查模型的输出到了上限被截断时给人看的话。TRUNCATED_ERROR 是没交成结果的原因：双盲复算的无法复算、反方审稿的「反查 +#: 未完成」、质询的报错都照它说，也是 verify_truncated 日志的原话。TRUNCATED_SETTLED 是截断前查到过东西、照常收尾了的， +#: 记进 notes 和日志。cut 是 budget.cut_words 的「核查模型的输出达到上限（N token）被截断」 +TRUNCATED_ERROR = "{cut}，核查未能完成。{advice}" +TRUNCATED_SETTLED = "{cut}，已按截断前查到的结果收尾，结果可能不完整。{advice}" +TRUNCATED_ADVICE = "通常是推理占满了输出额度，可在「设置 → 偏好设置 → 核查」中换用不带推理的模型" VERIFY_SYSTEM = ( "你是数据核查员：只读地查库，用查询结果完成交给你的任务。\n" @@ -129,7 +142,8 @@ class VerifyOutcome: source, tool, call_id}。 rationale：理由组装成的报告文档（evidence.compose_doc 的形状，violations 里有裸数字）。usage：{input_tokens, output_tokens, total_tokens, cost_usd, calls, unpriced_calls}。limit：触顶原因（LIMITS 之一），没触顶为 None。 - answered：模型是不是自己停下来的（没被上限掐断）。 + answered：模型是不是自己停下来的（没被上限掐断，也没因为输出到了上限被截断）。因为截断没交成结果的，error 是 + TRUNCATED_ERROR；截断前查到过东西、照常交了结果的，error 为 None，notes 里记一句 TRUNCATED_SETTLED。 """ result: dict[str, Any] | None @@ -686,13 +700,16 @@ async def _verify(state: _State, *, spec: ModelSpec, system: str, task: str, sou def left() -> float | None: return None if deadline is None else deadline - time.monotonic() - # ---- 模型 + # ---- 模型。额度没写的按 MAX_TOKENS：推理模型的思考算在输出额度里，4096 常常想完就没额度发查询、交结果了 + spec = spec.model_copy(update={"max_tokens": spec.max_tokens or MAX_TOKENS}) + small = FALLBACK_MAX_TOKENS if spec.max_tokens > FALLBACK_MAX_TOKENS else None try: async with SessionLocal() as session: - base_model, model_id = await get_chat_model(session, spec) + first, model_id = await get_chat_model(session, spec) except Exception as e: # noqa: BLE001 - 核查模型没配好：这回核不了，照实交回 why = not_configured(e) if isinstance(e, ProviderNotConfigured) else describe_exception(e) return outcome(error=f"核查模型不可用（{why}）") + base_model = _with_fallback(first, spec, small) state.model = model_id if daily_max_usd is Default.SETTINGS: from app.api.settings import verify_settings # 在这里导入：api.settings 导入了 engine 里的模块 @@ -743,7 +760,8 @@ async def _verify(state: _State, *, spec: ModelSpec, system: str, task: str, sou raw = got if isinstance(got, BaseMessage) else _split_structured(got)[0] except (asyncio.TimeoutError, Exception) as e: # noqa: BLE001 - 调用失败落在返回值里 failure = e - inp, out = _usage(raw) + # openai 的结构化输出被截断时抛异常，用量在异常里：额度是真花出去了 + inp, out = _usage(raw) if raw is not None else error_usage(failure) cost = await budget.settle(day, estimate, inp, out) usage["input_tokens"] += inp usage["output_tokens"] += out @@ -756,6 +774,10 @@ async def _verify(state: _State, *, spec: ModelSpec, system: str, task: str, sou if failure is not None: end["error"] = "超过核查的时长上限，已放弃这次调用" if isinstance(failure, asyncio.TimeoutError) \ else explain_model_error(failure, model_id) + if truncated_error(failure) or cut_off(raw): + # 到了上限被截断：不是「调用失败，请稍后重试」，重试解决不了额度问题 + limit = base_model.max_tokens + end.update(truncated=True, max_tokens=limit, error=cut_words("核查模型", limit)) await _announce(emit, EventType.LLM_END, **end) return got, raw, failure @@ -766,6 +788,7 @@ async def _verify(state: _State, *, spec: ModelSpec, system: str, task: str, sou said = "" attempts = 0 done: dict[str, str] = {} # 成功过的查询（工具名 + 参数）→ 编号：一模一样的不再执行 + cut: tuple[str, int | None] | None = None # 最近一次因为输出到了上限被截断的调用：(哪一步, 那次的额度) for _ in range(steps): remaining = left() if remaining is not None and remaining <= 0: @@ -781,11 +804,18 @@ async def _verify(state: _State, *, spec: ModelSpec, system: str, task: str, sou if isinstance(failure, asyncio.TimeoutError): stopped = "timeout_s" break + if truncated_error(failure): + cut = ("verify", base_model.max_tokens) + break if failure is not None or not isinstance(response, BaseMessage): return outcome(error=explain_model_error(failure, model_id) if failure else "核查模型没有给出回复") messages.append(response) calls = getattr(response, "tool_calls", None) or [] if not calls: + if cut_off(response): + # 话没说完、也没发查询就到了上限（推理模型想完了额度也用完了）：不是模型自己停下来的，不能当成说完了去收尾 + cut, said = ("verify", base_model.max_tokens), message_text(response) or said + break answered, said = True, message_text(response) break said = message_text(response) or said @@ -850,12 +880,35 @@ async def _verify(state: _State, *, spec: ModelSpec, system: str, task: str, sou await _announce(emit, EventType.LOG, level="warn", code="verify_limit", limit=stopped, message=f"核查已达上限（{words}），已停止查询，按已有结果收尾") - # ---- 收尾：按结构交结果。关掉 thinking：结构化输出靠强制工具调用，和思考模式不兼容 + async def report_cut(delivered: bool) -> str: + """被截断过：记一条 verify_truncated，返回给人看的话（没交成结果的原因，或者交了结果时记进 notes 的说明）。""" + purpose, limit = cut or ("verify", None) + text = (TRUNCATED_SETTLED if delivered else TRUNCATED_ERROR).format(cut=cut_words("核查模型", limit), + advice=TRUNCATED_ADVICE) + await _announce(emit, EventType.LOG, level="warn", code="verify_truncated", purpose=purpose, max_tokens=limit, + delivered=delivered, message=text) + return text + + async def give_up(error: str, limit: str | None = None) -> VerifyOutcome: + """没交成结果，原因照 error 说。工具循环里被截断过的，时间线上照记一条：收尾失败可能就是它带出来的。""" + if cut is not None: + await report_cut(False) + return outcome(limit=limit or stopped, answered=answered, error=error) + + if cut is not None and not queries: + # 一条查询都没成功就被截断：收尾只能交「查不到」，双盲复算会记成「盲算没有查到这个数」，原因却是输出额度。 + # 收尾那次多半同样被推理占满，不再为它多花一次额度，直接交回截断的原因 + return outcome(limit=stopped, answered=answered, error=await report_cut(False)) + + # ---- 收尾：按结构交结果。关掉 thinking：结构化输出靠强制工具调用，和思考模式不兼容（经 OpenAI 兼容接入的推理模型 + # 关不掉，所以额度和工具循环一样，退回小额度也和工具循环共用一个开关） + settle_spec = spec.model_copy(update={"thinking": "off"}) try: async with SessionLocal() as session: - settler, _ = await get_chat_model(session, spec.model_copy(update={"thinking": "off"})) + settle_model, _ = await get_chat_model(session, settle_spec) except Exception as e: # noqa: BLE001 - return outcome(limit=stopped, answered=answered, error=f"核查模型不可用（{describe_exception(e)}）") + return await give_up(f"核查模型不可用（{describe_exception(e)}）") + settler = _with_fallback(settle_model, settle_spec, small, shared_with=base_model) final = _final_messages(head, task, queries, said, words) try: structured = settler.with_structured_output(schema, include_raw=True) @@ -870,11 +923,12 @@ async def _verify(state: _State, *, spec: ModelSpec, system: str, task: str, sou parsed: dict[str, Any] | None = None failure: BaseException | None = None + settle_cut = False # 收尾那次到了上限被截断 if structured is not None: estimate = budget.estimate(_text_of(final) + schema_text, _OUT_FINAL) day, refused = await budget.reserve(estimate, kept=kept) if refused: - return outcome(limit=stopped or refused, answered=answered, error=f"已达{words_of(refused)}，没能交结果") + return await give_up(f"已达{words_of(refused)}，没能交结果", refused) got, raw, failure = await ask(final, structured.ainvoke, estimate, day or "", "verify_result", until_finish()) parsed = None if failure is not None else _parsed(_split_structured(got)[1]) if parsed is None and raw is not None: @@ -882,43 +936,67 @@ async def _verify(state: _State, *, spec: ModelSpec, system: str, task: str, sou for candidate in [message_text(raw), *(c.get("args") for c in getattr(raw, "tool_calls", None) or [])]: if (parsed := _parsed(candidate)) is not None: break - if parsed is None and not isinstance(failure, asyncio.TimeoutError): + settle_cut = parsed is None and (truncated_error(failure) or bool(cut_off(raw))) + if parsed is None and not isinstance(failure, asyncio.TimeoutError) and not settle_cut: # 不支持结构化输出的模型：普通调用一次，结构写进消息，从回复里把 JSON 抠出来。结构化那次已经做过的， - # 这一次是补救，照常按上限拦；没做过的，这一次就是收尾，循环里留过钱 + # 这一次是补救，照常按上限拦；没做过的，这一次就是收尾，循环里留过钱。结构化那次是被截断的不补救： + # 推理占满额度时普通调用多半一样被截断，白花一次额度 plain = _plain_messages(final, schema) remaining = until_finish() if remaining is None or remaining > 0: estimate = budget.estimate(_text_of(plain), _OUT_FINAL) day, refused = await budget.reserve(estimate, kept=kept and structured is None) if refused and structured is None: - return outcome(limit=stopped or refused, answered=answered, - error=f"已达{words_of(refused)}，没能交结果") + return await give_up(f"已达{words_of(refused)}，没能交结果", refused) if not refused: got, _, failure = await ask(plain, settler.ainvoke, estimate, day or "", "verify_result", remaining) parsed = None if failure is not None else _parsed(message_text(got) if got is not None else None) + settle_cut = parsed is None and (truncated_error(failure) or bool(cut_off(got))) if parsed is None: + if settle_cut: + cut = ("verify_result", settler.max_tokens) + return outcome(limit=stopped, answered=answered, error=await report_cut(False)) why = "收尾时超过时长上限，没能交结果" if isinstance(failure, asyncio.TimeoutError) else \ f"核查模型没有按结构交结果（{explain_model_error(failure, model_id)}）" if failure is not None else \ "核查模型没有按结构交结果" - return outcome(limit=stopped, answered=answered, error=why) + return await give_up(why) declared = list((resolved.get("properties") or {}) if _is_object(resolved) else {}) if declared and not any(k in parsed for k in declared): # 一个要求的字段都没交：不是「查不到」，是没按结构交。不这么分，字段名对不上的回复会被当成什么都没查到 - return outcome(limit=stopped, answered=answered, - error="核查模型没有按结构交结果（交回的内容里没有要求的字段：" + return await give_up("核查模型没有按结构交结果（交回的内容里没有要求的字段：" + "、".join(declared[:8]) + ("等" if len(declared) > 8 else "") + "）") # ---- 取值、组装理由 cells = _Cells(queries) result = cells.walk(resolved, parsed) if _is_object(resolved) else {} + if cut is not None and any(c["status"] != "resolved" for c in cells.cells): + # 工具循环被截断，截断前查到的里面没有要的数（或者引用指不到）：话没说完就停了，说「没查到」会让人以为数据里 + # 没有。不交这份残缺的结果，原因照截断说；查过的照记 + return outcome(cells=cells.cells, limit=stopped, answered=answered, error=await report_cut(False)) rationale = parsed.get(RATIONALE) doc = None if isinstance(rationale, str) and rationale.strip(): doc = compose_doc(rationale, build_catalog(nodes={}, ledger=ledger), node_id=owner, run_id=run_id or None) result[RATIONALE] = doc["markdown"] if doc else "" + if cut is not None: + # 截断前查到的够用：照常交结果，说明记进 notes，时间线上照记一条（结果可能不完整） + notes.append(await report_cut(True)) return outcome(result=result, cells=cells.cells, rationale=doc, limit=stopped, answered=answered) +def _with_fallback(model: Any, spec: ModelSpec, small: int | None, *, shared_with: BudgetFallback | None = None + ) -> BudgetFallback: + """按 spec 建好的模型包一层：服务方嫌额度大时按 small 重建一个再调，要退回时才建，平常一个都不多建。small 为 None + 是额度本来就不比它大，没有可退的。经本模块的 get_chat_model 建，测试换成剧本时小额度的那个也是剧本。""" + + async def build() -> Any: + async with SessionLocal() as session: + return (await get_chat_model(session, spec.model_copy(update={"max_tokens": small})))[0] + + return BudgetFallback(model, build=build if small else None, limits=(spec.max_tokens, small or spec.max_tokens), + shared_with=shared_with) + + async def _run_tool(tool: Any, name: str, args: dict[str, Any], call_id: str, remaining: float | None, emit: Emit | None, queries: list[dict[str, Any]], ledger: list[dict[str, Any]], owner: str) -> str | None: diff --git a/backend/tests/test_evidence_judge.py b/backend/tests/test_evidence_judge.py index 75a3b995..f46d6a62 100644 --- a/backend/tests/test_evidence_judge.py +++ b/backend/tests/test_evidence_judge.py @@ -6,7 +6,7 @@ - 一份报告合并成一次调用（超过每批的句数就分批），每句给 supported / partial / contradicted / insufficient / not_a_claim - 模型不支持结构化输出、或者给的不是合法结构时退回 JSON 解析，再失败就记「未裁判」，不编判定 -- 裁判模型关掉 thinking，max_tokens 4096 +- 裁判模型关掉 thinking，输出额度 16384（推理模型的思考也算在里面）；服务方嫌大时退回 4096，截断照实说（见 test_judge_truncation） - 预算（每份报告的句数、金额、时长，全局每日金额）每一项都能写 null 表示不限；触顶就停、已判的保留、 没判的记未裁判并说明是哪个上限——不报错、不让报告失败。金额超了按优先级截断：先判挂了依据的， 再判有方向词、因果词的 @@ -207,7 +207,7 @@ async def test_the_judge_model_runs_without_thinking(monkeypatch, stored): max_tokens=64000), budget=Budget(max_claims=None, max_cost_usd=None, timeout_s=30, daily_max_usd=None)) [spec] = judge.specs - assert spec.thinking == "off" and spec.max_tokens == MAX_TOKENS == 4096 + assert spec.thinking == "off" and spec.max_tokens == MAX_TOKENS == 16384 assert spec.model == PRICED and spec.timeout and spec.timeout <= 30 diff --git a/backend/tests/test_evidence_judge_api.py b/backend/tests/test_evidence_judge_api.py index 598d5cad..923aec55 100644 --- a/backend/tests/test_evidence_judge_api.py +++ b/backend/tests/test_evidence_judge_api.py @@ -350,7 +350,7 @@ async def test_the_judge_model_follows_node_then_settings(client, monkeypatch): row = await finish(weekly(claims="judge")) await judge(client, row.id, [unit_ids(doc_of(row))[CAUSAL]]) [spec] = models.specs - assert spec.model == "judge-from-settings" and spec.thinking == "off" and spec.max_tokens == 4096 + assert spec.model == "judge-from-settings" and spec.thinking == "off" and spec.max_tokens == 16384 row = await finish(weekly(claims="judge", judge={"model": "judge-from-node"})) body = (await judge(client, row.id, [unit_ids(doc_of(row))[CAUSAL]])).json() diff --git a/backend/tests/test_judge_settings.py b/backend/tests/test_judge_settings.py index 88fd6de3..80934b2d 100644 --- a/backend/tests/test_judge_settings.py +++ b/backend/tests/test_judge_settings.py @@ -17,6 +17,7 @@ from app.db.base import SessionLocal from app.db.models import Setting from app.engine.judge import ( JUDGE_DEFAULTS, + MAX_TOKENS, SETTING_GROUP, SPEND_KEY, Budget, @@ -166,7 +167,7 @@ async def test_the_judge_model_priority(): async with SessionLocal() as session: spec = await judge_model_spec(session, None) assert (spec.provider, spec.model) == (None, None), "什么都没配：默认 provider" - assert spec.thinking == "off" and spec.max_tokens == 4096 + assert spec.thinking == "off" and spec.max_tokens == MAX_TOKENS == 16384 session.add(Setting(key="copilot", value={"provider": "", "model": "copilot-model"})) await session.commit() diff --git a/backend/tests/test_verify_truncation.py b/backend/tests/test_verify_truncation.py new file mode 100644 index 00000000..26cf8d7e --- /dev/null +++ b/backend/tests/test_verify_truncation.py @@ -0,0 +1,334 @@ +"""核查 agent（engine/verify_agent.py）的输出额度：截断照实说，服务方嫌额度大时退回小额度。 + +核查（双盲复算、反方审稿、审批质询共用）原来固定 max_tokens 4096、thinking off。经 OpenAI 兼容接入的推理模型收不到 +thinking=off，思考就能把 4096 用光：这次调用 finish_reason=length，没有工具调用也没有 JSON。于是工具循环把它当成 +「模型说完了」去收尾，双盲复算记成「盲算没有查到这个数」；收尾被截断时记成「核查模型没有按结构交结果」；质询回答失败 +——都没说出真正的原因。这里守几件事： + +- 额度：默认 16384（judge.MAX_TOKENS，核查和裁判共用）；服务方嫌大时退回 4096 再试一次，之后工具循环、收尾都用小的， + 只在还没有任何输出时重来；别的 400 不重试 +- 工具循环里被截断（没有工具调用、话没说完）不算模型自己停下：一条查询都没成功时不再收尾，直接交回截断的原因； + 查到过东西的照常按查到的收尾，取不到要的数时同样交回截断的原因 +- 收尾被截断（Anthropic 的 stop_reason=max_tokens、openai 结构化输出抛的 LengthFinishReasonError、普通调用的 + finish_reason=length）：交回截断的原因，不再退回普通调用多花一次额度；截断那次的用量照记 +- 时间线上记一条 verify_truncated（节点里调用时），llm.end 标出这次被截断 +- 原因经现有的渠道交出去：双盲复算的无法复算原因、反方审稿的「反查未完成」、质询的报错；没有截断时原有说法不变 + +模型是剧本，数据源是现造的 SQLite 文件（借 test_verify_agent 的夹具），表名、数字都是虚构的。 +""" +from __future__ import annotations + +import json + +from langchain_core.messages import AIMessage, BaseMessage +from openai import LengthFinishReasonError +from openai.types import CompletionUsage +from openai.types.chat import ChatCompletion + +from app.core.events import EventType +from app.engine import judge as judging +from app.engine import verify_agent +from app.engine.inquiry import _error_words +from app.engine.nodes.challenge import _claim_of +from app.engine.nodes.dual_check import _slim +from tests.test_verify_agent import VALID_SUM, clean_spend, q, run, shop # noqa: F401 - 夹具按名字引入 + +CUT_16K = "核查模型的输出达到上限（16384 token）被截断" +ADVICE = "通常是推理占满了输出额度，可在「设置 → 偏好设置 → 核查」中换用不带推理的模型" +FOUND = {"value": {"alias": "Q1", "row": 0, "column": "total"}, "verdict": "一致", + "rationale": "未作废订单的金额合计是 [[v:Q1.r0.total]]。"} + + +def cut_reply(output: int = 16384) -> AIMessage: + """推理模型把额度全用在思考上：正文为空、没有工具调用，finish_reason=length（DeepSeek 经 LiteLLM 的样子）。""" + return AIMessage(content="", additional_kwargs={"reasoning_content": "先想想该查哪张表……"}, + response_metadata={"finish_reason": "length"}, + usage_metadata={"input_tokens": 900, "output_tokens": output, "total_tokens": 900 + output}) + + +def anthropic_cut() -> dict: + """结构化输出走强制工具调用（Anthropic）时被截断：原始回复 stop_reason=max_tokens，解析不出结果。""" + raw = AIMessage(content=[], response_metadata={"stop_reason": "max_tokens"}, + usage_metadata={"input_tokens": 700, "output_tokens": 16384, "total_tokens": 17084}) + return {"raw": raw, "parsed": None, "parsing_error": None} + + +def length_error(prompt: int = 800, completion: int = 16384) -> LengthFinishReasonError: + """openai 的结构化输出（json_schema）在 finish_reason=length 时抛的那个错。""" + usage = CompletionUsage(completion_tokens=completion, prompt_tokens=prompt, total_tokens=prompt + completion) + return LengthFinishReasonError(completion=ChatCompletion.model_construct( + id="x", choices=[], created=0, model="m", object="chat.completion", usage=usage)) + + +def query_step(name: str, sql: str = VALID_SUM, call_id: str = "c1") -> AIMessage: + return AIMessage(content="查一下。", tool_calls=[{"name": n, "args": a, "id": i} for n, a, i in q(name, sql, call_id)]) + + +class BudgetRejected(Exception): + """服务方嫌输出额度太大时的 400（DeepSeek chat 系只收 8K）。""" + + status_code = 400 + + +class Reasoner: + """核查模型的剧本。steps：工具循环里每一轮的回复，用完了回「查完了。」；structured / plain：收尾时结构化调用、 + 普通调用的回复（AIMessage、结构化结果，或者一个异常就抛它）。calls 按顺序记下每次调用是哪一种。""" + + def __init__(self, steps=(), *, structured=None, plain=None, unstructured: bool = False) -> None: + self.steps = list(steps) + self.structured = structured + self.plain = plain + self.unstructured = unstructured + self.calls: list[str] = [] + + def bind_tools(self, tools, **kwargs): + return _View(self, "step") + + def with_structured_output(self, schema, **kwargs): + if self.unstructured: + raise NotImplementedError("这个模型不支持结构化输出") + return _View(self, "final") + + async def ainvoke(self, messages, *args, **kwargs): + return self.answer("plain") + + def answer(self, kind: str): + self.calls.append(kind) + if kind == "step": + reply = self.steps.pop(0) if self.steps else AIMessage(content="查完了。") + else: + reply = self.structured if kind == "final" else self.plain + if isinstance(reply, BaseException): + raise reply + return reply + + +class _View: + def __init__(self, model: Reasoner, kind: str) -> None: + self.model, self.kind = model, kind + + async def ainvoke(self, messages, *args, **kwargs): + return self.model.answer(self.kind) + + +class Rejecting: + """一开口就被拒的模型：工具循环、收尾都一样。""" + + def __init__(self, message: str = "max_tokens: 16384 > 4096, which is the maximum allowed number of output tokens"): + self.message = message + self.calls = 0 + + def bind_tools(self, tools, **kwargs): + return self + + def with_structured_output(self, schema, **kwargs): + return self + + async def ainvoke(self, messages, *args, **kwargs): + self.calls += 1 + raise BudgetRejected(self.message) + + +class Models: + """按额度给模型：rejects_above 给了，额度比它大的一律交 Rejecting。specs 记下每次建模型的配置。""" + + def __init__(self, monkeypatch, model, *, rejects_above: int | None = None, rejecting: Rejecting | None = None): + self.specs = [] + self.rejecting = rejecting or Rejecting() + + async def fake(session, spec): + self.specs.append(spec) + if rejects_above is not None and (spec.max_tokens or 0) > rejects_above: + return self.rejecting, "reasoner-x" + return model, "reasoner-x" + + monkeypatch.setattr(verify_agent, "get_chat_model", fake) + + +def collector(): + events: list[tuple] = [] + return events, (lambda kind, **data: events.append((kind, data))) + + +def truncated_logs(events) -> list[dict]: + return [d for k, d in events if k == EventType.LOG and d.get("code") == "verify_truncated"] + + +# -------------------------------------------------------------------------- +# 工具循环里被截断 +# -------------------------------------------------------------------------- + + +async def test_thinking_that_eats_the_budget_in_the_loop_is_reported_as_truncation(monkeypatch, shop): + name, _ = shop + model = Reasoner([cut_reply()], structured={"value": None, "verdict": "无法判断", "rationale": "没有查询结果"}) + Models(monkeypatch, model) + events, emit = collector() + out = await run(name, emit=emit, node_id="dc") + assert out.result is None and out.queries == [] + assert out.error == f"{CUT_16K}，核查未能完成。{ADVICE}" + assert model.calls == ["step"], "一条查询都没有：收尾只能交「查不到」，不再为它多花一次额度" + assert out.answered is False, "被截断不是模型自己停下来的" + assert out.usage["output_tokens"] == 16384, "截断那次的额度是真花出去了" + [log] = truncated_logs(events) + assert log["level"] == "warn" and log["max_tokens"] == 16384 and log["delivered"] is False + assert log["purpose"] == "verify" and log["message"] == out.error + [end] = [d for k, d in events if k == EventType.LLM_END] + assert end["truncated"] is True and end["max_tokens"] == 16384 and CUT_16K in end["error"] + + +async def test_a_cut_after_a_query_still_settles_on_what_was_found(monkeypatch, shop): + name, _ = shop + model = Reasoner([query_step(name), cut_reply()], structured=FOUND) + Models(monkeypatch, model) + events, emit = collector() + out = await run(name, emit=emit, node_id="dc") + assert model.calls == ["step", "step", "final"] + assert out.error is None and out.result["value"] == 230, "查到的数照常交：截断前的查询是真查过的" + assert out.answered is False + assert any(CUT_16K in n and "已按截断前查到的结果收尾" in n for n in out.notes) + [log] = truncated_logs(events) + assert log["delivered"] is True and log["purpose"] == "verify" and log["max_tokens"] == 16384 + + +async def test_a_cut_that_leaves_the_number_unfound_is_reported_as_truncation(monkeypatch, shop): + """查过、可要的数没查到就被截断了：说「没查到」会让人以为数据里没有，原因要写截断。""" + name, _ = shop + model = Reasoner([query_step(name, "SELECT COUNT(*) AS n FROM orders"), cut_reply()], + structured={"value": None, "verdict": "无法判断", "rationale": "还没查到合计"}) + Models(monkeypatch, model) + events, emit = collector() + out = await run(name, emit=emit, node_id="dc") + assert out.result is None and out.error == f"{CUT_16K}，核查未能完成。{ADVICE}" + assert [c["status"] for c in out.cells] == ["missing"] and len(out.queries) == 1, "查过的照记" + [log] = truncated_logs(events) + assert log["delivered"] is False and log["purpose"] == "verify" + + +# -------------------------------------------------------------------------- +# 收尾被截断 +# -------------------------------------------------------------------------- + + +async def test_a_cut_off_settle_is_reported_and_not_retried_as_plain(monkeypatch, shop): + """Anthropic 系：强制工具调用写到一半到了上限。退回普通调用也多半被推理占满，不再白花一次额度。""" + name, _ = shop + model = Reasoner([query_step(name)], structured=anthropic_cut(), plain=AIMessage(content=json.dumps(FOUND))) + Models(monkeypatch, model) + events, emit = collector() + out = await run(name, emit=emit, node_id="dc") + assert model.calls == ["step", "step", "final"], "没有再退回普通调用" + assert out.result is None and out.error == f"{CUT_16K}，核查未能完成。{ADVICE}" + [log] = truncated_logs(events) + assert log["purpose"] == "verify_result" and log["delivered"] is False + end = [d for k, d in events if k == EventType.LLM_END][-1] + assert end["purpose"] == "verify_result" and end["truncated"] is True and CUT_16K in end["error"] + + +async def test_an_openai_length_error_in_the_settle_is_truncation_not_a_failed_call(monkeypatch, shop): + """openai 的结构化输出解析时抛 LengthFinishReasonError：不是「调用失败，请稍后重试」，是额度被推理占满。""" + name, _ = shop + model = Reasoner([query_step(name)], structured=length_error(completion=16384)) + Models(monkeypatch, model) + events, emit = collector() + out = await run(name, emit=emit, node_id="dc") + assert model.calls == ["step", "step", "final"] + assert out.error == f"{CUT_16K}，核查未能完成。{ADVICE}" + assert out.usage["output_tokens"] == 16384 and out.usage["input_tokens"] >= 800, "异常里带着的用量照记" + end = [d for k, d in events if k == EventType.LLM_END][-1] + assert "请稍后重试" not in end["error"] and CUT_16K in end["error"] and end["output_tokens"] == 16384 + + +async def test_a_plain_settle_cut_off_is_reported(monkeypatch, shop): + """不支持结构化输出的模型：收尾直接走普通调用，回复只有思考、finish_reason=length。""" + name, _ = shop + model = Reasoner([query_step(name)], plain=cut_reply(), unstructured=True) + Models(monkeypatch, model) + out = await run(name) + assert model.calls == ["step", "step", "plain"] + assert out.result is None and out.error == f"{CUT_16K}，核查未能完成。{ADVICE}" + + +async def test_without_truncation_the_old_wording_stays(monkeypatch, shop): + name, _ = shop + model = Reasoner([query_step(name)], structured="不是 JSON", plain=AIMessage(content="还是不是 JSON")) + Models(monkeypatch, model) + events, emit = collector() + out = await run(name, emit=emit, node_id="dc") + assert model.calls == ["step", "step", "final", "plain"], "格式不对照旧退回普通调用补救一次" + assert out.error == "核查模型没有按结构交结果" + assert truncated_logs(events) == [] + assert not any(d.get("truncated") for k, d in events if k == EventType.LLM_END) + + +# -------------------------------------------------------------------------- +# 额度与退回 +# -------------------------------------------------------------------------- + + +async def test_the_default_budget_leaves_room_for_reasoning(monkeypatch, shop): + name, _ = shop + models = Models(monkeypatch, Reasoner([query_step(name)], structured=FOUND)) + out = await run(name) + assert out.error is None and out.result["value"] == 230 + assert [s.max_tokens for s in models.specs] == [16384, 16384], "工具循环、收尾都按 16384，不多建模型" + assert models.specs[1].thinking == "off" + + +async def test_a_rejected_budget_falls_back_to_4096_for_the_loop_and_the_settle(monkeypatch, shop): + name, _ = shop + model = Reasoner([query_step(name)], structured=FOUND) + models = Models(monkeypatch, model, rejects_above=4096) + out = await run(name) + assert out.error is None and out.result["value"] == 230 + assert [s.max_tokens for s in models.specs] == [16384, 4096, 16384, 4096] + assert [s.thinking for s in models.specs][2:] == ["off", "off"], "收尾的小额度模型同样关掉思考" + assert models.rejecting.calls == 1, "退回过一次之后收尾直接用小的，不再先撞一次" + assert model.calls == ["step", "step", "final"] + + +async def test_a_cut_after_the_fallback_names_the_small_budget(monkeypatch, shop): + name, _ = shop + Models(monkeypatch, Reasoner([cut_reply(output=4096)]), rejects_above=4096) + out = await run(name) + assert out.error.startswith("核查模型的输出达到上限（4096 token）被截断") + + +async def test_other_rejections_are_not_retried_with_the_small_budget(monkeypatch, shop): + name, _ = shop + models = Models(monkeypatch, Reasoner(), rejects_above=4096, + rejecting=Rejecting("model 'reasoner-x' does not exist")) + out = await run(name) + assert out.result is None and "reasoner-x" in out.error + assert [s.max_tokens for s in models.specs] == [16384], "不是额度的问题：不拿小额度再撞一次" + + +async def test_the_verify_spec_asks_for_the_larger_budget(): + """核查模型的配置（api/settings.py 的 verify_model_spec 沿用证据裁判的额度）：16384，思考照旧关掉。""" + assert judging.MAX_TOKENS == 16384 and judging.FALLBACK_MAX_TOKENS == 4096 + + +# -------------------------------------------------------------------------- +# 原因经现有的渠道交出去 +# -------------------------------------------------------------------------- + + +async def test_the_reason_reaches_dual_check_challenge_and_inquiry(monkeypatch, shop): + name, _ = shop + Models(monkeypatch, Reasoner([cut_reply()])) + out = await run(name) + expected = f"{CUT_16K}，核查未能完成。{ADVICE}" + # 双盲复算：无法复算的原因 + slim = _slim(out) + assert slim["value"] is None and slim["reason"] == expected + # 反方审稿：这句记为无法判断，理由写「反查未完成」，没能反查的原因就是截断 + cand = judging.Candidate(unit="u1", priority=1, order=0, text="华南每周的销售额都高于华东", section="", + aliases=("Q1",)) + claim, why = _claim_of(cand, out) + assert claim["verdict"] == "inconclusive" and claim["rationale"] == f"反查未完成：{expected}" and why == expected + # 审批质询：提问人看到的报错照原话 + assert _error_words(out, timeout_s=60, daily_max_usd=1.0) == expected + + +def test_the_reply_shapes_used_here_are_messages(): + assert isinstance(cut_reply(), BaseMessage) and isinstance(anthropic_cut()["raw"], BaseMessage)