diff --git a/backend/app/api/catalog.py b/backend/app/api/catalog.py index a0945853..624a2c12 100644 --- a/backend/app/api/catalog.py +++ b/backend/app/api/catalog.py @@ -144,7 +144,8 @@ def _summary_row(name: str, meta: dict[str, Any] | None, entry: catalog.CatalogE "version": entry.version if entry else 0, "updated_at": _iso(entry.updated_at) if entry else None, "updated_by": entry.updated_by if entry else None, - # 这张表最近一次起草任务里的那一项：{status, job_id, finished_at, error, model_error}，没起草过是 null + # 这张表最近一次起草任务里的那一项：{status, job_id, finished_at, error, model_error, blank_columns}，没起草过 + # 是 null "last_draft": last_draft, } diff --git a/backend/app/data/catalog.py b/backend/app/data/catalog.py index f92f824b..cdb99192 100644 --- a/backend/app/data/catalog.py +++ b/backend/app/data/catalog.py @@ -1463,12 +1463,23 @@ def draft_structure(source: Any, table: str) -> TableDraft: #: 每次调用最多几张表、合计多少列（data/catalog_draft.py 按它分段）。实测每列约 50 个输出 token（不算思考）， #: 会思考的模型思考也算在输出额度里：一段太大，想完就没额度写正文。列数超过 MODEL_BATCH_COLUMNS 的宽表拆成 -#: 「表级一段 + 列分段」，每个列分段也不超过这个列数 +#: 「表级一段 + 列分段」，列分段数是列数除以它向上取整，各段列数平均（相差不超过 1），不留很小的尾段 MODEL_BATCH_TABLES = 4 MODEL_BATCH_COLUMNS = 40 +#: 一次起草里额度用完或超时累计这么多次，之后新开始的段列数上限减半（MODEL_BATCH_COLUMNS → 一半 → …）， +#: 最低 MODEL_MIN_PART_COLUMNS 列 +MODEL_SHRINK_AFTER = 2 +MODEL_MIN_PART_COLUMNS = 10 +#: 一段（连同它拆出来的、补问的、重试的）最多调用几次模型，到了次数还没写完的记没写出来 +MODEL_PART_MAX_CALLS = 8 +#: 提醒：最近 MODEL_NOTICE_WINDOW 次调用里有 MODEL_NOTICE_ON 次以上没写出来时提醒换模型；之后少于 MODEL_NOTICE_OFF +#: 次时清掉 +MODEL_NOTICE_WINDOW = 6 +MODEL_NOTICE_ON = 4 +MODEL_NOTICE_OFF = 2 #: 全进程同时进行的模型调用最多几个（所有起草任务、同步起草共用） MODEL_CONCURRENCY = 3 -#: 单次调用最多等多久（秒）。会思考的模型想得久，一段 40 列常常要几分钟 +#: 单次调用最多等多久（秒）。会思考的模型想得久，一段 40 列常常要几分钟；超时了拆小再问（同额度用完） MODEL_TIMEOUT_S = 420 #: 输出额度。会思考的模型（经 OpenAI 兼容网关的推理模型等）思考也算在里面：8192 时上百列的表常常想完就没额度 #: 写正文。保留思考（不关），加大额度；服务方嫌额度太大时退回 MODEL_FALLBACK_MAX_TOKENS（同 copilot、recipe_ai） @@ -1477,10 +1488,10 @@ MODEL_FALLBACK_MAX_TOKENS = 8192 #: 一段没写出来（额度用完、只给空壳）时拆成两半再问：最多拆几层、拆到最少几列为止 MODEL_SPLIT_DEPTH = 3 MODEL_SPLIT_MIN_COLUMNS = 5 -#: 止损：连续这么多段拆到底仍一项没写出来，这次起草后面的段不再拆小重试（一段 40 列拆到底要 15 次调用，模型明显 -#: 用不了时不该照这个倍数烧额度） +#: 止损：连续这么多段拆到底仍一项没写出来，这次起草后面的段不再拆小重试（模型明显用不了时不该一段一段拆到底烧额度； +#: 每段另有 MODEL_PART_MAX_CALLS 次封顶） MODEL_GIVE_UP_AFTER = 2 -#: 网络错误、429、5xx、超时最多重试几次，每次重试前等多久（秒，依次取用） +#: 网络错误、429、5xx 最多重试几次，每次重试前等多久（秒，依次取用）。超时不原样重试，当作这一段太大、拆小再问 MODEL_RETRIES = 2 MODEL_RETRY_DELAYS_S = (10.0, 30.0) #: 不指定表时起草使用次数最多的前几张 @@ -1794,6 +1805,8 @@ class TableDraftResult: status: str = "ok" #: 模型分几段起草（不用模型时为 None） parts_total: int | None = None + #: 模型拿不准、留空了的列（catalog_draft.ModelOutcome.blank，按表结构的顺序）；表的状态不因此变 + blank: list[str] = field(default_factory=list) @dataclass @@ -1837,6 +1850,7 @@ async def write_draft(session: AsyncSession, source: Any, table: str, modeled: A result.model_error = modeled.error if modeled.items: result.model_items = modeled.items + result.blank = list(getattr(modeled, "blank", None) or []) model_draft = TableDraft(notes=modeled.notes, covered=frozenset({"llm"}) if modeled.status == "ok" else frozenset()) for _attempt in range(3): @@ -2298,8 +2312,10 @@ __all__ = [ "CatalogEntry", "CatalogInvalid", "CatalogModelUnavailable", "CatalogPathError", "DRAFT_DEFAULT_TABLES", "DraftModel", "DraftReport", "INFERRED_MARK", "ITEM_SOURCES", "ITEM_STATUSES", "JoinEdge", "KIND_LABEL", "MEASURE_LABEL", "MODEL_BATCH_COLUMNS", "MODEL_BATCH_TABLES", "MODEL_CONCURRENCY", "MODEL_EMPTY_REASON", - "MODEL_FALLBACK_MAX_TOKENS", "MODEL_MAX_TOKENS", "MODEL_RETRIES", "MODEL_RETRY_DELAYS_S", "MODEL_SPLIT_DEPTH", - "MODEL_SPLIT_MIN_COLUMNS", "MODEL_TIMEOUT_S", "MergeStats", "NEW_RELATION_PATH", "PATCH_MAX_CHANGES", + "MODEL_FALLBACK_MAX_TOKENS", "MODEL_MAX_TOKENS", "MODEL_MIN_PART_COLUMNS", "MODEL_NOTICE_OFF", "MODEL_NOTICE_ON", + "MODEL_NOTICE_WINDOW", "MODEL_PART_MAX_CALLS", "MODEL_RETRIES", "MODEL_RETRY_DELAYS_S", "MODEL_SHRINK_AFTER", + "MODEL_SPLIT_DEPTH", "MODEL_SPLIT_MIN_COLUMNS", "MODEL_TIMEOUT_S", "MergeStats", "NEW_RELATION_PATH", + "PATCH_MAX_CHANGES", "PROFILE_NOTE_PREFIX", "PROFILE_VERIFY_COVERAGE", "WRITE_CONFLICT_ERROR", "default_tables", "missing_table_error", "pick_tables", "write_draft", "PatchChange", "PatchPlan", "QUERY_SNAPSHOT_KIND", "REVIEW_ACTIONS", "TABLE_FIELDS", "TABLE_FIELD_LABEL", diff --git a/backend/app/data/catalog_draft.py b/backend/app/data/catalog_draft.py index bbacc754..90c6b33e 100644 --- a/backend/app/data/catalog_draft.py +++ b/backend/app/data/catalog_draft.py @@ -9,15 +9,21 @@ **分段。** 每次调用最多 MODEL_BATCH_TABLES 张表、合计 MODEL_BATCH_COLUMNS 列：列数不超过它的表按顺序拼在一起 （整表段）；超过它的宽表拆成「表级一段 + 列分段」——表级那段给全部列名和类型做背景、只要表级的项（中文名、粒度、 -表类型、业务主键、业务日期）；列分段每段 MODEL_BATCH_COLUMNS 列，带着表名、表注释和表级起草结果做背景，只要这几列。 -最后合并成这张表的一份，照旧经 catalog._from_model 逐项校验（起草要求不能让模型绕过校验）。 +表类型、业务主键、业务日期）；列分段数是列数除以 MODEL_BATCH_COLUMNS 向上取整、各段列数平均（相差不超过 1，以前 +按 40 列定长切，留下几列的尾段，实测会思考的模型常对这种小段只给空壳），带着表名、表注释和表级起草结果做背景，只要 +这几列。最后合并成这张表的一份，照旧经 catalog._from_model 逐项校验（起草要求不能让模型绕过校验）。 + +**段大小随模型调整。** 一次起草里额度用完或超时累计 MODEL_SHRINK_AFTER 次，之后新开始的段列数上限减半（40 → 20 +→ 10，最低 MODEL_MIN_PART_COLUMNS）：新开始的段（整表段、列分段）超过上限就先按上限均分开再问。已经开始的段照旧 +按失败原因拆小。 **按失败原因处理，不原样重发。** -- 思考加输出用完额度（finish_reason=length、LengthFinishReasonError、正文为空而有思考）、只给空壳：把这一段拆成 - 两半再问（多表的段按表拆，单表的按列拆），最小 MODEL_SPLIT_MIN_COLUMNS 列，最多拆 MODEL_SPLIT_DEPTH 层。 - 拆不开的空壳（结构化输出给的）换纯文本问一次：有的网关的结构化输出只给空壳、纯文本却能写——写出来了，这次起草 - 后面都走纯文本。 +- 思考加输出用完额度（finish_reason=length、LengthFinishReasonError、正文为空而有思考）、只给空壳、超时（当作这一段 + 太大，不原样重试同样大小的段）：把这一段拆成两半再问（多表的段按表拆，单表的按列拆），最小 + MODEL_SPLIT_MIN_COLUMNS 列，最多拆 MODEL_SPLIT_DEPTH 层。拆不开的空壳再问一次：结构化输出给的换纯文本（有的网关的 + 结构化输出只给空壳、纯文本却能写——写出来了，这次起草后面都走纯文本），已经走纯文本的原样再问；仍是空壳才记没写 + 出来。 - 结构化输出回了一段说明文字（不是 JSON、也没停在额度上）：网关接受了结构化输出的参数却没照做（例如开了丢弃不认识 的参数），提示词里又没有格式说明，模型就写成了说明文字。这次起草后面改走纯文本，这一段换纯文本问一次。有思考 不等于额度用完：只有停在额度上、或者正文为空而有思考才算。 @@ -27,20 +33,30 @@ - 结构化输出这个接口不支持（400 / 422 提到 response_format、json_schema、tool……，或者模型类不支持）：这次起草 后面改走纯文本、从正文取 JSON（纯文本那条路另外说清 JSON 的格式）。 - 服务方嫌额度太大：退回 MODEL_FALLBACK_MAX_TOKENS 重来，这次起草后面都用它（catalog.DraftModel.shrink）。 -- 网络错误、429、5xx、超时：等一会儿重试（最多 MODEL_RETRIES 次，退避）。单次调用的时限是 MODEL_TIMEOUT_S。SDK +- 网络错误、429、5xx：等一会儿重试（最多 MODEL_RETRIES 次，退避）。单次调用的时限是 MODEL_TIMEOUT_S。SDK 自己不再重试（catalog.resolve_draft_model 建模型时 max_retries=0），实际重试次数和报错写的一致。 - 别的错误：这一段没写出来，原因照实写。 +- 封顶：一段（连同它拆出来的、补问的、重试的）最多调用 MODEL_PART_MAX_CALLS 次，到了次数还没写完的记没写出来。 - 止损：连续 MODEL_GIVE_UP_AFTER 段拆到底仍一项没写出来（模型明显用不了，例如思考总是吃满额度），这次起草后面的 - 段不再拆小重试、没写出来就直接记下，报错建议更换起草模型；有一段写出来了就恢复。不然一段拆到底要几十次调用， - 起草几十张宽表会烧掉数百万 token、跑几个小时。 + 段不再拆小重试、没写出来就直接记下，报错建议更换起草模型；有一段写出来了就恢复。 +- 提醒：最近 MODEL_NOTICE_WINDOW 次调用里有 MODEL_NOTICE_ON 次以上没写出来（额度用完、超时、空壳、出错），给一句 + 提醒（次数和原因，建议换起草模型、减少表或停止），经进度回调交给任务；之后少于 MODEL_NOTICE_OFF 次时清掉。以前 + 看不出调用在反复失败：一个任务 55 分钟只结束 1 张表，界面上只有段进度和用量。 + +**原因。** 没写出来的原因只写进这张表的 model_error：几段因为同一个原因没写出来时说「这几段」，每段写上列的范围 +（「第 5 段（列名 到 列名，共 33 列）」，列名照表结构原样），真拆小重试过的才补「已拆小重试仍没写出」。 + +**留空的列。** 请模型写了、模型也列出了，中文名和含义却都没给（或者给的都没通过校验）的列是模型拿不准，不补问， +记在 ModelOutcome.blank 里（表的状态不因此变）。已确认的、库注释已经说清的列不算（提示词里它们以此为准）。 **并发。** 全进程最多 MODEL_CONCURRENCY 个调用同时进行（model_gate：按事件循环各一个信号量，所有起草任务和同步 起草共用，先来先放）。一次起草里同时只起草 MODEL_CONCURRENCY 段（一段拆小、补缺都做完再起草下一段），同时在等 名额的调用也最多这么多个：不把几百段一下子都排上去，几次起草（不同数据源的任务、同步起草）的调用交替排队，后开始 -的不会等前一个排着的段全部调完。 +的不会等前一个排着的段全部调完。一次起草里按表的先后排队：宽表的表级那段写完，它的列分段排到已经排着的段后面 +（先来先服务），不插到窄表前面。 -**进度。** 调用方给 Hooks：开始调用、每次调用的用量、一段结束、一张表的模型部分全部结束。后台起草任务据此落库， -每张表一结束就并入目录（data/catalog_jobs.py）。 +**进度。** 调用方给 Hooks：开始调用、每次调用的结果（用量、没写出来的原因、当前的提醒和列数上限）、一段结束、一张表 +的模型部分全部结束。后台起草任务据此落库，每张表一结束就并入目录（data/catalog_jobs.py）。 交给模型的只有表结构（列、类型、主键、外键、唯一约束、注释）和使用者补充的背景，不发任何数据值。写给模型的文字都在 catalog.py 名字带 _SYSTEM / _PROMPT 的常量里；这里的 REASON_* 是给人看的（拼进这张表的 model_error）。 @@ -59,6 +75,7 @@ from dataclasses import dataclass, field from typing import Any from app.data import catalog as C +from app.data.draft_model import DRAFT_MODEL_PLACE from app.data.names import name_key logger = logging.getLogger(__name__) @@ -75,21 +92,33 @@ STRUCTURED_WORDS = ("response_format", "json_schema", "json_object", "tool", "fu #: 一张表的表级的项在「没写出来」记录里的键（不会和列名撞上） _TABLE = "\0table" -# ---- 一段没写出来的原因（给人看，拼进这张表的 model_error） -REASON_EXHAUSTED = "模型的思考和输出用完了额度（{budget} 个 token），这一段没写出来" +# ---- 一段没写出来的原因（给人看，拼进这张表的 model_error）。{seg} 在拼成一句时换成「这一段」或「这几段」：同一个 +#: 原因落在几段上（宽表的几个列分段）说「这几段」，同一段拆出来的几块仍是「这一段」 +SEG = "{seg}" +SEG_ONE = "这一段" +SEG_MANY = "这几段" +REASON_EXHAUSTED = "模型的思考和输出用完了额度（{budget} 个 token），{seg}没写出来" REASON_SPLIT = "；已拆小重试仍没写出" REASON_HOLLOW = "模型只给了空壳，没有写出可用的内容" -REASON_TRUNCATED = "模型的输出被截断，这一段没写完" +REASON_TRUNCATED = "模型的输出被截断，{seg}没写完" REASON_FORMAT = "模型的回复不是约定的格式" REASON_MISSING = "模型的回复里没有这张表" REASON_UNANSWERED = "模型的回复里漏了这些列，补问一次仍没有给出" -REASON_TIMEOUT = "模型 {seconds} 秒内没有写完这一段" +REASON_TIMEOUT = "模型 {seconds} 秒内没有写完{seg}" REASON_CALL = "调用模型出错：{line}" REASON_RETRIED = "（已重试 {n} 次）" REASON_CRASH = "起草这张表时出错：{line}" +#: 一段到了调用次数的上限（MODEL_PART_MAX_CALLS）还没写完 +REASON_CAPPED = "每段最多调用 {n} 次模型，{seg}到了次数仍没写完，已停止重试" #: 止损（连续几段拆到底仍一项没写出来，后面的段不再拆小重试）时补在原因后面 REASON_HALTED = "；模型连续 {n} 段都没写出来，已停止拆小重试，建议更换起草模型" +# ---- 提醒（最近几次调用多数没写出来），经进度回调交给任务（CatalogDraftJob.notice） +NOTICE = ("最近 {calls} 次调用有 {failed} 次没写出来（{kinds}）。可以在{place}换一个模型、减少这次起草的表，" + "或者停止这次起草") +#: 一次调用没写出来的原因种类 → 提醒里的说法（按这个顺序列） +FAILURE_LABEL = {"exhausted": "额度用完", "timeout": "超时", "hollow": "只给空壳", "error": "出错"} + # ========================================================================== # 对外的形状 @@ -111,7 +140,8 @@ class ModelOutcome: status：ok 各段都写出来了；partial 有的段写出来、有的没有（error 写清哪部分、为什么）；failed 一项都没写出来 （error 写原因）。notes 是写出来的那些（已逐项校验，llm / proposed）；items 是项数。parts_total 分几段起草 - （整表段是 1，宽表是 1 + 列分段数），parts_done 已经结束了几段。""" + （整表段是 1，宽表是 1 + 列分段数），parts_done 已经结束了几段。blank：模型拿不准、留空了的列（按表结构的顺序； + 表的状态不因此变）。""" status: str notes: dict[str, Any] @@ -119,6 +149,26 @@ class ModelOutcome: error: str | None parts_total: int parts_done: int + blank: list[str] = field(default_factory=list) + + +@dataclass(frozen=True) +class CallReport: + """一次调用的结果（进度回调 call_done 收到的）。 + + failure：没写出来的原因种类，写出来了是 None——exhausted 额度用完（一项都没捡回来）、timeout 超时、hollow 只给空壳 + （回复里没有可用的内容）、error 出错（会重试的那次也算一次）。超时的调用拿不到用量，token 记 0（网关那边可能已经 + 生成了）。notice、part_columns 是这次调用之后的：提醒（没有为 None）、新开始的段每段最多几列。""" + + input_tokens: int + output_tokens: int + failure: str | None + notice: str | None + part_columns: int + + @property + def timeout(self) -> bool: + return self.failure == "timeout" @dataclass @@ -126,22 +176,37 @@ class Hooks: """进度回调，都是 async，出错只记日志、不影响起草。 - call_started(表名列表)：一次调用拿到并发名额、马上要发出去了； - - usage(输入 token, 输出 token)：每次调用一回（拿不到用量的是 0），条数就是调用次数； + - call_done(CallReport)：每次调用一回（条数就是调用次数），带着用量、没写出来的原因、当前的提醒和列数上限； - part_done(表名)：这张表的一段结束了（写没写出来都算）； - table_done(表名, ModelOutcome)：这张表的模型部分全部结束。""" call_started: Callable[[list[str]], Awaitable[None]] | None = None - usage: Callable[[int, int], Awaitable[None]] | None = None + call_done: Callable[[CallReport], Awaitable[None]] | None = None part_done: Callable[[str], Awaitable[None]] | None = None table_done: Callable[[str, ModelOutcome], Awaitable[None]] | None = None def parts_for(columns: int) -> int: - """一张有这么多列的表分几段起草：不超过 MODEL_BATCH_COLUMNS 列是 1 段，否则是表级一段加列分段。""" + """一张有这么多列的表分几段起草：不超过 MODEL_BATCH_COLUMNS 列是 1 段，否则是表级一段加列分段（列数除以 + MODEL_BATCH_COLUMNS 向上取整）。""" width = max(1, int(C.MODEL_BATCH_COLUMNS)) return 1 if columns <= width else 1 + -(-columns // width) +def _even(columns: Sequence[str], parts: int) -> list[tuple[str, ...]]: + """按顺序切成 parts 段，各段列数平均（相差不超过 1，多出来的放在前面几段）。""" + columns = tuple(columns) + parts = max(1, min(int(parts), len(columns) or 1)) + size, extra = divmod(len(columns), parts) + out: list[tuple[str, ...]] = [] + start = 0 + for i in range(parts): + n = size + (1 if i < extra else 0) + out.append(columns[start:start + n]) + start += n + return out + + async def run_model_draft(model: Any, source: Any, tables: Sequence[str], *, existing: Mapping[str, Mapping[str, Any]] | None = None, background: Background | None = None, @@ -237,25 +302,36 @@ class _Reply: error: str | None = None #: 结构化输出回了一段说明文字（error 是「不是约定的格式」）：换纯文本还可能写出来 prose: bool = False + #: 这次调用超时了（当作这一段太大：拆小再问，不原样重试） + timeout: bool = False + #: 这一段的调用次数用完了，这次没有调用 + capped: bool = False + #: 这次调用的用量 (输入, 输出)，还没报给进度回调；没有调用（capped）的是 None + usage: tuple[int, int] | None = None @dataclass(frozen=True) class _Why: - """一部分没写出来的原因。base 给人看的原因；split 拆小重试过；halted 本来还能拆小，因为前面连续几段都没写出来 - （止损）才没拆。""" + """一部分没写出来的原因。base 给人看的原因（可以带 SEG，拼成一句时换成「这一段」「这几段」）；split 真拆小重试过； + halted 本来还能拆小，因为前面连续几段都没写出来（止损）才没拆。""" base: str split: bool = False halted: bool = False -def _join(whys: Iterable[_Why]) -> str: - """几处没写出来的原因拼成一句：同样的原因只说一次（最多说两种），拆小重试过、止损的说明各在最后补一次（只看说出来 - 的那几种）。不然宽表的表级那段拆不开、列分段拆过时，同一句原因会说两遍。""" - whys = list(whys) - shown = list(dict.fromkeys(w.base for w in whys))[:2] - told = [w for w in whys if w.base in shown] - text = "；".join(shown) +def _join(failures: Iterable[tuple[_Why, int]]) -> str: + """几处没写出来的原因拼成一句。failures 是 (原因, 属于这张表的第几段)。 + + 同样的原因只说一次（最多说两种）；同一个原因落在几段上说「这几段」，同一段拆出来的几块仍是「这一段」；拆小重试过、 + 止损的说明各在最后补一次（只看说出来的那几种）。不然宽表的表级那段拆不开、列分段拆过时，同一句原因会说两遍。""" + failures = list(failures) + parts: dict[str, set[int]] = {} + for why, part in failures: + parts.setdefault(why.base, set()).add(part) + shown = list(parts)[:2] + told = [w for w, _ in failures if w.base in shown] + text = "；".join(base.replace(SEG, SEG_MANY if len(parts[base]) > 1 else SEG_ONE) for base in shown) if any(w.split for w in told): text += REASON_SPLIT if any(w.halted for w in told): @@ -263,6 +339,22 @@ def _join(whys: Iterable[_Why]) -> str: return text +def _columns_text(order: Sequence[str], columns: Iterable[str], *, count: bool) -> str: + """说清是哪几列（列名照表结构原样，按表结构的顺序）：一到三列逐个写；更多的连着就写「头 到 尾」，不连着写前三个加 + 「等」。count：后面带上列数（「，共 33 列」）。""" + pos = {name: i for i, name in enumerate(order)} + cols = sorted(dict.fromkeys(columns), key=lambda c: pos.get(c, len(pos))) + n = len(cols) + if n <= 3: + text = "、".join(cols) + elif pos.get(cols[-1], -1) - pos.get(cols[0], 0) + 1 == n: + text = f"{cols[0]} 到 {cols[-1]}" + else: + head = "、".join(cols[:3]) + return f"{head} 等 {n} 列" if count else f"{head} 等" + return f"{text}，共 {n} 列" if count else text + + class _Cut: """被截断的容器（对象或数组）：value 里只有已经完整的成员，最后一个没写完的容器成员也包成 _Cut 留着，好往里捡。""" @@ -541,6 +633,12 @@ def _text_reply(raw: Any) -> _Reply: return _Reply(exhausted=cut or _thought(raw)) +def _usage_of(raw: Any) -> tuple[int, int]: + """一次调用的用量 (输入, 输出)：拿不到的是 0（测试的假模型没有原始回复）。""" + meta = getattr(raw, "usage_metadata", None) or {} + return int(meta.get("input_tokens") or 0), int(meta.get("output_tokens") or 0) + + def _unpack(out: Any) -> tuple[Any, Any]: """with_structured_output(include_raw=True) 交回 {raw, parsed, parsing_error}；测试的假模型直接交回解析好的对象。""" if isinstance(out, dict) and "raw" in out and ("parsed" in out or "parsing_error" in out): @@ -602,6 +700,12 @@ class _Table: """这一段请写的部分拿到了内容。""" return (piece.table_level and self.got_table) or any(c in self.got_columns for c in piece.columns) + def part_of(self) -> dict[str, int]: + """每一列（表级的项是 _TABLE）属于这张表的第几段。""" + out = {col: piece.part for piece in self.parts for col in piece.columns} + out[_TABLE] = 1 + return out + def _restrict(notes: Mapping[str, Any], piece: _Piece) -> dict[str, Any]: """只留这一段请模型写的部分：没请写表级的项就不要表级的项，列只要请写的那几列。""" @@ -624,6 +728,27 @@ class _Work: then: list[_Work] = field(default_factory=list) +@dataclass +class _Budget: + """一段（新开始的一段，连同它拆出来的、补问的、重试的）还能调用几次模型。""" + + left: int + + +def _pack(pieces: list[_Piece], cap: int) -> list[list[_Piece]]: + """多表的段按顺序重新拼成每段不超过 cap 列（超过 cap 的表自成一段）。""" + groups: list[list[_Piece]] = [] + used = 0 + for piece in pieces: + n = len(piece.columns) + if not groups or used + n > cap: + groups.append([]) + used = 0 + groups[-1].append(piece) + used += n + return groups + + def _halves(pieces: list[_Piece]) -> tuple[list[_Piece], list[_Piece]] | None: """拆成两半：多表的段按表拆；单表的按列拆，两半都至少 MODEL_SPLIT_MIN_COLUMNS 列（表级的项跟着前一半）。拆不开返回 None。""" if len(pieces) > 1: @@ -659,6 +784,12 @@ class _Run: self.gate = asyncio.Semaphore(max(1, int(C.MODEL_CONCURRENCY))) #: 连续几段拆到底仍一项没写出来（止损用，有一段写出来就清零） self.dead_streak = 0 + #: 新开始的段每段最多几列：额度用完或超时（strikes）累计 MODEL_SHRINK_AFTER 次减半一次 + self.part_columns = max(1, int(C.MODEL_BATCH_COLUMNS)) + self.strikes = 0 + #: 最近几次调用没写出来的原因种类（写出来了是 None），提醒看它 + self.recent: collections.deque[str | None] = collections.deque(maxlen=max(1, int(C.MODEL_NOTICE_WINDOW))) + self.notice: str | None = None @property def halted(self) -> bool: @@ -669,8 +800,9 @@ class _Run: async def run(self) -> dict[str, ModelOutcome]: """按计划的先后起草：同时只起草 MODEL_CONCURRENCY 段，一段（连同拆小、补缺）做完再取下一段；宽表的表级那段 - 写完，它的列分段排到最前面。不一下子把所有段（包括宽表的全部列分段）都排上去：那样后开始的另一次起草要等这次 - 排着的段全部调完；拆小重试也排在所有段的第一次调用后面，一段要等几乎所有调用都做完才结束，止损形同虚设。""" + 写完，它的列分段排到队尾（先来先服务：排在已经排着的窄表段后面，不插队）。不一下子把所有段（包括宽表的全部列 + 分段）都排上去：那样后开始的另一次起草要等这次排着的段全部调完；拆小重试也排在所有段的第一次调用后面，一段要 + 等几乎所有调用都做完才结束，止损形同虚设。""" queue: collections.deque[_Work] = collections.deque() for unit in self._plan(): if isinstance(unit, list): @@ -704,7 +836,8 @@ class _Run: return {name: self._outcome(t) for name, t in self.tables.items()} def _plan(self) -> list[list[str] | str]: - """分段：窄表按顺序拼成整表段（列表），宽表一个一项（表名）。各项按第一张表在 tables 里的位置排。""" + """分段：窄表按顺序拼成整表段（列表），宽表一个一项（表名），列分段均分（_even）。各项按第一张表在 tables 里的 + 位置排。""" width = max(1, int(C.MODEL_BATCH_COLUMNS)) most = max(1, int(C.MODEL_BATCH_TABLES)) units: list[list[str] | str] = [] @@ -715,8 +848,7 @@ class _Run: n = len(st.columns) if n > width: st.parts = [_Piece(name, (), True, 1)] + [ - _Piece(name, tuple(st.columns[i:i + width]), False, k + 2) - for k, i in enumerate(range(0, n, width))] + _Piece(name, cols, False, k + 2) for k, cols in enumerate(_even(st.columns, -(-n // width)))] units.append(name) continue st.parts = [_Piece(name, tuple(st.columns), True, 1)] @@ -728,20 +860,40 @@ class _Run: return units async def _work(self, work: _Work, queue: collections.deque[_Work], left: dict[str, int]) -> None: - """起草一段；它之后的段（宽表的列分段）排到最前面；一张表的段都结束了，这张表的模型部分就结束了。""" - await self._guarded(work.pieces) + """起草一段；它之后的段（宽表的列分段）排到队尾；一张表的段都结束了，这张表的模型部分就结束了。""" + await self._start(work.pieces) for piece in work.pieces: await self._part_done(piece.table) - queue.extendleft(reversed(work.then)) + queue.extend(work.then) for piece in work.pieces: left[piece.table] -= 1 if not left[piece.table]: await self._table_done(piece.table) + async def _start(self, pieces: list[_Piece]) -> None: + """开始起草一段。超过当前的列数上限（额度用完、超时多了会减半）的先按上限分开：多表的段按表重新拼，单张表的 + 按列均分（表级的项跟着第一块，先写它，后面几块拿它做背景）。分开的每一块各算一段（各有调用次数的上限）。""" + cap = max(1, self.part_columns) + if sum(len(p.columns) for p in pieces) <= cap: + await self._guarded(pieces) + return + if len(pieces) > 1: + await asyncio.gather(*(self._start(group) for group in _pack(pieces, cap))) + return + piece = pieces[0] + chunks = _even(piece.columns, -(-len(piece.columns) // cap)) + first = [_Piece(piece.table, chunks[0], piece.table_level, piece.part)] + rest = [[_Piece(piece.table, cols, False, piece.part)] for cols in chunks[1:]] + if piece.table_level: + await self._guarded(first) + await asyncio.gather(*(self._guarded(group) for group in rest)) + else: + await asyncio.gather(*(self._guarded(group) for group in [first, *rest])) + async def _guarded(self, pieces: list[_Piece]) -> None: """起草一段；意料之外的错误只记这一段没写出来，不影响别的段和别的表。""" try: - await self._segment(pieces, 0, False) + await self._segment(pieces, 0, False, _Budget(max(1, int(C.MODEL_PART_MAX_CALLS)))) except asyncio.CancelledError: raise except Exception as e: # noqa: BLE001 @@ -762,111 +914,132 @@ class _Run: # ---------------------------------------------------------------- 一段 - async def _segment(self, pieces: list[_Piece], depth: int, followup: bool, *, mode: str | None = None, - last: bool = False) -> None: + async def _segment(self, pieces: list[_Piece], depth: int, followup: bool, budget: _Budget, *, + halved: bool = False, mode: str | None = None, last: bool = False) -> None: """问一段，拿到的并进各表；按失败原因拆开、补缺、记下没写出来的。 depth：已经拆了几层（截断后只补剩下的也算一层）。followup：这一段是在补缺（补过就不再补，见 _after_followup）。 + budget：这一段还能调用几次。halved：这几块是真拆小（_split）出来的——原因里「已拆小重试」只在这时写。 mode / last：换纯文本问的那一次——拆不开的空壳、结构化输出回了说明文字（再不行就记没写出来）。""" - reply = await self._ask(pieces, mode) + reply = await self._ask(pieces, budget, mode) + if reply.capped: + self._fail(pieces, self._capped(halved)) + return + if reply.timeout: + await self._report(reply, wrote=False) + await self._split(pieces, depth, "timeout", followup, budget, halved) + return if reply.error is not None: + await self._report(reply, wrote=False) if reply.prose and not last: - await self._prose_to_text(pieces, depth, followup) + await self._prose_to_text(pieces, depth, followup, budget, halved) return - self._fail(pieces, reply.error) + self._fail(pieces, _Why(reply.error, split=halved)) return gains = self._take(pieces, reply.entries) anything = any(g.items for g in gains) + await self._report(reply, wrote=anything) if last and anything and self.mode != "text": # 结构化输出只给空壳、纯文本写出来了：这个模型的结构化输出靠不住，这次起草后面都走纯文本 self.mode = "text" logger.info("起草数据目录：结构化输出只给了空壳，纯文本写出来了，后面改走纯文本") if reply.exhausted or reply.truncated: - rest = self._remainder(pieces, gains, truncated=True) + rest = self._remainder(pieces, gains, truncated=True, followup=followup) if not rest: return kind = "exhausted" if reply.exhausted else "truncated" if not anything: - await self._split(rest, depth, kind, followup) + await self._split(rest, depth, kind, followup, budget, halved) elif depth < int(C.MODEL_SPLIT_DEPTH): - await self._segment(rest, depth + 1, followup) # 捡回了一部分：只补剩下的 + # 捡回了一部分：只补剩下的（不是拆小） + await self._segment(rest, depth + 1, followup, budget, halved=halved) else: - self._fail(rest, self._reason(kind, split=True)) + self._fail(rest, self._reason(kind, split=halved)) return if not anything and not followup: if last: halted = self._stopped(pieces, depth) - self._fail(pieces, lambda p: self._reason("hollow", split=depth > 0, piece=p, halted=halted)) + self._fail(pieces, lambda p: self._reason("hollow", split=halved, piece=p, halted=halted)) return - await self._split(pieces, depth, "hollow", followup) + await self._split(pieces, depth, "hollow", followup, budget, halved) return - rest = self._remainder(pieces, gains, truncated=False) + rest = self._remainder(pieces, gains, truncated=False, followup=followup) if not rest: return if followup: - await self._after_followup(rest, depth, last=last) + await self._after_followup(rest, depth, budget, halved, last=last) return - await self._segment(rest, depth, True) + await self._segment(rest, depth, True, budget, halved=halved) - async def _after_followup(self, rest: list[_Piece], depth: int, *, last: bool) -> None: + async def _after_followup(self, rest: list[_Piece], depth: int, budget: _Budget, halved: bool, *, + last: bool) -> None: """补问过仍然没给的部分（补过就不再补）。 - - 这张表已经拿到了内容：没列出来的列记没写出来（这张表是 partial，旧的模型起草项不删）。模型列出了、字段留空 - 的列是拿不准，不在这里（_remainder 只把没列出来的列交过来）； - - 这张表到现在还一项都没有（回复里没有这张表、只给了空壳）：照第一次只给空壳的规矩，拆小再问，拆不开换纯文本 - 问一次。拆出来的半段是新的一段（还能补缺一次），层数照算。""" + - 这张表已经拿到了内容：补问时仍没列出来的列记没写出来（这张表是 partial）。补问时列出了、字段留空的列是拿不准 + （记留空），不在这里（_remainder 只把没列出来的列交过来）； + - 这张表到现在还一项都没有（回复里没有这张表、只给了空壳）：照第一次只给空壳的规矩，拆小再问，拆不开再问一次。 + 拆出来的半段是新的一段（还能补缺一次），层数照算。""" given = [p for p in rest if self.tables[p.table].has_any()] empty = [p for p in rest if not self.tables[p.table].has_any()] if given: - self._fail(given, _Why(REASON_UNANSWERED)) + self._fail(given, _Why(REASON_UNANSWERED, split=halved)) if not empty: return if last: halted = self._stopped(empty, depth) - self._fail(empty, lambda p: self._reason("hollow", split=depth > 0, piece=p, halted=halted)) + self._fail(empty, lambda p: self._reason("hollow", split=halved, piece=p, halted=halted)) return - await self._split(empty, depth, "hollow", False) + await self._split(empty, depth, "hollow", False, budget, halved) - async def _prose_to_text(self, pieces: list[_Piece], depth: int, followup: bool) -> None: + async def _prose_to_text(self, pieces: list[_Piece], depth: int, followup: bool, budget: _Budget, + halved: bool) -> None: """结构化输出回了一段说明文字：网关接受了结构化输出的参数却没照做。这次起草后面都走纯文本（另带格式说明）， 这一段换纯文本问一次。纯文本额度用完了还能照常拆小（拆出来的半段也走纯文本，不再先问一遍结构化输出）。""" if self.mode != "text": self.mode = "text" logger.info("起草数据目录：结构化输出回了说明文字，后面改走纯文本") - await self._segment(pieces, depth, followup, mode="text", last=True) + await self._segment(pieces, depth, followup, budget, halved=halved, mode="text", last=True) def _stopped(self, pieces: list[_Piece], depth: int) -> bool: """这几段本来还能拆小，因为前面连续几段都没写出来（止损）才没拆。""" return self.halted and depth < int(C.MODEL_SPLIT_DEPTH) and _halves(pieces) is not None - async def _split(self, pieces: list[_Piece], depth: int, kind: str, followup: bool) -> None: + async def _split(self, pieces: list[_Piece], depth: int, kind: str, followup: bool, budget: _Budget, + halved: bool) -> None: + """拆成两半再问（额度用完、超时、只给空壳、截断后一项都没捡回来）。拆不开的空壳再问一次：结构化输出给的换纯文本， + 已经走纯文本的原样再问；别的原因拆不开就记没写出来。""" halves = _halves(pieces) if depth < int(C.MODEL_SPLIT_DEPTH) else None halted = halves is not None and self.halted if halves is not None and not halted: first, second = halves if (len(first) == len(second) == 1 and first[0].table == second[0].table and first[0].table_level): # 同一张表按列拆、前一半带着表级的项：先写前一半，后一半拿它做背景 - await self._segment(first, depth + 1, followup) - await self._segment(second, depth + 1, followup) + await self._segment(first, depth + 1, followup, budget, halved=True) + await self._segment(second, depth + 1, followup, budget, halved=True) return - results = await asyncio.gather(self._segment(first, depth + 1, followup), - self._segment(second, depth + 1, followup), return_exceptions=True) + results = await asyncio.gather(self._segment(first, depth + 1, followup, budget, halved=True), + self._segment(second, depth + 1, followup, budget, halved=True), + return_exceptions=True) for result in results: if isinstance(result, BaseException): raise result return - if kind == "hollow" and self.mode == "structured": - await self._segment(pieces, depth, followup, mode="text", last=True) + if kind == "hollow": + await self._segment(pieces, depth, followup, budget, halved=halved, mode="text", last=True) return - self._fail(pieces, lambda p: self._reason(kind, split=depth > 0, piece=p, halted=halted)) + self._fail(pieces, lambda p: self._reason(kind, split=halved, piece=p, halted=halted)) - def _remainder(self, pieces: list[_Piece], gains: list[_Gain], *, truncated: bool) -> list[_Piece]: + def _remainder(self, pieces: list[_Piece], gains: list[_Gain], *, truncated: bool, + followup: bool) -> list[_Piece]: """还没写出来的部分：回复里没有这张表、或者这张表一项可用内容都没有的整段重问；有内容的只问没列出来的列 （列出了、字段留空的算模型拿不准，不再问）。截断的回复里表级的项也没拿到的，表级的项一起重问（没截断时表级 - 空着算模型拿不准：这张表在回复里，表级的字段留空和列的字段留空一样）。""" + 空着算模型拿不准：这张表在回复里，表级的字段留空和列的字段留空一样）。 + + 补问（followup）时这张表已经有内容了：补问的列列出来了就算答过（字段留空是拿不准），只交回没列出来的列。""" out: list[_Piece] = [] for piece, gain in zip(pieces, gains): - if not gain.named or not gain.items: + answered_blank = followup and gain.named and self.tables[piece.table].has_any() + if not gain.named or (not gain.items and not answered_blank): out.append(piece) continue cols = tuple(c for c in piece.columns if c not in gain.answered) @@ -877,40 +1050,49 @@ class _Run: # ---------------------------------------------------------------- 调用 - async def _ask(self, pieces: list[_Piece], mode: str | None = None) -> _Reply: - """调一次模型（按原因重来：额度被拒、结构化输出不支持、网络错误……），交回读好的回复。""" + async def _ask(self, pieces: list[_Piece], budget: _Budget, mode: str | None = None) -> _Reply: + """调一次模型（按原因重来：额度被拒、结构化输出不支持、网络错误……），交回读好的回复。 + + 这一段的调用次数用完了交回 capped（不调用）。超时不原样重试：交回 timeout，由调用方当作这一段太大、拆小再问。 + 重来之前的那次调用在这里就报给进度回调；交回的那次由调用方读完内容再报（写没写出来要看校验之后的内容）。""" mode = mode or self.mode retried = 0 while True: + if budget.left <= 0: + return _Reply(capped=True) + budget.left -= 1 used = getattr(self.model, "budget", None) try: # 先占这次起草自己的名额，再去排全进程的：一次起草排在全进程名额上的调用最多 MODEL_CONCURRENCY 个 async with self.gate, model_gate(): await self._hook(self.hooks.call_started, [p.table for p in pieces]) reply, raw = await asyncio.wait_for(self._call(pieces, mode), timeout=C.MODEL_TIMEOUT_S) - # 用量记在时限之外：落库慢了不能把已经回来的这次调用算成超时 - await self._usage(raw) + reply.usage = _usage_of(raw) return reply except asyncio.CancelledError: raise except Exception as e: # noqa: BLE001 - 按原因分别处理 kind = classify_failure(e) if kind == "length": - return await self._length_reply(e) - await self._hook(self.hooks.usage, 0, 0) + return self._length_reply(e) + 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") 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") continue - if kind in ("transient", "timeout") and retried < int(C.MODEL_RETRIES): + if kind == "transient" and retried < int(C.MODEL_RETRIES): + await self._called((0, 0), "error") delays = tuple(C.MODEL_RETRY_DELAYS_S) or (0.0,) await _sleep(delays[min(retried, len(delays) - 1)]) retried += 1 continue - return _Reply(error=self._call_error(e, kind, retried)) + return _Reply(error=self._call_error(e, retried), usage=(0, 0)) async def _call(self, pieces: list[_Piece], mode: str) -> tuple[_Reply, Any]: """一次调用：(读好的回复, 原始回复)。原始回复给用量用（测试的假模型没有，是 None）。""" @@ -922,12 +1104,12 @@ class _Run: raw = await self.model.ainvoke(messages) return _text_reply(raw), raw - async def _length_reply(self, e: BaseException) -> _Reply: - """LengthFinishReasonError：额度用完了。异常里带着那次的回复，能捡回来的照捡。""" + @staticmethod + def _length_reply(e: BaseException) -> _Reply: + """LengthFinishReasonError：额度用完了。异常里带着那次的回复（和用量），能捡回来的照捡。""" completion = getattr(e, "completion", None) usage = getattr(completion, "usage", None) - await self._hook(self.hooks.usage, int(getattr(usage, "prompt_tokens", 0) or 0), - int(getattr(usage, "completion_tokens", 0) or 0)) + tokens = (int(getattr(usage, "prompt_tokens", 0) or 0), int(getattr(usage, "completion_tokens", 0) or 0)) try: text = completion.choices[0].message.content or "" # type: ignore[union-attr] except (AttributeError, IndexError, TypeError): @@ -935,8 +1117,8 @@ class _Run: if text.strip(): reply = _parse_text(text, True) if reply is not None and reply.entries: - return _Reply(entries=reply.entries, truncated=True, exhausted=True) - return _Reply(exhausted=True) + return _Reply(entries=reply.entries, truncated=True, exhausted=True, usage=tokens) + return _Reply(exhausted=True, usage=tokens) def _shrink(self) -> bool: shrink = getattr(self.model, "shrink", None) @@ -948,17 +1130,50 @@ class _Run: def _budget(self) -> int: return int(getattr(self.model, "budget", None) or C.MODEL_MAX_TOKENS) - def _call_error(self, e: BaseException, kind: str, retried: int) -> str: + @staticmethod + def _call_error(e: BaseException, retried: int) -> str: from app.core.errors import describe_exception, first_line tail = REASON_RETRIED.format(n=retried) if retried else "" - if kind == "timeout": - return REASON_TIMEOUT.format(seconds=_seconds(C.MODEL_TIMEOUT_S)) + tail return REASON_CALL.format(line=first_line(e) or describe_exception(e)) + tail - async def _usage(self, raw: Any) -> None: - meta = getattr(raw, "usage_metadata", None) or {} - await self._hook(self.hooks.usage, int(meta.get("input_tokens") or 0), int(meta.get("output_tokens") or 0)) + # ---------------------------------------------------------------- 每次调用之后：用量、段大小、提醒 + + async def _report(self, reply: _Reply, *, wrote: bool) -> None: + """交回的那次调用读完内容之后报给进度回调。""" + if reply.usage is None: + return + if reply.timeout: + failure: str | None = "timeout" + elif reply.error is not None: + failure = "error" + elif wrote: + failure = None + else: + failure = "exhausted" if reply.exhausted else "hollow" + await self._called(reply.usage, failure, strike=reply.timeout or reply.exhausted) + + async def _called(self, usage: tuple[int, int], failure: str | None, *, strike: bool = False) -> None: + """记下一次调用：额度用完或超时（strike）累计 MODEL_SHRINK_AFTER 次，新开始的段列数上限减半；最近几次调用多数 + 没写出来时提醒，少了清掉。报给进度回调。""" + self.recent.append(failure) + if strike: + self.strikes += 1 + width = max(1, int(C.MODEL_BATCH_COLUMNS)) + floor = min(width, max(1, int(C.MODEL_MIN_PART_COLUMNS))) + cap = max(floor, width >> (self.strikes // max(1, int(C.MODEL_SHRINK_AFTER)))) + if cap < self.part_columns: + self.part_columns = cap + logger.info("起草数据目录：额度用完或超时已有 %d 次，之后新开始的段每段最多 %d 列", self.strikes, cap) + failed = [f for f in self.recent if f is not None] + if len(failed) >= int(C.MODEL_NOTICE_ON) or (self.notice is not None and len(failed) >= int(C.MODEL_NOTICE_OFF)): + counts = collections.Counter(failed) + kinds = "、".join(f"{label} {counts[kind]} 次" for kind, label in FAILURE_LABEL.items() if counts[kind]) + self.notice = NOTICE.format(calls=len(self.recent), failed=len(failed), kinds=kinds, place=DRAFT_MODEL_PLACE) + else: + self.notice = None + await self._hook(self.hooks.call_done, CallReport(int(usage[0]), int(usage[1]), failure, self.notice, + self.part_columns)) # ---------------------------------------------------------------- 拿到的内容 @@ -1028,9 +1243,11 @@ class _Run: def _reason(self, kind: str, *, split: bool, piece: _Piece | None = None, halted: bool = False) -> _Why: if kind == "exhausted": - base = REASON_EXHAUSTED.format(budget=self._budget()) + base = REASON_EXHAUSTED.format(budget=self._budget(), seg=SEG) elif kind == "truncated": - base = REASON_TRUNCATED + base = REASON_TRUNCATED.format(seg=SEG) + elif kind == "timeout": + base = REASON_TIMEOUT.format(seconds=_seconds(C.MODEL_TIMEOUT_S), seg=SEG) else: st = self.tables[piece.table] if piece is not None else None if st is not None and st.whole(piece): @@ -1039,6 +1256,10 @@ class _Run: base = REASON_HOLLOW return _Why(base, split=split, halted=halted) + @staticmethod + def _capped(halved: bool) -> _Why: + return _Why(REASON_CAPPED.format(n=max(1, int(C.MODEL_PART_MAX_CALLS)), seg=SEG), split=halved) + # ---------------------------------------------------------------- 收尾 def _outcome(self, st: _Table) -> ModelOutcome: @@ -1046,31 +1267,57 @@ class _Run: failed_cols = [c for c in st.columns if c in st.failed and c not in st.got_columns] failed_table = _TABLE in st.failed and not st.got_table total = len(st.parts) or 1 + part_of = st.part_of() if not items: - whys = list(st.failed.values()) - reason = _join(whys) if whys else (C.MODEL_EMPTY_REASON if st.named else REASON_MISSING) + failures = [(why, part_of.get(key, 1)) for key, why in st.failed.items()] + reason = _join(failures) if failures else (C.MODEL_EMPTY_REASON if st.named else REASON_MISSING) return ModelOutcome("failed", {}, 0, reason, total, st.parts_done) notes = copy.deepcopy(st.notes) + blank = self._blank(st, set(failed_cols)) if failed_cols or failed_table: return ModelOutcome("partial", notes, items, self._partial_text(st, failed_cols, failed_table), total, - st.parts_done) - return ModelOutcome("ok", notes, items, None, total, st.parts_done) + st.parts_done, blank) + return ModelOutcome("ok", notes, items, None, total, st.parts_done, blank) + + def _blank(self, st: _Table, failed: set[str]) -> list[str]: + """模型拿不准、留空了的列：请它写了（不在没写出来的里面），中文名和含义却都没给（或者给的都没通过校验）。已确认 + 的、库注释已经说清的不算——提示词里它们以此为准，模型不写也不缺：已确认（有确证）的中文名或含义、库注释（按 + catalog.comment_role 记进目录的那种；导入表格的源是系统说明）。""" + drafted = st.notes.get("columns") or {} + confirmed = C._confirmed_only(st.existing).get("columns") or {} + comments = {str(c.get("name")): (c.get("comment") or "").strip() + for c in st.meta.get("columns") or [] if isinstance(c, dict)} + out: list[str] = [] + for col in st.columns: + if col in failed: + continue + if any(f in (drafted.get(col) or {}) or f in (confirmed.get(col) or {}) for f in ("label", "meaning")): + continue + text = comments.get(col) + if text and (self.system or C.comment_role(text, col) is not None): + continue + out.append(col) + return out @staticmethod def _partial_text(st: _Table, failed_cols: list[str], failed_table: bool) -> str: - """只写出一部分时说清哪部分没写出来、为什么、怎么办。""" + """只写出一部分时说清哪部分没写出来（第几段、哪几列）、为什么、怎么办。""" + part_of = st.part_of() keys = ([_TABLE] if failed_table else []) + failed_cols - reason = _join(st.failed[k] for k in keys) + reason = _join((st.failed[k], part_of.get(k, 1)) for k in keys) n = len(st.columns) if len(st.parts) > 1: - part_of = {c: p.part for p in st.parts for c in p.columns} - bad = sorted({part_of[c] for c in failed_cols if c in part_of} | ({1} if failed_table else set())) - head = f"这张表 {n} 列，分 {len(st.parts)} 段起草，第 {'、'.join(str(b) for b in bad)} 段没写出来" - head += f"（共 {len(failed_cols)} 列）" if failed_cols else "（表级的项）" + bad: dict[int, list[str]] = {} + for col in failed_cols: + bad.setdefault(part_of.get(col, 1), []).append(col) + said = ["第 1 段（表级的项）"] if failed_table else [] + said += [f"第 {k} 段（{_columns_text(st.columns, cols, count=True)}）" for k, cols in sorted(bad.items())] + head = f"这张表 {n} 列，分 {len(st.parts)} 段起草，{'、'.join(said)}没写出来" elif failed_cols and failed_table: - head = f"这张表 {n} 列，有表级的项和 {len(failed_cols)} 列没写出来" + head = (f"这张表 {n} 列，有表级的项和 {len(failed_cols)} 列（{_columns_text(st.columns, failed_cols, count=False)}）" + "没写出来") elif failed_cols: - head = f"这张表 {n} 列，有 {len(failed_cols)} 列没写出来" + head = f"这张表 {n} 列，有 {len(failed_cols)} 列（{_columns_text(st.columns, failed_cols, count=False)}）没写出来" else: head = f"这张表 {n} 列，表级的项没写出来" return f"{head}：{reason}。可以只重试这张表" @@ -1196,6 +1443,7 @@ class _Run: return C._notes_lines(confirmed, meta=st.meta, system_notes=False) -__all__ = ["BUDGET_WORDS", "Background", "Hooks", "ModelOutcome", "REASON_CALL", "REASON_EXHAUSTED", "REASON_FORMAT", - "REASON_HALTED", "REASON_HOLLOW", "REASON_MISSING", "REASON_RETRIED", "REASON_SPLIT", "REASON_TIMEOUT", - "REASON_TRUNCATED", "REASON_UNANSWERED", "classify_failure", "model_gate", "parts_for", "run_model_draft"] +__all__ = ["BUDGET_WORDS", "Background", "CallReport", "FAILURE_LABEL", "Hooks", "ModelOutcome", "NOTICE", "REASON_CALL", + "REASON_CAPPED", "REASON_EXHAUSTED", "REASON_FORMAT", "REASON_HALTED", "REASON_HOLLOW", "REASON_MISSING", + "REASON_RETRIED", "REASON_SPLIT", "REASON_TIMEOUT", "REASON_TRUNCATED", "REASON_UNANSWERED", "SEG", + "classify_failure", "model_gate", "parts_for", "run_model_draft"] diff --git a/backend/app/data/catalog_jobs.py b/backend/app/data/catalog_jobs.py index 4c580e8b..d708ed50 100644 --- a/backend/app/data/catalog_jobs.py +++ b/backend/app/data/catalog_jobs.py @@ -6,8 +6,9 @@ - **建任务**（create_job）：同一个数据源同时只有一个排队或进行中的任务（DraftBusy；部分唯一索引兜底）。不指定表时 按使用次数取前 20 张；表结构里没有的表照样建进任务、状态 skipped。起草要求（数据源级）和本次说明随任务记下快照。 - **在后台起草**（manager.start）：进程内的 asyncio 任务，引擎是 data/catalog_draft.py（分段、按失败原因处理）。 - 每张表一结束就并入目录（catalog.write_draft，规则和同步起草一样：人工确认、驳回过的不动，乐观锁冲突重读重并）； - 逐表的状态、到第几段、计数、用量随时落库，接口读的是库里的状态。 + 每张表一结束就并入目录（catalog.write_draft，规则和同步起草一样：人工确认、驳回过的不动，模型这次没给的项不删 + 以前起草的，乐观锁冲突重读重并）；逐表的状态、到第几段、计数、模型留空的列，任务的用量（含超时、没写出来的调用 + 次数）、当前的列数上限、提醒随时落库，接口读的是库里的状态。 - **停止**（cancel）：排队的直接 cancelled；进行中的停掉正在进行的模型调用，没开始和没写完的表记 cancelled，已经写进 目录的不回滚，任务最后是 cancelled。 - **重试**（retry_tables + create_job）：新任务，只含原任务里 failed、partial、cancelled 的表；用不用模型、本次说明沿用， @@ -55,6 +56,8 @@ GUIDANCE_MAX = 2000 NOTE_MAX = 1000 #: 目录清单上「这张表最近一次起草」只看最近这么多个任务 LAST_DRAFT_JOBS = 100 +#: 逐表最多记几个模型留空的列名 +BLANK_SAMPLE = 5 #: 停止进行中的任务时最多等它收尾多久（秒）；关停服务时另给（shutdown 的 grace） STOP_GRACE_S = 10.0 @@ -164,6 +167,9 @@ def task_dict(task: CatalogDraftTask) -> dict[str, Any]: "removed": task.removed, "version": task.version, "model_items": task.model_items, + # 模型拿不准、留空了的列（中文名和含义都没给）：几列、最多 5 个列名（按表结构的顺序）。表的状态不因此变 + "blank_columns": task.blank_columns or 0, + "blank_sample": list(task.blank_sample or []), "error": task.error, "model_error": task.model_error, "started_at": _iso(task.started_at), @@ -200,8 +206,14 @@ def job_dict(job: CatalogDraftJob, tasks: list[CatalogDraftTask]) -> dict[str, A "added": sum(t.added or 0 for t in tasks), "updated": sum(t.updated or 0 for t in tasks), "removed": sum(t.removed or 0 for t in tasks), + "blank_columns": sum(t.blank_columns or 0 for t in tasks), + # 超时的调用拿不到用量，token 按 0 算（网关那边可能已经生成了）；failed_calls 含额度用完、只给空壳、超时、出错 "usage": {"calls": job.calls or 0, "input_tokens": job.input_tokens or 0, - "output_tokens": job.output_tokens or 0}, + "output_tokens": job.output_tokens or 0, "timeouts": job.timeouts or 0, + "failed_calls": job.failed_calls or 0}, + # 新开始的段每段最多几列（不用模型、还没开始时为 null）；提醒：最近几次调用多数没写出来时的一句话，没有为 null + "part_columns": job.part_columns, + "notice": job.notice, "error": job.error, "tasks": [task_dict(t) for t in tasks], } @@ -225,20 +237,22 @@ async def job_brief(session: AsyncSession, job: CatalogDraftJob | None) -> dict[ async def last_drafts(session: AsyncSession, source_id: str) -> dict[str, dict[str, Any]]: - """每张表最近一次起草任务里的那一项：{表名: {status, job_id, finished_at, error, model_error}}。任务还在进行， - 就是进行中的状态。只看最近 LAST_DRAFT_JOBS 个任务；每张表取哪一项在库里挑好（清单页起草时会反复取）。""" + """每张表最近一次起草任务里的那一项：{表名: {status, job_id, finished_at, error, model_error, blank_columns}}。 + 任务还在进行，就是进行中的状态。只看最近 LAST_DRAFT_JOBS 个任务；每张表取哪一项在库里挑好（清单页起草时会反复取）。""" recent = (select(CatalogDraftJob.id).where(CatalogDraftJob.source_id == source_id) .order_by(CatalogDraftJob.created_at.desc()).limit(LAST_DRAFT_JOBS)).scalar_subquery() rank = func.row_number().over(partition_by=CatalogDraftTask.table_name, order_by=(CatalogDraftJob.created_at.desc(), CatalogDraftTask.seq)).label("rank") inner = (select(CatalogDraftTask.table_name, CatalogDraftTask.status, CatalogDraftTask.job_id, - CatalogDraftTask.finished_at, CatalogDraftTask.error, CatalogDraftTask.model_error, rank) + CatalogDraftTask.finished_at, CatalogDraftTask.error, CatalogDraftTask.model_error, + CatalogDraftTask.blank_columns, rank) .join(CatalogDraftJob, CatalogDraftJob.id == CatalogDraftTask.job_id) .where(CatalogDraftJob.id.in_(recent))).subquery() rows = (await session.execute(select(inner).where(inner.c.rank == 1))).mappings() return {row["table_name"]: {"status": row["status"], "job_id": row["job_id"], "finished_at": _iso(row["finished_at"]), "error": row["error"], - "model_error": row["model_error"]} for row in rows} + "model_error": row["model_error"], "blank_columns": row["blank_columns"] or 0} + for row in rows} # ========================================================================== @@ -463,7 +477,7 @@ class _JobRun: await self._write(self._skip, gone) if use_model and names: await self._write(self._set_model, model_id) - hooks = catalog_draft.Hooks(call_started=self._call_started, usage=self._usage, + hooks = catalog_draft.Hooks(call_started=self._call_started, call_done=self._call_done, part_done=self._part_done, table_done=self._table_done) await catalog_draft.run_model_draft(model, source, names, existing=existing, background=background, hooks=hooks) @@ -482,8 +496,8 @@ class _JobRun: self._started.update(fresh) await self._write(self._mark_running, fresh) - async def _usage(self, input_tokens: int, output_tokens: int) -> None: - await self._write(self._add_usage, input_tokens, output_tokens) + async def _call_done(self, report: catalog_draft.CallReport) -> None: + await self._write(self._add_call, report) async def _part_done(self, table: str) -> None: await self._write(self._bump_part, table) @@ -506,7 +520,8 @@ class _JobRun: async def _set_model(self, session: AsyncSession, model_id: str) -> None: await session.execute(update(CatalogDraftJob).where(CatalogDraftJob.id == self.job_id) - .values(model=model_id or None).execution_options(synchronize_session=False)) + .values(model=model_id or None, part_columns=max(1, int(catalog.MODEL_BATCH_COLUMNS))) + .execution_options(synchronize_session=False)) await session.commit() async def _skip(self, session: AsyncSession, names: list[str]) -> None: @@ -521,11 +536,16 @@ class _JobRun: .values(status="running", started_at=utcnow())) await session.commit() - async def _add_usage(self, session: AsyncSession, input_tokens: int, output_tokens: int) -> None: + async def _add_call(self, session: AsyncSession, report: catalog_draft.CallReport) -> None: + """一次调用：用量、超时、没写出来的次数累加，当前的列数上限和提醒照这次调用之后的写（按调用的先后写进来）。""" await session.execute( update(CatalogDraftJob).where(CatalogDraftJob.id == self.job_id) - .values(calls=CatalogDraftJob.calls + 1, input_tokens=CatalogDraftJob.input_tokens + int(input_tokens), - output_tokens=CatalogDraftJob.output_tokens + int(output_tokens)) + .values(calls=CatalogDraftJob.calls + 1, + input_tokens=CatalogDraftJob.input_tokens + int(report.input_tokens), + output_tokens=CatalogDraftJob.output_tokens + int(report.output_tokens), + timeouts=func.coalesce(CatalogDraftJob.timeouts, 0) + int(report.timeout), + failed_calls=func.coalesce(CatalogDraftJob.failed_calls, 0) + int(report.failure is not None), + part_columns=report.part_columns, notice=report.notice) .execution_options(synchronize_session=False)) await session.commit() @@ -551,6 +571,7 @@ class _JobRun: "status": result.status, "added": result.added, "updated": result.updated, "removed": result.removed, "version": result.version, "model_items": result.model_items, "error": result.error, "model_error": result.model_error, "finished_at": utcnow(), + "blank_columns": len(result.blank), "blank_sample": result.blank[:BLANK_SAMPLE] or None, } if outcome is not None: values["parts_done"] = outcome.parts_done @@ -638,8 +659,9 @@ class DraftJobs: manager = DraftJobs() -__all__ = ["ACTIVE", "DraftBusy", "DraftFinished", "DraftJobs", "FINISHED", "GUIDANCE_MAX", "INTERRUPTED_ERROR", - "MAX_TABLES", "NOTE_MAX", "NothingToRetry", "RETRYABLE", "TASK_DONE", "TASK_STATUSES", "TextTooLong", +__all__ = ["ACTIVE", "BLANK_SAMPLE", "DraftBusy", "DraftFinished", "DraftJobs", "FINISHED", "GUIDANCE_MAX", + "INTERRUPTED_ERROR", "MAX_TABLES", "NOTE_MAX", "NothingToRetry", "RETRYABLE", "TASK_DONE", "TASK_STATUSES", + "TextTooLong", "active_job", "cancel", "check_model", "clean_text", "create_job", "ensure_idle", "guidance_of", "job_brief", "job_dict", "job_out", "last_drafts", "latest_job", "load", "manager", "retry_tables", "save_guidance", "startup", "tasks_of"] diff --git a/backend/app/db/base.py b/backend/app/db/base.py index 68ee478b..f0da4ac6 100644 --- a/backend/app/db/base.py +++ b/backend/app/db/base.py @@ -204,6 +204,14 @@ _COLUMN_MIGRATIONS: list[tuple[str, str, str]] = [ # 没有选项，也不是定时任务发起的 ("runs", "options", "JSON DEFAULT '{}'"), ("runs", "schedule_id", "VARCHAR(32)"), + # 数据目录起草任务（验收后修正）：超时、没写出来的调用次数，当前的列数上限，提醒；逐表的模型留空的列。 + # 存量任务回填 0 和 NULL：升级前没记这些 + ("catalog_draft_jobs", "timeouts", "INTEGER DEFAULT 0"), + ("catalog_draft_jobs", "failed_calls", "INTEGER DEFAULT 0"), + ("catalog_draft_jobs", "part_columns", "INTEGER"), + ("catalog_draft_jobs", "notice", "TEXT"), + ("catalog_draft_tasks", "blank_columns", "INTEGER DEFAULT 0"), + ("catalog_draft_tasks", "blank_sample", "JSON"), ] #: 迁移补上的列要建的索引：(索引名, 表, 列)。create_all 不碰已有的表，表上新列的索引也就不会建。 diff --git a/backend/app/db/models.py b/backend/app/db/models.py index 932d02a7..a5438d84 100644 --- a/backend/app/db/models.py +++ b/backend/app/db/models.py @@ -379,10 +379,17 @@ class CatalogDraftJob(Base, TimestampMixin): created_by: Mapped[str | None] = mapped_column(String(100), default=None) started_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None) finished_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None) - # 用量：调用次数、输入和输出 token（拿不到用量的调用按 0 算） + # 用量：调用次数、输入和输出 token（拿不到用量的调用按 0 算，超时的调用也是 0）、超时几次、没写出来几次（额度用完、 + # 只给空壳、超时、出错） calls: Mapped[int] = mapped_column(Integer, default=0) input_tokens: Mapped[int] = mapped_column(Integer, default=0) output_tokens: Mapped[int] = mapped_column(Integer, default=0) + timeouts: Mapped[int] = mapped_column(Integer, default=0) + failed_calls: Mapped[int] = mapped_column(Integer, default=0) + # 新开始的段每段最多几列（额度用完、超时多了会减半；用模型时开始起草后写上） + part_columns: Mapped[int | None] = mapped_column(Integer, default=None) + # 提醒：最近几次调用多数没写出来时的一句话（建议换起草模型、减少表或停止），少了清掉 + notice: Mapped[str | None] = mapped_column(Text, default=None) # 任务整体出错的原因（逐表的原因在 catalog_draft_tasks） error: Mapped[str | None] = mapped_column(Text, default=None) # 重试建的任务记下是重试哪一个 @@ -411,6 +418,9 @@ class CatalogDraftTask(Base): removed: Mapped[int] = mapped_column(Integer, default=0) version: Mapped[int] = mapped_column(Integer, default=0) model_items: Mapped[int | None] = mapped_column(Integer, default=None) + # 模型拿不准、留空了的列（中文名和含义都没给）：几列，最多记 5 个列名。表的状态不因此变 + blank_columns: Mapped[int] = mapped_column(Integer, default=0) + blank_sample: Mapped[list[Any] | None] = mapped_column(default=None) # error：整张表没起草（表结构里没有、写入一直冲突、服务重启中断）；model_error：模型那部分没写出来的原因和办法 error: Mapped[str | None] = mapped_column(Text, default=None) model_error: Mapped[str | None] = mapped_column(Text, default=None) diff --git a/backend/tests/fixtures/catalog/fake_model.py b/backend/tests/fixtures/catalog/fake_model.py index 1c4206f1..1574076e 100644 --- a/backend/tests/fixtures/catalog/fake_model.py +++ b/backend/tests/fixtures/catalog/fake_model.py @@ -10,6 +10,7 @@ 不整列漏掉。list_all=False 时只给 replies 里有的列（模拟漏列的模型：漏掉的列会被补问一次，补问后仍没给的算没写出来）； - 直接 ainvoke 是纯文本那条路：同样的内容写成一段正文（AIMessage），前面带一句话； - fail 里的表所在的那一段抛异常，用来验证「一段失败只影响这一段」； +- hang=N：头 N 次调用一直不回（等起草引擎按时限掐掉，当作超时）； - calls 记下每次收到的提示词全文，用来核对交给模型的内容（只有结构，没有数据）；modes 记下每次走的是 structured 还是 text。 @@ -22,6 +23,7 @@ structured、text 模拟实测见过的几种回复（两条路各自设）： """ from __future__ import annotations +import asyncio import json from typing import Any @@ -105,10 +107,11 @@ def with_every_column(entry: dict, wanted: list[str]) -> dict: class FakeModel: def __init__(self, replies: dict[str, dict], fail: set[str] = frozenset(), *, - structured: str = "tables", text: str = "tables", list_all: bool = True) -> None: + structured: str = "tables", text: str = "tables", list_all: bool = True, hang: int = 0) -> None: self.replies, self.fail, self.calls = replies, set(fail), [] self.structured, self.text = structured, text self.list_all = list_all + self.hang = hang self.modes: list[str] = [] def with_structured_output(self, schema: Any, **_: Any) -> "_Structured": @@ -118,9 +121,17 @@ class FakeModel: async def ainvoke(self, messages: list[Any]) -> Any: from langchain_core.messages import AIMessage - reply = self._reply(messages, "text") + reply = await self._answer(messages, "text") return AIMessage(content="草稿如下：" + json.dumps(reply, ensure_ascii=False)) + async def _answer(self, messages: list[Any], mode: str) -> dict: + if self.hang > 0: + self.hang -= 1 + self.calls.append("\n".join(str(getattr(m, "content", m)) for m in messages)) + self.modes.append(mode) + await asyncio.sleep(3600) + return self._reply(messages, mode) + def _reply(self, messages: list[Any], mode: str) -> dict: text = "\n".join(str(getattr(m, "content", m)) for m in messages) self.calls.append(text) @@ -147,4 +158,4 @@ class _Structured: self.owner = owner async def ainvoke(self, messages: list[Any]) -> dict: - return self.owner._reply(messages, "structured") + return await self.owner._answer(messages, "structured") diff --git a/backend/tests/test_catalog_draft.py b/backend/tests/test_catalog_draft.py index 9bf50ea5..81a5629b 100644 --- a/backend/tests/test_catalog_draft.py +++ b/backend/tests/test_catalog_draft.py @@ -324,7 +324,7 @@ async def test_hollow_structured_output_retries_with_text(): got = await catalog.draft_with_model(model, _source(_scenic_tables()), ["visits"]) assert model.modes == ["structured", "text", "text"] assert got["visits"].notes == {"label": make_item("入园记录", "llm"), "kind": make_item("fact", "llm")} - assert got["visits"].covered == frozenset() and "有 5 列没写出来" in got["visits"].error + assert got["visits"].covered == frozenset() and "有 5 列（id 到 status）没写出来" in got["visits"].error async def test_hollow_structured_then_single_text_object(): diff --git a/backend/tests/test_catalog_draft_engine.py b/backend/tests/test_catalog_draft_engine.py index 4f17ee3f..59171b50 100644 --- a/backend/tests/test_catalog_draft_engine.py +++ b/backend/tests/test_catalog_draft_engine.py @@ -6,7 +6,11 @@ - 被截断的 JSON：已经完整的表、列对象捡回来，只补剩下的； - 只回一部分表：只补缺的，补一次； - 服务方嫌额度太大 → 退回 8192，后面都用它；结构化输出不支持 → 后面改走纯文本； -- 网络错误、5xx、超时 → 退避重试，最多 2 次，报错写明重试过几次。 +- 网络错误、5xx → 退避重试，最多 2 次，报错写明重试过几次；超时当作这一段太大，拆小再问，不原样重试。 + +验收后修正的几条（用假模型复现）：宽表的列分段均分、没有很小的尾段；拆不开的空壳纯文本也再问一次；额度用完或超时 +累计 2 次后新开始的段列数减半；一段最多调用 8 次；最近 6 次调用 4 次没写出来时提醒、少于 2 次时清掉；记超时和没写 +出来的调用；模型留空的列计数；窄表不被宽表的列分段插队；原因写清是哪几段、哪几列。 表名、列名、说明文字都是虚构的。 """ @@ -163,6 +167,34 @@ def _status(outcomes: dict) -> dict: return {t: o.status for t, o in outcomes.items()} +def _hollow(call: Call) -> Reply: + return Reply(obj={"tables": [{"table": t} for t in call.asked]}) + + +def _exhausted(seg: str = "这一段", budget: int = 32768) -> str: + return D.REASON_EXHAUSTED.format(budget=budget, seg=seg) + + +EXHAUSTED = Reply(text="", finish="length", thinking="……") + + +class Reports: + """进度回调 call_done 收到的每次调用的结果。""" + + def __init__(self) -> None: + self.items: list[D.CallReport] = [] + + async def __call__(self, report: D.CallReport) -> None: + self.items.append(report) + + def hooks(self) -> D.Hooks: + return D.Hooks(call_done=self) + + @property + def failures(self) -> list[str | None]: + return [r.failure for r in self.items] + + # -------------------------------------------------------------------------- # 思考用完额度、空壳：拆开再问 # -------------------------------------------------------------------------- @@ -220,9 +252,10 @@ async def test_hollow_shell_splits_the_pack_by_table(): assert all(c.asked[t][0] == "full" for c in model.calls[1:] for t in c.asked) -async def test_splitting_stops_at_three_levels_and_five_columns(): +async def test_splitting_stops_at_three_levels_and_five_columns(monkeypatch): + monkeypatch.setattr(catalog, "MODEL_PART_MAX_CALLS", 99) tables = {"wide40": _table(40)} - model = Scripted(lambda call: Reply(text="", finish="length", thinking="……")) + model = Scripted(lambda call: EXHAUSTED) got = await D.run_model_draft(model, _source(tables), ["wide40"]) o = got["wide40"] assert len(model.calls) == 1 + 2 + 4 + 8, "40 → 20 → 10 → 5 列，最多拆三层" @@ -236,6 +269,30 @@ async def test_splitting_stops_at_three_levels_and_five_columns(): assert got["t9"].error == "模型的思考和输出用完了额度（32768 个 token），这一段没写出来" +async def test_one_part_makes_at_most_eight_calls(monkeypatch): + """一段（连同它拆出来的、补问的、重试的）最多调用 8 次：40 列一直用完额度，拆到第 8 次就停，剩下没问的记没写出来、 + 原因写清。以前一段 40 列拆到底要 15 次。""" + monkeypatch.setattr(catalog, "MODEL_GIVE_UP_AFTER", 99) + assert catalog.MODEL_PART_MAX_CALLS == 8 + model = Scripted(lambda call: EXHAUSTED) + got = await D.run_model_draft(model, _source({"t40": _table(40)}), ["t40"]) + assert len(model.calls) == 8 + o = got["t40"] + assert o.status == "failed" + assert o.error == (_exhausted() + "；每段最多调用 8 次模型，这一段到了次数仍没写完，已停止重试" + + D.REASON_SPLIT) + + # 网络错误的重试也算在这 8 次里 + def flaky(call: Call) -> Reply: + raise ConnectError("All connection attempts failed") + + monkeypatch.setattr(catalog, "MODEL_PART_MAX_CALLS", 2) + model = Scripted(flaky) + got = await D.run_model_draft(model, _source({"t4": _table(4)}), ["t4"]) + assert len(model.calls) == 2 + assert got["t4"].error == "每段最多调用 2 次模型，这一段到了次数仍没写完，已停止重试" + + async def test_an_unsplittable_hollow_shell_tries_text_once_then_stays_on_text(): """有的网关的结构化输出只给空壳：拆不开时换纯文本问一次，写出来了，这次起草后面直接走纯文本。 @@ -255,6 +312,31 @@ async def test_an_unsplittable_hollow_shell_tries_text_once_then_stays_on_text() assert catalog.CATALOG_DRAFT_TEXT_PROMPT in model.calls[1].text, "纯文本那条路要说清 JSON 的格式" +def _text_only(handle_text: Callable[[Call], Reply]) -> Callable[[Call], Reply]: + """结构化输出不支持（这次起草一开始就改走纯文本），纯文本交给 handle_text。""" + def handle(call: Call) -> Reply: + if call.mode == "structured": + raise _Status(400, "This response_format type is unavailable now") + return handle_text(call) + return handle + + +async def test_an_unsplittable_hollow_shell_is_asked_again_in_text_mode_too(): + """已经走纯文本时，拆不开的段（不到 10 列）只给空壳：原样再问一次，写出来就用；仍是空壳才记没写出来。 + 以前纯文本模式下不补问，整张表直接记没写出来。""" + tables = {"t8": _table(8)} + ok = answering(tables) + model = Scripted(_text_only(lambda call: _hollow(call) if call.n == 1 else ok(call))) + got = await D.run_model_draft(model, _source(tables), ["t8"]) + assert [c.mode for c in model.calls] == ["structured", "text", "text"] + assert got["t8"].status == "ok" and len(got["t8"].notes["columns"]) == 8 + + model = Scripted(_text_only(_hollow)) + got = await D.run_model_draft(model, _source(tables), ["t8"]) + assert [c.mode for c in model.calls] == ["structured", "text", "text"], "再问一次，不多问" + assert got["t8"].status == "failed" and got["t8"].error == catalog.MODEL_EMPTY_REASON + + # -------------------------------------------------------------------------- # 截断、缺表、缺列 # -------------------------------------------------------------------------- @@ -358,12 +440,35 @@ async def test_columns_still_missing_after_the_followup_are_not_written(): assert len(model.calls) == 2 and model.calls[1].asked == {"t8": ("columns", ["c007", "c008"])} o = got["t8"] assert o.status == "partial" - assert o.error == "这张表 8 列，有 2 列没写出来：模型的回复里漏了这些列，补问一次仍没有给出。可以只重试这张表" + assert o.error == ("这张表 8 列，有 2 列（c007、c008）没写出来：模型的回复里漏了这些列，补问一次仍没有给出。" + "可以只重试这张表") assert set(o.notes["columns"]) == {f"c{i:03d}" for i in range(1, 7)} + assert o.blank == [], "没写出来的列不算留空" compat = (await catalog.draft_with_model(Scripted(handle), _source(tables), ["t8"]))["t8"] assert compat.covered == frozenset(), "没写全的不算完整起草过" +async def test_columns_listed_blank_in_the_followup_are_left_blank_not_failed(): + """补问时模型把漏的列列出来了、字段却留空：是拿不准（记留空），不算「漏了这些列」。""" + tables = {"t8": _table(8)} + ok = answering(tables) + + def handle(call: Call) -> Reply: + reply = ok(call) + entry = reply.obj["tables"][0] + if call.n == 0: + entry["columns"] = [c for c in entry["columns"] if c["name"] not in ("c003", "c007")] + else: + entry["columns"] = [{"name": c["name"]} for c in entry["columns"]] + return reply + + model = Scripted(handle) + got = await D.run_model_draft(model, _source(tables), ["t8"]) + assert model.calls[1].asked == {"t8": ("columns", ["c003", "c007"])} + o = got["t8"] + assert o.status == "ok" and o.error is None and o.blank == ["c003", "c007"] + + async def test_a_followup_that_comes_back_as_an_empty_shell_leaves_the_table_partial(): """40 列的表第一次只回了 20 列，补问只给一个空壳：没写出来的 20 列照实记下，表是 partial。""" tables = {"t40": _table(40)} @@ -382,11 +487,12 @@ async def test_a_followup_that_comes_back_as_an_empty_shell_leaves_the_table_par assert len(model.calls) == 2 assert model.calls[1].asked == {"t40": ("columns", [f"c{i:03d}" for i in range(21, 41)])} assert o.status == "partial" and len(o.notes["columns"]) == 20 - assert o.error == "这张表 40 列，有 20 列没写出来：模型的回复里漏了这些列，补问一次仍没有给出。可以只重试这张表" + assert o.error == ("这张表 40 列，有 20 列（c021 到 c040）没写出来：模型的回复里漏了这些列，补问一次仍没有给出。" + "可以只重试这张表") async def test_columns_listed_with_empty_fields_are_left_to_the_model(): - """列出了、字段留空的列是模型拿不准：不补问，也不算没写出来。""" + """列出了、字段留空的列是模型拿不准：不补问，也不算没写出来；记成留空（表的状态不因此变）。""" tables = {"t8": _table(8)} ok = answering(tables) @@ -402,6 +508,30 @@ async def test_columns_listed_with_empty_fields_are_left_to_the_model(): assert len(model.calls) == 1 assert got["t8"].status == "ok" and got["t8"].error is None assert set(got["t8"].notes["columns"]) == {f"c{i:03d}" for i in range(1, 7)} + assert got["t8"].blank == ["c007", "c008"] + + +async def test_blank_columns_are_the_ones_without_a_label_or_meaning(): + """留空的列：请模型写了、中文名和含义却都没给（只给了度量类型、单位，或者给的都没通过校验）。已确认的、库注释已经说清 + 的不算：给模型的提示词里它们以此为准，模型不写也不缺。""" + meta = _table(9) + meta["columns"][5]["comment"] = "会员等级" # c006：库注释就是中文名 + meta["columns"][8]["comment"] = "C009" # c009：注释就是列名，说明不了什么 + tables = {"t9": meta} + existing = {"t9": {"columns": {"c005": {"label": catalog.make_item("已确认的名字", "human")}}}} + + def handle(call: Call) -> Reply: + return Reply(obj={"tables": [{"table": "t9", "label": "某表", "columns": [ + {"name": "c001", "label": "第一列"}, + {"name": "c002", "meaning": "第二列的含义"}, + {"name": "c003", "measure": "attribute", "unit": "元"}, # 中文名、含义都没给 + {"name": "c004", "label": "=c001+c002"}, # 像公式：校验丢掉 + {"name": "c005"}, {"name": "c006"}, {"name": "c007"}, {"name": "c008"}, {"name": "c009"}]}]}) + + got = await D.run_model_draft(Scripted(handle), _source(tables), ["t9"], existing=existing) + o = got["t9"] + assert o.status == "ok" and o.error is None + assert o.blank == ["c003", "c004", "c007", "c008", "c009"] # -------------------------------------------------------------------------- @@ -521,7 +651,29 @@ async def test_other_errors_are_not_retried(): assert len(model.calls) == 1 and got["t4"].error == "调用模型出错：网关拒绝了这次请求" -async def test_a_call_past_the_time_limit_is_retried_then_reported(monkeypatch): +async def test_a_call_past_the_time_limit_is_split_not_retried(monkeypatch, _no_wait): + """超时当作这一段太大：不原样重试同样大小的段，拆成两半再问（同额度用完）。超时的调用拿不到用量，记 0。""" + monkeypatch.setattr(catalog, "MODEL_TIMEOUT_S", 0.05) + tables = {"t24": _table(24)} + ok = answering(tables, usage=(100, 50)) + + async def handle(call: Call) -> Reply: + if call.n == 0: + await asyncio.sleep(1) + return ok(call) + + model, reports = Scripted(handle), Reports() + got = await D.run_model_draft(model, _source(tables), ["t24"], hooks=reports.hooks()) + cols = [f"c{i:03d}" for i in range(1, 25)] + assert got["t24"].status == "ok" + assert [c.asked for c in model.calls] == [{"t24": ("full", None)}, {"t24": ("table+columns", cols[:12])}, + {"t24": ("columns", cols[12:])}] + assert reports.failures == ["timeout", None, None] + assert [(r.input_tokens, r.output_tokens) for r in reports.items] == [(0, 0), (100, 50), (100, 50)] + assert _no_wait == [], "不退避重试" + + +async def test_a_small_part_past_the_time_limit_is_not_retried(monkeypatch): monkeypatch.setattr(catalog, "MODEL_TIMEOUT_S", 0.05) async def slow(call: Call) -> Reply: @@ -530,8 +682,8 @@ async def test_a_call_past_the_time_limit_is_retried_then_reported(monkeypatch): model = Scripted(slow) got = await D.run_model_draft(model, _source({"t4": _table(4)}), ["t4"]) - assert len(model.calls) == 3 - assert got["t4"].error == "模型 0.05 秒内没有写完这一段（已重试 2 次）" + assert len(model.calls) == 1, "拆不开、也不原样重试" + assert got["t4"].status == "failed" and got["t4"].error == "模型 0.05 秒内没有写完这一段" # -------------------------------------------------------------------------- @@ -554,13 +706,30 @@ async def test_a_wide_table_is_split_into_a_table_part_and_column_parts(): first, *rest = model.calls assert first.asked == {"wide95": ("table", None)} assert catalog.CATALOG_DRAFT_PROMPT_LABELS["brief_columns"].format(n=95) in first.text - assert sorted(len(c.asked["wide95"][1]) for c in rest) == [15, 40, 40] + assert sorted(len(c.asked["wide95"][1]) for c in rest) == [31, 32, 32] for call in rest: assert call.asked["wide95"][0] == "columns" assert "wide95的中文名" in call.text and "虚构的宽表" in call.text, "列分段带着表注释和表级起草结果" assert {c for call in rest for c in call.asked["wide95"][1]} == {f"c{i:03d}" for i in range(1, 96)} +@pytest.mark.parametrize("n, sizes", [(164, [33, 33, 33, 33, 32]), (95, [32, 32, 31]), (41, [21, 20]), + (80, [40, 40]), (81, [27, 27, 27])]) +async def test_column_parts_are_even_without_a_small_tail(monkeypatch, n, sizes): + """列分段数 = 列数 / 40 向上取整，各段列数平均（相差不超过 1）：不再留下几列的尾段（实测会思考的模型常对这种小段只给 + 空壳）。表级那段照旧在最前面。""" + monkeypatch.setattr(catalog, "MODEL_CONCURRENCY", 1) + tables = {"wide": _table(n)} + model = Scripted(answering(tables)) + got = await D.run_model_draft(model, _source(tables), ["wide"]) + o = got["wide"] + assert o.status == "ok" and o.parts_total == 1 + len(sizes) == D.parts_for(n) + assert model.calls[0].asked == {"wide": ("table", None)} + parts = [c.asked["wide"][1] for c in model.calls[1:]] + assert [len(p) for p in parts] == sizes + assert [c for p in parts for c in p] == [f"c{i:03d}" for i in range(1, n + 1)], "按表结构的顺序连续切开" + + async def test_a_wide_table_with_one_failed_part_is_partial_and_says_which(): tables = {"wide95": _table(95)} ok = answering(tables) @@ -574,12 +743,69 @@ async def test_a_wide_table_with_one_failed_part_is_partial_and_says_which(): model = Scripted(handle) got = await D.run_model_draft(model, _source(tables), ["wide95"]) o = got["wide95"] - assert o.status == "partial" and len(o.notes["columns"]) == 55 - assert o.error == "这张表 95 列，分 4 段起草，第 3 段没写出来（共 40 列）：调用模型出错：网关拒绝。可以只重试这张表" + assert o.status == "partial" and len(o.notes["columns"]) == 63 + assert o.error == ("这张表 95 列，分 4 段起草，第 3 段（c033 到 c064，共 32 列）没写出来：调用模型出错：网关拒绝。" + "可以只重试这张表") compat = (await catalog.draft_with_model(Scripted(handle), _source(tables), ["wide95"]))["wide95"] assert compat.covered == frozenset() and compat.error == o.error, "没写全的不算完整起草过，旧的模型项不删" +async def test_reasons_name_each_part_and_say_these_parts(monkeypatch): + """几段因为同一个原因没写出来：原因里写「这几段」；每段写上列的范围（列名照表结构原样）；拆过的才补「已拆小重试」。""" + monkeypatch.setattr(catalog, "MODEL_SHRINK_AFTER", 99) + tables = {"wide95": _table(95)} + ok = answering(tables) + + def handle(call: Call) -> Reply: + part, cols = call.asked["wide95"] + if part == "columns" and cols[0] >= "c033": # 第 3、4 段（c033 起）一直用完额度 + return EXHAUSTED + return ok(call) + + got = await D.run_model_draft(Scripted(handle), _source(tables), ["wide95"]) + o = got["wide95"] + assert o.status == "partial" and len(o.notes["columns"]) == 32 + assert o.error == ("这张表 95 列，分 4 段起草，第 3 段（c033 到 c064，共 32 列）、第 4 段（c065 到 c095，共 31 列）没写出来：" + + _exhausted("这几段") + D.REASON_SPLIT + "。可以只重试这张表") + + +async def test_a_part_that_was_never_halved_does_not_say_it_was(): + """被截断后只补剩下的（不是拆小）：补的那次也用完了额度、又拆不开，原因里不写「已拆小重试」。""" + tables = {"t9": _table(9)} + full = json.dumps({"tables": [{"table": "t9", **_reply("t9", tables["t9"])}]}, ensure_ascii=False) + cut = full[:full.index('"c005"') + 8] + + def handle(call: Call) -> Reply: + return Reply(text=cut, finish="length") if call.n == 0 else EXHAUSTED + + model = Scripted(handle) + got = await D.run_model_draft(model, _source(tables), ["t9"]) + o = got["t9"] + assert len(model.calls) == 2 and model.calls[1].asked == {"t9": ("columns", [f"c{i:03d}" for i in range(5, 10)])} + assert o.status == "partial" + assert o.error == "这张表 9 列，有 5 列（c005 到 c009）没写出来：" + _exhausted() + "。可以只重试这张表" + + +@pytest.mark.parametrize("missing, said", [ + (["c004"], "有 1 列（c004）"), + (["c003", "c007"], "有 2 列（c003、c007）"), + (["c002", "c004", "c006", "c008"], "有 4 列（c002、c004、c006 等）"), + (["c003", "c004", "c005", "c006"], "有 4 列（c003 到 c006）"), +]) +async def test_columns_not_written_are_named(missing, said): + tables = {"t8": _table(8)} + ok = answering(tables) + + def handle(call: Call) -> Reply: + reply = ok(call) + for entry in reply.obj["tables"]: + entry["columns"] = [c for c in entry.get("columns", []) if c["name"] not in missing] + return reply + + got = await D.run_model_draft(Scripted(handle), _source(tables), ["t8"]) + assert got["t8"].error == f"这张表 8 列，{said}没写出来：{D.REASON_UNANSWERED}。可以只重试这张表" + + async def test_tables_are_packed_up_to_40_columns_and_4_tables(): widths = {"ta": 15, "tb": 15, "tc": 15, "td": 5, "te": 5, "tf": 5, "tg": 5, "th": 41} tables = {name: _table(n) for name, n in widths.items()} @@ -598,13 +824,10 @@ async def test_tables_are_packed_up_to_40_columns_and_4_tables(): # -------------------------------------------------------------------------- -def _hollow(call: Call) -> Reply: - return Reply(obj={"tables": [{"table": t} for t in call.asked]}) - - async def test_the_same_reason_is_said_once_even_when_some_parts_were_split(monkeypatch): """宽表的表级那段拆不开、列分段拆过：原因只说一次，拆过的在最后补一句。""" monkeypatch.setattr(catalog, "MODEL_GIVE_UP_AFTER", 99) + monkeypatch.setattr(catalog, "MODEL_PART_MAX_CALLS", 99) model = Scripted(_hollow) got = await D.run_model_draft(model, _source({"wide45": _table(45)}), ["wide45"]) o = got["wide45"] @@ -613,14 +836,19 @@ async def test_the_same_reason_is_said_once_even_when_some_parts_were_split(monk def test_reasons_are_joined_by_their_base(): + """(原因, 第几段)：同样的原因只说一次；同一个原因落在几段上写「这几段」，同一段拆出来的几块仍是「这一段」。""" W = D._Why - exhausted = D.REASON_EXHAUSTED.format(budget=32768) - assert D._join([W(D.REASON_HOLLOW), W(D.REASON_HOLLOW, split=True)]) == D.REASON_HOLLOW + D.REASON_SPLIT - assert D._join([W(exhausted, split=True), W(D.REASON_HOLLOW)]) == \ - f"{exhausted}；{D.REASON_HOLLOW}{D.REASON_SPLIT}" - assert D._join([W("甲"), W("乙"), W("丙", split=True)]) == "甲；乙", "没说出来的原因不补后缀" - assert D._join([W(D.REASON_HOLLOW, halted=True), W(D.REASON_HOLLOW)]) == \ + ex = D.REASON_EXHAUSTED.format(budget=32768, seg=D.SEG) + assert D._join([(W(D.REASON_HOLLOW), 1), (W(D.REASON_HOLLOW, split=True), 2)]) == \ + D.REASON_HOLLOW + D.REASON_SPLIT + assert D._join([(W(ex, split=True), 2), (W(D.REASON_HOLLOW), 1)]) == \ + f"{_exhausted()}；{D.REASON_HOLLOW}{D.REASON_SPLIT}" + assert D._join([(W(ex), 2), (W(ex), 3)]) == _exhausted("这几段") + assert D._join([(W(ex), 2), (W(ex, split=True), 2)]) == _exhausted() + D.REASON_SPLIT + assert D._join([(W("甲"), 1), (W("乙"), 1), (W("丙", split=True), 1)]) == "甲；乙", "没说出来的原因不补后缀" + assert D._join([(W(D.REASON_HOLLOW, halted=True), 1), (W(D.REASON_HOLLOW), 2)]) == \ D.REASON_HOLLOW + D.REASON_HALTED.format(n=catalog.MODEL_GIVE_UP_AFTER) + assert D._join([(W("调用模型出错：{坏的}"), 1)]) == "调用模型出错：{坏的}", "报错原文里的花括号原样留着" def _by_table(model: Scripted) -> dict[str, int]: @@ -643,7 +871,7 @@ async def test_after_two_segments_in_a_row_come_to_nothing_the_rest_are_not_spli model = Scripted(lambda call: exhausted) got = await D.run_model_draft(model, _source(tables), list(tables)) assert _by_table(model) == {"t1": 3, "t2": 3, "t3": 1, "t4": 1} - budget = D.REASON_EXHAUSTED.format(budget=32768) + budget = _exhausted() assert got["t1"].error == budget + D.REASON_SPLIT assert got["t3"].error == budget + "；模型连续 2 段都没写出来，已停止拆小重试，建议更换起草模型" @@ -657,23 +885,23 @@ async def test_after_two_segments_in_a_row_come_to_nothing_the_rest_are_not_spli async def test_the_column_parts_of_a_wide_table_stop_splitting_too(monkeypatch): """宽表的列分段一段一段排队起草（一段拆到底再起草下一段），止损在宽表里同样起作用。 - 95 列：表级那段拆不开（结构化、纯文本各一次）；第 2 段 40 列拆到底（15 次结构化 + 8 次纯文本）；这时已经连续 2 段 - 没写出来，第 3、4 段只问一次、换纯文本一次。""" + 95 列：表级那段拆不开（结构化、纯文本各一次）；第 2 段 32 列拆小、到 8 次封顶；这时已经连续 2 段没写出来，第 3、4 段 + 只问一次、换纯文本一次。""" monkeypatch.setattr(catalog, "MODEL_CONCURRENCY", 1) model = Scripted(_hollow) got = await D.run_model_draft(model, _source({"wide95": _table(95)}), ["wide95"]) - assert len(model.calls) == 2 + 23 + 2 + 2 - assert got["wide95"].error == (D.REASON_HOLLOW + D.REASON_SPLIT - + "；模型连续 2 段都没写出来，已停止拆小重试，建议更换起草模型") + assert len(model.calls) == 2 + 8 + 2 + 2 + assert got["wide95"].error == (D.REASON_HOLLOW + "；每段最多调用 8 次模型，这一段到了次数仍没写完，已停止重试" + + D.REASON_SPLIT + "；模型连续 2 段都没写出来，已停止拆小重试，建议更换起草模型") async def test_a_wide_table_that_never_comes_back_is_bounded_by_the_first_wave(): - """默认并发（3 段同时起草）：294 列一直只给空壳，最多是表级那段 2 次、头 3 个列分段拆到底（各 23 次）、其余 5 段 - 各 2 次。以前所有列分段一下子都排上去、一层一层地拆，要 168 次。""" + """默认并发（3 段同时起草）：294 列（8 个列分段）一直只给空壳，最多是表级那段 2 次、头 3 个列分段各 8 次（封顶）、 + 其余 5 段各 2 次。以前所有列分段一下子都排上去、一层一层地拆，要 168 次。""" model = Scripted(_hollow) got = await D.run_model_draft(model, _source({"wide294": _table(294)}), ["wide294"]) - assert got["wide294"].status == "failed" - assert len(model.calls) <= 2 + 3 * 23 + 5 * 2 + assert got["wide294"].status == "failed" and got["wide294"].parts_total == 9 + assert len(model.calls) <= 2 + 3 * 8 + 5 * 2 async def test_a_big_draft_does_not_hold_the_model_ahead_of_a_later_one(monkeypatch): @@ -707,6 +935,85 @@ async def test_a_big_draft_does_not_hold_the_model_ahead_of_a_later_one(monkeypa assert order == ["a1", "b1", "a2", "a3", "a4"] +def _order(model: Scripted) -> list[tuple[str, str]]: + return [(name, part) for call in model.calls for name, (part, _) in call.asked.items()] + + +async def test_narrow_tables_are_not_cut_in_by_the_column_parts_of_a_wide_one(monkeypatch): + """一个任务里按表的先后排队：宽表的表级那段写完，它的列分段排到已经排着的窄表段后面（先来先服务），不插队。 + 以前插到队首，窄表要等宽表的列分段全部派完。""" + monkeypatch.setattr(catalog, "MODEL_CONCURRENCY", 1) + monkeypatch.setattr(catalog, "MODEL_BATCH_TABLES", 1) + tables = {"wide": _table(45), "na": _table(4), "nb": _table(4)} + model = Scripted(answering(tables)) + got = await D.run_model_draft(model, _source(tables), ["wide", "na", "nb"]) + assert set(_status(got).values()) == {"ok"} + assert _order(model) == [("wide", "table"), ("na", "full"), ("nb", "full"), ("wide", "columns"), + ("wide", "columns")] + + +async def test_after_two_exhausted_or_timed_out_calls_new_parts_are_halved(monkeypatch): + """额度用完或超时累计 2 次，之后新开始的段列数上限减半（40 → 20 → 10，最低 10）；已经开始的段照旧（按原因拆小）。 + 每次调用的结果里带着当前的上限（part_columns）。""" + monkeypatch.setattr(catalog, "MODEL_CONCURRENCY", 1) + monkeypatch.setattr(catalog, "MODEL_BATCH_TABLES", 1) + tables = {f"t{i}": _table(40) for i in range(1, 4)} + ok = answering(tables) + + def handle(call: Call) -> Reply: + (_, (part, cols)), = call.asked.items() + width = 40 if part == "full" else len(cols) + return EXHAUSTED if width > 10 else ok(call) # 这个模型一次只写得出 10 列 + + model, reports = Scripted(handle), Reports() + got = await D.run_model_draft(model, _source(tables), list(tables), hooks=reports.hooks()) + assert set(_status(got).values()) == {"ok"} + asked = {name: [(part, len(cols or [])) for c in model.calls for n, (part, cols) in c.asked.items() if n == name] + for name in tables} + # t1：整张 40 列开始，用完额度拆成 20、20，又各拆成 10、10 + assert asked["t1"][0] == ("full", 0) and len(asked["t1"]) == 7 + # 第 2 次用完额度后上限是 20：t2 一开始就按 20 列切成两段（各自再拆一次） + assert asked["t2"][0] == ("table+columns", 20) and ("columns", 20) in asked["t2"] + # 第 4 次之后上限是 10：t3 一开始就切成 4 段 10 列，一次都没用完额度 + assert asked["t3"] == [("table+columns", 10), ("columns", 10), ("columns", 10), ("columns", 10)] + caps = [r.part_columns for r in reports.items] + assert list(dict.fromkeys(caps)) == [40, 20, 10] + assert caps[:2] == [40, 20], "第 2 次用完额度的那次调用之后就是 20" + assert min(caps) == 10 == catalog.MODEL_MIN_PART_COLUMNS + + +async def test_the_notice_comes_when_four_of_six_calls_fail_and_goes_when_fewer_than_two(monkeypatch): + """最近 6 次调用有 4 次以上没写出来：提醒（写明次数和原因，建议换起草模型、减少表或停止）；之后最近 6 次里没写出来 + 的少于 2 次就清掉。""" + monkeypatch.setattr(catalog, "MODEL_CONCURRENCY", 1) + monkeypatch.setattr(catalog, "MODEL_BATCH_TABLES", 1) + tables = {f"t{i}": _table(4) for i in range(1, 10)} + ok = answering(tables) + + def handle(call: Call) -> Reply: + name = next(iter(call.asked)) + if name in ("t1", "t2"): + raise RuntimeError("网关拒绝了这次请求") + if name == "t3": + return EXHAUSTED + if name == "t4": + return _hollow(call) + return ok(call) + + reports = Reports() + got = await D.run_model_draft(Scripted(handle), _source(tables), list(tables), hooks=reports.hooks()) + assert reports.failures == ["error", "error", "exhausted", "hollow", "hollow"] + [None] * 5 + notices = [r.notice for r in reports.items] + assert notices[:3] == [None] * 3 + assert notices[3] == ("最近 4 次调用有 4 次没写出来（额度用完 1 次、只给空壳 1 次、出错 2 次）。可以在「设置 → 偏好设置」" + "的「起草模型」换一个模型、减少这次起草的表，或者停止这次起草") + assert notices[4].startswith("最近 5 次调用有 5 次没写出来") + assert notices[7].startswith("最近 6 次调用有 3 次没写出来"), "还有 2 次以上没写出来：留着，次数照实写" + assert notices[8].startswith("最近 6 次调用有 2 次没写出来") + assert notices[9] is None, "最近 6 次只有 1 次没写出来：清掉" + assert got["t9"].status == "ok" + + # -------------------------------------------------------------------------- # 起草要求、用量 # -------------------------------------------------------------------------- @@ -747,13 +1054,11 @@ async def test_usage_is_added_up_through_the_hooks(): return Reply(text="", finish="length", thinking="……", usage=(100, 32768)) return ok(call) - seen: list[tuple[int, int]] = [] - - async def usage(i: int, o: int) -> None: - seen.append((i, o)) - - await D.run_model_draft(Scripted(handle), _source(tables), ["orders_x"], hooks=D.Hooks(usage=usage)) - assert seen == [(100, 32768), (120, 60), (120, 60)] + reports = Reports() + await D.run_model_draft(Scripted(handle), _source(tables), ["orders_x"], hooks=reports.hooks()) + assert [(r.input_tokens, r.output_tokens, r.failure) for r in reports.items] == \ + [(100, 32768, "exhausted"), (120, 60, None), (120, 60, None)] + assert {r.notice for r in reports.items} == {None} and {r.part_columns for r in reports.items} == {40} def test_budget_words_match_the_other_drafts(): diff --git a/backend/tests/test_catalog_draft_jobs.py b/backend/tests/test_catalog_draft_jobs.py index a8a961bb..c72cc9b0 100644 --- a/backend/tests/test_catalog_draft_jobs.py +++ b/backend/tests/test_catalog_draft_jobs.py @@ -6,7 +6,9 @@ - 停止：排队的直接 cancelled；进行中的停掉正在进行的模型调用，没写完的表记 cancelled，已经写进目录的不回滚；已结束的 409 - 服务启动时把没结束的任务标成 interrupted（没结束的表记 cancelled、写明服务重启），不自动续跑；关停时同样 - 重试：只含原任务里 failed、partial、cancelled 的表，本次说明沿用、起草要求取数据源当前的；没有可重试的 422 -- 目录清单：每行 last_draft，顶层 guidance、active_job +- 目录清单：每行 last_draft（带模型留空的列数），顶层 guidance、active_job +- 重新起草时模型这次留空、给的值没通过校验：以前起草好的模型项留着，删除只在列已经不在表结构里时才有 +- 用量记超时次数、没写出来的调用次数；任务记当前的列数上限、提醒（最近 6 次调用 4 次没写出来）；逐表记模型留空的列 - 起草要求存在数据源 options 里（上传表格的源存在它自己那一行），数据源接口不显示、整份回写 options 时留着，拼连接串时摘掉 表名、列名、说明文字都是虚构的。 @@ -261,11 +263,136 @@ async def test_a_followup_that_comes_back_empty_keeps_the_earlier_model_items(cl task = job["tasks"][0] assert model.calls == 2 assert job["status"] == "partial" and task["status"] == "partial" and task["removed"] == 0 - assert "有 4 列没写出来" in task["model_error"] and "可以只重试这张表" in task["model_error"] + assert "有 4 列（c05 到 c08）没写出来" in task["model_error"] and "可以只重试这张表" in task["model_error"] after = (await client.get(f"{_base(sid)}/tickets")).json()["notes"]["columns"] assert {c for c, items in after.items() if "label" in items} == {f"c{i:02d}" for i in range(1, 9)} +def _tickets(n: int = 6) -> dict: + return {"tickets": {"qualified": "tickets", "columns": [_col(f"c{i:02d}") for i in range(1, n + 1)], + "primary_key": ["c01"], "comment": None, "is_view": False, "foreign_keys": []}} + + +async def _draft(client, sid: str, monkeypatch, model, tables: list[str]) -> dict: + use(monkeypatch, model) + r = await client.post(f"{_base(sid)}/draft-jobs", json={"tables": tables, "use_model": True}) + assert r.status_code == 202, r.text + await finished(r.json()["job"]["id"]) + return await get_job(client, sid, r.json()["job"]["id"]) + + +async def test_a_redraft_keeps_what_the_model_left_blank_this_time(client, monkeypatch): + """重新起草（比如补了起草要求再起草一次）时模型这次留空、或者给的值没通过校验：以前起草好的模型项留着，不算删除； + 给了合格新值的才替换。以前整库重跑一次删掉几百项，表仍记「已完成」。""" + sid = await _source(_tickets()) + full = {"label": "门票", "kind": "fact", + "columns": [{"name": f"c{i:02d}", "label": f"第{i}列", "meaning": f"第{i}列的含义"} for i in range(1, 7)]} + first = await _draft(client, sid, monkeypatch, FakeModel({"tickets": full}), ["tickets"]) + assert first["tasks"][0]["status"] == "ok" and first["blank_columns"] == 0 + before = (await client.get(f"{_base(sid)}/tickets")).json()["notes"] + + # 这一次：表的中文名换了；c03 的中文名像公式（校验丢掉）、含义换了；c04 只给了中文名；其余列列出了但都留空 + again = {"label": "门票明细", "columns": [{"name": "c03", "label": "=c01+c02", "meaning": "新的含义"}, + {"name": "c04", "label": "第4列"}]} + second = await _draft(client, sid, monkeypatch, FakeModel({"tickets": again}), ["tickets"]) + task = second["tasks"][0] + assert task["status"] == "ok" and task["removed"] == 0 and second["removed"] == 0 + assert (task["added"], task["updated"]) == (0, 2) + after = (await client.get(f"{_base(sid)}/tickets")).json()["notes"] + assert after["label"]["value"] == "门票明细" and after["kind"] == before["kind"] + assert after["columns"]["c03"]["label"] == before["columns"]["c03"]["label"] + assert after["columns"]["c03"]["meaning"]["value"] == "新的含义" + for col in ("c01", "c02", "c05", "c06"): + assert after["columns"][col] == before["columns"][col], col + # 模型这次留空的列照实记下（表的状态不因此变） + assert task["blank_columns"] == 4 and task["blank_sample"] == ["c01", "c02", "c05", "c06"] + + +async def test_model_items_of_a_column_gone_from_the_schema_are_removed(client, monkeypatch): + sid = await _source(_tickets()) + full = {"label": "门票", "columns": [{"name": f"c{i:02d}", "label": f"第{i}列"} for i in range(1, 7)]} + await _draft(client, sid, monkeypatch, FakeModel({"tickets": full}), ["tickets"]) + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + row.schema_cache = {"tables": _tickets(5)} # 重新探查过：c06 已经没有了 + await session.commit() + job = await _draft(client, sid, monkeypatch, FakeModel({"tickets": {"label": "门票"}}), ["tickets"]) + assert job["tasks"][0]["removed"] == 1 + after = (await client.get(f"{_base(sid)}/tickets")).json()["notes"] + assert set(after["columns"]) == {"c01", "c02", "c03", "c04", "c05"} + + +async def test_usage_counts_timeouts_and_failed_calls(client, monkeypatch): + """超时的调用拆小再问，用量里记超时几次、没写出来几次（超时的调用拿不到用量，token 记 0）；额度用完或超时累计 2 次， + 新开始的段列数上限减半，任务上记着当前的上限。""" + monkeypatch.setattr(catalog, "MODEL_TIMEOUT_S", 0.05) + sid = await _source(_tickets(24)) + replies = {"tickets": {"label": "门票", "columns": [{"name": f"c{i:02d}", "label": f"第{i}列"} + for i in range(1, 25)]}} + model = FakeModel(replies, hang=2) + job = await _draft(client, sid, monkeypatch, model, ["tickets"]) + assert job["status"] == "succeeded" and job["tasks"][0]["status"] == "ok" + assert job["usage"] == {"calls": 5, "input_tokens": 0, "output_tokens": 0, "timeouts": 2, "failed_calls": 2} + assert job["part_columns"] == 20 and job["notice"] is None + # 24 列整张问超时 → 前 12 列（带表级的项）超时 → 拆成 6、6 → 后 12 列 + assert [len(requested(c)["tickets"][1] or []) for c in model.calls] == [0, 12, 6, 6, 12] + + plain = await _draft(client, sid, monkeypatch, FakeModel(replies), ["tickets"]) + assert plain["part_columns"] == 40 and plain["usage"]["timeouts"] == plain["usage"]["failed_calls"] == 0 + + +async def test_the_notice_is_kept_on_the_job(client, monkeypatch): + """最近 6 次调用有 4 次以上没写出来：任务上写一句提醒（次数、原因、建议换起草模型）；不用模型的任务没有。""" + monkeypatch.setattr(catalog, "MODEL_BATCH_TABLES", 1) + tables = {f"t{i}": {"qualified": f"t{i}", "columns": [_col("id"), _col("v")], "primary_key": ["id"], + "comment": None, "is_view": False, "foreign_keys": []} for i in range(1, 6)} + sid = await _source(tables) + replies = {name: {"label": f"表{name}"} for name in tables} + job = await _draft(client, sid, monkeypatch, FakeModel(replies, fail={"t1", "t2", "t3", "t4"}), list(tables)) + assert job["usage"]["calls"] == 5 and job["usage"]["failed_calls"] == 4 + assert job["notice"] == ("最近 5 次调用有 4 次没写出来（出错 4 次）。可以在「设置 → 偏好设置」的「起草模型」换一个模型、" + "减少这次起草的表，或者停止这次起草") + plain = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["t1"]})).json()["job"] + await finished(plain["id"]) + plain = await get_job(client, sid, plain["id"]) + assert plain["notice"] is None and plain["part_columns"] is None and plain["blank_columns"] == 0 + + +async def test_blank_columns_reach_the_task_the_job_and_the_list(client, monkeypatch): + """模型列出了、中文名和含义都没给的列：逐表记列数和最多 5 个列名，任务上是合计，目录清单的 last_draft 也带列数。""" + tables = {**_tickets(8), "gates": _tables()["gates"]} + sid = await _source(tables) + replies = {"tickets": {"label": "门票", "columns": [{"name": "c01", "label": "编号"}]}, + "gates": {"label": "闸口", "columns": [{"name": "id", "label": "编号"}]}} + job = await _draft(client, sid, monkeypatch, FakeModel(replies), ["tickets", "gates"]) + by = {t["table_name"]: t for t in job["tasks"]} + assert by["tickets"]["status"] == "ok" and by["tickets"]["blank_columns"] == 7 + assert by["tickets"]["blank_sample"] == ["c02", "c03", "c04", "c05", "c06"] + assert by["gates"]["blank_columns"] == 1 and by["gates"]["blank_sample"] == ["gate_code"] + assert job["blank_columns"] == 8 + rows = {r["table_name"]: r for r in (await client.get(_base(sid))).json()["tables"]} + assert rows["tickets"]["last_draft"]["blank_columns"] == 7 and rows["gates"]["last_draft"]["blank_columns"] == 1 + + +def test_old_job_tables_get_the_new_columns(): + """升级前建的起草任务表：新列补上，存量行的计数是 0、提醒和列数上限为空。""" + from sqlalchemy import create_engine, text + + from app.db.base import _migrate + + db = create_engine("sqlite://") + with db.begin() as conn: + conn.execute(text("CREATE TABLE catalog_draft_jobs (id VARCHAR(32) PRIMARY KEY, calls INTEGER)")) + conn.execute(text("CREATE TABLE catalog_draft_tasks (id VARCHAR(32) PRIMARY KEY, removed INTEGER)")) + conn.execute(text("INSERT INTO catalog_draft_jobs (id, calls) VALUES ('j1', 3)")) + conn.execute(text("INSERT INTO catalog_draft_tasks (id, removed) VALUES ('t1', 0)")) + _migrate(conn) + _migrate(conn) + assert conn.execute(text("SELECT timeouts, failed_calls, part_columns, notice FROM catalog_draft_jobs")) \ + .one() == (0, 0, None, None) + assert conn.execute(text("SELECT blank_columns, blank_sample FROM catalog_draft_tasks")).one() == (0, None) + + async def test_a_job_that_cannot_start_is_failed_with_the_reason(client, monkeypatch): """建任务时起草模型还能用，开始起草时已经用不了（接入刚被停用）：任务整体 failed，error 写原因，没开始的表也记 failed。""" sid = await _source()