1. 凌晨的 Cohere 批处理 401为什么这次只需要改 Key 与 Base URLCohere 批处理在第 17 个分片报401 invalid api keyCO_API_KEY失效。先别重写任务TaoToken 官网https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_intro注册拿 Key再把 Base URL 设为https://taotoken.net/api。最近 Cohere 与 Aleph Alpha 签署最终协议、计划统一品牌运营模型供应格局出现新变量对批处理开发者来说最务实的动作不是等公告而是把 Key 与 Base URL 从.env层解耦让任务能在五分钟内切换。本文以“Cohere 批处理换 Key”为主线给出可复制的.env对照、迁移脚本、冒烟验证与排障清单。你不需要改业务流程只需要把供应商配置抽出来把旧 Key 替换成 TaoToken 的YOUR_API_KEY并把客户端 Base URL 指向https://taotoken.net/api。很多批处理项目最初是这样写的import os import cohere co cohere.Client(os.getenv(CO_API_KEY)) response co.chat( modelcommand-r, message把这条工单总结成一句话 )这类代码的痛点是Key、模型名、Base URL 散落在业务文件里。一旦 Key 失效或者供应商侧策略变化你就要全仓搜索CO_API_KEY、COHERE_API_KEY、cohere.Client。更麻烦的是部分 SDK 并不支持显式指定 Base URL导致“换 Key”变成“换调用方式”。所以这次迁移目标很明确在 TaoToken 官网注册并创建 Key得到YOUR_API_KEY。把.env中 Cohere 相关变量改成 TaoToken 变量。把客户端 Base URL 改为https://taotoken.net/api。用迁移脚本扫描旧引用保留备份。用冒烟脚本验证单请求与 3 条批量样本。如果同时使用 Claude Code、Codex 或 CC Switch按各自配置文件写入不要把ANTHROPIC_*套到 Codex。这套流程的核心不是“重写批处理”而是“把供应商配置变成可替换层”。下面直接进入可复制步骤。2. .env 对照Cohere 变量到 TaoToken 变量的最小改动批处理开发者最关心的通常是原来.env里那些变量怎么改下面给出一张最小对照表。旧变量名可能因项目而异但核心只有三类Key、Base URL、模型名。旧变量新变量示例说明COHERE_API_KEYTAOTOKEN_API_KEYYOUR_API_KEY在 TaoToken 控制台创建CO_API_KEYTAOTOKEN_API_KEYYOUR_API_KEY兼容旧脚本中的短变量名COHERE_BASE_URLTAOTOKEN_BASE_URLhttps://taotoken.net/api不加 UTM直接用于客户端COHERE_MODELTAOTOKEN_MODELYOUR_MODEL_ID从模型列表或控制台确认COHERE_BATCH_SIZETAOTOKEN_BATCH_SIZE20批大小按限流调整COHERE_TIMEOUTTAOTOKEN_TIMEOUT60单位秒COHERE_MAX_RETRIESTAOTOKEN_MAX_RETRIES3配合指数退避新的.env可以写成这样# TaoToken 配置 TAOTOKEN_API_KEYYOUR_API_KEY TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODELYOUR_MODEL_ID # 批处理参数 TAOTOKEN_BATCH_SIZE20 TAOTOKEN_TIMEOUT60 TAOTOKEN_MAX_RETRIES3 # 兼容旧代码的过渡变量迁移完成后可删除 # CO_API_KEYYOUR_API_KEY # COHERE_API_KEYYOUR_API_KEY # COHERE_MODELYOUR_MODEL_ID如果你还没有 Key先到 TaoToken 官网注册https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_env_register注册后进入控制台创建 Key。创建时建议按项目命名例如cohere-batch-migrate方便后续轮换。不要把真实 Key 写进.env.example也不要提交到 Git。.env.example只保留占位符TAOTOKEN_API_KEYYOUR_API_KEY TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODELYOUR_MODEL_ID TAOTOKEN_BATCH_SIZE20这里有一个容易忽略的点TAOTOKEN_BASE_URL的值是https://taotoken.net/api不要在末尾追加/v1也不要改成其他路径。很多 404 都来自 Base URL 拼接错误。客户端通常会在 Base URL 后拼接/chat/completions之类的路径重复/v1会导致请求落到错误路由。另外模型名不要凭记忆写。批处理任务里经常把模型名硬编码成command-r、command-r-plus之类。迁移到 TaoToken 后应该从控制台或模型对话页面确认可用模型 ID再写入TAOTOKEN_MODELYOUR_MODEL_ID。如果你只是先跑通可以在模型对话页面验证同一个提示词确认模型可用后再回填到.env。3. 五分钟迁移脚本扫描旧 Key 引用并生成 .env 对照真正耗时的不是改一行 Key而是找到所有旧引用。下面这个 Python 脚本可以放在项目根目录执行。它会扫描常见文本文件找出CO_API_KEY、COHERE_API_KEY、COHERE_MODEL、cohere.Client等引用备份现有.env并生成.env.taotoken对照文件。#!/usr/bin/env python3 migrate_cohere_env.py 用途扫描 Cohere 旧 Key 引用备份 .env生成 TaoToken 迁移对照。 建议先本地执行确认输出后再手动替换代码。 from pathlib import Path import shutil import re from datetime import datetime ROOT Path.cwd() ENV_FILE ROOT / .env BACKUP_DIR ROOT / .migration_backup TAOTOKEN_ENV ROOT / .env.taotoken TEXT_EXTENSIONS { .py, .js, .ts, .tsx, .jsx, .json, .toml, .yaml, .yml, .env, .example, .md, .txt, .sh, .bash } PATTERNS { CO_API_KEY: re.compile(rCO_API_KEY), COHERE_API_KEY: re.compile(rCOHERE_API_KEY), COHERE_MODEL: re.compile(rCOHERE_MODEL), COHERE_BASE_URL: re.compile(rCOHERE_BASE_URL), cohere.Client: re.compile(rcohere\.Client), cohere.ClientV2: re.compile(rcohere\.ClientV2), } def iter_text_files(): for p in ROOT.rglob(*): if p.is_dir(): continue if any(part in {.git, node_modules, __pycache__, .venv, venv} for part in p.parts): continue if p.suffix.lower() in TEXT_EXTENSIONS or p.name.startswith(.env): yield p def scan(): hits [] for file in iter_text_files(): try: content file.read_text(encodingutf-8, errorsignore) except Exception: continue for name, pattern in PATTERNS.items(): for match in pattern.finditer(content): line_no content[:match.start()].count(\n) 1 hits.append((name, file.relative_to(ROOT), line_no)) return hits def backup_env(): BACKUP_DIR.mkdir(exist_okTrue) if ENV_FILE.exists(): stamp datetime.now().strftime(%Y%m%d_%H%M%S) target BACKUP_DIR / f.env.backup_{stamp} shutil.copy2(ENV_FILE, target) print(f[备份] {ENV_FILE} - {target}) else: print([提示] 当前目录没有 .env将生成 .env.taotoken 模板) def write_taotoken_env(): content # TaoToken 迁移配置 TAOTOKEN_API_KEYYOUR_API_KEY TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODELYOUR_MODEL_ID TAOTOKEN_BATCH_SIZE20 TAOTOKEN_TIMEOUT60 TAOTOKEN_MAX_RETRIES3 # 旧变量对照确认迁移完成后删除 # CO_API_KEYYOUR_API_KEY # COHERE_API_KEYYOUR_API_KEY # COHERE_MODELYOUR_MODEL_ID # COHERE_BASE_URLhttps://taotoken.net/api TAOTOKEN_ENV.write_text(content, encodingutf-8) print(f[生成] {TAOTOKEN_ENV}) def main(): print( Cohere - TaoToken 迁移扫描 ) hits scan() if not hits: print(未发现 Cohere 旧变量引用。) else: print(\n发现以下引用) for name, file, line_no in hits: print(f - {name:18} {file}:{line_no}) backup_env() write_taotoken_env() print(\n下一步) print(1. 对比 .env 与 .env.taotoken把 TAOTOKEN_* 写入 .env。) print(2. 把业务代码中的旧 Key 读取改为 TAOTOKEN_API_KEY。) print(3. 把 Base URL 改为 https://taotoken.net/api。) print(4. 执行 smoke_test.py 验证。) if __name__ __main__: main()执行方式python3 migrate_cohere_env.py输出示例 Cohere - TaoToken 迁移扫描 发现以下引用 - CO_API_KEY batch_job.py:12 - COHERE_MODEL batch_job.py:18 - cohere.Client batch_job.py:25 [备份] /project/.env - /project/.migration_backup/.env.backup_20250101_120000 [生成] /project/.env.taotoken这个脚本不会自动改业务代码这是有意为之。批处理任务通常涉及重试、限流、日志、幂等自动替换容易把cohere.Client改成不可用的写法。更安全的做法是先让脚本帮你找出所有引用再按下一节把调用层替换成 OpenAI 兼容方式。如果你在团队里协作可以把扫描结果贴到迁移 PR 描述中。确认.env.taotoken内容后再把TAOTOKEN_API_KEY写入本地.env。Key 只存在本地或密钥管理服务不要进入代码仓库。4. 批处理客户端改造把 Cohere 调用换成 OpenAI 兼容 Base URL迁移到 TaoToken 后最稳妥的调用方式是使用 OpenAI 兼容客户端Base URL 指向https://taotoken.net/api。这样你不需要依赖 Cohere SDK 是否支持自定义 Base URL也不需要在业务代码里保留两套客户端。假设原来的批处理逻辑是逐条调用 Cohere现在可以改成下面这样# batch_process.py import os import time import json from pathlib import Path from dotenv import load_dotenv from openai import OpenAI load_dotenv() API_KEY os.getenv(TAOTOKEN_API_KEY) BASE_URL os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) MODEL os.getenv(TAOTOKEN_MODEL, YOUR_MODEL_ID) BATCH_SIZE int(os.getenv(TAOTOKEN_BATCH_SIZE, 20)) TIMEOUT float(os.getenv(TAOTOKEN_TIMEOUT, 60)) MAX_RETRIES int(os.getenv(TAOTOKEN_MAX_RETRIES, 3)) if not API_KEY or API_KEY YOUR_API_KEY: raise RuntimeError(请先在 .env 中设置 TAOTOKEN_API_KEY) client OpenAI( api_keyAPI_KEY, base_urlBASE_URL, timeoutTIMEOUT, max_retriesMAX_RETRIES, ) def summarize_ticket(ticket: str) - str: resp client.chat.completions.create( modelMODEL, messages[ {role: system, content: 你是工单摘要助手输出一句话中文摘要。}, {role: user, content: ticket}, ], temperature0.2, ) return resp.choices[0].message.content.strip() def run_batch(rows: list[str], output_path: str batch_result.jsonl): success, failed 0, 0 with open(output_path, w, encodingutf-8) as f: for i in range(0, len(rows), BATCH_SIZE): chunk rows[i:i BATCH_SIZE] for idx, row in enumerate(chunk, starti): try: result summarize_ticket(row) record {index: idx, ok: True, result: result} success 1 except Exception as e: record {index: idx, ok: False, error: str(e)} failed 1 f.write(json.dumps(record, ensure_asciiFalse) \n) f.flush() print(f进度{min(i BATCH_SIZE, len(rows))}/{len(rows)}成功 {success}失败 {failed}) time.sleep(0.5) print(f完成输出{output_path}) if __name__ __main__: sample_rows [ 客户反馈登录后页面空白清理缓存后恢复。, 订单支付成功但状态未同步需要人工核查。, API 返回 429建议降低并发并增加重试。, ] run_batch(sample_rows, batch_result_smoke.jsonl)这段代码的关键点base_url使用https://taotoken.net/api不加 UTM不加多余/v1。api_key使用YOUR_API_KEY实际值从.env读取。模型名使用YOUR_MODEL_ID不要硬编码旧 Cohere 模型名。批处理按BATCH_SIZE分块每块后短暂 sleep降低 429 概率。失败记录单独写入 JSONL便于断点续跑。如果你原来的代码必须使用 Cohere SDK可以先检查该 SDK 是否支持base_url参数。支持则传入https://taotoken.net/api不支持则不要强行改直接换到上面的 OpenAI 兼容调用。批处理的核心逻辑是读入、调用、写结果客户端替换不会影响外层幂等设计。另外一个实用技巧是把调用层封装成函数这样迁移时只改一个文件# provider_client.py import os from openai import OpenAI def get_client(): return OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api), timeoutfloat(os.getenv(TAOTOKEN_TIMEOUT, 60)), max_retriesint(os.getenv(TAOTOKEN_MAX_RETRIES, 3)), )业务文件只保留from provider_client import get_client client get_client()这样下次再换供应商只需要改provider_client.py和.env。5. 五分钟冒烟验证单请求、小样本、日志检查迁移后不要直接跑全量批处理。先用单请求确认 Key、Base URL、模型名三项配置正确。# smoke_test.py import os from dotenv import load_dotenv from openai import OpenAI load_dotenv() client OpenAI( api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api), timeout30, ) resp client.chat.completions.create( modelos.getenv(TAOTOKEN_MODEL, YOUR_MODEL_ID), messages[ {role: user, content: 只回复四个字迁移成功} ], ) print(resp.choices[0].message.content)执行python3 smoke_test.py如果返回“迁移成功”或类似内容说明 Key、Base URL、模型名基本正确。如果报 401检查.env是否被正确加载如果报 404检查 Base URL 是否为https://taotoken.net/api以及模型名是否为控制台中的可用 ID如果报 429降低并发增加退避。第二步用 3 条样本跑小批量python3 batch_process.py观察batch_result_smoke.jsonl{index: 0, ok: true, result: 客户登录后页面空白清理缓存后恢复。} {index: 1, ok: true, result: 支付成功但订单状态未同步需人工核查。} {index: 2, ok: true, result: API 返回 429建议降低并发并增加重试。}第三步检查日志中是否还有旧变量。可以用grep -R CO_API_KEY\|COHERE_API_KEY\|COHERE_MODEL . \ --exclude-dir.git \ --exclude-dirnode_modules \ --exclude-dir.venv如果只剩注释或备份文件就可以逐步删掉旧变量。注意命令由你在本地执行不要在生产数据库或关键目录里直接跑破坏性命令。6. 排障401、404、429、超时与 JSON 解析失败批处理换 Key 最常见的错误不是模型能力问题而是配置细节。下面按错误码给出排查顺序。6.1 401 invalid api key可能原因.env中TAOTOKEN_API_KEY仍是YOUR_API_KEY。代码读取的是旧变量CO_API_KEY但旧值已失效。环境变量优先级覆盖了.env例如 shell 中已有旧 Key。Key 前后有空格或引号。排查python3 -c import os; from dotenv import load_dotenv; load_dotenv(); print(os.getenv(TAOTOKEN_API_KEY)[:8], ...)确认输出不是None也不是YOUR_API。如果使用 CI/CD检查 secrets 名称是否已改为TAOTOKEN_API_KEY。6.2 404 not found可能原因Base URL 写成了https://taotoken.net/api/v1。Base URL 末尾多了/chat/completions。模型名YOUR_MODEL_ID未替换。客户端自动拼接路径与 Base URL 冲突。正确配置TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODELYOUR_MODEL_ID如果你在代码里写base_urlhttps://taotoken.net/api/v1请改回https://taotoken.net/api。6.3 429 too many requests批处理并发过高时容易触发。建议把TAOTOKEN_BATCH_SIZE从 50 降到 20 或 10。在每批之间time.sleep(0.5)到time.sleep(2)。对 429 使用指数退避。import random import time from openai import RateLimitError def call_with_backoff(fn, max_retries5): for attempt in range(max_retries): try: return fn() except RateLimitError: wait min(2 ** attempt random.random(), 30) print(f触发限流等待 {wait:.1f}s 后重试) time.sleep(wait) raise RuntimeError(重试次数耗尽)6.4 超时与连接错误批处理单条文本很长时可以适当提高TAOTOKEN_TIMEOUT例如 90 秒。但不要无限重试建议设置最大重试次数并把失败记录写入死信文件。client OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api), timeout90, max_retries3, )6.5 JSON 解析失败有些旧代码会解析 Cohere SDK 的返回结构迁移到 OpenAI 兼容结构后字段不同。正确读取方式是text resp.choices[0].message.content不要继续读resp.text或旧 SDK 的字段。把解析层集中到一个函数里后续替换成本最低。7. Claude Code / Codex / CC Switch同一把 Key 的三套配置批处理开发者往往也会用 Claude Code 或 Codex 辅助排障。TaoToken 的 Key 可以统一管理但配置文件要分开写。记住原则Claude Code 用settings.json和ANTHROPIC_*Codex 用config.toml不要把ANTHROPIC_*套到 Codex。7.1 Claude Codesettings.jsonClaude Code 的配置可以放在settings.json中核心是三个变量Base URL、API Key、模型。{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_API_KEY: YOUR_API_KEY, ANTHROPIC_MODEL: YOUR_MODEL_ID } }如果你使用外部配置文件也可以写成环境变量export ANTHROPIC_BASE_URLhttps://taotoken.net/api export ANTHROPIC_API_KEYYOUR_API_KEY export ANTHROPIC_MODELYOUR_MODEL_ID更多 Claude Code 接入细节可参考文档https://taotoken.net/doc/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_claude_code_doc7.2 Codexconfig.tomlCodex 不使用ANTHROPIC_*。在config.toml中配置模型供应商与 Base URLmodel YOUR_MODEL_ID model_provider taotoken [model_providers.taotoken] name TaoToken base_url https://taotoken.net/api env_key TAOTOKEN_API_KEY然后确保 shell 中已设置export TAOTOKEN_API_KEYYOUR_API_KEY这样 Codex 会从TAOTOKEN_API_KEY读取 Key并请求https://taotoken.net/api。再次强调不要把ANTHROPIC_BASE_URL或ANTHROPIC_API_KEY写进 Codex 配置。7.3 CC Switch 三件套如果你用 CC Switch 管理多套配置建议固定三件套Base URLhttps://taotoken.net/apiAPI KeyYOUR_API_KEYModelYOUR_MODEL_ID在 CC Switch 中分别建立“批处理验证”“Claude Code”“Codex”三个预设避免把不同工具的变量混在一起。切换预设后重新打开终端或执行一次冒烟请求确认环境变量已生效。8. 文末 CTA从模型对话验证到 Coding Plan 与 Key 管理到这里五分钟迁移的主线已经完成.env中旧 Cohere Key 改为TAOTOKEN_API_KEYYOUR_API_KEY。Base URL 改为https://taotoken.net/api。用迁移脚本扫描旧引用并备份.env。用 OpenAI 兼容客户端替换批处理调用层。用单请求和 3 条小样本冒烟验证。按 401、404、429、超时、JSON 解析失败逐项排障。如果使用 Claude Code、Codex、CC Switch按各自配置文件写入。如果你还没有完成注册和 Key 创建可以从这里开始https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_footer_register建议按以下顺序验证先在模型对话页面确认模型可用https://taotoken.net/models/detail/chat?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_chat如果你需要长期跑批处理与 Coding 任务查看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_plan进入控制台创建或轮换 Keyhttps://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_keys需要配置 Claude Code 时对照官方文档https://taotoken.net/doc/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_claude_code最后给一个迁移检查清单方便你直接贴到 PR 描述里[ ] .env 已备份 [ ] TAOTOKEN_API_KEYYOUR_API_KEY 已写入本地 [ ] TAOTOKEN_BASE_URLhttps://taotoken.net/api [ ] TAOTOKEN_MODELYOUR_MODEL_ID [ ] 旧 CO_API_KEY / COHERE_API_KEY 已从代码中移除或注释 [ ] smoke_test.py 返回正常 [ ] batch_process.py 小样本成功 [ ] 429 退避策略已确认 [ ] Claude Code / Codex / CC Switch 配置未混用变量批处理换 Key 不需要推倒重来。把 Key 和 Base URL 抽到.env与调用层五分钟足以完成一次可回滚的迁移。 五分钟给 Cohere 批处理换 Key，TaoToken 配进 .env
1. 凌晨的 Cohere 批处理 401为什么这次只需要改 Key 与 Base URLCohere 批处理在第 17 个分片报401 invalid api keyCO_API_KEY失效。先别重写任务TaoToken 官网https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_intro注册拿 Key再把 Base URL 设为https://taotoken.net/api。最近 Cohere 与 Aleph Alpha 签署最终协议、计划统一品牌运营模型供应格局出现新变量对批处理开发者来说最务实的动作不是等公告而是把 Key 与 Base URL 从.env层解耦让任务能在五分钟内切换。本文以“Cohere 批处理换 Key”为主线给出可复制的.env对照、迁移脚本、冒烟验证与排障清单。你不需要改业务流程只需要把供应商配置抽出来把旧 Key 替换成 TaoToken 的YOUR_API_KEY并把客户端 Base URL 指向https://taotoken.net/api。很多批处理项目最初是这样写的import os import cohere co cohere.Client(os.getenv(CO_API_KEY)) response co.chat( modelcommand-r, message把这条工单总结成一句话 )这类代码的痛点是Key、模型名、Base URL 散落在业务文件里。一旦 Key 失效或者供应商侧策略变化你就要全仓搜索CO_API_KEY、COHERE_API_KEY、cohere.Client。更麻烦的是部分 SDK 并不支持显式指定 Base URL导致“换 Key”变成“换调用方式”。所以这次迁移目标很明确在 TaoToken 官网注册并创建 Key得到YOUR_API_KEY。把.env中 Cohere 相关变量改成 TaoToken 变量。把客户端 Base URL 改为https://taotoken.net/api。用迁移脚本扫描旧引用保留备份。用冒烟脚本验证单请求与 3 条批量样本。如果同时使用 Claude Code、Codex 或 CC Switch按各自配置文件写入不要把ANTHROPIC_*套到 Codex。这套流程的核心不是“重写批处理”而是“把供应商配置变成可替换层”。下面直接进入可复制步骤。2. .env 对照Cohere 变量到 TaoToken 变量的最小改动批处理开发者最关心的通常是原来.env里那些变量怎么改下面给出一张最小对照表。旧变量名可能因项目而异但核心只有三类Key、Base URL、模型名。旧变量新变量示例说明COHERE_API_KEYTAOTOKEN_API_KEYYOUR_API_KEY在 TaoToken 控制台创建CO_API_KEYTAOTOKEN_API_KEYYOUR_API_KEY兼容旧脚本中的短变量名COHERE_BASE_URLTAOTOKEN_BASE_URLhttps://taotoken.net/api不加 UTM直接用于客户端COHERE_MODELTAOTOKEN_MODELYOUR_MODEL_ID从模型列表或控制台确认COHERE_BATCH_SIZETAOTOKEN_BATCH_SIZE20批大小按限流调整COHERE_TIMEOUTTAOTOKEN_TIMEOUT60单位秒COHERE_MAX_RETRIESTAOTOKEN_MAX_RETRIES3配合指数退避新的.env可以写成这样# TaoToken 配置 TAOTOKEN_API_KEYYOUR_API_KEY TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODELYOUR_MODEL_ID # 批处理参数 TAOTOKEN_BATCH_SIZE20 TAOTOKEN_TIMEOUT60 TAOTOKEN_MAX_RETRIES3 # 兼容旧代码的过渡变量迁移完成后可删除 # CO_API_KEYYOUR_API_KEY # COHERE_API_KEYYOUR_API_KEY # COHERE_MODELYOUR_MODEL_ID如果你还没有 Key先到 TaoToken 官网注册https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_env_register注册后进入控制台创建 Key。创建时建议按项目命名例如cohere-batch-migrate方便后续轮换。不要把真实 Key 写进.env.example也不要提交到 Git。.env.example只保留占位符TAOTOKEN_API_KEYYOUR_API_KEY TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODELYOUR_MODEL_ID TAOTOKEN_BATCH_SIZE20这里有一个容易忽略的点TAOTOKEN_BASE_URL的值是https://taotoken.net/api不要在末尾追加/v1也不要改成其他路径。很多 404 都来自 Base URL 拼接错误。客户端通常会在 Base URL 后拼接/chat/completions之类的路径重复/v1会导致请求落到错误路由。另外模型名不要凭记忆写。批处理任务里经常把模型名硬编码成command-r、command-r-plus之类。迁移到 TaoToken 后应该从控制台或模型对话页面确认可用模型 ID再写入TAOTOKEN_MODELYOUR_MODEL_ID。如果你只是先跑通可以在模型对话页面验证同一个提示词确认模型可用后再回填到.env。3. 五分钟迁移脚本扫描旧 Key 引用并生成 .env 对照真正耗时的不是改一行 Key而是找到所有旧引用。下面这个 Python 脚本可以放在项目根目录执行。它会扫描常见文本文件找出CO_API_KEY、COHERE_API_KEY、COHERE_MODEL、cohere.Client等引用备份现有.env并生成.env.taotoken对照文件。#!/usr/bin/env python3 migrate_cohere_env.py 用途扫描 Cohere 旧 Key 引用备份 .env生成 TaoToken 迁移对照。 建议先本地执行确认输出后再手动替换代码。 from pathlib import Path import shutil import re from datetime import datetime ROOT Path.cwd() ENV_FILE ROOT / .env BACKUP_DIR ROOT / .migration_backup TAOTOKEN_ENV ROOT / .env.taotoken TEXT_EXTENSIONS { .py, .js, .ts, .tsx, .jsx, .json, .toml, .yaml, .yml, .env, .example, .md, .txt, .sh, .bash } PATTERNS { CO_API_KEY: re.compile(rCO_API_KEY), COHERE_API_KEY: re.compile(rCOHERE_API_KEY), COHERE_MODEL: re.compile(rCOHERE_MODEL), COHERE_BASE_URL: re.compile(rCOHERE_BASE_URL), cohere.Client: re.compile(rcohere\.Client), cohere.ClientV2: re.compile(rcohere\.ClientV2), } def iter_text_files(): for p in ROOT.rglob(*): if p.is_dir(): continue if any(part in {.git, node_modules, __pycache__, .venv, venv} for part in p.parts): continue if p.suffix.lower() in TEXT_EXTENSIONS or p.name.startswith(.env): yield p def scan(): hits [] for file in iter_text_files(): try: content file.read_text(encodingutf-8, errorsignore) except Exception: continue for name, pattern in PATTERNS.items(): for match in pattern.finditer(content): line_no content[:match.start()].count(\n) 1 hits.append((name, file.relative_to(ROOT), line_no)) return hits def backup_env(): BACKUP_DIR.mkdir(exist_okTrue) if ENV_FILE.exists(): stamp datetime.now().strftime(%Y%m%d_%H%M%S) target BACKUP_DIR / f.env.backup_{stamp} shutil.copy2(ENV_FILE, target) print(f[备份] {ENV_FILE} - {target}) else: print([提示] 当前目录没有 .env将生成 .env.taotoken 模板) def write_taotoken_env(): content # TaoToken 迁移配置 TAOTOKEN_API_KEYYOUR_API_KEY TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODELYOUR_MODEL_ID TAOTOKEN_BATCH_SIZE20 TAOTOKEN_TIMEOUT60 TAOTOKEN_MAX_RETRIES3 # 旧变量对照确认迁移完成后删除 # CO_API_KEYYOUR_API_KEY # COHERE_API_KEYYOUR_API_KEY # COHERE_MODELYOUR_MODEL_ID # COHERE_BASE_URLhttps://taotoken.net/api TAOTOKEN_ENV.write_text(content, encodingutf-8) print(f[生成] {TAOTOKEN_ENV}) def main(): print( Cohere - TaoToken 迁移扫描 ) hits scan() if not hits: print(未发现 Cohere 旧变量引用。) else: print(\n发现以下引用) for name, file, line_no in hits: print(f - {name:18} {file}:{line_no}) backup_env() write_taotoken_env() print(\n下一步) print(1. 对比 .env 与 .env.taotoken把 TAOTOKEN_* 写入 .env。) print(2. 把业务代码中的旧 Key 读取改为 TAOTOKEN_API_KEY。) print(3. 把 Base URL 改为 https://taotoken.net/api。) print(4. 执行 smoke_test.py 验证。) if __name__ __main__: main()执行方式python3 migrate_cohere_env.py输出示例 Cohere - TaoToken 迁移扫描 发现以下引用 - CO_API_KEY batch_job.py:12 - COHERE_MODEL batch_job.py:18 - cohere.Client batch_job.py:25 [备份] /project/.env - /project/.migration_backup/.env.backup_20250101_120000 [生成] /project/.env.taotoken这个脚本不会自动改业务代码这是有意为之。批处理任务通常涉及重试、限流、日志、幂等自动替换容易把cohere.Client改成不可用的写法。更安全的做法是先让脚本帮你找出所有引用再按下一节把调用层替换成 OpenAI 兼容方式。如果你在团队里协作可以把扫描结果贴到迁移 PR 描述中。确认.env.taotoken内容后再把TAOTOKEN_API_KEY写入本地.env。Key 只存在本地或密钥管理服务不要进入代码仓库。4. 批处理客户端改造把 Cohere 调用换成 OpenAI 兼容 Base URL迁移到 TaoToken 后最稳妥的调用方式是使用 OpenAI 兼容客户端Base URL 指向https://taotoken.net/api。这样你不需要依赖 Cohere SDK 是否支持自定义 Base URL也不需要在业务代码里保留两套客户端。假设原来的批处理逻辑是逐条调用 Cohere现在可以改成下面这样# batch_process.py import os import time import json from pathlib import Path from dotenv import load_dotenv from openai import OpenAI load_dotenv() API_KEY os.getenv(TAOTOKEN_API_KEY) BASE_URL os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) MODEL os.getenv(TAOTOKEN_MODEL, YOUR_MODEL_ID) BATCH_SIZE int(os.getenv(TAOTOKEN_BATCH_SIZE, 20)) TIMEOUT float(os.getenv(TAOTOKEN_TIMEOUT, 60)) MAX_RETRIES int(os.getenv(TAOTOKEN_MAX_RETRIES, 3)) if not API_KEY or API_KEY YOUR_API_KEY: raise RuntimeError(请先在 .env 中设置 TAOTOKEN_API_KEY) client OpenAI( api_keyAPI_KEY, base_urlBASE_URL, timeoutTIMEOUT, max_retriesMAX_RETRIES, ) def summarize_ticket(ticket: str) - str: resp client.chat.completions.create( modelMODEL, messages[ {role: system, content: 你是工单摘要助手输出一句话中文摘要。}, {role: user, content: ticket}, ], temperature0.2, ) return resp.choices[0].message.content.strip() def run_batch(rows: list[str], output_path: str batch_result.jsonl): success, failed 0, 0 with open(output_path, w, encodingutf-8) as f: for i in range(0, len(rows), BATCH_SIZE): chunk rows[i:i BATCH_SIZE] for idx, row in enumerate(chunk, starti): try: result summarize_ticket(row) record {index: idx, ok: True, result: result} success 1 except Exception as e: record {index: idx, ok: False, error: str(e)} failed 1 f.write(json.dumps(record, ensure_asciiFalse) \n) f.flush() print(f进度{min(i BATCH_SIZE, len(rows))}/{len(rows)}成功 {success}失败 {failed}) time.sleep(0.5) print(f完成输出{output_path}) if __name__ __main__: sample_rows [ 客户反馈登录后页面空白清理缓存后恢复。, 订单支付成功但状态未同步需要人工核查。, API 返回 429建议降低并发并增加重试。, ] run_batch(sample_rows, batch_result_smoke.jsonl)这段代码的关键点base_url使用https://taotoken.net/api不加 UTM不加多余/v1。api_key使用YOUR_API_KEY实际值从.env读取。模型名使用YOUR_MODEL_ID不要硬编码旧 Cohere 模型名。批处理按BATCH_SIZE分块每块后短暂 sleep降低 429 概率。失败记录单独写入 JSONL便于断点续跑。如果你原来的代码必须使用 Cohere SDK可以先检查该 SDK 是否支持base_url参数。支持则传入https://taotoken.net/api不支持则不要强行改直接换到上面的 OpenAI 兼容调用。批处理的核心逻辑是读入、调用、写结果客户端替换不会影响外层幂等设计。另外一个实用技巧是把调用层封装成函数这样迁移时只改一个文件# provider_client.py import os from openai import OpenAI def get_client(): return OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api), timeoutfloat(os.getenv(TAOTOKEN_TIMEOUT, 60)), max_retriesint(os.getenv(TAOTOKEN_MAX_RETRIES, 3)), )业务文件只保留from provider_client import get_client client get_client()这样下次再换供应商只需要改provider_client.py和.env。5. 五分钟冒烟验证单请求、小样本、日志检查迁移后不要直接跑全量批处理。先用单请求确认 Key、Base URL、模型名三项配置正确。# smoke_test.py import os from dotenv import load_dotenv from openai import OpenAI load_dotenv() client OpenAI( api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api), timeout30, ) resp client.chat.completions.create( modelos.getenv(TAOTOKEN_MODEL, YOUR_MODEL_ID), messages[ {role: user, content: 只回复四个字迁移成功} ], ) print(resp.choices[0].message.content)执行python3 smoke_test.py如果返回“迁移成功”或类似内容说明 Key、Base URL、模型名基本正确。如果报 401检查.env是否被正确加载如果报 404检查 Base URL 是否为https://taotoken.net/api以及模型名是否为控制台中的可用 ID如果报 429降低并发增加退避。第二步用 3 条样本跑小批量python3 batch_process.py观察batch_result_smoke.jsonl{index: 0, ok: true, result: 客户登录后页面空白清理缓存后恢复。} {index: 1, ok: true, result: 支付成功但订单状态未同步需人工核查。} {index: 2, ok: true, result: API 返回 429建议降低并发并增加重试。}第三步检查日志中是否还有旧变量。可以用grep -R CO_API_KEY\|COHERE_API_KEY\|COHERE_MODEL . \ --exclude-dir.git \ --exclude-dirnode_modules \ --exclude-dir.venv如果只剩注释或备份文件就可以逐步删掉旧变量。注意命令由你在本地执行不要在生产数据库或关键目录里直接跑破坏性命令。6. 排障401、404、429、超时与 JSON 解析失败批处理换 Key 最常见的错误不是模型能力问题而是配置细节。下面按错误码给出排查顺序。6.1 401 invalid api key可能原因.env中TAOTOKEN_API_KEY仍是YOUR_API_KEY。代码读取的是旧变量CO_API_KEY但旧值已失效。环境变量优先级覆盖了.env例如 shell 中已有旧 Key。Key 前后有空格或引号。排查python3 -c import os; from dotenv import load_dotenv; load_dotenv(); print(os.getenv(TAOTOKEN_API_KEY)[:8], ...)确认输出不是None也不是YOUR_API。如果使用 CI/CD检查 secrets 名称是否已改为TAOTOKEN_API_KEY。6.2 404 not found可能原因Base URL 写成了https://taotoken.net/api/v1。Base URL 末尾多了/chat/completions。模型名YOUR_MODEL_ID未替换。客户端自动拼接路径与 Base URL 冲突。正确配置TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODELYOUR_MODEL_ID如果你在代码里写base_urlhttps://taotoken.net/api/v1请改回https://taotoken.net/api。6.3 429 too many requests批处理并发过高时容易触发。建议把TAOTOKEN_BATCH_SIZE从 50 降到 20 或 10。在每批之间time.sleep(0.5)到time.sleep(2)。对 429 使用指数退避。import random import time from openai import RateLimitError def call_with_backoff(fn, max_retries5): for attempt in range(max_retries): try: return fn() except RateLimitError: wait min(2 ** attempt random.random(), 30) print(f触发限流等待 {wait:.1f}s 后重试) time.sleep(wait) raise RuntimeError(重试次数耗尽)6.4 超时与连接错误批处理单条文本很长时可以适当提高TAOTOKEN_TIMEOUT例如 90 秒。但不要无限重试建议设置最大重试次数并把失败记录写入死信文件。client OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api), timeout90, max_retries3, )6.5 JSON 解析失败有些旧代码会解析 Cohere SDK 的返回结构迁移到 OpenAI 兼容结构后字段不同。正确读取方式是text resp.choices[0].message.content不要继续读resp.text或旧 SDK 的字段。把解析层集中到一个函数里后续替换成本最低。7. Claude Code / Codex / CC Switch同一把 Key 的三套配置批处理开发者往往也会用 Claude Code 或 Codex 辅助排障。TaoToken 的 Key 可以统一管理但配置文件要分开写。记住原则Claude Code 用settings.json和ANTHROPIC_*Codex 用config.toml不要把ANTHROPIC_*套到 Codex。7.1 Claude Codesettings.jsonClaude Code 的配置可以放在settings.json中核心是三个变量Base URL、API Key、模型。{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_API_KEY: YOUR_API_KEY, ANTHROPIC_MODEL: YOUR_MODEL_ID } }如果你使用外部配置文件也可以写成环境变量export ANTHROPIC_BASE_URLhttps://taotoken.net/api export ANTHROPIC_API_KEYYOUR_API_KEY export ANTHROPIC_MODELYOUR_MODEL_ID更多 Claude Code 接入细节可参考文档https://taotoken.net/doc/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_claude_code_doc7.2 Codexconfig.tomlCodex 不使用ANTHROPIC_*。在config.toml中配置模型供应商与 Base URLmodel YOUR_MODEL_ID model_provider taotoken [model_providers.taotoken] name TaoToken base_url https://taotoken.net/api env_key TAOTOKEN_API_KEY然后确保 shell 中已设置export TAOTOKEN_API_KEYYOUR_API_KEY这样 Codex 会从TAOTOKEN_API_KEY读取 Key并请求https://taotoken.net/api。再次强调不要把ANTHROPIC_BASE_URL或ANTHROPIC_API_KEY写进 Codex 配置。7.3 CC Switch 三件套如果你用 CC Switch 管理多套配置建议固定三件套Base URLhttps://taotoken.net/apiAPI KeyYOUR_API_KEYModelYOUR_MODEL_ID在 CC Switch 中分别建立“批处理验证”“Claude Code”“Codex”三个预设避免把不同工具的变量混在一起。切换预设后重新打开终端或执行一次冒烟请求确认环境变量已生效。8. 文末 CTA从模型对话验证到 Coding Plan 与 Key 管理到这里五分钟迁移的主线已经完成.env中旧 Cohere Key 改为TAOTOKEN_API_KEYYOUR_API_KEY。Base URL 改为https://taotoken.net/api。用迁移脚本扫描旧引用并备份.env。用 OpenAI 兼容客户端替换批处理调用层。用单请求和 3 条小样本冒烟验证。按 401、404、429、超时、JSON 解析失败逐项排障。如果使用 Claude Code、Codex、CC Switch按各自配置文件写入。如果你还没有完成注册和 Key 创建可以从这里开始https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_footer_register建议按以下顺序验证先在模型对话页面确认模型可用https://taotoken.net/models/detail/chat?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_chat如果你需要长期跑批处理与 Coding 任务查看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_plan进入控制台创建或轮换 Keyhttps://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_keys需要配置 Claude Code 时对照官方文档https://taotoken.net/doc/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_claude_code最后给一个迁移检查清单方便你直接贴到 PR 描述里[ ] .env 已备份 [ ] TAOTOKEN_API_KEYYOUR_API_KEY 已写入本地 [ ] TAOTOKEN_BASE_URLhttps://taotoken.net/api [ ] TAOTOKEN_MODELYOUR_MODEL_ID [ ] 旧 CO_API_KEY / COHERE_API_KEY 已从代码中移除或注释 [ ] smoke_test.py 返回正常 [ ] batch_process.py 小样本成功 [ ] 429 退避策略已确认 [ ] Claude Code / Codex / CC Switch 配置未混用变量批处理换 Key 不需要推倒重来。把 Key 和 Base URL 抽到.env与调用层五分钟足以完成一次可回滚的迁移。