把 LangChain Agent 中间件的模型 base_url 改到 TaoToken，token_usage 照常统计

LangChain Agent 中间件里，after_model 钩子负责统计每次模型调用的 token_usage，可当主模型、摘要模型、审核模型分散在多个官方接口时，这套统计会被拆散到不同控制台，Key 也要各管各的。TaoToken 提供一条兼容通道，官网在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end ，先在那里创建 Key，再回来看中间件代码怎么改。所有模型的 base_url 指向 https://taotoken.net/api，token_usage 照常从 response_metadata 读取，但全部记在一把 Key 名下。本文延续《LangChain Agent 中间件实战》那套钩子机制，用同一段 create_agent 代码把模型通道换掉，让对话摘要、内容审核、Token 统计都对齐到一个 base_url 上。

1. 中间件管得住流程，管不住分散的 Token 账本

1.1 模型调用链路上的拦截点

一个 Agent 的真实执行路径并不只是“输入—回答”那么短。用户消息先进来，模型要决定是否调用工具；工具返回结果后，模型还要继续推理；最后才生成最终回答。LangChain v1 的中间件能在这些节点之间插入逻辑：模型调用前改请求、模型调用后读响应、工具执行前做权限校验、工具执行后做异常转换、Agent 启动和结束时做状态清理。

可以用一张表归纳这些拦截位置和它们能做的事：

中间件解决的核心问题是横切关注点：日志、审核、重试、统计这些逻辑如果不拆出来，全部会写进 Agent 主循环里。但拆出来之后还会遇到另一个问题：模型请求到底发给了谁，统计数字又由谁来汇总。

1.2 base_url 不统一，token_usage 就对不上账

after_model 钩子里读response.result[-1].response_metadata["token_usage"]["completion_tokens"]是 OpenAI 兼容接口的标准返回格式。可如果主模型用的是 A 平台的官方接口，摘要模型用的是 B 平台的官方接口，两边返回的 token_usage 字段结构也许一样，但控制台各自独立，拿到的是两套账单。统计脚本要从两个后台分别拉数据，再手工拼接；Key 也要分散保存，每多一个模型来源，就多一组需要轮换的密钥。

TaoToken 的做法是让这些请求统一指向同一个 base_url。主模型、摘要模型、审核模型都填https://taotoken.net/api，API Key 用同一把。模型响应仍然保留 OpenAI 兼容结构，token_usage照常出现在response_metadata里，只是它的来源从多个平台变成了一个通道。这样 after_model 钩子本身不用改，你只需要改模型客户端的 base_url 和 api_key 两处配置。

2. 把 create_agent 的模型客户端统一指向 TaoToken

2.1 先拿 Key，再抄模型 ID

去 TaoToken 注册后，在控制台创建一把 API Key。复制出来的 Key 形如YOUR_API_KEY，后续代码里所有模型的 api_key 栏都填它，不用为每个模型分别申请。模型 ID 不用猜，以 https://taotoken.net/?utm_source=taotoken_aicg_blog_end 模型广场当时列出的为准；同一个语义的模型在广场上会给出规范的模型标识，复制后直接填进 ChatOpenAI 的 model 参数。

建议先别急着写代码，先在网页上把 Key 和模型 ID 各测一遍，能省掉不少后面 debug 的时间。

2.2 ChatOpenAI 的 base_url 填 https://taotoken.net/api

在 LangChain 里，Base URL 是跟着模型客户端走的。先把主模型建出来：

from langchain_openai import ChatOpenAI # 模型 ID 请以 TaoToken 模型广场当时列表为准 MODEL_ID = "YOUR_MODEL_ID" main_llm = ChatOpenAI( model=MODEL_ID, api_key="YOUR_API_KEY", base_url="https://taotoken.net/api", )

注意https://taotoken.net/api末尾不要加/v1。很多从官方文档复制习惯的人会自动补上/v1，LangChain 的 OpenAI 兼容客户端会自行拼接路径，多余的/v1会直接 404。然后创建 Agent：

