1. 从一次 Prefect 流水线的 429 报警说起为什么把 LLM 调用统一到 TaoToken最近央视财经关于 Token 调用量快速攀升的报道引发了不少讨论但作为工作流工程师真正让我从椅子上弹起来的不是新闻而是 Prefect task 里反复出现的429 Too Many Requests和APITimeoutError。我们的流水线原本把多个模型的调用散落在不同 task 里有的用环境变量有的写死在代码里有的走内部网关结果一到批量跑报告就互相抢配额重试策略也各写各的。后来我把这一层统一收口到 TaoToken先去 TaoToken 官网 获取 Key再把所有 OpenAI 兼容调用的 Base URL 改成https://taotoken.net/api。这篇文章复盘的就是这条 Prefect 流程的改造过程包含可运行的 Flow 代码、运行报告生成方式以及 Claude Code、Codex、CC Switch 的配置对齐。目标不是写一篇行业评论而是让你能在自己的 Prefect 项目里复现同样的接入与排障路径。先说结论Prefect 负责“流程状态”TaoToken 负责“模型调用入口”两者之间用一层薄薄的 task 封装隔开。这样做的好处是重试、缓存、并发、日志、报告都可以留在 Prefect 侧而 Key、Base URL、模型名、超时、限流这些供应商相关的东西只在一个地方出现。下面从最基础的 Key 与 Base URL 开始。2. 接入 TaoToken 的三件套Key、Base URL 与 Prefect Secret Block2.1 先拿 Key不要写进代码不管你是在本地跑 Flow还是把 Flow 部署到 Prefect work pool第一步都是拿 Key。打开 TaoToken 官网登录后进入控制台在 API Keys 页面创建一个新的 Key。创建时建议按用途命名例如prefect-batch-report、prefect-nightly-eval这样后面在运行报告里看到调用来源时更容易排查。拿到 Key 之后不要直接写进flow.py。Prefect 提供了 Secret block适合存放这类凭证。你可以用 Python 代码创建from prefect.blocks.system import Secret Secret(valueYOUR_API_KEY).save(taotoken-api-key, overwriteTrue) print(taotoken-api-key 已保存)如果你们团队使用环境变量注入也可以保留一条 fallbackexport TAOTOKEN_API_KEYYOUR_API_KEY export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_MODELYOUR_MODEL_NAME注意TAOTOKEN_BASE_URL的值就是https://taotoken.net/api不要额外加 UTM 参数。UTM 只用于官网链接追踪API 请求不需要。2.2 Prefect 侧的最小配置结构我在 Flow 里通常只留三个配置入口import os from prefect.blocks.system import Secret BASE_URL os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) DEFAULT_MODEL os.getenv(TAOTOKEN_MODEL, YOUR_MODEL_NAME) def get_api_key() - str: try: return Secret.load(taotoken-api-key).get() except Exception: key os.getenv(TAOTOKEN_API_KEY) if not key: raise RuntimeError(未找到 TAOTOKEN_API_KEY请先在 TaoToken 官网创建 Key) return key这段代码看起来简单但它解决了几个常见问题本地开发可以用环境变量部署到 Prefect 后可以用 Secret block。Base URL 只出现一次后面所有 task 都引用同一个常量。模型名不写死在 task 签名里方便从模型对话页切换。如果你还没有创建 Key可以直接走 API Keys 控制台创建后复制到YOUR_API_KEY的位置。2.3 为什么强调 OpenAI 兼容调用Prefect 的 task 里最常见的调用方式是openaiSDK 或httpx。TaoToken 的 Base URL 是https://taotoken.net/api在 SDK 里配置后请求会走统一的入口。这样做的好处是你不需要为每个模型写一套 SDK也不需要把供应商差异散落到业务代码里。对于工作流工程师来说能在一个 task 里稳定拿到prompt_tokens、completion_tokens、total_tokens比多写几行适配代码重要得多。3. Prefect Flow 实战把 LLM 调用封装成可重试、可统计的 task3.1 完整 Flow 代码下面是一份可以直接复制运行的flow.py。它做了几件事从 Prefect Secret 或环境变量读取 Key。用openaiSDK 指向 TaoToken Base URL。对429、5xx、超时做重试。统计每个 item 的延迟与 token。生成 JSON 与 Markdown 运行报告。不连接任何生产数据库所有输出都落到本地reports/目录。# flow.py import json import os import time from datetime import datetime, timezone from pathlib import Path from openai import OpenAI, APIStatusError, APITimeoutError from prefect import flow, task, get_run_logger from prefect.blocks.system import Secret from prefect.runtime import flow_run BASE_URL os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) DEFAULT_MODEL os.getenv(TAOTOKEN_MODEL, YOUR_MODEL_NAME) REPORT_DIR Path(reports) def get_api_key() - str: try: return Secret.load(taotoken-api-key).get() except Exception: key os.getenv(TAOTOKEN_API_KEY) if not key: raise RuntimeError(未找到 TAOTOKEN_API_KEY请先创建 Key) return key def build_client() - OpenAI: return OpenAI( api_keyget_api_key(), base_urlBASE_URL, timeout60.0, max_retries0, ) task( retries3, retry_delay_seconds[5, 15, 45], log_printsTrue, ) def call_model(prompt: str, model: str, tag: str default) - dict: logger get_run_logger() client build_client() started time.perf_counter() try: response client.chat.completions.create( modelmodel, messages[ { role: system, content: 你是 Prefect 工作流中的结构化文本节点只输出 JSON。, }, {role: user, content: prompt}, ], temperature0.2, max_tokens1024, ) except APIStatusError as exc: logger.error( TaoToken APIStatusError status%s body%s, exc.status_code, exc.response.text[:500], ) raise except APITimeoutError as exc: logger.error(TaoToken 请求超时: %s, exc) raise latency_ms round((time.perf_counter() - started) * 1000, 2) usage response.usage return { tag: tag, requested_model: model, returned_model: response.model, content: response.choices[0].message.content, latency_ms: latency_ms, prompt_tokens: getattr(usage, prompt_tokens, 0), completion_tokens: getattr(usage, completion_tokens, 0), total_tokens: getattr(usage, total_tokens, 0), } def write_report(report: dict) - None: REPORT_DIR.mkdir(exist_okTrue) run_id report[flow_run_id] or local json_path REPORT_DIR / ftaotoken-report-{run_id}.json md_path REPORT_DIR / ftaotoken-report-{run_id}.md json_path.write_text( json.dumps(report, ensure_asciiFalse, indent2), encodingutf-8, ) lines [ f# Prefect TaoToken 运行报告, , f- Flow Run ID: {report[flow_run_id]}, f- Flow Run Name: {report[flow_run_name]}, f- 模型: {report[model]}, f- 条目数: {report[count]}, f- 总 Token: {report[total_tokens]}, f- 平均延迟: {report[avg_latency_ms]} ms, f- 生成时间: {report[generated_at]}, , | # | Tag | 延迟(ms) | Prompt | Completion | Total |, |---|-----|----------|--------|------------|-------|, ] for idx, item in enumerate(report[items], start1): lines.append( f| {idx} | {item[tag]} | {item[latency_ms]} | f{item[prompt_tokens]} | {item[completion_tokens]} | f{item[total_tokens]} | ) md_path.write_text(\n.join(lines), encodingutf-8) print(f报告已写入: {json_path} 和 {md_path}) flow(nameprefect-taotoken-batch, log_printsTrue) def batch_flow(prompts: list[str], model: str DEFAULT_MODEL) - dict: logger get_run_logger() logger.info(使用 Base URL: %s, BASE_URL) logger.info(使用模型: %s, model) items [] for idx, prompt in enumerate(prompts): result call_model(prompt, model, tagfitem-{idx}) items.append(result) total_tokens sum(item[total_tokens] for item in items) avg_latency round( sum(item[latency_ms] for item in items) / max(len(items), 1), 2, ) report { flow_run_id: flow_run.id, flow_run_name: flow_run.name, model: model, count: len(items), total_tokens: total_tokens, avg_latency_ms: avg_latency, generated_at: datetime.now(timezone.utc).isoformat(), items: items, } write_report(report) return report if __name__ __main__: sample_prompts [ 把这句话改写成适合工单的标题Prefect 任务超时。, 输出一个 JSON包含字段 status 和 reason接口返回 429。, 把这段日志归类为网络、鉴权、模型、限流、未知。, ] batch_flow(sample_prompts)运行方式export TAOTOKEN_API_KEYYOUR_API_KEY export TAOTOKEN_MODELYOUR_MODEL_NAME python flow.py如果你已经把 Key 存进 Prefect Secret则只需要设置模型名export TAOTOKEN_MODELYOUR_MODEL_NAME python flow.py运行完成后reports/目录下会出现 JSON 和 Markdown 两份报告。JSON 适合接入后续的监控或 BIMarkdown 适合贴在复盘文档里。3.2 重试策略为什么放在 task 上Prefect 的 task 重试和 OpenAI SDK 自带的重试不要叠在一起。我的做法是把 SDK 的max_retries0关掉把重试完全交给 Prefect。原因有三个Prefect 的重试会记录在 Flow Run 里能看到第几次重试、间隔多久。Prefect 的retry_delay_seconds可以写成[5, 15, 45]比 SDK 的线性退避更可控。当同一个 Flow 里既有普通 HTTP task又有 LLM task 时统一在 Prefect 侧看重试次数更直观。如果你使用 Prefect 3 的缓存策略也可以给call_model加cache_policy。但注意LLM 调用通常不适合无条件缓存建议只对“完全相同的 prompt 模型 参数”做短时间缓存例如from datetime import timedelta from prefect.cache_policies import INPUTS task( retries3, retry_delay_seconds[5, 15, 45], cache_policyINPUTS, cache_expirationtimedelta(hours2), ) def call_model(prompt: str, model: str, tag: str default) - dict: ...这样在重跑同一批 prompt 时可以直接复用结果减少不必要的调用。4. Claude Code、Codex 与 CC Switch 的配置对齐Prefect 流水线解决的是批处理但工作流工程师日常还会用 Claude Code 写 Flow、用 Codex 补配置、用 CC Switch 切供应商。这三者的配置文件不同千万不要把ANTHROPIC_*套到 Codex 上。4.1 Claude Codesettings.json 里写 ANTHROPIC_*Claude Code 的配置通常放在~/.claude/settings.json或项目级.claude/settings.json。你要做的是把 Base URL 指向https://taotoken.net/apiKey 用YOUR_API_KEY占位{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_AUTH_TOKEN: YOUR_API_KEY, ANTHROPIC_MODEL: YOUR_MODEL_NAME, ANTHROPIC_SMALL_FAST_MODEL: YOUR_FAST_MODEL_NAME } }改完后重启 Claude Code然后在会话里输入/status或类似命令检查当前模型与端点。如果出现401优先检查ANTHROPIC_AUTH_TOKEN是否有多余空格如果出现404检查ANTHROPIC_BASE_URL是否误写成了带/v1的地址。产品事实里给定的 Base URL 就是https://taotoken.net/api不要自行拼接。4.2 Codexconfig.toml 用 model_providersCodex 使用~/.codex/config.toml配置结构和 Claude Code 完全不同。下面是一个 provider 示例model YOUR_MODEL_NAME model_provider taotoken [model_providers.taotoken] name TaoToken base_url https://taotoken.net/api env_key TAOTOKEN_API_KEY wire_api chat然后在 shell 里导出 Keyexport TAOTOKEN_API_KEYYOUR_API_KEY再运行 Codex。注意这里用的是TAOTOKEN_API_KEY不是ANTHROPIC_AUTH_TOKEN。如果你之前把 Claude Code 的变量复制到 Codex会出现鉴权失败或找不到 provider 的问题。4.3 CC Switch 三件套provider 条目、当前选中项、settings.json如果你用 CC Switch 管理多套 Claude Code 配置重点检查三件套providers 列表里有一条 TaoToken 条目。当前选中的 provider 是 TaoToken。Claude Code 的settings.json里的env与 CC Switch 写入的一致。一个常见的 CC Switch provider 条目如下{ name: TaoToken, settingsConfig: { env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_AUTH_TOKEN: YOUR_API_KEY, ANTHROPIC_MODEL: YOUR_MODEL_NAME } } }切换后不要只重启终端最好把 Claude Code 完全退出再打开。如果仍然报旧 Key 的 401检查系统环境变量里是否残留了旧的ANTHROPIC_AUTH_TOKEN它的优先级可能高于配置文件。5. 调用量抬升后的排障与压测429、超时、成本与观测当 Prefect 流水线从每天几十次调用变成成百上千次调用时最先暴露的通常不是模型效果而是限流、超时和统计缺失。下面是一张我在复盘时常用的排障表。现象优先检查Prefect 侧动作TaoToken 侧动作401 UnauthorizedKey 是否为空、是否复制错检查 Secret block 与环境变量在控制台重新创建 Key404 Not FoundBase URL 是否写成/api/v1确认BASE_URL常量查看模型名是否可用429 Too Many Requests并发是否过高降低并发、增加 task 重试间隔查看当前套餐与限额APITimeoutError单次请求是否过长设置timeout60拆分 prompt换用更小模型或流式Token 统计为 0响应对象是否被裁剪打印response.usage确认模型是否返回 usage5.1 429 的处理429 不一定意味着 Key 有问题可能是瞬时并发过高。Prefect 里最简单的做法是给 task 加retries和retry_delay_seconds同时把 Flow 的并发压下来。如果你使用 Prefect 的并发限制可以给 task 加tags然后在 work pool 侧限制并发。不要在代码里写while True无限重试那会把一次限流放大成雪崩。5.2 超时的处理LLM 请求超时通常有两种原因prompt 太长或者模型输出太长。我的做法是在 Prefect task 里显式设置timeout60.0。对长文本先做切分再用多个 task 并发处理最后汇总。对不需要完整输出的场景设置较小的max_tokens。对必须流式的场景使用streamTrue并在 task 里逐块读取避免一次性等待。一个流式请求的本地验证命令如下你可以先在终端确认 TaoToken 端点可达curl -sS https://taotoken.net/api/chat/completions \ -H Authorization: Bearer YOUR_API_KEY \ -H Content-Type: application/json \ -d { model: YOUR_MODEL_NAME, messages: [{role: user, content: 只回复 ok}], stream: false }这个命令只是本地验证不要把生产 Key 写进脚本历史记录。如果你在 CI 里执行请用 secret 注入。5.3 运行报告要记录什么可复现的复盘不仅要有代码还要有报告。建议运行报告至少记录flow_run_id与flow_run_name使用的模型名和 Base URL每个 item 的延迟、prompt_tokens、completion_tokens、total_tokens重试次数与失败原因成功、失败、跳过数量生成时间与代码版本上面的 Flow 代码已经把这些字段写进了 JSON 和 Markdown。你可以把reports/目录挂到 CI artifact或者在 Prefect 里再接一个通知 task把 Markdown 报告发到团队频道。6. 复盘清单从“能跑通”到“可观测”改造完成后我通常会按下面这份清单做一次复盘。它不是新闻评论而是工作流工程师的检查表。配置收敛Base URL 是否只出现一次Key 是否只从 Secret 或环境变量读取重试边界SDK 重试是否关闭Prefect 重试是否覆盖 429、5xx、超时并发控制Flow 并发是否超过 TaoToken 侧限额是否有 work pool 限制统计完整每个 task 是否返回 usage汇总报告是否包含总 token 和平均延迟失败可见失败 task 是否能在 Prefect UI 里看到 error body 片段本地可复现新同学能否只靠文档和YOUR_API_KEY跑通一份样例工具配置对齐Claude Code 用ANTHROPIC_*Codex 用config.tomlCC Switch 三件套是否一致安全边界没有把 Key 写进代码、没有让 Flow 直连生产库、没有把 SQL 执行放到模型侧。如果你准备把 Prefect 流程从零搭起来建议先跑一个最小 Flow三个 prompt、一个模型、一份报告。跑通后再逐步加并发、缓存、通知和部署。不要一上来就把所有业务逻辑塞进一个 task那样后面排障会很痛苦。7. 下一步从模型对话到 Coding Plan把 Key 和文档补齐到这里Prefect Flow 的改造路径已经完整获取 Key、设置 Base URL、封装 task、加重试与统计、生成运行报告、对齐 Claude Code 与 Codex 配置。如果你还没有开始可以按下面的顺序操作先去 模型对话 验证模型名和基础调用。如果长期跑批处理或 Coding Agent查看 Coding Plan 选择合适的使用方式。然后到 创建 API Key 生成YOUR_API_KEY写入 Prefect Secret block。如果你同时使用 Claude Code参考 Claude Code 文档 完成settings.json配置。最后再提醒一次Prefect 侧负责流程编排、重试、缓存、日志和报告TaoToken 侧负责统一模型入口。Base URL 用https://taotoken.net/apiKey 用YOUR_API_KEY不要把 UTM 参数带进 API 请求。按上面的 Flow 代码跑一遍你应该能拿到一份带 token 统计和延迟明细的运行报告然后再根据报告去调并发、重试和模型选择。 Prefect 流程改用 TaoToken，复盘千倍级增长
1. 从一次 Prefect 流水线的 429 报警说起为什么把 LLM 调用统一到 TaoToken最近央视财经关于 Token 调用量快速攀升的报道引发了不少讨论但作为工作流工程师真正让我从椅子上弹起来的不是新闻而是 Prefect task 里反复出现的429 Too Many Requests和APITimeoutError。我们的流水线原本把多个模型的调用散落在不同 task 里有的用环境变量有的写死在代码里有的走内部网关结果一到批量跑报告就互相抢配额重试策略也各写各的。后来我把这一层统一收口到 TaoToken先去 TaoToken 官网 获取 Key再把所有 OpenAI 兼容调用的 Base URL 改成https://taotoken.net/api。这篇文章复盘的就是这条 Prefect 流程的改造过程包含可运行的 Flow 代码、运行报告生成方式以及 Claude Code、Codex、CC Switch 的配置对齐。目标不是写一篇行业评论而是让你能在自己的 Prefect 项目里复现同样的接入与排障路径。先说结论Prefect 负责“流程状态”TaoToken 负责“模型调用入口”两者之间用一层薄薄的 task 封装隔开。这样做的好处是重试、缓存、并发、日志、报告都可以留在 Prefect 侧而 Key、Base URL、模型名、超时、限流这些供应商相关的东西只在一个地方出现。下面从最基础的 Key 与 Base URL 开始。2. 接入 TaoToken 的三件套Key、Base URL 与 Prefect Secret Block2.1 先拿 Key不要写进代码不管你是在本地跑 Flow还是把 Flow 部署到 Prefect work pool第一步都是拿 Key。打开 TaoToken 官网登录后进入控制台在 API Keys 页面创建一个新的 Key。创建时建议按用途命名例如prefect-batch-report、prefect-nightly-eval这样后面在运行报告里看到调用来源时更容易排查。拿到 Key 之后不要直接写进flow.py。Prefect 提供了 Secret block适合存放这类凭证。你可以用 Python 代码创建from prefect.blocks.system import Secret Secret(valueYOUR_API_KEY).save(taotoken-api-key, overwriteTrue) print(taotoken-api-key 已保存)如果你们团队使用环境变量注入也可以保留一条 fallbackexport TAOTOKEN_API_KEYYOUR_API_KEY export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_MODELYOUR_MODEL_NAME注意TAOTOKEN_BASE_URL的值就是https://taotoken.net/api不要额外加 UTM 参数。UTM 只用于官网链接追踪API 请求不需要。2.2 Prefect 侧的最小配置结构我在 Flow 里通常只留三个配置入口import os from prefect.blocks.system import Secret BASE_URL os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) DEFAULT_MODEL os.getenv(TAOTOKEN_MODEL, YOUR_MODEL_NAME) def get_api_key() - str: try: return Secret.load(taotoken-api-key).get() except Exception: key os.getenv(TAOTOKEN_API_KEY) if not key: raise RuntimeError(未找到 TAOTOKEN_API_KEY请先在 TaoToken 官网创建 Key) return key这段代码看起来简单但它解决了几个常见问题本地开发可以用环境变量部署到 Prefect 后可以用 Secret block。Base URL 只出现一次后面所有 task 都引用同一个常量。模型名不写死在 task 签名里方便从模型对话页切换。如果你还没有创建 Key可以直接走 API Keys 控制台创建后复制到YOUR_API_KEY的位置。2.3 为什么强调 OpenAI 兼容调用Prefect 的 task 里最常见的调用方式是openaiSDK 或httpx。TaoToken 的 Base URL 是https://taotoken.net/api在 SDK 里配置后请求会走统一的入口。这样做的好处是你不需要为每个模型写一套 SDK也不需要把供应商差异散落到业务代码里。对于工作流工程师来说能在一个 task 里稳定拿到prompt_tokens、completion_tokens、total_tokens比多写几行适配代码重要得多。3. Prefect Flow 实战把 LLM 调用封装成可重试、可统计的 task3.1 完整 Flow 代码下面是一份可以直接复制运行的flow.py。它做了几件事从 Prefect Secret 或环境变量读取 Key。用openaiSDK 指向 TaoToken Base URL。对429、5xx、超时做重试。统计每个 item 的延迟与 token。生成 JSON 与 Markdown 运行报告。不连接任何生产数据库所有输出都落到本地reports/目录。# flow.py import json import os import time from datetime import datetime, timezone from pathlib import Path from openai import OpenAI, APIStatusError, APITimeoutError from prefect import flow, task, get_run_logger from prefect.blocks.system import Secret from prefect.runtime import flow_run BASE_URL os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) DEFAULT_MODEL os.getenv(TAOTOKEN_MODEL, YOUR_MODEL_NAME) REPORT_DIR Path(reports) def get_api_key() - str: try: return Secret.load(taotoken-api-key).get() except Exception: key os.getenv(TAOTOKEN_API_KEY) if not key: raise RuntimeError(未找到 TAOTOKEN_API_KEY请先创建 Key) return key def build_client() - OpenAI: return OpenAI( api_keyget_api_key(), base_urlBASE_URL, timeout60.0, max_retries0, ) task( retries3, retry_delay_seconds[5, 15, 45], log_printsTrue, ) def call_model(prompt: str, model: str, tag: str default) - dict: logger get_run_logger() client build_client() started time.perf_counter() try: response client.chat.completions.create( modelmodel, messages[ { role: system, content: 你是 Prefect 工作流中的结构化文本节点只输出 JSON。, }, {role: user, content: prompt}, ], temperature0.2, max_tokens1024, ) except APIStatusError as exc: logger.error( TaoToken APIStatusError status%s body%s, exc.status_code, exc.response.text[:500], ) raise except APITimeoutError as exc: logger.error(TaoToken 请求超时: %s, exc) raise latency_ms round((time.perf_counter() - started) * 1000, 2) usage response.usage return { tag: tag, requested_model: model, returned_model: response.model, content: response.choices[0].message.content, latency_ms: latency_ms, prompt_tokens: getattr(usage, prompt_tokens, 0), completion_tokens: getattr(usage, completion_tokens, 0), total_tokens: getattr(usage, total_tokens, 0), } def write_report(report: dict) - None: REPORT_DIR.mkdir(exist_okTrue) run_id report[flow_run_id] or local json_path REPORT_DIR / ftaotoken-report-{run_id}.json md_path REPORT_DIR / ftaotoken-report-{run_id}.md json_path.write_text( json.dumps(report, ensure_asciiFalse, indent2), encodingutf-8, ) lines [ f# Prefect TaoToken 运行报告, , f- Flow Run ID: {report[flow_run_id]}, f- Flow Run Name: {report[flow_run_name]}, f- 模型: {report[model]}, f- 条目数: {report[count]}, f- 总 Token: {report[total_tokens]}, f- 平均延迟: {report[avg_latency_ms]} ms, f- 生成时间: {report[generated_at]}, , | # | Tag | 延迟(ms) | Prompt | Completion | Total |, |---|-----|----------|--------|------------|-------|, ] for idx, item in enumerate(report[items], start1): lines.append( f| {idx} | {item[tag]} | {item[latency_ms]} | f{item[prompt_tokens]} | {item[completion_tokens]} | f{item[total_tokens]} | ) md_path.write_text(\n.join(lines), encodingutf-8) print(f报告已写入: {json_path} 和 {md_path}) flow(nameprefect-taotoken-batch, log_printsTrue) def batch_flow(prompts: list[str], model: str DEFAULT_MODEL) - dict: logger get_run_logger() logger.info(使用 Base URL: %s, BASE_URL) logger.info(使用模型: %s, model) items [] for idx, prompt in enumerate(prompts): result call_model(prompt, model, tagfitem-{idx}) items.append(result) total_tokens sum(item[total_tokens] for item in items) avg_latency round( sum(item[latency_ms] for item in items) / max(len(items), 1), 2, ) report { flow_run_id: flow_run.id, flow_run_name: flow_run.name, model: model, count: len(items), total_tokens: total_tokens, avg_latency_ms: avg_latency, generated_at: datetime.now(timezone.utc).isoformat(), items: items, } write_report(report) return report if __name__ __main__: sample_prompts [ 把这句话改写成适合工单的标题Prefect 任务超时。, 输出一个 JSON包含字段 status 和 reason接口返回 429。, 把这段日志归类为网络、鉴权、模型、限流、未知。, ] batch_flow(sample_prompts)运行方式export TAOTOKEN_API_KEYYOUR_API_KEY export TAOTOKEN_MODELYOUR_MODEL_NAME python flow.py如果你已经把 Key 存进 Prefect Secret则只需要设置模型名export TAOTOKEN_MODELYOUR_MODEL_NAME python flow.py运行完成后reports/目录下会出现 JSON 和 Markdown 两份报告。JSON 适合接入后续的监控或 BIMarkdown 适合贴在复盘文档里。3.2 重试策略为什么放在 task 上Prefect 的 task 重试和 OpenAI SDK 自带的重试不要叠在一起。我的做法是把 SDK 的max_retries0关掉把重试完全交给 Prefect。原因有三个Prefect 的重试会记录在 Flow Run 里能看到第几次重试、间隔多久。Prefect 的retry_delay_seconds可以写成[5, 15, 45]比 SDK 的线性退避更可控。当同一个 Flow 里既有普通 HTTP task又有 LLM task 时统一在 Prefect 侧看重试次数更直观。如果你使用 Prefect 3 的缓存策略也可以给call_model加cache_policy。但注意LLM 调用通常不适合无条件缓存建议只对“完全相同的 prompt 模型 参数”做短时间缓存例如from datetime import timedelta from prefect.cache_policies import INPUTS task( retries3, retry_delay_seconds[5, 15, 45], cache_policyINPUTS, cache_expirationtimedelta(hours2), ) def call_model(prompt: str, model: str, tag: str default) - dict: ...这样在重跑同一批 prompt 时可以直接复用结果减少不必要的调用。4. Claude Code、Codex 与 CC Switch 的配置对齐Prefect 流水线解决的是批处理但工作流工程师日常还会用 Claude Code 写 Flow、用 Codex 补配置、用 CC Switch 切供应商。这三者的配置文件不同千万不要把ANTHROPIC_*套到 Codex 上。4.1 Claude Codesettings.json 里写 ANTHROPIC_*Claude Code 的配置通常放在~/.claude/settings.json或项目级.claude/settings.json。你要做的是把 Base URL 指向https://taotoken.net/apiKey 用YOUR_API_KEY占位{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_AUTH_TOKEN: YOUR_API_KEY, ANTHROPIC_MODEL: YOUR_MODEL_NAME, ANTHROPIC_SMALL_FAST_MODEL: YOUR_FAST_MODEL_NAME } }改完后重启 Claude Code然后在会话里输入/status或类似命令检查当前模型与端点。如果出现401优先检查ANTHROPIC_AUTH_TOKEN是否有多余空格如果出现404检查ANTHROPIC_BASE_URL是否误写成了带/v1的地址。产品事实里给定的 Base URL 就是https://taotoken.net/api不要自行拼接。4.2 Codexconfig.toml 用 model_providersCodex 使用~/.codex/config.toml配置结构和 Claude Code 完全不同。下面是一个 provider 示例model YOUR_MODEL_NAME model_provider taotoken [model_providers.taotoken] name TaoToken base_url https://taotoken.net/api env_key TAOTOKEN_API_KEY wire_api chat然后在 shell 里导出 Keyexport TAOTOKEN_API_KEYYOUR_API_KEY再运行 Codex。注意这里用的是TAOTOKEN_API_KEY不是ANTHROPIC_AUTH_TOKEN。如果你之前把 Claude Code 的变量复制到 Codex会出现鉴权失败或找不到 provider 的问题。4.3 CC Switch 三件套provider 条目、当前选中项、settings.json如果你用 CC Switch 管理多套 Claude Code 配置重点检查三件套providers 列表里有一条 TaoToken 条目。当前选中的 provider 是 TaoToken。Claude Code 的settings.json里的env与 CC Switch 写入的一致。一个常见的 CC Switch provider 条目如下{ name: TaoToken, settingsConfig: { env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_AUTH_TOKEN: YOUR_API_KEY, ANTHROPIC_MODEL: YOUR_MODEL_NAME } } }切换后不要只重启终端最好把 Claude Code 完全退出再打开。如果仍然报旧 Key 的 401检查系统环境变量里是否残留了旧的ANTHROPIC_AUTH_TOKEN它的优先级可能高于配置文件。5. 调用量抬升后的排障与压测429、超时、成本与观测当 Prefect 流水线从每天几十次调用变成成百上千次调用时最先暴露的通常不是模型效果而是限流、超时和统计缺失。下面是一张我在复盘时常用的排障表。现象优先检查Prefect 侧动作TaoToken 侧动作401 UnauthorizedKey 是否为空、是否复制错检查 Secret block 与环境变量在控制台重新创建 Key404 Not FoundBase URL 是否写成/api/v1确认BASE_URL常量查看模型名是否可用429 Too Many Requests并发是否过高降低并发、增加 task 重试间隔查看当前套餐与限额APITimeoutError单次请求是否过长设置timeout60拆分 prompt换用更小模型或流式Token 统计为 0响应对象是否被裁剪打印response.usage确认模型是否返回 usage5.1 429 的处理429 不一定意味着 Key 有问题可能是瞬时并发过高。Prefect 里最简单的做法是给 task 加retries和retry_delay_seconds同时把 Flow 的并发压下来。如果你使用 Prefect 的并发限制可以给 task 加tags然后在 work pool 侧限制并发。不要在代码里写while True无限重试那会把一次限流放大成雪崩。5.2 超时的处理LLM 请求超时通常有两种原因prompt 太长或者模型输出太长。我的做法是在 Prefect task 里显式设置timeout60.0。对长文本先做切分再用多个 task 并发处理最后汇总。对不需要完整输出的场景设置较小的max_tokens。对必须流式的场景使用streamTrue并在 task 里逐块读取避免一次性等待。一个流式请求的本地验证命令如下你可以先在终端确认 TaoToken 端点可达curl -sS https://taotoken.net/api/chat/completions \ -H Authorization: Bearer YOUR_API_KEY \ -H Content-Type: application/json \ -d { model: YOUR_MODEL_NAME, messages: [{role: user, content: 只回复 ok}], stream: false }这个命令只是本地验证不要把生产 Key 写进脚本历史记录。如果你在 CI 里执行请用 secret 注入。5.3 运行报告要记录什么可复现的复盘不仅要有代码还要有报告。建议运行报告至少记录flow_run_id与flow_run_name使用的模型名和 Base URL每个 item 的延迟、prompt_tokens、completion_tokens、total_tokens重试次数与失败原因成功、失败、跳过数量生成时间与代码版本上面的 Flow 代码已经把这些字段写进了 JSON 和 Markdown。你可以把reports/目录挂到 CI artifact或者在 Prefect 里再接一个通知 task把 Markdown 报告发到团队频道。6. 复盘清单从“能跑通”到“可观测”改造完成后我通常会按下面这份清单做一次复盘。它不是新闻评论而是工作流工程师的检查表。配置收敛Base URL 是否只出现一次Key 是否只从 Secret 或环境变量读取重试边界SDK 重试是否关闭Prefect 重试是否覆盖 429、5xx、超时并发控制Flow 并发是否超过 TaoToken 侧限额是否有 work pool 限制统计完整每个 task 是否返回 usage汇总报告是否包含总 token 和平均延迟失败可见失败 task 是否能在 Prefect UI 里看到 error body 片段本地可复现新同学能否只靠文档和YOUR_API_KEY跑通一份样例工具配置对齐Claude Code 用ANTHROPIC_*Codex 用config.tomlCC Switch 三件套是否一致安全边界没有把 Key 写进代码、没有让 Flow 直连生产库、没有把 SQL 执行放到模型侧。如果你准备把 Prefect 流程从零搭起来建议先跑一个最小 Flow三个 prompt、一个模型、一份报告。跑通后再逐步加并发、缓存、通知和部署。不要一上来就把所有业务逻辑塞进一个 task那样后面排障会很痛苦。7. 下一步从模型对话到 Coding Plan把 Key 和文档补齐到这里Prefect Flow 的改造路径已经完整获取 Key、设置 Base URL、封装 task、加重试与统计、生成运行报告、对齐 Claude Code 与 Codex 配置。如果你还没有开始可以按下面的顺序操作先去 模型对话 验证模型名和基础调用。如果长期跑批处理或 Coding Agent查看 Coding Plan 选择合适的使用方式。然后到 创建 API Key 生成YOUR_API_KEY写入 Prefect Secret block。如果你同时使用 Claude Code参考 Claude Code 文档 完成settings.json配置。最后再提醒一次Prefect 侧负责流程编排、重试、缓存、日志和报告TaoToken 侧负责统一模型入口。Base URL 用https://taotoken.net/apiKey 用YOUR_API_KEY不要把 UTM 参数带进 API 请求。按上面的 Flow 代码跑一遍你应该能拿到一份带 token 统计和延迟明细的运行报告然后再根据报告去调并发、重试和模型选择。