1. llm_pipeline.py 的校验-重试为什么先把 Token 烧光llm_pipeline.py校验-重试想换 Base URL第一件能落地的事是去 TaoToken 建一把 Keyhttps://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_content 。不过先别急着改代码把「换通道」和「重试逻辑」两件事分开看才知道这次改动的边界在哪。原文里LLMPipeline用AsyncOpenAI客户端跑summarize和extract_keywords两步每步都要过JSONSchemaValidator校验不过就走_execute_step里那个while retry_count max_retries把上一轮的报错拼成修正提示重新发一次并且按指数退避拉长间隔。这个设计的初衷没问题问题是它把「模型输出不稳定」的成本全部转嫁到了 Token 上。1.1 一次 summarize 失败账单上是三次完整调用_execute_step的循环不是只发一个「请修正」的短消息。为了让模型知道错在哪绝大多数实现会把原始长文本、上一轮的完整输出、以及 validator 抛出的错误信息一起塞回 context。这意味着重试一次输入侧的长度基本翻倍max_retries2时同一段素材可能被完整送进去三遍。如果这一步还跑在按量计费的官方通道上重试次数越多越接近「花三倍钱拿一份兜底结果」的尴尬局面。更隐蔽的是失败本身不会被立刻叫停。retry_count到顶之后代码通常回落到fallback_output返回一个结构合法但内容空壳的对象流程继续往下走。看起来 pipeline 没崩但token_usage已经记了一大笔而这笔钱对应的产出质量和直接跳过这一步差不多。所以在做接入改造前先把「重试次数」「单次输入长度」「兜底触发频率」三个数字记下来改造完才有对照。1.2 JSONSchemaValidator 只认结构不认你走哪条通道很多人把「校验老失败」和「通道不给力」混在一起。其实JSONSchemaValidator是纯本地的活儿它拿到的是模型返回的字符串json.loads之后对着 schema 检查字段名、类型、必填项、长度约束。Base URL 换成什么它都不关心也不会因为你换了接入点就网开一面。这个边界必须说清楚因为它是判断「行不行」的关键。TaoToken 在这条链路里只做两件事给你一把 API Key给你一个统一的兼容入口https://taotoken.net/api。它不替JSONSchemaValidator做校验不替while循环做重试也不替fallback_output做兜底。你的重试策略、退避曲线、降级逻辑全都还留在自己的代码里。换句话说换通道解决的是「能不能稳定调到模型」和「账单口径是否统一」不解决「模型输出为什么不符合 schema」——后者要从 prompt 约束和 schema 宽严上找原因。1.3 真正值得改的只有客户端构造那一处把上面两层拆开之后改造范围就很小了LLMPipeline内部逻辑一行不动只把AsyncOpenAI(...)的构造参数换掉。原来那里可能是空参数走环境变量也可能是硬编码了某个官方地址现在改成显式传入base_urlhttps://taotoken.net/api和一把从官网创建的 Key其余全部保留。这样做还有一个副作用是好的因为只动了一处回滚成本极低。真遇到问题把构造函数改回去就能对比出到底是通道问题还是 schema 问题。下面几节就按这个思路把 Key 的创建、客户端的改法、StepResult的验证逐个落地。2. 把 AsyncOpenAI() 那一步从直连挪出来2.1 先在模型广场确认模型 ID 再动手打开 TaoToken 官网 注册登录进控制台创建一把 API Key把它记成YOUR_API_KEY占位真实值写进环境变量不要提交进仓库。接着去模型广场看一眼当前可用的模型列表把要做summarize和extract_keywords的那个模型 ID 抄下来——本文所有示例里的模型 ID 都写成占位符实际取值以 模型广场 当时的列表为准不要照着网上随便一个带日期后缀的名字填。# 本地开发环境先设好别写进代码 export TAOTOKEN_API_KEYYOUR_API_KEY提示Key 只在创建时完整显示一次复制之后立刻存进密码管理器或本地.env后面settings类文件里一律引用变量名不引用明文。2.2 环境变量里 Base URL 和 Key 分开存如果项目本来就用.env管理配置加两行就够了。这里要注意一个高频错误填进工具的地址是https://taotoken.net/api末尾不要加/v1。AsyncOpenAI的 SDK 自己会拼/chat/completions多写一层路径进不去。# .env TAOTOKEN_API_KEYYOUR_API_KEY TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODELYOUR_MODEL_ID官网页面地址和接口地址是两回事注册、建 Key、看用量、查模型走https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_content填进代码和环境变量的 Base URL 用https://taotoken.net/api。把这两个混在一起最常见的结果就是请求打到落地页上然后收到一段 HTML。2.3 为什么不用 SDK 自带的重试AsyncOpenAI构造函数里有max_retries参数默认会自己吞掉一部分 5xx 和超时。放到这个 pipeline 里很危险SDK 悄悄重试一轮_execute_step又重试一轮两套策略叠加retry_count根本对不上实际请求次数指数退避也白写了。所以构造函数里显式设max_retries0把重试权完整收回给业务层账单和日志才对得上。3. LLMPipeline 的客户端与 schema 怎么落盘3.1 客户端构造只改两行假设原来的写法是client AsyncOpenAI()依赖环境变量里的官方配置。改完长这样# llm_pipeline/client.py import os from openai import AsyncOpenAI client AsyncOpenAI( api_keyos.environ[TAOTOKEN_API_KEY], # 从 https://taotoken.net 控制台创建 base_urlos.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api), timeout90.0, max_retries0, # 重试统一交给 _execute_step避免和 SDK 策略叠加 ) MODEL_ID os.environ[TAOTOKEN_MODEL] # 以模型广场当时的列表为准timeout给到 90 秒是有原因的summarize这类任务输入长、输出也长退避之后第二次请求又紧跟着发60 秒经常不够。如果你的素材普遍超过几千字可以把这一步的输入先做一次截断或分块重试时才不会越滚越贵。3.2 两个 schema宽严要分开调summarize和extract_keywords的输出结构不同校验失败的常见原因也不同。前者容易在字段缺失上翻车后者容易在数组长度和重复词上翻车。分开写 schema重试时的修正提示才有针对性。# llm_pipeline/schemas.py SUMMARIZE_SCHEMA { type: object, required: [summary, key_points], properties: { summary: {type: string, minLength: 30}, key_points: { type: array, items: {type: string, minLength: 4}, minItems: 2, maxItems: 8, }, }, additionalProperties: False, } KEYWORDS_SCHEMA { type: object, required: [keywords], properties: { keywords: { type: array, items: {type: string, minLength: 2}, minItems: 3, maxItems: 15, uniqueItems: True, } }, additionalProperties: False, }additionalProperties: False是个双刃剑。它能让输出更干净但也会让模型多写一个无关字段就直接判失败多消耗一次重试。第一次接入时建议先放开这一条等输出稳定了再收紧。3.3 _execute_step退避和兜底保持原样改造的重点是「通道换、策略不换」。循环结构、修正提示的拼法、指数退避的基数全部沿用原来的实现只在请求参数上补一个response_format让模型尽量吐 JSON。# llm_pipeline/pipeline.py import asyncio, json from dataclasses import dataclass, field from jsonschema import Draft202012Validator, ValidationError from .client import client, MODEL_ID from .schemas import SUMMARIZE_SCHEMA, KEYWORDS_SCHEMA dataclass class StepResult: ok: bool data: dict retry_count: int 0 token_usage: dict field(default_factorydict) class LLMPipeline: def __init__(self, model: str MODEL_ID, max_retries: int 2): self.model model self.max_retries max_retries def _build_messages(self, task: str, text: str, last_error: str | None): sys f你是文本处理助手只输出 JSON任务{task}。 user text if not last_error else ( f{text}\n\n上一次输出未通过校验错误{last_error}\n 请只修正结构问题不要新增解释文字。 ) return [{role: system, content: sys}, {role: user, content: user}] async def _execute_step(self, task, text, schema, fallback): retry_count 0 total_tokens 0 last_error None while retry_count self.max_retries: resp await client.chat.completions.create( modelself.model, messagesself._build_messages(task, text, last_error), response_format{type: json_object}, temperature0.2 if retry_count 0 else 0.0, ) total_tokens resp.usage.total_tokens try: payload json.loads(resp.choices[0].message.content) Draft202012Validator(schema).validate(payload) return StepResult(True, payload, retry_count, {total_tokens: total_tokens}) except (json.JSONDecodeError, ValidationError) as e: last_error str(e)[:300] retry_count 1 if retry_count self.max_retries: await asyncio.sleep(0.5 * (2 ** (retry_count - 1))) return StepResult(False, fallback, retry_count, {total_tokens: total_tokens}) async def summarize(self, text: str): return await self._execute_step( 输出 summary 与 key_points, text, SUMMARIZE_SCHEMA, {summary: , key_points: []}) async def extract_keywords(self, text: str): return await self._execute_step( 输出 keywords 数组, text, KEYWORDS_SCHEMA, {keywords: []})注意total_tokens是累加的不是最后一次的。这个细节决定后面能不能看懂账单如果只记最后一轮重试造成的额外消耗就完全看不见了。fallback依然由业务层自己给没有人替你决定兜底内容长什么样。4. 跑 main()用 StepResult 对账 token_usage 和 retry_count4.1 第一次跑先确认两步都有结果# main.py import asyncio from llm_pipeline.pipeline import LLMPipeline TEXT 把这里换成你自己的测试素材长度尽量接近真实场景。 async def main(): pipe LLMPipeline() r1 await pipe.summarize(TEXT) r2 await pipe.extract_keywords(TEXT) for name, r in ((summarize, r1), (extract_keywords, r2)): print(f{name}: ok{r.ok} retry{r.retry_count} ftokens{r.token_usage.get(total_tokens)}) asyncio.run(main())第一次运行的目标不是拿到漂亮结果而是确认两个StepResult都能回来。如果summarize的okFalse先看retry_count是否等于max_retries——等于说明每一轮都被 schema 挡下来了问题多半在 prompt 或 schema小于则说明请求本身就没成功去看异常信息。4.2 用 token_usage 判断这次重试值不值retry_count0且okTrue是最理想的一次请求、一次通过total_tokens就是这次输入加输出的真实消耗。retry_count1时如果total_tokens是零重试情况的两倍左右说明修正提示把上下文撑长了这条路径就有优化空间——比如把原文本换成摘要预览再重试而不是整段重发。还有一种情况要警惕okFalse且total_tokens很高。这说明兜底结果是自己花钱买来的pipeline 表面没报错实际白跑。把这两个字段接进日志跑几天就能看出哪一类素材最容易触发重试。4.3 顺手去控制台核对一次用量服务端记的用量和本地total_tokens对不上时先排除是不是 SDK 层重试被悄悄打开了。想看清楚每把 Key 的调用情况登录 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_content 进控制台看用量明细把这段时间的请求数和自己日志里的次数比一比差额通常就是隐藏重试。5. 换到兼容通道后常见的几类报错5.1 401Key 没读到或者复制缺了尾巴AuthenticationError在这个场景里八成不是 Key 无效而是没读进来。检查三件事环境变量名和代码里os.environ[...]是否一致.env是否真的被加载很多项目要显式load_dotenv()从控制台复制时有没有漏掉末尾字符。改完 Key 记得重启进程长驻服务不会自动读新环境变量。5.2 校验依然反复失败这不是通道的问题如果retry_count每次都跑满last_error又集中在同一类字段上那就是 prompt 和 schema 的匹配问题。三条实用做法把additionalProperties先放开在 system prompt 里直接给一个字段完整的 JSON 例子把minItems、maxItems这类硬边界放宽一档。改完再跑一次main()看retry_count有没有下降。5.3 404 与路径base_url 别多写一层NotFoundError常见于把 Base URL 写成https://taotoken.net/api/v1SDK 再拼一次/chat/completions就找不到路由。正确值就是https://taotoken.net/api。另一种 404 是模型 ID 抄错了——去模型广场核对一遍别用记忆里的名字。现象优先检查处理401环境变量是否加载重启进程后重试404base_url是否带/v1改回https://taotoken.net/api校验跑满重试schema 约束是否过严放宽边界并补 JSON 示例用量对不上SDK 层是否自动重试构造函数设max_retries06. 两步跑顺之后把 pipeline 接到真实任务上summarize与extract_keywords都能一次通过说明客户端、Key、模型 ID 这三样配对了。接下来可以做的事有几件先拿几条真实素材跑批量观察retry_count的分布把StepResult落进表里方便按素材类型统计兜底率再考虑要不要给extract_keywords单独换一个更省的小模型。想先确认模型 ID 和 Key 没配错去 TaoToken 模型对话 用同一把 Key 发一条测试消息最直接。如果这个 pipeline 后面还要长期跑批可以看一眼 Coding Plan 的套餐是否覆盖得住需要再开一把 Key 分环境用就在 控制台 API Keys 里创建。要是打算把同样的通道挪到 Claude Code 里做代码侧的辅助环境变量对照表在 Claude Code 接入文档。最后提醒一句fallback_output触发的时候别让它静默过去。那代表这一段素材的处理质量已经不达标了把retry_count和total_tokens一起打出来比事后翻账单要省事得多。
问题解析与实现)