from langchain.agents import create_agent from langgraph.checkpoint.memory import InMemorySaver agent = create_agent( model=main_llm, tools=[weather_tool, calculator_tool], checkpointer=InMemorySaver(), )

这样 agent 内部所有对主模型的调用都走 TaoToken 通道，invoke返回值里的response_metadata会携带该通道返回的 token 用量。

2.3 动态切换模型时只换 model，不换 Key

LangChain 中间件里常见的做法是根据对话长度切换模型。原文里dynamic_model_selection通过request.override(model=...)换模型，放在 TaoToken 通道下这段逻辑不需要改动，只需要在 handler 里换成不同的 ChatOpenAI 实例：

MODELS = { "basic": ChatOpenAI( model="YOUR_BASIC_MODEL_ID", api_key="YOUR_API_KEY", base_url="https://taotoken.net/api", ), "advanced": ChatOpenAI( model="YOUR_ADVANCED_MODEL_ID", api_key="YOUR_API_KEY", base_url="https://taotoken.net/api", ), } from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse @wrap_model_call def dynamic_model_selection( request: ModelRequest, handler, ) -> ModelResponse: message_count = len(request.state["messages"]) selected = MODELS["advanced"] if message_count > 10 else MODELS["basic"] return handler(request.override(model=selected))

这些实例共用同一个 api_key 和 base_url，只有 model 字段不同。切换模型不会切断统计链路，因为请求始终发往同一个 base_url，token_usage 最终都会回到同一处。根据状态过滤工具也是同样的逻辑，request.override(tools=...)只改工具列表，不影响模型通道和 Key。

3. 摘要中间件和审核中间件也复用同一通道

3.1 SummarizationMiddleware 的模型同样指向 TaoToken

对话摘要的用法是：当上下文达到触发条件时，用较轻量的摘要模型压缩早期消息，再把“历史摘要 + 最近消息”交给主模型。这个摘要模型虽然任务不同，但它同样是一次模型调用，也要被统计。

构造摘要模型时保持同一套配置：

