1. Cohere 批处理插入 TaoTokenendpoint 放 SDK、任务配置还是 workerCohere 与 Aleph Alpha 签署最终协议、合并后以 Cohere 品牌全球运营这条热点让不少数据流水线开发者重新检查模型出口层。问题很具体原来 Cohere 批处理任务直连默认域名现在要在中间插入 TaoTokenendpoint 到底放在哪里如果放在业务 payload 里每条请求都要改如果放在 SDK 初始化里批处理 worker 可能绕过如果只放在环境变量里老代码可能不认。准备 Key 时打开 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_intro 获取Base URL 统一用 https://taotoken.net/api不要把 UTM 查询串写进工具配置。本文从数据流水线开发者视角把 Cohere 批处理任务拆成“客户端初始化、任务定义、worker 出口”三层给出可复制的 endpoint 片段、批处理命令和耗时对照方法。目标不是评论合并事件而是让现有 Cohere 批处理链路能安全切到 TaoToken 出口并且可回滚、可观测。先说结论endpoint 不应该散落在每条输入数据里而应该集中在 worker 的“出口解析层”。业务数据只保留任务类型、模型名、提示词和业务 ID真正的请求地址由TAOTOKEN_BASE_URL、路径解析器和鉴权头统一生成。这样做的直接好处是当你要把 Cohere 批处理从默认出口切到 TaoToken 时只改环境变量或一份egress.yaml不用回刷历史 JSONL。Cohere 批处理任务通常有三个位置可以插入出口配置位置能放什么风险推荐程度业务 payloadendpoint、base_url每条数据都可能不同污染输入集不推荐SDK 初始化base_url、api_key换 SDK 或换 worker 时容易漏推荐一部分worker 环境变量 / 配置中心BASE_URL、API_KEY、路径模板与部署环境绑定可审计最推荐最稳的组合是SDK 初始化只接收base_url和api_key任务定义只声明task_kind例如chat、embed、rerankworker 在发出请求前用endpoint_for(task_kind)拼出最终 URL。最终 URL 的统一形态是{TAOTOKEN_BASE_URL}{PATH}其中TAOTOKEN_BASE_URL固定为https://taotoken.net/apiPATH按任务类型取值。这样就算 Cohere 批处理任务内部换了 SDK出口也不会散。2. 先准备 TaoToken Key 与 Base URL工具配置只认干净地址在准备 TaoToken Key 时打开 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_key 获取 Key也可以直接到 API Keys 页面创建和管理https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_keys 。拿到 Key 后先不要写进代码统一放进环境变量或流水线 secret。本文所有示例都用YOUR_API_KEY占位实际执行时替换成你自己的 Key。export TAOTOKEN_API_KEYYOUR_API_KEY export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_CHAT_PATH/v1/chat/completions export TAOTOKEN_EMBED_PATH/v1/embeddings export TAOTOKEN_RERANK_PATH/v1/rerank注意两点。第一Base URL 用https://taotoken.net/api不要在后面拼?utm_source...。工具配置里的 base URL 是给 SDK 拼接路径用的带查询串会导致部分 HTTP 客户端把参数带到每个请求里轻则鉴权异常重则路径错乱。第二不要把 Key 写进 JSONL 输入文件。批处理输入集经常会被复制、归档、传到对象存储一旦 Key 进入输入集就是泄漏。可以先做一个最小连通性检查。不同账号和模型列表路径以控制台文档为准下面命令只用于确认网络、Key 和 Base URL 三件事curl -sS ${TAOTOKEN_BASE_URL}/v1/models \ -H Authorization: Bearer ${TAOTOKEN_API_KEY} \ -H Content-Type: application/json | head -c 800如果返回 401先检查TAOTOKEN_API_KEY是否真的导出到了当前 shell如果返回 404先检查是不是在 Base URL 后面重复加了/v1例如写成了https://taotoken.net/api/v1/v1/chat/completions。如果返回 429说明并发或频率已经触到限制后面的批处理 worker 要加并发闸门。3. endpoint 片段怎么写base_url、chat、embedding、rerank 分开管Cohere 批处理里经常不止一种任务有的批次做文本生成有的批次做 embedding有的批次做 rerank。把所有任务都指向同一个 chat endpoint 是常见错误。建议写一份出口配置让 worker 按task_kind选择路径egress: provider: taotoken base_url: https://taotoken.net/api auth: type: bearer env: TAOTOKEN_API_KEY endpoints: chat: /v1/chat/completions embeddings: /v1/embeddings rerank: /v1/rerank models: /v1/models timeout_seconds: 120 retry: max_attempts: 3 backoff_seconds: 1.5 retry_on_status: [429, 500, 502, 503, 504]如果你的 Cohere SDK 版本支持base_url参数可以在客户端初始化时传进去。下面是一个 Cohere Python 客户端的示意写法实际参数名以你锁定的 SDK 版本为准import os import cohere co cohere.ClientV2( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api), ) # 后续批处理任务继续调用你原来的业务方法。 # 出口层已经由 base_url 指向 TaoToken业务代码不需要到处改 URL。如果 Cohere SDK 或内部封装不直接暴露base_url就不要硬改 SDK 源码。更干净的做法是在 worker 里加一个出口解析器import os BASE_URL os.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api).rstrip(/) PATH_MAP { chat: os.environ.get(TAOTOKEN_CHAT_PATH, /v1/chat/completions), embed: os.environ.get(TAOTOKEN_EMBED_PATH, /v1/embeddings), rerank: os.environ.get(TAOTOKEN_RERANK_PATH, /v1/rerank), models: /v1/models, } def endpoint_for(task_kind: str) - str: if task_kind not in PATH_MAP: raise ValueError(funsupported task_kind: {task_kind}) path PATH_MAP[task_kind] return f{BASE_URL}/{path.lstrip(/)} if __name__ __main__: print(endpoint_for(chat)) print(endpoint_for(embed)) print(endpoint_for(rerank))这样 endpoint 放在哪里就清楚了Base URL 放环境变量路径放PATH_MAP具体任务通过task_kind选择。业务 JSONL 里最多写task_kind: chat不要写完整 URL。批处理任务回放时只要环境变量一致历史输入集仍然可用。4. 可复现批处理命令JSONL 入口、并发闸门、输出耗时下面给一个可以直接落地的批处理 worker 示例。它读取 JSONL每行一个任务按--concurrency控制并发调用 TaoToken 的 OpenAI 兼容 chat endpoint并把每条请求的耗时写回输出 JSONL。如果你的 Cohere 批处理请求体不是 OpenAI Chat 格式只需要替换payload部分保留endpoint_for、鉴权头、并发闸门和耗时记录。输入文件input.jsonl示例{id:doc-001,task_kind:chat,model:your-model-id,prompt:把这条客服记录归类为退款、物流或产品咨询只输出类别。} {id:doc-002,task_kind:chat,model:your-model-id,prompt:从下面的评论中抽取三个关键词。} {id:doc-003,task_kind:chat,model:your-model-id,prompt:判断用户情绪是正向、中性还是负向。}批处理脚本batch_worker.pyimport argparse import asyncio import json import os import time from pathlib import Path import httpx BASE_URL os.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api).rstrip(/) API_KEY os.environ.get(TAOTOKEN_API_KEY, YOUR_API_KEY) CHAT_PATH os.environ.get(TAOTOKEN_CHAT_PATH, /v1/chat/completions) def endpoint() - str: return f{BASE_URL}/{CHAT_PATH.lstrip(/)} async def call_one(client: httpx.AsyncClient, sem: asyncio.Semaphore, row: dict, idx: int) - dict: async with sem: started time.perf_counter() payload { model: row.get(model, your-model-id), messages: [ {role: system, content: row.get(system, 你是一个批处理文本处理器。)}, {role: user, content: row[prompt]}, ], temperature: row.get(temperature, 0.2), } headers { Authorization: fBearer {API_KEY}, Content-Type: application/json, } try: resp await client.post(endpoint(), jsonpayload, headersheaders) resp.raise_for_status() data resp.json() content data[choices][0][message][content] return { id: row.get(id, idx), task_kind: row.get(task_kind, chat), output: content, elapsed: round(time.perf_counter() - started, 4), status: ok, } except Exception as exc: return { id: row.get(id, idx), task_kind: row.get(task_kind, chat), output: None, elapsed: round(time.perf_counter() - started, 4), status: error, error: str(exc), } async def run(args: argparse.Namespace) - None: lines Path(args.input).read_text(encodingutf-8).splitlines() rows [json.loads(line) for line in lines if line.strip()] sem asyncio.Semaphore(args.concurrency) limits httpx.Limits(max_connectionsargs.concurrency * 2, max_keepalive_connectionsargs.concurrency) timeout httpx.Timeout(args.timeout) async with httpx.AsyncClient(limitslimits, timeouttimeout) as client: tasks [call_one(client, sem, row, i) for i, row in enumerate(rows)] results [] for fut in asyncio.as_completed(tasks): results.append(await fut) with Path(args.output).open(w, encodingutf-8) as f: for item in results: f.write(json.dumps(item, ensure_asciiFalse) \n) ok sum(1 for item in results if item[status] ok) print(ftotal{len(results)} ok{ok} error{len(results)-ok} output{args.output}) def main() - None: parser argparse.ArgumentParser() parser.add_argument(--input, requiredTrue) parser.add_argument(--output, requiredTrue) parser.add_argument(--concurrency, typeint, default4) parser.add_argument(--timeout, typefloat, default120) args parser.parse_args() asyncio.run(run(args)) if __name__ __main__: main()执行命令export TAOTOKEN_API_KEYYOUR_API_KEY export TAOTOKEN_BASE_URLhttps://taotoken.net/api python batch_worker.py \ --input input.jsonl \ --output output.jsonl \ --concurrency 4 \ --timeout 120批处理任务最容易忽略的是输出原子性。如果 worker 在写output.jsonl时被中断可能留下半行 JSON。生产环境建议先写output.jsonl.tmp全部完成后再mv成正式文件或者按批次分片写output/part-0001.jsonl再用合并脚本汇总。失败重试时用id去重避免同一条业务数据被重复计费。5. 耗时对照串行、并发、重试三组基准怎么跑不要凭感觉调并发。建议至少跑三组串行、并发 4、并发 8。用同一个小批量输入集保持模型和提示词一致只改--concurrency。命令如下python batch_worker.py --input input.jsonl --output out_c1.jsonl --concurrency 1 python batch_worker.py --input input.jsonl --output out_c4.jsonl --concurrency 4 python batch_worker.py --input input.jsonl --output out_c8.jsonl --concurrency 8统计平均耗时和最大耗时jq -r select(.elapsed ! null) | .elapsed out_c4.jsonl \ | awk {s$1; n; if($1max)max$1} END{printf count%d avg%.3fs max%.3fs\n, n, s/n, max}下面是一个示例表格格式用于说明你应该记录哪些字段。具体数值必须以你本地实测为准因为模型、网络、提示词长度和限流策略都会影响结果。模式并发批大小平均单条耗时最大单条耗时总耗时备注串行11001.42s1.88s142s基线最稳定并发41001.61s2.20s41s通常最佳平衡并发81001.95s2.90s27s可能触发 429并发 重试81002.10s4.50s35s加入退避后更稳如果你发现并发 8 的错误率明显升高不要继续加并发。先把--concurrency降到 4再检查错误 JSON 中的status和error。常见情况是 429 和超时前者需要退避后者需要增大--timeout或减少单条提示词长度。批处理任务的目标不是把单批压到最短而是让整条流水线可重跑、可对账、可恢复。6. 在流水线机器上配置 Claude Code、Codex 与 CC Switch同一个流水线节点上除了 Cohere 批处理你可能还会用 Claude Code 或 Codex 辅助排查日志。它们可以共用 TaoToken Key但配置文件不要混用。核心原则是Claude Code 用settings.json和ANTHROPIC_*Codex 用config.toml不要把ANTHROPIC_*写进 Codex。Claude Code 的~/.claude/settings.json可以这样配置{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_AUTH_TOKEN: YOUR_API_KEY, ANTHROPIC_MODEL: your-model-id } }这里的ANTHROPIC_BASE_URL继续使用干净的https://taotoken.net/api不要附加查询串。ANTHROPIC_AUTH_TOKEN放你的 TaoToken Key生产环境可以用 secret 注入或本机环境变量覆盖。Codex 的~/.codex/config.toml单独配置model your-model-id model_provider taotoken [model_providers.taotoken] name TaoToken base_url https://taotoken.net/api env_key TAOTOKEN_API_KEY wire_api chatCodex 不读ANTHROPIC_*它走model_providers里的base_url和env_key。如果你把 Claude Code 的环境变量复制给 Codex通常不会生效还会让排障方向跑偏。CC Switch 这类切换工具可以按“三件套”理解。Claude Code 侧三件套是ANTHROPIC_BASE_URLhttps://taotoken.net/apiANTHROPIC_AUTH_TOKENYOUR_API_KEYANTHROPIC_MODEL从模型列表选实际模型 IDCodex 侧三件套是model_providertaotokenbase_urlhttps://taotoken.net/apienv_keyTAOTOKEN_API_KEY通用三件套则是 Provider 名称、Base URL、Key 环境变量。只要记住这条边界CC Switch 里切换配置时就不会把不同工具的字段混在一起。需要看 Claude Code 的完整配置说明时可以打开文末的 Claude Code 文档 deep link。7. 常见排障401、404、429、超时与半成品文件第一类问题是 401。批处理 worker 报 401 时先看请求头有没有带上Authorization: Bearer ${TAOTOKEN_API_KEY}。有些团队在本地 shell 里export了变量但流水线任务使用 systemd、cron 或容器时并没有继承。建议在 worker 启动日志里只打印 Key 是否存在和长度不要打印完整 Keyimport os key os.environ.get(TAOTOKEN_API_KEY, ) print({has_key: bool(key), key_length: len(key)})第二类问题是 404。最常见的拼接错误是 Base URL 已经带了/api代码里又写死/api/v1/chat/completions最终变成/api/api/v1/...。另一个常见错误是CHAT_PATH前多余斜杠或缺少斜杠。使用BASE_URL.rstrip(/)和path.lstrip(/)后再拼接能避免大部分路径问题。第三类问题是 429。批处理任务的并发不是越高越好。TaoToken 出口和上游模型都可能有限流。推荐的退避策略是遇到 429 后等待base * 2 ** attempt秒并加上随机抖动如果同一批 429 比例超过 10%直接把并发减半。不要无限重试否则会把整条流水线拖死。第四类问题是超时。长文本批处理容易超过 60 秒。可以先把--timeout设为 120再按任务 P95 调整。如果单条经常超过 120 秒更合理的做法是把长文本先切块而不是把超时无限放大。第五类问题是半成品文件。批处理中断后不要直接复用output.jsonl继续追加。推荐每次运行写独立分片例如output/run-20250101-001.jsonl然后用id去重合并。合并命令可以在本地执行cat output/run-*.jsonl | jq -s group_by(.id) | map(.[0])[] output/merged.jsonl如果错误记录也需要保留可以单独写errors.jsonl不要把错误和结果混在一起。这样后续对账时成功记录、失败记录、重试记录能分开统计。8. CTA把 Cohere 批处理出口固定到 TaoToken 后的下一步当 Cohere 批处理任务已经通过TAOTOKEN_BASE_URLhttps://taotoken.net/api跑通下一步不是继续在代码里硬编码 URL而是把出口配置收进部署模板。这样无论 Cohere 后续品牌和架构如何变化你的流水线只关心 TaoToken 这一层出口。如果你还没有确认模型返回格式建议先从模型对话开始验证https://taotoken.net/models/detail/chat?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_chat 。它可以快速确认 Key、Base URL 和模型 ID 是否匹配避免把连通性问题带进批处理。如果你需要长期跑批处理同时还要在 Claude Code、Codex 等工具之间切换可以查看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_plan 。把批处理出口和日常编码工具链分开管理排障会轻松很多。创建和管理 Key 从 API Keys 页面进入https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_keys 。生产环境建议一个环境一个 Key至少区分开发、预发、生产Key 只放在 secret 里YOUR_API_KEY只作为本地占位。Claude Code 的完整配置说明在https://taotoken.net/doc/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_claudecode 。注意 Claude Code 用settings.json和ANTHROPIC_*Codex 用config.toml两者不要混用。最后再回到官网入口https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_cta 。先获取 Key再把 Base URL 固定为https://taotoken.net/api然后用本文的批处理命令跑一组小样本。看到total、ok、elapsed三列数据后再逐步放大并发和批大小。这样插入 TaoToken 出口层就不是一次冒险改动而是一条可回滚、可复现、可对照的数据流水线。 在 Cohere 批处理里插 TaoToken，endpoint 放哪
1. Cohere 批处理插入 TaoTokenendpoint 放 SDK、任务配置还是 workerCohere 与 Aleph Alpha 签署最终协议、合并后以 Cohere 品牌全球运营这条热点让不少数据流水线开发者重新检查模型出口层。问题很具体原来 Cohere 批处理任务直连默认域名现在要在中间插入 TaoTokenendpoint 到底放在哪里如果放在业务 payload 里每条请求都要改如果放在 SDK 初始化里批处理 worker 可能绕过如果只放在环境变量里老代码可能不认。准备 Key 时打开 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_intro 获取Base URL 统一用 https://taotoken.net/api不要把 UTM 查询串写进工具配置。本文从数据流水线开发者视角把 Cohere 批处理任务拆成“客户端初始化、任务定义、worker 出口”三层给出可复制的 endpoint 片段、批处理命令和耗时对照方法。目标不是评论合并事件而是让现有 Cohere 批处理链路能安全切到 TaoToken 出口并且可回滚、可观测。先说结论endpoint 不应该散落在每条输入数据里而应该集中在 worker 的“出口解析层”。业务数据只保留任务类型、模型名、提示词和业务 ID真正的请求地址由TAOTOKEN_BASE_URL、路径解析器和鉴权头统一生成。这样做的直接好处是当你要把 Cohere 批处理从默认出口切到 TaoToken 时只改环境变量或一份egress.yaml不用回刷历史 JSONL。Cohere 批处理任务通常有三个位置可以插入出口配置位置能放什么风险推荐程度业务 payloadendpoint、base_url每条数据都可能不同污染输入集不推荐SDK 初始化base_url、api_key换 SDK 或换 worker 时容易漏推荐一部分worker 环境变量 / 配置中心BASE_URL、API_KEY、路径模板与部署环境绑定可审计最推荐最稳的组合是SDK 初始化只接收base_url和api_key任务定义只声明task_kind例如chat、embed、rerankworker 在发出请求前用endpoint_for(task_kind)拼出最终 URL。最终 URL 的统一形态是{TAOTOKEN_BASE_URL}{PATH}其中TAOTOKEN_BASE_URL固定为https://taotoken.net/apiPATH按任务类型取值。这样就算 Cohere 批处理任务内部换了 SDK出口也不会散。2. 先准备 TaoToken Key 与 Base URL工具配置只认干净地址在准备 TaoToken Key 时打开 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_key 获取 Key也可以直接到 API Keys 页面创建和管理https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_keys 。拿到 Key 后先不要写进代码统一放进环境变量或流水线 secret。本文所有示例都用YOUR_API_KEY占位实际执行时替换成你自己的 Key。export TAOTOKEN_API_KEYYOUR_API_KEY export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_CHAT_PATH/v1/chat/completions export TAOTOKEN_EMBED_PATH/v1/embeddings export TAOTOKEN_RERANK_PATH/v1/rerank注意两点。第一Base URL 用https://taotoken.net/api不要在后面拼?utm_source...。工具配置里的 base URL 是给 SDK 拼接路径用的带查询串会导致部分 HTTP 客户端把参数带到每个请求里轻则鉴权异常重则路径错乱。第二不要把 Key 写进 JSONL 输入文件。批处理输入集经常会被复制、归档、传到对象存储一旦 Key 进入输入集就是泄漏。可以先做一个最小连通性检查。不同账号和模型列表路径以控制台文档为准下面命令只用于确认网络、Key 和 Base URL 三件事curl -sS ${TAOTOKEN_BASE_URL}/v1/models \ -H Authorization: Bearer ${TAOTOKEN_API_KEY} \ -H Content-Type: application/json | head -c 800如果返回 401先检查TAOTOKEN_API_KEY是否真的导出到了当前 shell如果返回 404先检查是不是在 Base URL 后面重复加了/v1例如写成了https://taotoken.net/api/v1/v1/chat/completions。如果返回 429说明并发或频率已经触到限制后面的批处理 worker 要加并发闸门。3. endpoint 片段怎么写base_url、chat、embedding、rerank 分开管Cohere 批处理里经常不止一种任务有的批次做文本生成有的批次做 embedding有的批次做 rerank。把所有任务都指向同一个 chat endpoint 是常见错误。建议写一份出口配置让 worker 按task_kind选择路径egress: provider: taotoken base_url: https://taotoken.net/api auth: type: bearer env: TAOTOKEN_API_KEY endpoints: chat: /v1/chat/completions embeddings: /v1/embeddings rerank: /v1/rerank models: /v1/models timeout_seconds: 120 retry: max_attempts: 3 backoff_seconds: 1.5 retry_on_status: [429, 500, 502, 503, 504]如果你的 Cohere SDK 版本支持base_url参数可以在客户端初始化时传进去。下面是一个 Cohere Python 客户端的示意写法实际参数名以你锁定的 SDK 版本为准import os import cohere co cohere.ClientV2( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api), ) # 后续批处理任务继续调用你原来的业务方法。 # 出口层已经由 base_url 指向 TaoToken业务代码不需要到处改 URL。如果 Cohere SDK 或内部封装不直接暴露base_url就不要硬改 SDK 源码。更干净的做法是在 worker 里加一个出口解析器import os BASE_URL os.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api).rstrip(/) PATH_MAP { chat: os.environ.get(TAOTOKEN_CHAT_PATH, /v1/chat/completions), embed: os.environ.get(TAOTOKEN_EMBED_PATH, /v1/embeddings), rerank: os.environ.get(TAOTOKEN_RERANK_PATH, /v1/rerank), models: /v1/models, } def endpoint_for(task_kind: str) - str: if task_kind not in PATH_MAP: raise ValueError(funsupported task_kind: {task_kind}) path PATH_MAP[task_kind] return f{BASE_URL}/{path.lstrip(/)} if __name__ __main__: print(endpoint_for(chat)) print(endpoint_for(embed)) print(endpoint_for(rerank))这样 endpoint 放在哪里就清楚了Base URL 放环境变量路径放PATH_MAP具体任务通过task_kind选择。业务 JSONL 里最多写task_kind: chat不要写完整 URL。批处理任务回放时只要环境变量一致历史输入集仍然可用。4. 可复现批处理命令JSONL 入口、并发闸门、输出耗时下面给一个可以直接落地的批处理 worker 示例。它读取 JSONL每行一个任务按--concurrency控制并发调用 TaoToken 的 OpenAI 兼容 chat endpoint并把每条请求的耗时写回输出 JSONL。如果你的 Cohere 批处理请求体不是 OpenAI Chat 格式只需要替换payload部分保留endpoint_for、鉴权头、并发闸门和耗时记录。输入文件input.jsonl示例{id:doc-001,task_kind:chat,model:your-model-id,prompt:把这条客服记录归类为退款、物流或产品咨询只输出类别。} {id:doc-002,task_kind:chat,model:your-model-id,prompt:从下面的评论中抽取三个关键词。} {id:doc-003,task_kind:chat,model:your-model-id,prompt:判断用户情绪是正向、中性还是负向。}批处理脚本batch_worker.pyimport argparse import asyncio import json import os import time from pathlib import Path import httpx BASE_URL os.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api).rstrip(/) API_KEY os.environ.get(TAOTOKEN_API_KEY, YOUR_API_KEY) CHAT_PATH os.environ.get(TAOTOKEN_CHAT_PATH, /v1/chat/completions) def endpoint() - str: return f{BASE_URL}/{CHAT_PATH.lstrip(/)} async def call_one(client: httpx.AsyncClient, sem: asyncio.Semaphore, row: dict, idx: int) - dict: async with sem: started time.perf_counter() payload { model: row.get(model, your-model-id), messages: [ {role: system, content: row.get(system, 你是一个批处理文本处理器。)}, {role: user, content: row[prompt]}, ], temperature: row.get(temperature, 0.2), } headers { Authorization: fBearer {API_KEY}, Content-Type: application/json, } try: resp await client.post(endpoint(), jsonpayload, headersheaders) resp.raise_for_status() data resp.json() content data[choices][0][message][content] return { id: row.get(id, idx), task_kind: row.get(task_kind, chat), output: content, elapsed: round(time.perf_counter() - started, 4), status: ok, } except Exception as exc: return { id: row.get(id, idx), task_kind: row.get(task_kind, chat), output: None, elapsed: round(time.perf_counter() - started, 4), status: error, error: str(exc), } async def run(args: argparse.Namespace) - None: lines Path(args.input).read_text(encodingutf-8).splitlines() rows [json.loads(line) for line in lines if line.strip()] sem asyncio.Semaphore(args.concurrency) limits httpx.Limits(max_connectionsargs.concurrency * 2, max_keepalive_connectionsargs.concurrency) timeout httpx.Timeout(args.timeout) async with httpx.AsyncClient(limitslimits, timeouttimeout) as client: tasks [call_one(client, sem, row, i) for i, row in enumerate(rows)] results [] for fut in asyncio.as_completed(tasks): results.append(await fut) with Path(args.output).open(w, encodingutf-8) as f: for item in results: f.write(json.dumps(item, ensure_asciiFalse) \n) ok sum(1 for item in results if item[status] ok) print(ftotal{len(results)} ok{ok} error{len(results)-ok} output{args.output}) def main() - None: parser argparse.ArgumentParser() parser.add_argument(--input, requiredTrue) parser.add_argument(--output, requiredTrue) parser.add_argument(--concurrency, typeint, default4) parser.add_argument(--timeout, typefloat, default120) args parser.parse_args() asyncio.run(run(args)) if __name__ __main__: main()执行命令export TAOTOKEN_API_KEYYOUR_API_KEY export TAOTOKEN_BASE_URLhttps://taotoken.net/api python batch_worker.py \ --input input.jsonl \ --output output.jsonl \ --concurrency 4 \ --timeout 120批处理任务最容易忽略的是输出原子性。如果 worker 在写output.jsonl时被中断可能留下半行 JSON。生产环境建议先写output.jsonl.tmp全部完成后再mv成正式文件或者按批次分片写output/part-0001.jsonl再用合并脚本汇总。失败重试时用id去重避免同一条业务数据被重复计费。5. 耗时对照串行、并发、重试三组基准怎么跑不要凭感觉调并发。建议至少跑三组串行、并发 4、并发 8。用同一个小批量输入集保持模型和提示词一致只改--concurrency。命令如下python batch_worker.py --input input.jsonl --output out_c1.jsonl --concurrency 1 python batch_worker.py --input input.jsonl --output out_c4.jsonl --concurrency 4 python batch_worker.py --input input.jsonl --output out_c8.jsonl --concurrency 8统计平均耗时和最大耗时jq -r select(.elapsed ! null) | .elapsed out_c4.jsonl \ | awk {s$1; n; if($1max)max$1} END{printf count%d avg%.3fs max%.3fs\n, n, s/n, max}下面是一个示例表格格式用于说明你应该记录哪些字段。具体数值必须以你本地实测为准因为模型、网络、提示词长度和限流策略都会影响结果。模式并发批大小平均单条耗时最大单条耗时总耗时备注串行11001.42s1.88s142s基线最稳定并发41001.61s2.20s41s通常最佳平衡并发81001.95s2.90s27s可能触发 429并发 重试81002.10s4.50s35s加入退避后更稳如果你发现并发 8 的错误率明显升高不要继续加并发。先把--concurrency降到 4再检查错误 JSON 中的status和error。常见情况是 429 和超时前者需要退避后者需要增大--timeout或减少单条提示词长度。批处理任务的目标不是把单批压到最短而是让整条流水线可重跑、可对账、可恢复。6. 在流水线机器上配置 Claude Code、Codex 与 CC Switch同一个流水线节点上除了 Cohere 批处理你可能还会用 Claude Code 或 Codex 辅助排查日志。它们可以共用 TaoToken Key但配置文件不要混用。核心原则是Claude Code 用settings.json和ANTHROPIC_*Codex 用config.toml不要把ANTHROPIC_*写进 Codex。Claude Code 的~/.claude/settings.json可以这样配置{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_AUTH_TOKEN: YOUR_API_KEY, ANTHROPIC_MODEL: your-model-id } }这里的ANTHROPIC_BASE_URL继续使用干净的https://taotoken.net/api不要附加查询串。ANTHROPIC_AUTH_TOKEN放你的 TaoToken Key生产环境可以用 secret 注入或本机环境变量覆盖。Codex 的~/.codex/config.toml单独配置model your-model-id model_provider taotoken [model_providers.taotoken] name TaoToken base_url https://taotoken.net/api env_key TAOTOKEN_API_KEY wire_api chatCodex 不读ANTHROPIC_*它走model_providers里的base_url和env_key。如果你把 Claude Code 的环境变量复制给 Codex通常不会生效还会让排障方向跑偏。CC Switch 这类切换工具可以按“三件套”理解。Claude Code 侧三件套是ANTHROPIC_BASE_URLhttps://taotoken.net/apiANTHROPIC_AUTH_TOKENYOUR_API_KEYANTHROPIC_MODEL从模型列表选实际模型 IDCodex 侧三件套是model_providertaotokenbase_urlhttps://taotoken.net/apienv_keyTAOTOKEN_API_KEY通用三件套则是 Provider 名称、Base URL、Key 环境变量。只要记住这条边界CC Switch 里切换配置时就不会把不同工具的字段混在一起。需要看 Claude Code 的完整配置说明时可以打开文末的 Claude Code 文档 deep link。7. 常见排障401、404、429、超时与半成品文件第一类问题是 401。批处理 worker 报 401 时先看请求头有没有带上Authorization: Bearer ${TAOTOKEN_API_KEY}。有些团队在本地 shell 里export了变量但流水线任务使用 systemd、cron 或容器时并没有继承。建议在 worker 启动日志里只打印 Key 是否存在和长度不要打印完整 Keyimport os key os.environ.get(TAOTOKEN_API_KEY, ) print({has_key: bool(key), key_length: len(key)})第二类问题是 404。最常见的拼接错误是 Base URL 已经带了/api代码里又写死/api/v1/chat/completions最终变成/api/api/v1/...。另一个常见错误是CHAT_PATH前多余斜杠或缺少斜杠。使用BASE_URL.rstrip(/)和path.lstrip(/)后再拼接能避免大部分路径问题。第三类问题是 429。批处理任务的并发不是越高越好。TaoToken 出口和上游模型都可能有限流。推荐的退避策略是遇到 429 后等待base * 2 ** attempt秒并加上随机抖动如果同一批 429 比例超过 10%直接把并发减半。不要无限重试否则会把整条流水线拖死。第四类问题是超时。长文本批处理容易超过 60 秒。可以先把--timeout设为 120再按任务 P95 调整。如果单条经常超过 120 秒更合理的做法是把长文本先切块而不是把超时无限放大。第五类问题是半成品文件。批处理中断后不要直接复用output.jsonl继续追加。推荐每次运行写独立分片例如output/run-20250101-001.jsonl然后用id去重合并。合并命令可以在本地执行cat output/run-*.jsonl | jq -s group_by(.id) | map(.[0])[] output/merged.jsonl如果错误记录也需要保留可以单独写errors.jsonl不要把错误和结果混在一起。这样后续对账时成功记录、失败记录、重试记录能分开统计。8. CTA把 Cohere 批处理出口固定到 TaoToken 后的下一步当 Cohere 批处理任务已经通过TAOTOKEN_BASE_URLhttps://taotoken.net/api跑通下一步不是继续在代码里硬编码 URL而是把出口配置收进部署模板。这样无论 Cohere 后续品牌和架构如何变化你的流水线只关心 TaoToken 这一层出口。如果你还没有确认模型返回格式建议先从模型对话开始验证https://taotoken.net/models/detail/chat?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_chat 。它可以快速确认 Key、Base URL 和模型 ID 是否匹配避免把连通性问题带进批处理。如果你需要长期跑批处理同时还要在 Claude Code、Codex 等工具之间切换可以查看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_plan 。把批处理出口和日常编码工具链分开管理排障会轻松很多。创建和管理 Key 从 API Keys 页面进入https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_keys 。生产环境建议一个环境一个 Key至少区分开发、预发、生产Key 只放在 secret 里YOUR_API_KEY只作为本地占位。Claude Code 的完整配置说明在https://taotoken.net/doc/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_claudecode 。注意 Claude Code 用settings.json和ANTHROPIC_*Codex 用config.toml两者不要混用。最后再回到官网入口https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentcohere_batch_cta 。先获取 Key再把 Base URL 固定为https://taotoken.net/api然后用本文的批处理命令跑一组小样本。看到total、ok、elapsed三列数据后再逐步放大并发和批大小。这样插入 TaoToken 出口层就不是一次冒险改动而是一条可回滚、可复现、可对照的数据流水线。