From 720f842b823f24aaca16603eefb940f3fa38f397 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sun, 4 Oct 2026 12:27:38 +0800 Subject: [PATCH] =?UTF-8?q?=E7=BB=93=E8=AE=BA=E5=8F=A5=E8=A3=81=E5=88=A4?= =?UTF-8?q?=E5=B0=8F=E6=89=B9=E5=88=86=E6=B3=A2=E5=9C=B0=E9=97=AE=EF=BC=9A?= =?UTF-8?q?=E4=B8=80=E6=AC=A1=E8=B0=83=E7=94=A8=E6=9C=80=E5=A4=9A=208=20?= =?UTF-8?q?=E5=8F=A5=EF=BC=88=E5=8E=9F=E6=9D=A5=2040=20=E5=8F=A5=EF=BC=89?= =?UTF-8?q?=EF=BC=8C=E7=AC=AC=E4=B8=80=E6=89=B9=E5=8D=95=E7=8B=AC=E9=97=AE?= =?UTF-8?q?=E3=80=81=E4=B9=8B=E5=90=8E=E6=AF=8F=E6=B3=A2=E6=9C=80=E5=A4=9A?= =?UTF-8?q?=204=20=E6=89=B9=E5=90=8C=E6=97=B6=E9=97=AE=EF=BC=8C=E5=87=A0?= =?UTF-8?q?=E6=89=B9=E5=85=B1=E7=94=A8=E5=89=A9=E4=B8=8B=E7=9A=84=E6=97=B6?= =?UTF-8?q?=E9=95=BF=EF=BC=9B=E8=B6=85=E6=97=B6=E5=8F=AA=E4=B8=A2=E6=B2=A1?= =?UTF-8?q?=E5=9B=9E=E6=9D=A5=E7=9A=84=E9=82=A3=E5=87=A0=E6=89=B9=EF=BC=8C?= =?UTF-8?q?=E5=9B=9E=E6=9D=A5=E7=9A=84=E5=88=A4=E5=AE=9A=E7=85=A7=E6=A0=B7?= =?UTF-8?q?=E7=95=99=E4=B8=8B=E3=80=82=E7=9C=9F=E5=AE=9E=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E9=87=8C=E6=8E=A8=E7=90=86=E6=A8=A1=E5=9E=8B=E6=8B=BF=E7=9D=80?= =?UTF-8?q?=2040=20=E5=8F=A5=E7=9A=84=E6=91=98=E5=BD=95=E5=9C=A8=2030=20?= =?UTF-8?q?=E7=A7=92=E7=9A=84=E6=97=B6=E9=95=BF=E4=B8=8A=E9=99=90=E5=86=85?= =?UTF-8?q?=E6=B2=A1=E5=9B=9E=E6=9D=A5=EF=BC=8C=E4=BA=A4=E7=BB=99=E5=AE=A1?= =?UTF-8?q?=E6=89=B9=E7=9A=84=E6=8A=A5=E5=91=8A=E4=B8=80=E5=8F=A5=E9=83=BD?= =?UTF-8?q?=E6=B2=A1=E5=88=A4=E6=88=90=EF=BC=8C=E5=AE=A1=E6=89=B9=E5=8D=A1?= =?UTF-8?q?=E5=8F=AA=E8=83=BD=E5=86=99=E5=85=A8=E9=83=A8=E6=9C=AA=E8=A3=81?= =?UTF-8?q?=E5=88=A4=EF=BC=9B=E5=90=8C=E4=B8=80=E4=B8=AA=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=205=20=E5=8F=A5=E4=B8=80=E6=89=B9=203=20=E7=A7=92=E5=A4=9A?= =?UTF-8?q?=E5=B0=B1=E5=9B=9E=E6=9D=A5=E4=BA=86=E3=80=82=E9=87=91=E9=A2=9D?= =?UTF-8?q?=E4=B8=8A=E9=99=90=E6=8C=89=E5=B7=B2=E8=8A=B1=E7=9A=84=E5=8A=A0?= =?UTF-8?q?=E8=BF=99=E4=B8=80=E6=B3=A2=E5=B7=B2=E5=AE=9A=E4=B8=8B=E7=9A=84?= =?UTF-8?q?=E4=BC=B0=E7=AE=97=E7=AE=97=EF=BC=9B=E7=AC=AC=E4=B8=80=E6=89=B9?= =?UTF-8?q?=E8=B0=83=E4=B8=8D=E9=80=9A=EF=BC=88=E9=89=B4=E6=9D=83=E3=80=81?= =?UTF-8?q?=E7=BD=91=E5=85=B3=E6=8C=82=E4=BA=86=EF=BC=89=E6=97=B6=E5=8F=AA?= =?UTF-8?q?=E6=92=9E=E8=BF=99=E4=B8=80=E6=AC=A1=EF=BC=8C=E5=90=8E=E9=9D=A2?= =?UTF-8?q?=E7=9A=84=E7=BB=9F=E4=B8=80=E8=AE=B0=E6=9C=AA=E8=A3=81=E5=88=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/engine/judge.py | 166 ++++++++++++++++----------- backend/tests/test_evidence_judge.py | 32 ++++++ 2 files changed, 128 insertions(+), 70 deletions(-) diff --git a/backend/app/engine/judge.py b/backend/app/engine/judge.py index 8e8d67a8..1dc0e67d 100644 --- a/backend/app/engine/judge.py +++ b/backend/app/engine/judge.py @@ -98,8 +98,12 @@ MISSING_MAX = 30 #: 判定的规则版本：新判的判定带着它（rule_version），也进 request_key。报告节点的断点里换了规则的旧判定不复用； #: 按需裁判遇到早于当前规则的判定（没有版本、版本更小、旧取值 unsupported）重判一次 RULES_VERSION = 2 -#: 一次调用最多判几句：再多，4096 个输出令牌装不下每句一条判定加理由 -BATCH_SIZE = 40 +#: 一次调用最多判几句。以前是 40：真实运行里推理模型拿着 40 句的摘录 30 秒没回来，整份报告一句都没判成 +#: （同一个模型 5 句一批 3 秒多就回来了）。批小了，超时最多丢掉没回来的那几批 +BATCH_SIZE = 8 +#: 一波最多几批同时问：几批共用剩下的时长，句数上限内的句子一两波就判完。第一批单独问：接入调不通（鉴权、 +#: 网关挂了）时只撞这一次，不让一整波一起失败 +PARALLEL_BATCHES = 4 MAX_TOKENS = 4096 #: 证据摘录：SQL 前几个字、每次查询最多几行几列、每段知识库片段最多几个字、整份引用的检索给几段。 #: 列数上限只管没被引用的列：句子引用的格子所在的列一律给（遮罩的除外），再用别的列补满到这么多列 @@ -1180,80 +1184,102 @@ async def run_request(request: JudgeRequest, *, spec: ModelSpec, budget: Budget, emit(EventType.LOG, level="warn", code="judge_unpriced", message=note) spent = 0.0 + width = 1 while pending: remaining = None if budget.timeout_s is None else budget.timeout_s - (time.monotonic() - started) if remaining is not None and remaining <= 0: stop(pending, "timeout_s") break - batch = pending[:BATCH_SIZE] - cost = request.cost(model_id, batch) if priced else None - if cost is not None: - caps = [] - if budget.max_cost_usd is not None: - caps.append((budget.max_cost_usd - spent, "max_cost_usd")) - if budget.daily_max_usd is not None: - caps.append((budget.daily_max_usd - (await daily_spend())["usd"], "daily_max_usd")) - if caps: - cap, binding = min(caps, key=lambda x: x[0]) - trimmed = False - while batch and (cost := request.cost(model_id, batch) or 0.0) > cap + 1e-12: - batch, trimmed = batch[:-1], True # 按优先级从后往前丢：先判挂了依据的 - if not batch: - stop(pending, binding) - break - if trimmed: - # 这一批已经是剩下的钱装得下的最多句子，后面的句子更装不下 - stop(pending[len(batch):], binding) - pending = pending[:len(batch)] - pending = pending[len(batch):] - charged = cost or 0.0 - day = await _charge(charged, limit=budget.daily_max_usd if priced else None, priced=priced) - if day is None: # 估算之后、记账之前别的裁判先花掉了今天剩下的钱 - stop([*batch, *pending], "daily_max_usd") - break - estimated += charged - began = time.perf_counter() - emit(EventType.LLM_START, model=model_id, structured=True, purpose="judge", message_count=2, - units=len(batch)) - items: list[Any] | None = None - raws: list[BaseMessage] = [] - problem: str | None = None - calls = 1 - try: - async with asyncio.timeout(remaining): - items, raws, problem, calls = await _ask(model, request.messages(batch)) - except TimeoutError: - problem = "timeout_s" - tokens = [_usage_of(r, model_id) for r in raws] - inp, out = sum(t["input_tokens"] for t in tokens), sum(t["output_tokens"] for t in tokens) - # 回复没带用量（有的网关不给、调用超时）就按估算记：宁可多记，不能让每日上限漏掉 - actual = (pricing.estimate_cost(model_id, inp, out) if inp or out else charged) if priced else 0.0 - await _settle(day, actual - charged) - spent += actual - usage["input_tokens"] += inp - usage["output_tokens"] += out - usage["calls"] += calls - usage["cost_usd"] += actual - 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 is not None and problem != "format": - reason = "timeout_s" if problem == "timeout_s" else "error" - detail = problem.partition(":")[2] - emit(EventType.LLM_END, **end, error=detail or _why_unjudged(reason, label, click=click)) - if reason == "error": - # 调用本身失败（鉴权、网关挂了）：后面几批多半一样，不再一次次去撞 - details["error"] = detail - stop([*batch, *pending], reason, detail) + # 一波最多 PARALLEL_BATCHES 批：逐批按金额上限定下来（这一波已经定下的按估算先占着），再一起问 + wave: list[tuple[list[Candidate], float, Any]] = [] + reserved = 0.0 + while pending and len(wave) < width: + batch = pending[:BATCH_SIZE] + cost = request.cost(model_id, batch) if priced else None + if cost is not None: + caps = [] + if budget.max_cost_usd is not None: + caps.append((budget.max_cost_usd - spent - reserved, "max_cost_usd")) + if budget.daily_max_usd is not None: + caps.append((budget.daily_max_usd - (await daily_spend())["usd"], "daily_max_usd")) + if caps: + cap, binding = min(caps, key=lambda x: x[0]) + trimmed = False + while batch and (cost := request.cost(model_id, batch) or 0.0) > cap + 1e-12: + batch, trimmed = batch[:-1], True # 按优先级从后往前丢：先判挂了依据的 + if not batch: + stop(pending, binding) + pending = [] + break + if trimmed: + # 这一批已经是剩下的钱装得下的最多句子，后面的句子更装不下 + stop(pending[len(batch):], binding) + pending = pending[:len(batch)] + pending = pending[len(batch):] + charged = cost or 0.0 + day = await _charge(charged, limit=budget.daily_max_usd if priced else None, priced=priced) + if day is None: # 估算之后、记账之前别的裁判先花掉了今天剩下的钱 + stop([*batch, *pending], "daily_max_usd") + pending = [] break - stop(batch, reason, detail) - continue - emit(EventType.LLM_END, **end) - if items is None: - stop(batch, "format") - continue - _take(items, batch, request, verdicts, judge=judge, post_seal=post_seal) - stop(batch, "missing") + estimated += charged + reserved += charged + wave.append((batch, charged, day)) + if not wave: + break + + began = time.perf_counter() + for batch, _, _ in wave: + emit(EventType.LLM_START, model=model_id, structured=True, purpose="judge", message_count=2, + units=len(batch)) + tasks = [asyncio.create_task(_ask(model, request.messages(batch))) for batch, _, _ in wave] + done, late = await asyncio.wait(tasks, timeout=remaining) + for task in late: + task.cancel() + if late: + await asyncio.gather(*late, return_exceptions=True) + failed: str | None = None + for (batch, charged, day), task in zip(wave, tasks): + items: list[Any] | None = None + raws: list[BaseMessage] = [] + problem: str | None = None + calls = 1 + if task in done: + items, raws, problem, calls = task.result() + else: + problem = "timeout_s" + tokens = [_usage_of(r, model_id) for r in raws] + inp, out = sum(t["input_tokens"] for t in tokens), sum(t["output_tokens"] for t in tokens) + # 回复没带用量（有的网关不给、调用超时）就按估算记：宁可多记，不能让每日上限漏掉 + actual = (pricing.estimate_cost(model_id, inp, out) if inp or out else charged) if priced else 0.0 + await _settle(day, actual - charged) + spent += actual + usage["input_tokens"] += inp + usage["output_tokens"] += out + usage["calls"] += calls + usage["cost_usd"] += actual + 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 is not None and problem != "format": + reason = "timeout_s" if problem == "timeout_s" else "error" + detail = problem.partition(":")[2] + emit(EventType.LLM_END, **end, error=detail or _why_unjudged(reason, label, click=click)) + if reason == "error": + # 调用本身失败（鉴权、网关挂了）：后面几批多半一样，不再一次次去撞 + details["error"] = failed = detail + stop(batch, reason, detail) + continue + emit(EventType.LLM_END, **end) + if items is None: + stop(batch, "format") + continue + _take(items, batch, request, verdicts, judge=judge, post_seal=post_seal) + stop(batch, "missing") + if failed is not None: + stop(pending, "error", failed) + break + width = PARALLEL_BATCHES for c in request.cands: # 兜底：每个候选都有判定 verdicts.setdefault(c.unit, _unjudged("missing", _why_unjudged("missing", label), judge=judge, diff --git a/backend/tests/test_evidence_judge.py b/backend/tests/test_evidence_judge.py index 1c3c421c..75a3b995 100644 --- a/backend/tests/test_evidence_judge.py +++ b/backend/tests/test_evidence_judge.py @@ -17,6 +17,7 @@ """ from __future__ import annotations +import asyncio import json import re @@ -464,6 +465,37 @@ async def test_cost_truncation_follows_priority(monkeypatch, stored): assert {status_of(outcome, doc, t)["status"] for t in (CITED, CITED2)} == {"supported"} +async def test_a_slow_batch_times_out_alone_and_the_others_keep_their_verdicts(monkeypatch, stored): + """真实运行里推理模型拿着 40 句一批，30 秒没回来，整份报告一句都没判成（同一个模型 5 句一批 3 秒多就回来）。 + 现在小批：第一批单独问，之后几批同时问、共用剩下的时长；超时只丢没回来的那一批，回来的判定照样留下。""" + Judge(monkeypatch) + catalog = make_catalog(stored) + raw = "".join(f"第 {i} 个区域的销售额增长来自新客。[[see:m:gmv]]" for i in range(BATCH_SIZE * 3)) + doc = compose_doc(raw, catalog) + real_ask = judge_mod._ask + asked: list[int] = [] + running = peak = 0 + + async def ask(model, messages): + nonlocal running, peak + asked.append(len(re.findall(r"^\[(u\d+)\]", str(messages[-1].content), re.M))) + running += 1 + peak = max(peak, running) + try: + await asyncio.sleep(5 if len(asked) == 3 else 0.05) # 第三批卡住 + return await real_ask(model, messages) + finally: + running -= 1 + + monkeypatch.setattr(judge_mod, "_ask", ask) + outcome = await judge_doc(doc, catalog, spec=SPEC, + budget=Budget(max_claims=None, max_cost_usd=None, timeout_s=1, daily_max_usd=None)) + assert asked == [BATCH_SIZE] * 3 + assert peak == 2, "第一批单独问，后两批同时问" + assert outcome["counts"]["supported"] == BATCH_SIZE * 2 + assert outcome["unjudged"] == {"timeout_s": BATCH_SIZE} and outcome["limits_hit"] == ["timeout_s"] + + async def test_unlimited_never_truncates(monkeypatch, stored): """每一项都写 null：句数再多也全判（按每批上限分批），金额、时长、每日都不拦。""" judge = Judge(monkeypatch)