from langchain.agents.middleware import SummarizationMiddleware summarizer_llm = ChatOpenAI( model="YOUR_SUMMARIZER_MODEL_ID", # 以模型广场列表为准 api_key="YOUR_API_KEY", base_url="https://taotoken.net/api", ) summarization = SummarizationMiddleware( model=summarizer_llm, trigger=("tokens", 4000), keep=("messages", 20), )

把摘要中间件加进 create_agent 的 middleware 列表：

agent = create_agent( model=main_llm, tools=[weather_tool, calculator_tool], checkpointer=InMemorySaver(), middleware=[summarization], )

这样主模型和摘要模型共用一把 Key。after_model 钩子统计 token_usage 时，摘要模型产生的消耗也出现在同一套汇总里，不会因为来源不同被漏掉。trigger和keep的配置可以参照原文的几种组合：按 token 数量、按消息数量、按上下文窗口比例触发，按需调整即可。

3.2 内容审核中间件的模型接入

面向用户开放的 Agent 还需要审核输入、输出和工具返回结果。LangChain 的 OpenAIModerationMiddleware 负责这个场景，它的模型客户端同样可以指向 TaoToken：

from langchain_openai.middleware import OpenAIModerationMiddleware moderation_llm = ChatOpenAI( model="YOUR_MODERATION_MODEL_ID", api_key="YOUR_API_KEY", base_url="https://taotoken.net/api", ) moderation = OpenAIModerationMiddleware( model=moderation_llm, check_input=True, check_output=True, check_tool_results=True, exit_behavior="end", violation_message="请求或响应触发了内容安全策略。", )

不同版本的 LangChain v1 对中间件 model 参数的类型要求略有差异，有的直接传字符串模型名，有的传实例。只要最终构造模型时带上了同一个 base_url，审核模型产生的 token_usage 也能并入总账。exit_behavior按业务需要设置为end、error或replace：对外服务多数选择end，需要统一异常处理的后端选择error，替换后继续则用replace，与原文讨论的安全策略保持一致。

4. after_model 里照常读 token_usage

4.1 节点风格：在模型返回后更新状态

节点风格钩子按生命周期触发，适合记录调用次数和 Token 汇总。下面这个 after_model 钩子从模型返回的最后一条消息里读取 completion_tokens，并累加到 AgentState：

from typing import Any from langchain.agents import AgentState from langchain.agents.middleware import after_model from langgraph.runtime import Runtime @after_model def track_token_usage( state: AgentState, runtime: Runtime, ) -> dict[str, Any] | None: last_message = state["messages"][-1] usage = last_message.response_metadata.get("token_usage", {}) completion_tokens = usage.get("completion_tokens", 0) total = state.get("total_completion_tokens", 0) + completion_tokens return { "total_completion_tokens": total, "last_completion_tokens": completion_tokens, }

钩子返回的字典会合并进 AgentState。这个钩子跟模型供应商无关，只要响应是 OpenAI 兼容格式并且包含 token_usage 字段就能正常工作。TaoToken 返回的响应完整保留这套字段，所以节点钩子一行都不用改，统计口径自动收口。

4.2 包装风格：原样读取 response_metadata

更贴近底层的方式是用 wrap_model_call 包裹模型调用，在 handler 返回后直接读响应：

from collections.abc import Callable from langchain.agents.middleware import ( ExtendedModelResponse, ModelRequest, ModelResponse, wrap_model_call, ) from langgraph.types import Command @wrap_model_call def track_token_usage_completion( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ExtendedModelResponse: response = handler(request) usage = response.result[-1].response_metadata.get("token_usage", {}) tokens = usage.get("completion_tokens", 0) return ExtendedModelResponse( model_response=response, command=Command( update={"last_completion_tokens": tokens}, ), )

response.result[-1]是模型返回的最后一条消息，response_metadata["token_usage"]["completion_tokens"]是本次调用的补全 Token 数。改为.get是防止个别模型不返回该字段时整个 Agent 崩溃；走 TaoToken 通道后，字段结构保持一致，这段代码长期不用动。

包装风格钩子还可以额外控制流程：在统计 token 的同时通过 Command 更新状态，或在满足条件时附带jump_to提前结束。但要注意，包装风格不能直接改变执行方向，它只能决定是否调用 handler，以及把什么状态写进下一步。若想在满足策略条件时直接结束 Agent，用节点风格的jump_to="end"更清晰。

5. 扩展 AgentState 并关注中间件顺序

5.1 自定义字段记录 Token 的汇总结果

如果要在多次模型调用之间共享累计数据，可以像原文那样扩展 AgentState：

from typing import NotRequired from langchain.agents import AgentState class TrackingState(AgentState): last_completion_tokens: NotRequired[int] total_completion_tokens: NotRequired[int]

然后在钩子上声明 state_schema：

@after_model(state_schema=TrackingState) def add_token_stats(state: TrackingState, runtime: Runtime) -> dict[str, Any]: last_message = state["messages"][-1] usage = last_message.response_metadata.get("token_usage", {}) return { "last_completion_tokens": usage.get("completion_tokens", 0), "total_completion_tokens": state.get("total_completion_tokens", 0) + usage.get("completion_tokens", 0), }

如果一个中间件需要同时处理 before_agent、after_agent、wrap_model_call 等多个钩子，类式中间件比多个装饰器更容易组织。你可以继承 AgentMiddleware，把统计逻辑放在 wrap_model_call 里，把汇总结果放在 after_agent 里。选择标准仍然是：单一职责用装饰器，多钩子协作用类。这两组字段不只是展示数字，还能在 Agent 结束前由 after_agent 钩子统一写入日志，或作为后续路由决策的输入。

5.2 顺序不对，统计口径就变

LangChain v1 的多个中间件按注册顺序嵌套。注册[middleware_a, middleware_b]后，前置钩子按 A.before → B.before 执行，后置钩子按 B.after → A.after 返回。这个顺序会直接影响 token_usage 的统计结果：

• 摘要中间件先于统计钩子执行时，摘要模型产生的 Token 会被统计进去，因为统计钩子在模型调用后能看到摘要模型的响应。
• 审核中间件如果先于统计钩子执行，审核模型的输出也会被统计；但如果审核中间件拦截并替换了最终结果，统计钩子读到的是替换后的内容。
• 包装风格里，如果外层中间件捕获异常并提前返回，内层统计钩子可能根本没机会执行。

一个比较稳的注册顺序是：把统计类钩子放在 middleware 列表靠后的位置，让它能包裹住摘要、审核、动态模型这些中间件。这样无论下游如何换模型、做摘要、审内容，统计钩子都能拿到最终的 token_usage。

agent = create_agent( model=main_llm, tools=[weather_tool, calculator_tool], checkpointer=InMemorySaver(), middleware=[summarization, moderation, track_token_usage], )

如果统计钩子是节点风格的 after_model，它本身不参与嵌套，但你需要确认它和其它中间件都在 middleware 列表里，且没有被前面的中间件短路。

6. 跑通 invoke 后回 TaoToken 控制台对账

6.1 先用网页对话验证 Key 和模型 ID

配置改完后，先用一段简单的 invoke 跑通链路：

config = {"configurable": {"thread_id": "user-session-001"}} result = agent.invoke( {"messages": [{"role": "user", "content": "今天天气怎么样"}]}, config=config, ) print(result["messages"][-1].response_metadata.get("token_usage"))

如果打印出的 token_usage 正常出现，说明模型调用已经走通 TaoToken 通道。若想进一步确认模型 ID 是否可用，可以在 TaoToken 模型对话 里用同一把 Key 发一条相同内容的消息，两边行为一致就说明配置没问题。

6.2 回控制台核对该次调用的 completion_tokens

跑通之后，打开 https://taotoken.net/?utm_source=taotoken_aicg_blog_end 的用量记录，找到刚才那次调用的 completion_tokens，和 print 出来的数字对照。一致则说明 Agent 请求确实经过统一通道，统计也收口到了同一处。以后主模型、摘要模型、审核模型的 Key 和用量都能在这一处管理，不需要再切换多个平台后台。

这一步很值得养成习惯：每次改完模型配置后都回控制台对一次账，既能确认 base_url 填对了，又能提前发现模型 ID 是否匹配。Token 统计不是写完后看一眼就结束的，它应该成为 Agent 上线后的例行检查项。

6.3 常见报错与处理

配置过程中比较常见的错误集中在三点：

• 401 Unauthorized：API Key 复制不完整，或者 Key 创建后没有保存完整。回 API Keys 页面重新创建一把，替换掉代码里的YOUR_API_KEY。
• 404 Not Found：base_url 末尾加了/v1。确认 ChatOpenAI 的 base_url 是https://taotoken.net/api，不带/v1。
• response_metadata 里没有 token_usage 字段：模型 ID 对应的模型可能不返回 OpenAI 兼容的用量结构，到模型广场换一个兼容模型 ID 再试。

排障时注意，不要为了图省事在 base_url 里附带查询参数，接口地址只填https://taotoken.net/api。官网落地页和接口地址是两回事：注册、建 Key、看用量走 https://taotoken.net/?utm_source=taotoken_aicg_blog_end ；填进代码的 base_url 只写接口地址。把这个边界守住，配置就不会乱。

当你把这次改造真正用起来，可以先用 模型对话 验证刚才打印的这句消息，确认通道稳定；再根据日常调用量评估 Coding Plan 是否够用。Key 始终从 API Keys 页面 创建和轮换，不要把旧 Key 继续留在代码里。中间件画出的控制边界，最终要靠统一通道来兜底。