diff --git a/backend/app/engine/judge.py b/backend/app/engine/judge.py index 27ba26c0..b43d2f87 100644 --- a/backend/app/engine/judge.py +++ b/backend/app/engine/judge.py @@ -39,7 +39,7 @@ from dataclasses import asdict, dataclass from datetime import datetime from typing import Any -from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage +from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage from sqlalchemy import select from app.core.artifact_store import canonical_json, content_hash @@ -70,6 +70,7 @@ from app.engine.merge_query import MERGE_SOURCE from app.engine.nodes.llm import _split_structured, _usage_of from app.engine.state import message_text 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, get_chat_model logger = logging.getLogger(__name__) @@ -131,9 +132,12 @@ SHORT_CHARS = 6 _OUT_PER_UNIT = 120 _OUT_BASE = 20 -#: 未裁判的原因。前四个是上限，报告里写「已到上限」 +#: 未裁判的原因。前四个是上限，报告里写「已到上限」；truncated 是裁判模型的输出到了上限被截断（推理模型的思考也算在 +#: 额度里），和格式不对分开记：说成「格式有误」，人会去查提示词，真正该换的是模型 LIMITS = ("max_claims", "max_cost_usd", "daily_max_usd", "timeout_s") -REASONS = (*LIMITS, "error", "format", "missing", "on_demand") +REASONS = (*LIMITS, "error", "format", "truncated", "missing", "on_demand") +#: 输出被截断时给人看的建议，接在 budget.cut_words 的「裁判模型的输出达到上限（N token）被截断」后面 +TRUNCATED_ADVICE = "通常是推理占满了输出额度，可换用不带推理的裁判模型" JUDGE_ROLE = "你是报告核查员：逐句判断报告里的话，能不能被它挂的证据支持。你不写报告，也不改报告。" @@ -1071,7 +1075,10 @@ async def _ask(model: Any, messages: list[BaseMessage]) -> tuple[list[Any] | Non """问一次裁判：(判定清单, 模型的原始回复, 没成的原因, 调了几次模型)。 先走结构化输出；不成（模型不支持、给的不是合法结构）先从原始回复里找 JSON，再退回纯文本调用、 - 从回复里抠 JSON。原因以 error: 开头的是调用本身失败，format 是格式不对。超时由调用方掐。 + 从回复里抠 JSON。原因以 error: 开头的是调用本身失败，format 是格式不对，truncated 是输出到了上限被截断 + （Anthropic 的 stop_reason=max_tokens、OpenAI 系的 finish_reason=length，以及 openai 结构化输出抛的 + LengthFinishReasonError）。结构化调用被截断的不再退回纯文本：推理占满额度时纯文本调用多半一样被截断，白花一次 + 额度。超时由调用方掐。 """ raws: list[BaseMessage] = [] try: @@ -1086,17 +1093,47 @@ async def _ask(model: Any, messages: list[BaseMessage]) -> tuple[list[Any] | Non for found in [message_text(raw), *(c.get("args") for c in getattr(raw, "tool_calls", None) or [])]: if (items := _items_of(found)) is not None: return items, raws, None, 1 - except Exception: # noqa: BLE001 - 不支持结构化输出的模型退回文本解析 - pass + if cut_off(raw): + return None, raws, _TRUNCATED, 1 + except Exception as e: # noqa: BLE001 - 不支持结构化输出的模型退回文本解析 + if truncated_error(e): + return None, [*raws, *_spent(e)], _TRUNCATED, 1 try: reply = await model.ainvoke(messages) except Exception as e: # noqa: BLE001 + if truncated_error(e): + return None, [*raws, *_spent(e)], _TRUNCATED, 2 return None, raws, f"error:{describe_exception(e)}", 2 raws.append(reply) items = _items_of(message_text(reply)) + if items is None and cut_off(reply): + return None, raws, _TRUNCATED, 2 return items, raws, None if items is not None else "format", 2 +#: _ask 交回的没成的原因：输出到了上限被截断 +_TRUNCATED = "truncated" + + +def _spent(error: BaseException) -> list[BaseMessage]: + """截断时 openai 的异常带着这次的用量：折成一条空回复交给记账（额度是真花出去了，每日上限不能漏掉它）。""" + inp, out = error_usage(error) + if not (inp or out): + return [] + return [AIMessage(content="", usage_metadata={"input_tokens": inp, "output_tokens": out, "total_tokens": inp + out})] + + +def _with_fallback(model: Any, spec: ModelSpec) -> BudgetFallback: + """裁判模型包一层：服务方嫌 MAX_TOKENS 太大时按 FALLBACK_MAX_TOKENS 重建一个再问（要退回时才建，平常不多建），之后 + 各批都用小的。经本模块的 get_chat_model 建：测试换成剧本时，小额度的那个也是剧本。""" + + async def build() -> Any: + async with SessionLocal() as session: + return (await get_chat_model(session, spec.model_copy(update={"max_tokens": FALLBACK_MAX_TOKENS})))[0] + + return BudgetFallback(model, build=build, limits=(spec.max_tokens, FALLBACK_MAX_TOKENS)) + + def _unjudged(reason: str, rationale: str, *, judge: str | None, post_seal: bool) -> dict[str, Any]: return {"status": "unjudged", "rationale": rationale, "judge": judge, "post_seal": post_seal, "reason": reason} @@ -1124,6 +1161,8 @@ def _why_unjudged(reason: str, budget: Budget, detail: str = "", *, click: bool return f"裁判失败：{detail}"[:120] if reason == "format": return detail or "裁判返回的结果格式有误，这句未裁判" + if reason == "truncated": + return f"{detail or cut_words('裁判模型', None)}，这句未裁判。{TRUNCATED_ADVICE}" if reason == "missing": return "裁判未对这句给出判定" return "探索运行按需裁判：点开这句并请模型判断后才会裁判" @@ -1157,6 +1196,7 @@ async def run_request(request: JudgeRequest, *, spec: ModelSpec, budget: Budget, try: async with SessionLocal() as session: model, model_id = await get_chat_model(session, spec) + model = _with_fallback(model, spec) except Exception as e: # noqa: BLE001 - 裁判模型没配好：这回判不了，报告照常 why = not_configured(e) if isinstance(e, ProviderNotConfigured) else describe_exception(e) details["error"] = f"裁判模型不可用（{why}）" @@ -1269,6 +1309,14 @@ async def run_request(request: JudgeRequest, *, spec: ModelSpec, budget: Budget, end = {"model": model_id, "purpose": "judge", "duration_ms": int((time.perf_counter() - began) * 1000), "input_tokens": inp, "output_tokens": out, "total_tokens": inp + out, "cost_usd": round(actual, 6), "calls": calls, "units": len(batch)} + if problem == _TRUNCATED: + # 输出到了上限被截断：这一批记未裁判，原因照实说是额度，不说「格式有误」「调用失败」。别的批照常问 + # （同格式不对）：截断取决于这一批要想多久，不像鉴权、网关出错那样后面几批多半一样 + limit = model.max_tokens + details["truncated"] = cut_words("裁判模型", limit) + emit(EventType.LLM_END, **end, error=details["truncated"], truncated=True, max_tokens=limit) + stop(batch, "truncated", details["truncated"]) + continue if problem is not None and problem != "format": reason = "timeout_s" if problem == "timeout_s" else "error" detail = problem.partition(":")[2] @@ -1402,6 +1450,8 @@ def _outcome(request: JudgeRequest, verdicts: dict[str, dict[str, Any]], budget: text = f"有 {n} 句结论未裁判：裁判调用失败（{details.get('error') or '原因不明'}）" elif reason == "format": text = f"有 {n} 句结论未裁判：裁判返回的结果格式有误" + elif reason == "truncated": + text = f"有 {n} 句结论未裁判：{details.get('truncated') or cut_words('裁判模型', None)}，{TRUNCATED_ADVICE}" else: text = f"有 {n} 句结论未获得裁判结果" gaps.append((reason, text)) diff --git a/backend/tests/test_judge_truncation.py b/backend/tests/test_judge_truncation.py new file mode 100644 index 00000000..cf0af450 --- /dev/null +++ b/backend/tests/test_judge_truncation.py @@ -0,0 +1,236 @@ +"""结论句裁判（engine/judge.py）的输出额度：截断照实说，服务方嫌额度大时退回小额度。 + +裁判原来固定 max_tokens 4096、thinking off。经 OpenAI 兼容接入的推理模型收不到 thinking=off，思考就能把 4096 用光： +结构化调用写不出判定（openai 抛 LengthFinishReasonError，Anthropic 的强制工具调用写到一半），退回纯文本调用也一样， +最后记成「裁判返回的结果格式有误」——没说出真正的原因。这里守几件事： + +- 额度：16384（MAX_TOKENS）；服务方嫌大时退回 4096（FALLBACK_MAX_TOKENS）再问一次，之后各批都用小的，只撞一次 +- 输出到了上限被截断（stop_reason=max_tokens、finish_reason=length、LengthFinishReasonError）：这一批记未裁判，原因 + 是 truncated，理由、缺口、llm.end、judge_failed 日志都写明「裁判模型的输出达到上限（N token）被截断」；结构化调用 + 被截断的不再退回纯文本多花一次额度；截断那次的用量照记 +- 没有截断时原有说法不变（格式不对仍是 format） + +模型都是剧本；证据借 test_evidence_judge 的虚构夹具（口径卡、一次查询、一次检索）。 +""" +from __future__ import annotations + +import json +import re + +from langchain_core.messages import AIMessage +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 judge_mod +from app.engine.evidence import compose_doc +from app.engine.judge import BATCH_SIZE, FALLBACK_MAX_TOKENS, MAX_TOKENS, REASONS, judge_doc +from tests.test_evidence_judge import PRICED, RAW, SPEC, UNLIMITED, clean_settings, make_catalog, stored # noqa: F401 + +CUT_16K = "裁判模型的输出达到上限（16384 token）被截断" +ADVICE = "通常是推理占满了输出额度，可换用不带推理的裁判模型" + + +def length_error(prompt: int = 3000, completion: int = 16384) -> LengthFinishReasonError: + 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 anthropic_cut(units) -> dict: + """强制工具调用写到一半到了上限：原始回复 stop_reason=max_tokens，解析不出判定。""" + raw = AIMessage(content=[], response_metadata={"stop_reason": "max_tokens"}, + usage_metadata={"input_tokens": 3000, "output_tokens": 16384, "total_tokens": 19384}) + return {"raw": raw, "parsed": None, "parsing_error": None} + + +def thinking_only(units) -> AIMessage: + """纯文本调用：额度全用在思考上，正文为空，finish_reason=length（推理模型经 OpenAI 兼容接入的样子）。""" + return AIMessage(content="", additional_kwargs={"reasoning_content": "逐句看证据……"}, + response_metadata={"finish_reason": "length"}, + usage_metadata={"input_tokens": 3000, "output_tokens": 4096, "total_tokens": 7096}) + + +def supported(units) -> dict: + items = [{"unit": u, "verdict": "supported", "rationale": "证据支持", "missing": "", "used": []} for u in units] + return {"raw": AIMessage(content="", response_metadata={"finish_reason": "stop"}), "parsed": {"items": items}} + + +class BudgetRejected(Exception): + """服务方嫌输出额度太大时的 400。""" + + status_code = 400 + + +class Reasoner: + """裁判模型的剧本。structured(units) / plain(units) 给出结构化调用、退回纯文本调用的回复（返回异常就抛它）； + unstructured 为真时不支持结构化输出。calls 记下每次调用的种类和判的句子。""" + + def __init__(self, *, structured=supported, plain=None, unstructured: bool = False) -> None: + self.structured, self.plain, self.unstructured = structured, plain, unstructured + self.calls: list[tuple[str, list[str]]] = [] + + def with_structured_output(self, schema, **kwargs): + if self.unstructured: + raise NotImplementedError("这个模型不支持结构化输出") + return _View(self) + + async def ainvoke(self, messages, *args, **kwargs): + return self.answer("plain", messages) + + def answer(self, kind: str, messages): + units = re.findall(r"^\[(u\d+)\]", str(messages[-1].content), re.M) + self.calls.append((kind, units)) + make = self.structured if kind == "structured" else self.plain + reply = make(units) + if isinstance(reply, BaseException): + raise reply + return reply + + +class _View: + def __init__(self, model: Reasoner) -> None: + self.model = model + + async def ainvoke(self, messages, *args, **kwargs): + return self.model.answer("structured", messages) + + +class Rejecting: + def __init__(self) -> None: + self.calls = 0 + + def with_structured_output(self, schema, **kwargs): + return self + + async def ainvoke(self, messages, *args, **kwargs): + self.calls += 1 + raise BudgetRejected("max_tokens: 16384 > 4096, which is the maximum allowed number of output tokens") + + +class Models: + def __init__(self, monkeypatch, model, *, rejects_above: int | None = None) -> None: + self.specs = [] + self.rejecting = 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, PRICED + return model, PRICED + + monkeypatch.setattr(judge_mod, "get_chat_model", fake) + + +async def judged(stored, raw: str = RAW): + catalog = make_catalog(stored) + events: list[tuple] = [] + outcome = await judge_doc(compose_doc(raw, catalog), catalog, spec=SPEC, budget=UNLIMITED, + emit=lambda kind, **data: events.append((kind, data))) + return outcome, events + + +# -------------------------------------------------------------------------- +# 截断 +# -------------------------------------------------------------------------- + + +async def test_a_cut_off_structured_reply_is_unjudged_as_truncation_without_a_plain_retry(monkeypatch, stored): + model = Reasoner(structured=anthropic_cut, plain=lambda units: AIMessage(content=json.dumps({"items": []}))) + Models(monkeypatch, model) + outcome, events = await judged(stored) + assert [kind for kind, _ in model.calls] == ["structured"], "判定写到一半到了上限：不再退回纯文本多花一次额度" + verdicts = list(outcome["verdicts"].values()) + assert {(v["status"], v["reason"]) for v in verdicts} == {("unjudged", "truncated")} + assert verdicts[0]["rationale"] == f"{CUT_16K}，这句未裁判。{ADVICE}" + assert outcome["unjudged"] == {"truncated": 4} and outcome["limits_hit"] == [] + assert outcome["gaps"] == [f"有 4 句结论未裁判：{CUT_16K}，{ADVICE}"] + [end] = [d for k, d in events if k == EventType.LLM_END] + assert end["purpose"] == "judge" and end["error"] == CUT_16K and end["truncated"] is True + assert end["max_tokens"] == 16384 and end["output_tokens"] == 16384 + [failed] = [d for k, d in events if k == EventType.LOG and d.get("code") == "judge_failed"] + assert failed["message"] == f"结论句裁判未完成：有 4 句结论未裁判：{CUT_16K}，{ADVICE}" + + +async def test_an_openai_length_error_is_truncation_and_its_tokens_are_counted(monkeypatch, stored): + model = Reasoner(structured=lambda units: length_error(completion=16384)) + Models(monkeypatch, model) + outcome, events = await judged(stored) + assert [kind for kind, _ in model.calls] == ["structured"] + assert outcome["unjudged"] == {"truncated": 4} + assert outcome["usage"]["output_tokens"] == 16384 and outcome["cost_usd"] > 0, "截断那次的额度是真花出去了" + end = [d for k, d in events if k == EventType.LLM_END][-1] + assert "调用" not in end["error"] and end["error"] == CUT_16K + + +async def test_a_plain_reply_with_only_thinking_is_truncation(monkeypatch, stored): + """不支持结构化输出的模型直接退回纯文本调用：回复只有思考、finish_reason=length。""" + model = Reasoner(plain=thinking_only, unstructured=True) + Models(monkeypatch, model) + outcome, _ = await judged(stored) + assert [kind for kind, _ in model.calls] == ["plain"] and outcome["calls"] == 2 + assert outcome["unjudged"] == {"truncated": 4} + assert outcome["gaps"] == [f"有 4 句结论未裁判：{CUT_16K}，{ADVICE}"] + + +async def test_a_truncated_batch_does_not_stop_the_others(monkeypatch, stored): + """只有一批被截断：别的批照常判（和格式不对一样，不像调用失败那样后面的都不再问）。""" + first: list[bool] = [] + + def once(units): + if not first: + first.append(True) + return anthropic_cut(units) + return supported(units) + + Models(monkeypatch, Reasoner(structured=once)) + raw = "".join(f"第 {i} 个区域的销售额增长来自新客。[[see:m:gmv]]" for i in range(BATCH_SIZE + 3)) + outcome, _ = await judged(stored, raw) + assert outcome["unjudged"] == {"truncated": BATCH_SIZE} + assert outcome["counts"]["supported"] == 3 + + +async def test_without_truncation_format_problems_keep_their_wording(monkeypatch, stored): + model = Reasoner(structured=lambda units: {"raw": AIMessage(content="不是 JSON", + response_metadata={"finish_reason": "stop"}), + "parsed": None}, + plain=lambda units: AIMessage(content="还是不是", response_metadata={"finish_reason": "stop"})) + Models(monkeypatch, model) + outcome, events = await judged(stored) + assert [kind for kind, _ in model.calls] == ["structured", "plain"] + assert outcome["unjudged"] == {"format": 4} + assert outcome["gaps"] == ["有 4 句结论未裁判：裁判返回的结果格式有误"] + assert not any(d.get("truncated") for k, d in events if k == EventType.LLM_END) + + +# -------------------------------------------------------------------------- +# 额度与退回 +# -------------------------------------------------------------------------- + + +async def test_the_judge_asks_for_the_larger_budget(monkeypatch, stored): + models = Models(monkeypatch, Reasoner()) + outcome, _ = await judged(stored) + assert outcome["counts"]["supported"] == 4 + assert [s.max_tokens for s in models.specs] == [MAX_TOKENS] == [16384], "平常只建一个模型" + assert "truncated" in REASONS and FALLBACK_MAX_TOKENS == 4096 + + +async def test_a_rejected_budget_falls_back_to_4096_once_for_all_batches(monkeypatch, stored): + model = Reasoner() + models = Models(monkeypatch, model, rejects_above=4096) + raw = "".join(f"第 {i} 个区域的销售额增长来自新客。[[see:m:gmv]]" for i in range(BATCH_SIZE * 2 + 3)) + outcome, _ = await judged(stored, raw) + assert outcome["counts"]["supported"] == BATCH_SIZE * 2 + 3 and outcome["gaps"] == [] + assert [s.max_tokens for s in models.specs] == [16384, 4096] + assert models.rejecting.calls == 1, "第一批退回过，后面几批直接用小的" + assert [kind for kind, _ in model.calls] == ["structured"] * 3 + + +async def test_a_truncation_after_the_fallback_names_the_small_budget(monkeypatch, stored): + Models(monkeypatch, Reasoner(plain=thinking_only, unstructured=True), rejects_above=4096) + outcome, _ = await judged(stored) + [verdict, *_] = outcome["verdicts"].values() + assert verdict["reason"] == "truncated" + assert verdict["rationale"].startswith("裁判模型的输出达到上限（4096 token）被截断")