From a490365bcfcf955d94401a03ad106b9e4111c2bc Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Wed, 14 Oct 2026 09:59:19 +0800 Subject: [PATCH] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=E9=97=AD?= =?UTF-8?q?=E7=8E=AF=E7=AC=AC=E4=B8=80=E6=AD=A5=EF=BC=88=E5=90=8E=E7=AB=AF?= =?UTF-8?q?=20A=EF=BC=89=E5=A4=8D=E6=A0=B8=E4=BF=AE=E6=AD=A3=EF=BC=9A?= =?UTF-8?q?=E6=8C=89=E6=AF=94=E4=BE=8B=E6=AD=A2=E6=8D=9F=E5=8F=AA=E7=9C=8B?= =?UTF-8?q?=E6=AF=8F=E6=AC=A1=E8=B0=83=E7=94=A8=E7=9A=84=E6=9C=80=E7=BB=88?= =?UTF-8?q?=E7=BB=93=E5=B1=80=E3=80=82=E6=9C=8D=E5=8A=A1=E6=96=B9=E5=AB=8C?= =?UTF-8?q?=E9=A2=9D=E5=BA=A6=E5=A4=AA=E5=A4=A7=E9=80=80=E5=9B=9E=208192?= =?UTF-8?q?=E3=80=81=E7=BB=93=E6=9E=84=E5=8C=96=E8=BE=93=E5=87=BA=E4=B8=8D?= =?UTF-8?q?=E6=94=AF=E6=8C=81=E6=88=96=E5=9B=9E=E4=BA=86=E8=AF=B4=E6=98=8E?= =?UTF-8?q?=E6=96=87=E5=AD=97=E6=94=B9=E8=B5=B0=E7=BA=AF=E6=96=87=E6=9C=AC?= =?UTF-8?q?=E3=80=81429=20/=205xx=20=E9=80=80=E9=81=BF=E9=87=8D=E8=AF=95?= =?UTF-8?q?=E4=B9=8B=E5=89=8D=E7=9A=84=E9=82=A3=E6=AC=A1=E7=85=A7=E6=97=A7?= =?UTF-8?q?=E8=AE=B0=E8=BF=9B=E7=94=A8=E9=87=8F=E5=92=8C=E6=8F=90=E9=86=92?= =?UTF-8?q?=EF=BC=8C=E4=B8=8D=E8=BF=9B=E6=AD=A2=E6=8D=9F=E7=9A=84=E7=AA=97?= =?UTF-8?q?=E5=8F=A3=E2=80=94=E2=80=94=E5=AE=83=E4=B8=8D=E8=8A=B1=E9=A2=9D?= =?UTF-8?q?=E5=BA=A6=EF=BC=8C=E4=B9=9F=E8=AF=B4=E6=98=8E=E4=B8=8D=E4=BA=86?= =?UTF-8?q?=E6=A8=A1=E5=9E=8B=E5=86=99=E4=B8=8D=E5=87=BA=E6=9D=A5=EF=BC=9B?= =?UTF-8?q?=E9=87=8D=E8=AF=95=E7=94=A8=E5=AE=8C=E4=BB=8D=E5=87=BA=E9=94=99?= =?UTF-8?q?=E7=9A=84=E9=82=A3=E4=B8=80=E6=AC=A1=E7=85=A7=E5=B8=B8=E7=AE=97?= =?UTF-8?q?=E3=80=82=E4=BB=A5=E5=89=8D=E9=A2=9D=E5=BA=A6=E4=B8=8A=E9=99=90?= =?UTF-8?q?=208192=E3=80=81=E4=B8=8D=E6=94=AF=E6=8C=81=20json=5Fschema=20?= =?UTF-8?q?=E7=9A=84=20OpenAI=20=E5=85=BC=E5=AE=B9=E7=BD=91=E5=85=B3?= =?UTF-8?q?=E4=B8=8A=EF=BC=8C=E5=B9=B6=E5=8F=91=E7=9A=84=E5=A4=B4=E5=87=A0?= =?UTF-8?q?=E6=AC=A1=E8=B0=83=E7=94=A8=E5=85=A8=E6=98=AF=E8=BF=99=E7=A7=8D?= =?UTF-8?q?=E9=87=8D=E6=9D=A5=EF=BC=8C=E7=AC=AC=E4=B8=80=E6=AC=A1=E9=A2=9D?= =?UTF-8?q?=E5=BA=A6=E7=94=A8=E5=AE=8C=E5=B0=B1=E5=87=91=E6=BB=A1=2010=20?= =?UTF-8?q?=E6=AC=A1=E9=87=8C=207=20=E6=AC=A1=EF=BC=8C=E9=82=A3=E4=B8=80?= =?UTF-8?q?=E6=AE=B5=E7=9B=B4=E6=8E=A5=E8=AE=B0=E6=B2=A1=E5=86=99=E5=87=BA?= =?UTF-8?q?=E6=9D=A5=E3=80=81=E4=B8=8D=E5=86=8D=E6=8B=86=E5=B0=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/data/catalog_draft.py | 40 ++++--- backend/tests/test_catalog_draft_engine.py | 121 ++++++++++++++++++++- 2 files changed, 143 insertions(+), 18 deletions(-) diff --git a/backend/app/data/catalog_draft.py b/backend/app/data/catalog_draft.py index 07e83877..181cf73c 100644 --- a/backend/app/data/catalog_draft.py +++ b/backend/app/data/catalog_draft.py @@ -47,7 +47,10 @@ 的块也不再问），报错建议更换起草模型，直到比例降下去。以前按「连续几段拆到底」「接连超时」算，偶尔一次写出来就清零： 实测会思考的模型在宽表上单次 1 到 7 分钟、四分之一超时，90 分钟停不下来。几处只管通信的调整照旧：服务方嫌额度 太大退回小额度、结构化输出不支持或回了说明文字改走纯文本、网络错误退避重试；结构化输出只给空壳的，这次起草还没换 - 纯文本问过的照旧问一次（看看是不是结构化输出的问题）。 + 纯文本问过的照旧问一次（看看是不是结构化输出的问题）。止损的窗口只记每次调用的最终结局：上面这几种重来之前的那次 + 不花额度、也说明不了模型写不出来，照旧报给进度回调（用量、提醒照算），不进止损的窗口——以前都记成没写出来，额度上限 + 8192、不支持 json_schema 的 OpenAI 兼容网关上，并发的头几次调用全是这种重来，第一次额度用完就凑满 10 次里 7 次， + 那一段直接记没写出来、不再拆小。重试用完仍出错的那一次是最终结局，照常算。 - 同一张表连续 MODEL_TABLE_GIVE_UP_AFTER 个原始段都一项没写出来：这张表剩下还没开始的段跳过不问（宽表一段一段问下去 只是烧额度），原因写清跳过了几段、可以换一个起草模型后只重试这张表。中间有一段写出来了就不算「接连」。 - 提醒：最近 MODEL_NOTICE_WINDOW 次调用里有 MODEL_NOTICE_ON 次以上没写出来（额度用完、超时、空壳、出错），给一句 @@ -1093,8 +1096,10 @@ class _Run: await self._split(pieces, depth, "timeout", followup, budget, halved, chunk) return if reply.error is not None: - await self._report(reply, wrote=False) - if reply.prose and not last: + # 结构化输出回了说明文字、这一块马上改走纯文本再问：只管通信的重来，不进止损的窗口 + redo = reply.prose and not last + await self._report(reply, wrote=False, halting=not redo) + if redo: await self._prose_to_text(pieces, depth, followup, budget, halved, chunk) return self._fail(pieces, _Why(reply.error, split=halved)) @@ -1265,8 +1270,9 @@ class _Run: 这一段的调用次数用完了交回 capped（不调用）。optional（拆出来的块）：拿到名额时已经止损，交回 skipped（不调用， 次数退回）——等名额的时候前面的调用多数没写出来，不再拆小重试。超时不原样重试：交回 timeout，由调用方当作这一段 - 太大、拆小再问。重来之前的那次调用在这里就报给进度回调；交回的那次由调用方读完内容再报（写没写出来要看校验之后 - 的内容）。服务方嫌额度太大、结构化输出不支持、网络错误的重来止损期间照旧（只管通信，不是模型写不出来）。""" + 太大、拆小再问。重来之前的那次调用在这里就报给进度回调（记进用量和提醒，不进止损的窗口：halting=False）；交回的 + 那次由调用方读完内容再报（写没写出来要看校验之后的内容）。服务方嫌额度太大、结构化输出不支持、网络错误的重来止损 + 期间照旧（只管通信，不是模型写不出来）。""" mode = mode or self.mode retried = 0 while True: @@ -1293,16 +1299,16 @@ class _Run: if kind == "timeout": return _Reply(timeout=True, usage=(0, 0)) if kind == "budget" and (getattr(self.model, "budget", None) != used or self._shrink()): - await self._called((0, 0), "error") + await self._called((0, 0), "error", halting=False) continue if kind == "unsupported" and mode == "structured": if self.mode == "structured": logger.info("起草数据目录：结构化输出用不了（%s），后面改走纯文本", str(e)[:200]) self.mode = mode = "text" - await self._called((0, 0), "error") + await self._called((0, 0), "error", halting=False) continue if kind == "transient" and retried < int(C.MODEL_RETRIES): - await self._called((0, 0), "error") + await self._called((0, 0), "error", halting=False) delays = tuple(C.MODEL_RETRY_DELAYS_S) or (0.0,) await _sleep(delays[min(retried, len(delays) - 1)]) retried += 1 @@ -1354,8 +1360,9 @@ class _Run: # ---------------------------------------------------------------- 每次调用之后：用量、段大小、提醒 - async def _report(self, reply: _Reply, *, wrote: bool) -> None: - """交回的那次调用读完内容之后报给进度回调。""" + async def _report(self, reply: _Reply, *, wrote: bool, halting: bool = True) -> None: + """交回的那次调用读完内容之后报给进度回调。halting：这次是不是这一块这次调用的最终结局（进不进止损的窗口，见 + _called）。""" if reply.usage is None: return if reply.timeout: @@ -1366,13 +1373,18 @@ class _Run: failure = None else: failure = "exhausted" if reply.exhausted else "hollow" - await self._called(reply.usage, failure, strike=reply.timeout or reply.exhausted) + await self._called(reply.usage, failure, strike=reply.timeout or reply.exhausted, halting=halting) - async def _called(self, usage: tuple[int, int], failure: str | None, *, strike: bool = False) -> None: + async def _called(self, usage: tuple[int, int], failure: str | None, *, strike: bool = False, + halting: bool = True) -> None: """记下一次调用：额度用完或超时（strike）累计 MODEL_SHRINK_AFTER 次，新开始的段列数上限减半；最近几次调用多数 - 没写出来时提醒，少了清掉；止损按最近 MODEL_HALT_WINDOW 次调用的比例（_halted）。报给进度回调。""" + 没写出来时提醒，少了清掉；止损按最近 MODEL_HALT_WINDOW 次调用的比例（_halted）。报给进度回调。 + + halting=False：只管通信的重来之前的那次（服务方嫌额度太大、结构化输出不支持或回了说明文字、429 / 5xx 退避重试）。 + 照旧记进用量和提醒，不进止损的窗口：它不花额度，也说明不了模型写不出来；止损只看每次调用的最终结局。""" self.recent.append(failure) - self.window.append(failure is not None) + if halting: + self.window.append(failure is not None) halted = self._halted() if halted != self.was_halted: self.was_halted = halted diff --git a/backend/tests/test_catalog_draft_engine.py b/backend/tests/test_catalog_draft_engine.py index aadb8ced..39438de0 100644 --- a/backend/tests/test_catalog_draft_engine.py +++ b/backend/tests/test_catalog_draft_engine.py @@ -1285,15 +1285,17 @@ async def test_a_wide_table_that_wrote_nothing_names_every_part(): async def test_a_halted_part_that_could_be_split_is_not_asked_again_in_text_mode(monkeypatch): - """止损以后每块只问一次：已经走纯文本的空壳不再原样再问、也不拆小。t1：结构化不支持 1 次、纯文本 1 次、拆成两半各问 - 一次再原样再问一次（6 次都没写出来）；t2 的第一次是第 7 次没写出来，之后止损。""" + """止损以后每块只问一次：已经走纯文本的空壳不再原样再问、也不拆小。t1：结构化不支持 1 次（只管通信的重来，不进 + 止损的窗口）、纯文本 1 次、拆成两半各问一次再原样再问一次（5 次没写出来）；t2 的第一次是第 6 次，照常拆小，前一半 + 是第 7 次没写出来，之后止损：后一半不再问；t3 只问一次、不拆小。""" monkeypatch.setattr(catalog, "MODEL_CONCURRENCY", 1) monkeypatch.setattr(catalog, "MODEL_BATCH_TABLES", 1) tables = {f"t{i}": _table(10) for i in range(1, 5)} model = Scripted(_text_only(_hollow)) got = await D.run_model_draft(model, _source(tables), list(tables)) - assert _by_table(model) == {"t1": 6, "t2": 1, "t3": 1, "t4": 1} - assert got["t2"].error == catalog.MODEL_EMPTY_REASON + HALTED + assert _by_table(model) == {"t1": 6, "t2": 2, "t3": 1, "t4": 1} + assert got["t2"].error == D.REASON_HOLLOW + D.REASON_SPLIT + HALTED + assert got["t3"].error == catalog.MODEL_EMPTY_REASON + HALTED async def test_while_halted_structured_output_still_gets_one_probe_in_text(monkeypatch): @@ -1524,3 +1526,114 @@ async def test_only_the_parts_that_were_split_say_so(monkeypatch): assert o.status == "partial" assert o.error == ("这张表 95 列，分 4 段起草，第 3 段（c033 到 c064，共 32 列）、第 4 段（c065 到 c095，共 31 列）没写出来：" + _exhausted() + "；调用模型出错：网关拒绝；其中第 3 段已拆小重试仍没写出。可以只重试这张表") + + +# -------------------------------------------------------------------------- +# 复核后的修正：止损的窗口只记每次调用的最终结局。通信层的重来（服务方嫌额度太大退回 8192、结构化输出不支持或回了 +# 说明文字改走纯文本、429 / 5xx 退避重试）不花额度，也说明不了模型写不出来：照旧报给进度回调（用量、提醒照算）， +# 不进止损的窗口。以前都记成没写出来，OpenAI 兼容网关（额度上限 8192、不支持 json_schema）上并发的头几次调用全是 +# 重来，第一次额度用完就凑满 10 次里 7 次，那一段直接记没写出来、不再拆小。 +# -------------------------------------------------------------------------- + + +def _six_tables() -> dict: + """6 张 40 列的表：每张自成一段，默认并发 3 段同时起草。""" + return {f"t{i}": _table(40, prefix=f"t{i}_") for i in range(6)} + + +def _halved(model: Scripted, table: str) -> bool: + """这张表被拆成两半问过（表级的项加前 20 列那一半）。""" + return any(part == "table+columns" and len(cols or []) == 20 + for call in model.calls for name, (part, cols) in call.asked.items() if name == table) + + +async def test_budget_and_structured_output_fallbacks_do_not_halt_a_concurrent_draft(): + tables = _six_tables() + ok = answering(tables) + spent: set[str] = set() + + async def refuse(call: Call) -> Reply: + await asyncio.sleep(0.01) + raise _Status(400, "max_tokens: 32768 > 8192, which is the maximum allowed number of output tokens") + + async def small(call: Call) -> Reply: + await asyncio.sleep(0.01) + if call.mode == "structured": + raise _Status(400, "This response_format type is unavailable now") + if "t0" in call.asked and "t0" not in spent: + spent.add("t0") + return EXHAUSTED + return ok(call) + + big, sm, reports = Scripted(refuse), Scripted(small), Reports() + got = await D.run_model_draft(catalog.DraftModel(big, sm, "m"), _source(tables), list(tables), + hooks=reports.hooks()) + assert _status(got) == dict.fromkeys(tables, "ok"), {t: o.error for t, o in got.items()} + assert _halved(sm, "t0"), "额度用完的那一段照常拆小重试" + assert len(big.calls) == 3 and reports.failures.count("error") == 6, "退回小额度、改走纯文本的重来照旧报给进度回调" + assert reports.failures.count("exhausted") == 1 + + +async def test_rate_limit_retries_do_not_halt_the_draft(): + tables = _six_tables() + ok = answering(tables) + limited: dict[str, int] = {} + spent: set[str] = set() + + async def handle(call: Call) -> Reply: + await asyncio.sleep(0.01) + name = next(iter(call.asked)) + if name in ("t0", "t1", "t2") and limited.get(name, 0) < 2: + limited[name] = limited.get(name, 0) + 1 + raise _Status(429, "Too Many Requests") + if name == "t0" and not spent: + spent.add(name) + return EXHAUSTED + return ok(call) + + model, reports = Scripted(handle), Reports() + got = await D.run_model_draft(model, _source(tables), list(tables), hooks=reports.hooks()) + assert _status(got) == dict.fromkeys(tables, "ok"), {t: o.error for t, o in got.items()} + assert _halved(model, "t0"), "额度用完的那一段照常拆小重试" + assert reports.failures.count("error") == 6, "429 之后重试成功的那几次照旧报给进度回调" + + +async def test_structured_output_that_comes_back_as_prose_does_not_count_toward_halting(monkeypatch): + """结构化输出回了说明文字、这一块改走纯文本再问：也是通信方式的问题，那一次不进止损的窗口（止损门槛调低到 3 次， + 并发的 3 块都回说明文字也不止损）；纯文本写出来的、没写出来的照常算。""" + monkeypatch.setattr(catalog, "MODEL_HALT_ON", 3) + tables = _six_tables() + ok = answering(tables) + spent: set[str] = set() + + async def handle(call: Call) -> Reply: + await asyncio.sleep(0.01) + if call.mode == "structured": + return Reply(text="好的。这张表记录的是一笔笔业务，各列的含义如下：第一列是编号……") + if "t0" in call.asked and "t0" not in spent: + spent.add("t0") + return EXHAUSTED + return ok(call) + + model, reports = Scripted(handle), Reports() + got = await D.run_model_draft(model, _source(tables), list(tables), hooks=reports.hooks()) + assert _status(got) == dict.fromkeys(tables, "ok"), {t: o.error for t, o in got.items()} + assert _halved(model, "t0") + assert reports.failures.count("error") == 3, "回了说明文字的那几次照旧报给进度回调" + + +async def test_a_call_that_still_fails_after_its_retries_counts_toward_halting(monkeypatch): + """重试用完仍然出错的那一次是这次调用的最终结局，照常进止损的窗口：一直 503 的网关，止损照样起作用。""" + monkeypatch.setattr(catalog, "MODEL_CONCURRENCY", 1) + monkeypatch.setattr(catalog, "MODEL_BATCH_TABLES", 1) + tables = {f"t{i}": _table(10) for i in range(1, 9)} + ok = answering(tables) + + def handle(call: Call) -> Reply: + if set(call.asked) & {"t1", "t2", "t3", "t4", "t5", "t6", "t7"}: + raise _Status(503, "Service Unavailable") + return EXHAUSTED if "t8" in call.asked else ok(call) + + got = await D.run_model_draft(Scripted(handle), _source(tables), list(tables)) + assert got["t7"].error == "调用模型出错：Service Unavailable（已重试 2 次）" + assert got["t8"].error == _exhausted() + HALTED, "最近 10 次的最终结局有 7 次以上没写出来：t8 不再拆小"