1. 工具堆到 50 个,模型开始“装傻”是怎么回事
如果你正在用 LangChain 搭 Agent,并且工具数量已经上到几十个,大概率遇到过这种场面:用户只是说了句“你好”,模型却要先把 50 个工具的 JSON Schema 从头读一遍;用户问“帮我算个平均数”,模型在几十段工具描述里翻来翻去,最后选了个八竿子打不着的工具,或者干脆不调用工具直接编答案。这不是模型变笨了,而是你一次性把太多无关信息塞进了它的上下文窗口。
LangChain 1.1 的 Middleware 机制给了我们一个很干净的解法:在每次模型调用之前拦截请求,根据当前 Agent 的状态动态决定这次到底暴露哪些工具。这正是 Claude Skills 的核心思路——渐进式披露,按需加载。本文要做的,就是把原文那套wrap_model_call+request.override(tools=filtered_tools)的动态过滤逻辑完整复现出来,同时把模型通道接到 TaoToken 上,让你拿到 Key 之后能直接跑通整个 Agent。
适合谁看:已经写过基础 LangChain Agent、手里工具超过 10 个、想控制 token 消耗和提升工具选择准确率的开发者。你需要对 Python 和 LangChain 的@tool装饰器有基本了解,剩下的步骤我都会给全。
整篇文章的结构是这样:先讲清楚传统“全量工具暴露”的痛点,再配置 TaoToken 的模型通道,然后一步步写 SkillState、Loader 工具、SkillMiddleware,最后用一组销售数据跑测试,通过日志验证动态过滤确实生效。TaoToken 在这里只负责模型通道的 Key 和 Base URL,不参与 Middleware 的过滤逻辑,两者职责分开,后面配置时我会特别标注。
2. 把 LangChain Agent 的模型通道接到 TaoToken
原文 3.2 那一步是直接配 DeepSeek 官方通道,这里我们改成走 TaoToken。原因很简单:你只需要一个 Key,就能在 LangChain 的兼容 OpenAI 客户端里调用 DeepSeek-v3.2 这类模型,不用为每个模型单独维护一套鉴权。TaoToken 在这里的角色就是模型通道提供方,给你 Key 和 Base URL,Middleware 的过滤逻辑完全不受影响。
先打开 https://taotoken.net/?utm_source=taotoken_aicg_blog_end 创建 Key。登录后在控制台里找到 API Keys 页面,新建一个 Key 并复制保存。这个 Key 就是你后面填进 LangChain 模型客户端的凭证。
拿到 Key 之后,在项目根目录建一个.env文件,把 Key 写进去:
# .env TAOTOKEN_API_KEY=sk-你的TaoToken密钥注意 Base URL 的写法,这是最容易踩坑的地方。TaoToken 的 API 地址是https://taotoken.net/api,不带/v1,也不要把官网那个带 UTM 参数的地址填进去。官网地址是给人看的,Base URL 是给程序调用的,两者不能混。如果你把?utm_source=...那一长串填进 Base URL,请求会直接 404 或者鉴权失败。
在 LangChain 里,如果你用的是兼容 OpenAI 的 ChatModel,配置大概是这样:
import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv(override=True) model = ChatOpenAI( model="deepseek-v3.2", api_key=os.getenv("TAOTOKEN_API_KEY"), base_url="https://taotoken.net/api", temperature=0.7, )如果你沿用原文那个自定义的DeepSeekReasonerChatModel适配器,改法也一样:把api_key指向 TaoToken 的 Key,把base_url指向https://taotoken.net/api,模型名按你实际要用的填。适配器内部怎么处理推理字段是它自己的事,通道层只认 Key 和 Base URL。
这里再强调一次职责边界:TaoToken 只出现在模型通道配置里。SkillMiddleware读的是request.state里的skills_loaded,跟模型走哪个通道没有任何关系。你换成别的兼容通道,Middleware 代码一行都不用动。
3. 可复制配置:SkillState、Loader 工具与 SkillMiddleware
这一节是全文的技术主体,把原文 3.5 到 3.9 的代码完整走一遍。我按依赖顺序拆开,你可以直接照着敲。
3.1 定义 SkillState 状态
Agent 需要在多轮调用之间记住“哪些技能已经加载了”,所以我们要在状态里加一个skills_loaded字段。用Annotated配一个累加 reducer,保证新加载的技能是追加而不是覆盖:
from typing import Annotated, List from langgraph.graph import MessagesState def skill_list_accumulator(current: List[str], new: List[str]) -> List[str]: if not current: return new combined = current + [s for s in new if s not in current] return combined class SkillState(MessagesState): skills_loaded: Annotated[List[str], skill_list_accumulator] = []MessagesState负责消息历史,skills_loaded负责技能清单,两者互不干扰。
3.2 定义 Loader 工具和功能工具
工具分三类:Loader 工具始终可见,用来加载技能;数据分析和文本处理工具只在对应技能加载后才可见。Loader 工具返回Command,在更新消息的同时把技能名写进skills_loaded:
from langgraph.types import Command from langchain_core.messages import ToolMessage from langchain_core.tools import tool @tool def skill_data_analysis(runtime) -> Command: """加载数据分析技能。""" instructions = "数据分析技能已加载,可用工具:calculate_statistics、generate_chart" return Command(update={ "messages": [ToolMessage(content=instructions, tool_call_id=runtime.tool_call_id)], "skills_loaded": ["data_analysis"] }) @tool def skill_text_processing(runtime) -> Command: """加载文本处理技能。""" instructions = "文本处理技能已加载,可用工具:summarize_text、extract_keywords" return Command(update={ "messages": [ToolMessage(content=instructions, tool_call_id=runtime.tool_call_id)], "skills_loaded": ["text_processing"] })功能工具就是普通的@tool,这里给两个数据分析的示例:
@tool def calculate_statistics(numbers: List[float]) -> str: """计算一组数字的统计信息。""" import statistics if not numbers: return "错误:数字列表为空" return f"统计结果: mean={statistics.mean(numbers)}, max={max(numbers)}" @tool def generate_chart(data: List[float], chart_type: str = "bar") -> str: """根据数据生成图表(模拟)。""" return f"已生成 {chart_type} 图表,包含 {len(data)} 个数据点"把工具分组,方便后面做映射:
LOADER_TOOLS = [skill_data_analysis, skill_text_processing] DATA_ANALYSIS_TOOLS = [calculate_statistics, generate_chart] ALL_TOOLS = LOADER_TOOLS + DATA_ANALYSIS_TOOLS3.3 工具映射与过滤函数
过滤逻辑的核心是一张“技能到工具”的映射表,加上一个根据skills_loaded拼装工具列表的函数:
SKILL_TOOL_MAPPING = { "data_analysis": DATA_ANALYSIS_TOOLS, } def get_tools_for_skills(skills_loaded: List[str]) -> List: tools = list(LOADER_TOOLS) for skill_name in skills_loaded: if skill_name in SKILL_TOOL_MAPPING: tools.extend(SKILL_TOOL_MAPPING[skill_name]) return tools注意 Loader 工具永远在列表里,因为模型需要它们来触发技能加载。
3.4 实现 SkillMiddleware
这是整个方案的关键。wrap_model_call在每次模型调用前执行,从request.state读skills_loaded,算出过滤后的工具列表,再用request.override(tools=filtered_tools)替换掉原始请求里的工具:
from typing import Callable, List from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse class SkillMiddleware(AgentMiddleware): def __init__(self, verbose: bool = True): super().__init__() self.verbose = verbose self.call_count = 0 def _get_skills_from_state(self, request: ModelRequest) -> List[str]: if hasattr(request, "state") and request.state is not None: if isinstance(request.state, dict): return request.state.get("skills_loaded", []) return getattr(request.state, "skills_loaded", []) return [] def wrap_model_call( self, request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse: self.call_count += 1 skills_loaded = self._get_skills_from_state(request) filtered_tools = get_tools_for_skills(skills_loaded) if self.verbose: print(f"[SkillMiddleware] 第 {self.call_count} 次模型调用") print(f"skills_loaded: {skills_loaded}") print(f"过滤后工具: {[t.name for t in filtered_tools]}") filtered_request = request.override(tools=filtered_tools) return handler(filtered_request)request.override()返回的是一个新请求对象,原始请求不变,这样多个 Middleware 串联时不会互相污染。
3.5 用 create_agent 组装
最后把所有组件装进 Agent:
from langchain.agents import create_agent SYSTEM_PROMPT = """你是一个智能助手。 1. 你有两类工具:Skill Loader 和功能工具。 2. 当用户请求某个功能时,如果当前没有对应功能工具,先调用 Skill Loader 加载技能。 3. 加载后,使用新获得的工具完成任务。""" agent = create_agent( model=model, tools=ALL_TOOLS, middleware=(SkillMiddleware(verbose=True),), state_schema=SkillState, system_prompt=SYSTEM_PROMPT, )tools=ALL_TOOLS是把所有工具注册进去,但真正每次发给模型的是 Middleware 过滤后的子集。这就是“注册全量、暴露按需”的写法。
4. 验证请求:跑销售数据测试,看日志确认过滤生效
配置写完了,得用实际请求验证。构造一个销售数据统计的输入,初始skills_loaded为空:
from langchain_core.messages import HumanMessage test_input = { "messages": [HumanMessage(content="我有一组销售数据 [150, 200, 180, 220, 190],请帮我计算统计信息")], "skills_loaded": [] } result = agent.invoke(test_input)跑起来之后,重点看SkillMiddleware打印的日志。预期会看到两次模型调用:
第一次调用时,skills_loaded是空列表,过滤后只剩 2 个 Loader 工具:
[SkillMiddleware] 第 1 次模型调用 skills_loaded: [] 过滤后工具: ['skill_data_analysis', 'skill_text_processing']模型看到只有 Loader 工具,判断需要数据分析能力,于是调用skill_data_analysis。这个工具返回Command,把data_analysis写进skills_loaded。
第二次调用时,skills_loaded已经变成['data_analysis'],过滤后工具增加到 4 个:
[SkillMiddleware] 第 2 次模型调用 skills_loaded: ['data_analysis'] 过滤后工具: ['skill_data_analysis', 'skill_text_processing', 'calculate_statistics', 'generate_chart']模型这次看到了calculate_statistics,用它算出平均值 188.0、最大值 220,返回最终答案。整个过程里,模型第一次只面对 2 个工具,第二次面对 4 个工具,而不是一上来就面对全部工具。如果你把工具数量放大到 50 个,这个差距会非常明显。
验证成功的标志就是日志里这两次调用的工具数量变化:2 个变 4 个,且skills_loaded从空变成['data_analysis']。如果第二次调用工具数量没变,说明状态没写进去或者 Middleware 没读到,往下看排查部分。
5. 本篇常见错排查
5.1 Base URL 填错导致 404 或鉴权失败
最常见的错误是把官网地址https://taotoken.net/?utm_source=...填进了base_url。Base URL 必须是https://taotoken.net/api,不带/v1,不带任何查询参数。如果你在日志里看到 404 或者invalid api key,先检查这一项。另外确认.env里的 Key 没有多余空格,load_dotenv(override=True)要放在读取环境变量之前。
5.2 skills_loaded 一直是空,工具数量不变
如果第二次调用日志里skills_loaded还是[],通常是 Loader 工具没有正确返回Command。检查两点:一是Command的update字典里键名必须是skills_loaded,跟SkillState的字段名完全一致;二是ToolMessage的tool_call_id要取自runtime.tool_call_id,写错会导致消息更新失败,状态也就带不出来。
5.3 Middleware 没生效,模型看到全部工具
如果日志里根本没有[SkillMiddleware]的输出,说明 Middleware 没被注册进 Agent。检查create_agent的middleware参数是不是传了元组(SkillMiddleware(verbose=True),),注意末尾的逗号。另外确认state_schema=SkillState传对了,否则request.state里读不到自定义字段。
5.4 工具名冲突或重复注册
ALL_TOOLS里如果出现同名工具,LangChain 在绑定工具时可能报错或者行为异常。确保每个@tool函数的名称唯一。Loader 工具和功能工具不要重名,映射表里的技能名也要和 Loader 写入的字符串一致,大小写敏感。
5.5 模型不调用 Loader 直接编答案
有时候模型看到用户问题后,不调用 Loader 而是直接凭训练知识回答。这通常是系统提示词没写清楚。把SYSTEM_PROMPT里的规则强调一下:当前没有对应功能工具时,必须先调用 Skill Loader。如果还是不行,可以在提示词里明确列出 Loader 工具的名字,降低模型的选择难度。
6. 继续复现:从拿 Key 到跑通动态工具加载
到这里,整条链路已经跑通了:从 https://taotoken.net/?utm_source=taotoken_aicg_blog_end 创建 Key,把 LangChain 模型客户端的 Base URL 填成https://taotoken.net/api,再配上SkillState、Loader 工具和SkillMiddleware,最后用销售数据测试验证了动态过滤生效。TaoToken 负责模型通道,Middleware 负责工具过滤,两者各司其职。
如果你后面要接更多技能,只需要在SKILL_TOOL_MAPPING里加一条映射,再写一个对应的 Loader 工具,Middleware 的代码不用改。工具数量继续涨,每次模型调用看到的仍然只是当前技能相关的子集。
想直接调模型对话验证通道是否通,可以走模型对话入口;长期跑编码类 Agent、需要稳定额度的话,看 Coding Plan;接入过程中遇到鉴权或参数问题,去 API Keys 页面和接入文档对照检查。Key 拿到手之后,剩下的就是把这篇的代码跑一遍,看日志里那两行工具数量从 2 变 4。