1. 生产环境 Agent 成本为什么会失控如果你正在维护一个跑在生产环境的 Agent 服务大概率遇到过这种场景上线第一周账单还算正常第二周开始翻倍第三周直接超出预算三倍但业务量并没有明显增长。你打开账单页面只看到一个总 Token 数却完全不知道这些 Token 花在了哪个环节、哪个用户、哪次工具调用上。这就是生产环境 Agent 成本失控的典型特征——消耗不透明、归因不清晰、调度不灵活。一个完整的 Agent 单轮交互Token 消耗至少来自五个地方主模型输入、主模型输出、工具描述与历史工具记录、工具内部调用的子模型、长期记忆压缩后的拼接内容。任何一处没有做精细化统计账单就会像滚雪球一样膨胀。我见过最夸张的一个案例某客服 Agent 把 12 个工具的描述文档约 25k Token在每一轮工具链推理时都完整传一遍单次会话 8 轮工具调用光工具描述就烧掉 200k Token。还有的 Agent 用无限制的 ConversationBufferMemory 拼接历史对话一个长会话拼到 100k 以上输入 Token而其中 80% 的历史内容对当前问题毫无帮助。这篇内容面向后端与 AI 平台工程师交付一套可复制的方案用 TaoToken 统一 Key 接入多家模型做 Token 消耗埋点与统计再基于统计结果配置混合模型调度策略。目标不是让你把成本压到最低而是让每一分 Token 花得可解释、可控制、可优化。2. TaoToken 前置准备统一 Key 与接入配置TaoToken 的核心价值在于一个 Key 调用多家模型这对混合模型调度来说是刚需。你不需要为每个模型厂商维护一套鉴权、一套 SDK、一套计费口径所有请求走同一个入口返回统一的 usage 字段埋点统计的复杂度直接降一个数量级。2.1 获取 API Key 与确认接入地址先到控制台创建 API Key。建议按环境拆分开发环境一个 Key预发一个生产一个方便后续按 Key 维度做成本归因。创建入口在控制台的 API Keys 页面生成后立即复制保存页面刷新后不再完整显示。接入地址统一使用https://taotoken.net/api兼容 OpenAI 风格的/v1/chat/completions路径。如果你用的是 Anthropic 风格的 SDKTaoToken 也提供对应的兼容端点具体路径在接入文档里有完整说明。2.2 环境变量与基础客户端配置不要把 Key 硬编码在代码里。用环境变量管理生产环境配合密钥管理服务注入export TAOTOKEN_API_KEYsk-你的实际Key export TAOTOKEN_BASE_URLhttps://taotoken.net/apiPython 侧用 openai SDK 即可因为接口兼容import os from openai import OpenAI client OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], ) def chat(model: str, messages: list, **kwargs): resp client.chat.completions.create( modelmodel, messagesmessages, **kwargs, ) return resp这段代码的关键点是base_url指向 TaoTokenmodel参数填你要调度的具体模型名。混合调度的本质就是在这里根据任务类型动态替换model的值。2.3 模型清单与单价对照在配置调度策略前先把你实际会用到的模型列一张表标注单价和适用场景。下面是一个参考骨架具体单价以控制台实时数据为准模型输入单价每 1k Token输出单价每 1k Token适用子任务轻量模型 A低低意图识别、分类、简单抽取中量模型 B中中多轮对话、摘要、格式转换重量模型 C高高复杂推理、长文档解析、代码生成这张表是你后续写调度决策函数的依据。没有这张表调度就是拍脑袋。3. 可复制配置Token 消耗埋点与统计脚本埋点是整个方案的地基。没有埋点你连哪个环节贵都不知道更谈不上优化。TaoToken 的响应体里带有标准 usage 字段包含prompt_tokens、completion_tokens、total_tokens直接拿来用。3.1 埋点数据结构设计每次请求记录以下字段写入你的日志系统或时序数据库import time import uuid import json import logging logger logging.getLogger(agent.token) def traced_chat( model: str, messages: list, biz_scene: str, session_id: str, sub_task: str, **kwargs, ): trace_id str(uuid.uuid4()) start time.time() resp client.chat.completions.create( modelmodel, messagesmessages, **kwargs, ) latency_ms int((time.time() - start) * 1000) usage resp.usage record { trace_id: trace_id, model: model, biz_scene: biz_scene, session_id: session_id, sub_task: sub_task, prompt_tokens: usage.prompt_tokens, completion_tokens: usage.completion_tokens, total_tokens: usage.total_tokens, latency_ms: latency_ms, ts: int(time.time()), } logger.info(json.dumps(record, ensure_asciiFalse)) return respbiz_scene填业务场景如订单查询财报解析sub_task填子任务如意图识别SQL转译。这两个字段是后续归因分析的核心维度。3.2 按维度聚合的统计脚本日志落盘后用一段脚本做聚合。下面按业务场景 × 模型两个维度统计 Token 消耗和成本import json from collections import defaultdict PRICE { light-model: {in: 0.0005, out: 0.0015}, mid-model: {in: 0.003, out: 0.006}, heavy-model: {in: 0.01, out: 0.03}, } def aggregate(log_path: str): agg defaultdict(lambda: {in: 0, out: 0, cost: 0.0, calls: 0}) with open(log_path) as f: for line in f: r json.loads(line) key (r[biz_scene], r[model]) p PRICE.get(r[model], {in: 0, out: 0}) agg[key][in] r[prompt_tokens] agg[key][out] r[completion_tokens] agg[key][cost] ( r[prompt_tokens] / 1000 * p[in] r[completion_tokens] / 1000 * p[out] ) agg[key][calls] 1 for (scene, model), v in sorted(agg.items(), keylambda x: -x[1][cost]): print(f{scene:20s} {model:15s} calls{v[calls]:6d} fin{v[in]:10d} out{v[out]:10d} cost${v[cost]:.4f})跑完这段脚本你会得到一张按成本降序排列的表。排在最上面的那几行就是你的成本黑洞。3.3 混合模型调度策略骨架有了统计数据就可以写调度决策函数。核心逻辑是根据子任务类型和输入长度选择满足质量要求的最低成本模型。def pick_model(sub_task: str, input_tokens: int, sla: str normal) - str: # 简单分类任务轻量模型足够 if sub_task in (intent, classify, extract_simple): return light-model # 中等复杂度且输入不长 if sub_task in (summarize, format, multi_turn) and input_tokens 4000: return mid-model # 复杂推理或长文档必须上重量模型 if sub_task in (reasoning, doc_parse, code_gen): return heavy-model # 兜底按 SLA 决定 return heavy-model if sla strict else mid-model这个骨架可以直接用也可以替换成更复杂的决策逻辑比如引入贝叶斯优化或负载预测。但建议先从规则版跑起来拿到真实数据后再迭代。4. 验证请求与成功结果配置写完后必须做一次端到端验证确认三件事请求能通、usage 能拿到、统计脚本能跑出结果。4.1 最小验证请求resp traced_chat( modellight-model, messages[{role: user, content: 把这句话分类我要退货}], biz_scenecustomer_service, session_idtest-session-001, sub_taskintent, ) print(resp.choices[0].message.content) print(resp.usage)预期输出类似退货意图 CompletionUsage(prompt_tokens18, completion_tokens6, total_tokens24)看到usage字段有值说明埋点数据源没问题。4.2 统计脚本验证把上面几次请求的日志喂给聚合脚本预期输出customer_service light-model calls 1 in 18 out 6 cost$0.0000如果成本显示为 0检查 PRICE 表里的单价是否填对以及日志里的 model 名是否和 PRICE 的 key 一致。4.3 混合调度验证构造三个不同子任务的请求确认调度函数选出的模型符合预期print(pick_model(intent, 50)) # 期望 light-model print(pick_model(summarize, 2000)) # 期望 mid-model print(pick_model(doc_parse, 30000)) # 期望 heavy-model三个输出都对说明调度骨架生效。接下来就是把它接入你的 Agent 主流程替换掉原来写死的模型名。5. 本篇常见错排查5.1 usage 字段为 None 或缺失部分兼容接口在流式模式下不返回 usage。如果你用了streamTrue需要在请求里显式加上stream_options{include_usage: True}否则最后一块 chunk 里不会有 usage。非流式模式下如果仍然缺失检查 base_url 是否拼写正确以及 model 名是否是 TaoToken 支持的名称。5.2 统计脚本成本算出来偏高或偏低九成是单价表没更新。模型单价会调整PRICE 字典必须和控制台实时数据对齐。建议把单价表抽成独立配置文件每月核对一次。另一个常见原因是把prompt_tokens和completion_tokens的单价用反了输出单价通常比输入高 2 到 6 倍用反会导致成本严重失真。5.3 调度函数选错模型导致质量下降规则版调度最容易踩的坑是把本该用重量模型的子任务误判为轻量。典型信号是用户反馈回答变敷衍了或格式不对了。排查方法是按sub_task维度对比调度前后的准确率如果某个子任务准确率掉了超过 3 个百分点就把它从轻量模型挪回中量或重量模型。调度策略永远是在成本和质量的曲线上找平衡点不是一味求低。5.4 日志量过大导致存储成本反超 Token 成本埋点日志如果每条都写全量 messages日志体积会迅速膨胀。建议只记录元数据Token 数、模型、场景、延迟不记录 messages 原文。需要排查具体内容时用 trace_id 去业务库里关联查询。日志保留周期设 30 天即可历史数据聚合后归档。5.5 多环境 Key 混用导致归因混乱开发、预发、生产共用一个 Key统计时无法区分环境成本归因直接失效。务必按环境拆分 Key并在埋点记录里加上env字段。这个字段在排查为什么预发环境消耗了生产级别的 Token时特别有用。6. 下一步把调度策略跑成常态到这里你已经有了统一接入、埋点统计、调度骨架和排查清单。接下来最重要的一步是让统计脚本定期跑起来比如每天凌晨聚合前一天的数据输出成本 Top 10 的业务场景和模型组合。连续跑一周你就能看出哪些场景在恶化、哪些调度规则需要调整。如果你要长期维护多个 Agent 服务建议把调度策略和成本监控做成独立模块而不是散落在各个业务代码里。TaoToken 的 Coding Plan 适合需要长期编码和 Agent 调度的场景模型对话入口可以用来快速验证不同模型在同一 Prompt 下的输出质量和 Token 消耗差异接入文档里有完整的端点说明和参数列表。先把埋点跑通再谈优化顺序不要反。