mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
数据目录闭环第一步（后端 A）复核修正：按比例止损只看每次调用的最终结局。服务方嫌额度太大退回 8192、结构化输出不支持或回了说明文字改走纯文本、429 / 5xx 退避重试之前的那次照旧记进用量和提醒，不进止损的窗口——它不花额度，也说明不了模型写不出来；重试用完仍出错的那一次照常算。以前额度上限 8192、不支持 json_schema 的 OpenAI 兼容网关上，并发的头几次调用全是这种重来，第一次额度用完就凑满 10 次里 7 次，那一段直接记没写出来、不再拆小
This commit is contained in:
ThearchyHelios
@@ -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
|
||||
|
||||
@@ -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 不再拆小"
|
||||
|
||||
Reference in New Issue
Block a user