
1. 多 Agent 协作 Token 膨胀到底出在哪如果你正在用 CrewAI 或者 LangGraph 搭多智能体系统大概率遇到过这个现象单 Agent 跑一个任务消耗 2k Token换成 planner / executor / reviewer 三个角色协作同样的任务直接飙到 8k 甚至 12k。这不是错觉CrewAI 官方文档在「预算与成本控制」一节里就明确写过多 Agent 协作会产生 3~5 倍的 Token 膨胀。问题在于很多人第一反应是去优化 Prompt、砍上下文却忽略了一个更底层的原因——每个 Agent 各自持有一个模型客户端各自连不同的厂商端点成本边界和调用边界从一开始就是散的。我试过在一个三节点的 StateGraph 里planner 走某厂商的强模型、executor 走另一个厂商的小模型、reviewer 又换一家结果就是三个 API Key、三套计费口径、三种限流策略LangSmith 的 trace 里能看到调用成功但你根本说不清哪个角色在持续烧 Token。更麻烦的是当你想做「模型分级路由」——Planner 用强模型、Executor 用小模型——你得在三个控制台之间来回切换改一个 base_url 就要重新申请一次 Key。这篇就是对着「多 Agent 协作 Token 膨胀」这个现象来改配置的。核心动作只有一个把每个 Agent 模型客户端的 base_url 统一指向同一个入口Key 也只用一个。编排、状态流转、条件路由仍然由 CrewAI / LangGraph 自己完成TaoToken 只负责提供 Key 和 Base URL。改完之后你可以在 trace 里逐个节点核对调用找出到底是哪个角色在膨胀再补上单任务预算和超限阻断。适合谁看已经跑通单 Agent、正在往多 Agent 协作迁移或者已经被 Token 账单吓到过的开发者。不需要你重写编排逻辑只需要改几行客户端初始化代码。2. 前置准备一个 Key 管住所有 Agent 的模型入口在动手改 base_url 之前先把入口统一掉。打开 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_end 注册账号然后在控制台里创建一个 API Key。这个 Key 就是你后面所有 Agent 共用的凭证不用再为 planner、executor、reviewer 分别去不同厂商申请。创建 Key 的入口在控制台的 API Keys 页面直接访问 https://taotoken.net/console 就能看到。生成之后先复制保存后面配置环境变量要用。这里要强调一个容易踩的坑Base URL 填https://taotoken.net/api不要带/v1。很多 OpenAI 兼容客户端默认会自己在后面拼/v1/chat/completions如果你手动写成https://taotoken.net/api/v1最终请求路径就会变成/api/v1/v1/chat/completions直接 404。这一点在接入文档里有说明地址是 https://taotoken.net/doc 配置前建议扫一眼。环境变量建议这样组织把 Key 和 Base URL 分开存方便后面在多个 Agent 之间复用export TAOTOKEN_API_KEYsk-你的Key export TAOTOKEN_BASE_URLhttps://taotoken.net/api如果你用的是.env文件配合 python-dotenv写法一样只是去掉 export。这样做的目的是后面无论 CrewAI 的 Agent 还是 LangGraph 的节点初始化模型客户端时都从这两个变量读改一处就全局生效。模型分级路由也在同一个入口下调整。你不需要为「Planner 走强模型、Executor 走小模型」去开两个账号只需要在创建客户端时传不同的model参数即可。比如 planner 用gpt-4o这类强模型executor 用gpt-4o-mini这类小模型两者共用同一个 base_url 和同一个 Key。这样成本口径统一trace 里也能按 model 字段区分。3. 可复制配置CrewAI 与 LangGraph 双份改法先看 CrewAI。CrewAI 底层用的是 LiteLLM 做模型调用所以最干净的方式是通过环境变量让 LiteLLM 走统一入口而不是在每个 Agent 里硬编码。你可以在 Crew 启动前设置import os from crewai import Agent, Task, Crew, Process os.environ[OPENAI_API_KEY] os.environ[TAOTOKEN_API_KEY] os.environ[OPENAI_API_BASE] os.environ[TAOTOKEN_BASE_URL] planner Agent( rolePlanner, goal拆解用户任务为可执行步骤, backstory你负责规划不直接执行, llmgpt-4o, verboseTrue, ) executor Agent( roleExecutor, goal按计划执行具体步骤, backstory你负责执行遇到问题上报, llmgpt-4o-mini, verboseTrue, ) reviewer Agent( roleReviewer, goal校验执行结果是否符合预期, backstory你负责质量把关, llmgpt-4o-mini, verboseTrue, ) crew Crew( agents[planner, executor, reviewer], tasks[...], processProcess.sequential, verboseTrue, )关键点在于OPENAI_API_BASE这个变量LiteLLM 会读取它作为所有 OpenAI 兼容调用的默认端点。三个 Agent 虽然llm不同但都走同一个 base_urlKey 也只有一个。这样 Token 膨胀的账就集中在一个地方不会散落到三个厂商。再看 LangGraph。原文 4.1 用 StateGraph 把 planner / executor / reviewer 三个节点接进同一张图每个节点各自持有模型客户端。改法是把客户端初始化抽出来统一从环境变量读import os from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, END from typing import TypedDict class TaskState(TypedDict): task: str plan: list results: dict done: bool def build_llm(model: str, temperature: float 0.2): return ChatOpenAI( modelmodel, temperaturetemperature, api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], ) planner_llm build_llm(gpt-4o) executor_llm build_llm(gpt-4o-mini) reviewer_llm build_llm(gpt-4o-mini) def planner_agent(state: TaskState): resp planner_llm.invoke(f拆解任务{state[task]}) return {plan: resp.content.split(\n)} def executor_agent(state: TaskState): resp executor_llm.invoke(f执行计划{state[plan]}) return {results: {output: resp.content}} def reviewer_agent(state: TaskState): resp reviewer_llm.invoke(f校验结果{state[results]}) return {done: 通过 in resp.content} workflow StateGraph(TaskState) workflow.add_node(planner, planner_agent) workflow.add_node(executor, executor_agent) workflow.add_node(reviewer, reviewer_agent) workflow.add_edge(planner, executor) workflow.add_edge(executor, reviewer) workflow.add_conditional_edges( reviewer, lambda state: done if state[done] else retry, {done: END, retry: executor}, ) workflow.set_entry_point(planner) app workflow.compile() result app.invoke({task: 分析Q2销售数据})注意base_url传的是https://taotoken.net/api不带/v1。ChatOpenAI内部会自己拼/chat/completions。如果你用的是其他 LangChain 兼容客户端参数名可能是openai_api_base但值是一样的。模型分级路由在这里体现得很清楚planner 用强模型负责规划executor 和 reviewer 用小模型负责执行和校验。三个客户端共用同一个 Key 和 base_url但 model 字段不同。这样你在 trace 里既能按节点看调用也能按 model 看成本分布。4. 验证请求从 trace 里逐个节点核对配置改完先别急着跑完整任务用一个最小请求验证链路通不通。最直接的方式是用 curl 打一次 chat completionscurl https://taotoken.net/api/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: gpt-4o-mini, messages: [{role: user, content: 回复 OK}] }如果返回里有正常的choices字段说明 Key 和 Base URL 都没问题。如果返回 401检查 Key 是否复制完整如果返回 404八成是 base_url 多写了/v1。链路通了之后跑一次完整的 LangGraph 任务然后打开 LangSmith 或 LangFuse 的 trace 面板。你要重点看三件事第一每个节点的调用是否成功。planner、executor、reviewer 三个节点应该各自有一条 LLM 调用记录状态都是 success。如果某个节点报错先看它的 model 字段是不是写错了。第二哪个角色在持续消耗 Token。在 trace 里按节点展开看每个节点的 prompt tokens 和 completion tokens。多 Agent 协作的膨胀往往集中在某一个角色上——常见的是 reviewer 因为反复校验、或者 executor 因为重试循环导致 Token 远超预期。第三重试路径是否被触发。LangGraph 的条件路由里reviewer 不通过会回到 executor 重试。如果 trace 里看到 executor 被调用了多次说明校验逻辑太严或者 executor 输出质量不够这时候要么调 reviewer 的判定 Prompt要么给 executor 换更强的模型。实测下来把三个 Agent 的 base_url 统一之后最直观的变化是成本口径清晰了。以前三个厂商三份账单现在一个入口按 model 字段就能拆出 planner 和 executor 各自的消耗占比。原文提到的「单任务预算」和「超限阻断」也有了落地基础——你可以在每个节点调用前检查累计 Token超过阈值就抛异常终止而不是等账单出来才发现。5. 本篇常见错排查报错一404 Not Found路径里出现两个 /v1这是最高频的坑。原因就是 base_url 写成了https://taotoken.net/api/v1而客户端又自动拼了/v1/chat/completions。解决方法是把 base_url 改回https://taotoken.net/api不带/v1。CrewAI 的 LiteLLM 和 LangChain 的 ChatOpenAI 都是这个规则。报错二401 UnauthorizedKey 无效先确认环境变量有没有真正加载。在 Python 里打印os.environ.get(TAOTOKEN_API_KEY)看是不是 None。如果是 None说明.env没被读取或者 export 的 shell 和跑代码的 shell 不是同一个。另外检查 Key 有没有多余空格复制的时候容易带上换行。报错三某个 Agent 调用成功另一个失败如果 planner 成功但 executor 失败先看两个 Agent 的 model 字段。有些模型名在不同入口下的可用性不一样确认你填的 model 是当前入口支持的。另外检查是不是某个 Agent 硬编码了旧的 base_url没有走统一的环境变量。CrewAI 里如果 Agent 初始化时显式传了llm对象而不是字符串那个对象可能还带着旧端点。报错四Token 消耗没有下降反而更高统一入口本身不会自动降低 Token 消耗它只是让消耗可见、可管。如果改完之后发现总 Token 没降去 trace 里看是不是重试循环变多了。常见原因是 reviewer 的判定太严格导致 executor 反复重跑。这时候要调的是校验逻辑不是 base_url。另外确认模型分级路由有没有生效——如果 executor 还在用强模型成本自然下不来。报错五LangSmith 里看不到 traceLangSmith 需要在环境变量里配LANGCHAIN_TRACING_V2true和LANGCHAIN_API_KEY。这两个和 TaoToken 的 Key 是两回事别混。如果 trace 一直不出现先确认这两个变量有没有设再看网络能不能通到 LangSmith 的端点。6. 把预算和阻断补上链路跑通、trace 能看之后最后一步是把原文说的单任务预算和超限阻断落地。思路很简单在 StateGraph 的状态里加一个token_used字段每次节点调用后累加超过阈值就路由到 END 或者抛异常。class TaskState(TypedDict): task: str plan: list results: dict done: bool token_used: int TOKEN_BUDGET 20000 def planner_agent(state: TaskState): resp planner_llm.invoke(f拆解任务{state[task]}) used state.get(token_used, 0) resp.usage_metadata[total_tokens] if used TOKEN_BUDGET: raise RuntimeError(fToken 预算超限{used}) return {plan: resp.content.split(\n), token_used: used}每个节点都做同样的累加和检查这样无论哪个角色在膨胀都会在超限时立刻中断而不是跑完整个流程才发现账单爆了。配合 trace 里的节点级消耗你就能定位到是 planner 规划太啰嗦、还是 executor 重试太多、还是 reviewer 校验太严然后针对性优化。如果你后面要长期跑多 Agent 的编码或 Agent 任务可以看一下 Coding Plan 这个入口 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 它更适合持续性的编码场景。日常调试模型调用是否正常用模型对话页面 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 快速验证就行。Key 的管理和新建都在 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 接入细节以文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 为准。Claude Code 相关的接入配置在 https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaudecodeutm_campaignrewrite 有单独说明。改完这一轮你手里应该有一个统一的模型入口、一份按节点可查的 trace、以及一个能自动阻断的超限逻辑。多 Agent 协作的 Token 膨胀不会凭空消失但至少从「不知道为什么贵」变成了「知道贵在哪、能管住」。