简介
文章介绍如何使用LangChain 1.1的Middleware机制实现Claude Skills动态工具过滤，让AI Agent在运行时智能选择所需工具，避免上下文窗口耗尽和"选择困难"。文章详细讲解环境配置、状态管理、中间件实现到Agent创建的完整过程，帮助开发者减少Token消耗，提升模型推理效率与准确性，构建高效可扩展的AI Agent。
LangChain 1.1 实现 Claude Skills 动态工具过滤：从原理到实践
在搭建复杂的 Agent 时，我们往往需要接入几十甚至上百个工具。如果把所有工具一次性丢给大模型（LLM），不仅会迅速耗尽上下文窗口（Context Window），还会加剧“选择困难”，让模型在冗长的工具描述中迷失方向，表现出明显的“降智”现象。
本文将带你从零开始，深入理解并通过 LangChain 1.1 实现 Claude Skills 的核心功能——动态工具过滤，教你如何让 AI Agent 在运行时智能地选择真正需要的工具，从而在面对大量工具时，既避免无效信息占满上下文窗口，又缓解“选择困难”，显著降低 Token 消耗，并提升模型在复杂任务中的 推理效率与准确性。
完整代码免费赠送👇
- Claude Code Skills 背景介绍
Vibe Coding (氛围编程) 的兴起标志着从传统的代码补全（如 GitHub Copilot）向 代理式编程（Agentic Coding） 的范式转变。开发者不再受困于语法细节，而是转向更高层次的意图表达。
Skill：特定领域的“操作手册”
在 Claude Code 的架构中，Skill（技能） 扮演着至关重要的角色。如果说 Claude 模型是大脑，MCP (Model Context Protocol) 是连接外部世界的手脚，那么 Skill 就是存储特定领域专业知识的操作手册。
虽然 Claude Opus/Sonnet 拥有广泛的编程知识，但它并不了解某家特定初创公司的内部部署脚本、某种冷门框架的特殊配置，或者某个团队特定的代码审查规范。传统的解决方案是将这些信息全部塞入系统提示词（System Prompt）或上下文窗口中，但这会导致两个严重问题：
- 上下文窗口迅速耗尽：增加了推理成本（Token Economics）。
- 注意力分散：过多的无关信息会干扰模型的注意力，导致“迷失中间”（Lost in the Middle）现象。
Skill 通过引入 动态加载（Dynamic Loading）和渐进式披露（Progressive Disclosure） 机制，优雅地解决了这一难题。它允许开发者将海量的程序性知识（Procedural Knowledge）封装在本地文件系统中，Agent 仅在识别到用户意图与某个 Skill 匹配时，才会按需加载相关的指令和脚本。
- Claude Skills 给 Agent 开发带来的启发
2.1 传统大模型的工具调用困境
传统的 AI Agent 在处理任务时，通常会将 所有可用的工具（Tools）一次性暴露给大语言模型。想象一下，如果你有 50 个工具，每次模型调用都需要处理这 50 个工具的描述信息：
传统架构：┌─────────────────────────────────────────────────────────────┐│ Agent 启动时加载所有 50 个工具 ││ ││ 用户: "你好" ││ 模型收到: 50 个工具描述 + 用户消息 ← 浪费大量 Token！ ││ ││ 用户: "计算 1+1" ││ 模型收到: 50 个工具描述 + 用户消息 ← 还是 50 个工具！ │└─────────────────────────────────────────────────────────────┘这会带来几个严重问题：
- Token 消耗巨大：每个工具的描述可能有几百个 token，50 个工具就是上万个 token。
- 大模型困惑：面对过多选择，模型容易选错工具或产生幻觉。
- 响应延迟：处理大量工具描述需要更长时间。
- 成本高昂：API 调用按 token 计费，浪费严重。
完整的大模型学习和面试资料已经上传带到CSDN的官方了，有需要的朋友可以扫描下方二维码免费领取【保证100%免费】👇👇
2.2 Claude Skills 的核心思想
Claude Skills 的核心思想是：让模型在每次调用时只看到「相关的」工具，而不是全部工具。这就像一个智能助手，只有当你说"我要分析数据"时，才会把数据分析相关的工具拿出来；说"我要处理 PDF"时，才会展示 PDF 处理工具。
Claude Skills 架构：┌─────────────────────────────────────────────────────────────┐│ Agent 启动时只加载 2 个 "Loader" 工具 ││ ││ 用户: "你好" ││ 模型收到: 2 个工具描述 + 用户消息 ← 只有 2 个工具！ ││ ││ 用户: "分析这组数据 [1,2,3]" ││ 第一次调用: 2 个工具 → AI 调用 skill_data_analysis ││ 技能加载后: skills_loaded = ['data_analysis'] ││ 第二次调用: 4 个工具 (2 Loaders + 2 数据分析) ││ → AI 使用 calculate_statistics 完成任务 │└─────────────────────────────────────────────────────────────┘接下来，我们就通过底层技术来复现这个高价值的 Agent 开发模式。
2.3 为什么选择 LangChain 1.0+
LangChain 1.1 版本最大的优势就是在 LangGraph 之上构建并集成了革命性的 Middleware API。
LangChain middleware文档：https://docs.langchain.com/oss/python/langchain/middleware/overview
这个 Middleware API 允许我们在 Agent 的执行流程中插入自定义逻辑，实现：
- 动态工具过滤：在每次模型调用前修改工具列表。
- 状态管理：通过 state_schema 追踪运行时状态。
- 请求拦截：使用 request.override() 修改请求参数。
在 LangChain 1.1 之前，实现动态工具过滤需要复杂的 hack（比如重写 Agent 类）。现在，通过官方支持的 Middleware API，我们可以优雅地实现这一功能。
| 特性 | LangChain 0.x | LangChain 1.0+ |
| 工具过滤 | 需要 hack | 官方 Middleware 支持 |
| 状态管理 | 手动实现 | state_schema 内置 |
| 请求修改 | 不支持 | request.override() |
下图展示了使用 LangChain 1.0 复现 Claude Skills 的核心工作流程：
从图中可以看到，Middleware 是整个系统的核心。它在模型调用之前拦截请求，根据当前状态（skills_loaded）动态过滤工具列表，然后将过滤后的请求传递给模型。
- 从零复现 Claude Skills 动态工具过滤
开箱即用的 Claude Skills 动态工具过滤功能👇，扫码加入 赋范空间 免费领取
3.1 环境配置与依赖安装
首先，我们需要安装 LangChain 1.0 的核心库。确保你的环境中安装了以下包：
pip install -U langchain langgraph langchain-openai pdfplumber pandas numpy matplotlib python-dotenv接下来，导入我们需要的所有核心组件。特别注意 langchain.agents.middleware 模块，这是 LangChain 1.0 新增的关键模块。
# 基础库导入import osimport sysfrom pathlib import Pathfrom typing import List, Callable, Any, Optionalfrom typing_extensions import TypedDict# 加载环境变量from dotenv import load_dotenvload_dotenv(override=True)# LangChain 1.0 核心导入from langchain.agents import create_agentfrom langchain.agents.middleware import ( AgentMiddleware, ModelRequest, ModelResponse,)from langchain_core.tools import BaseTool, toolfrom langchain_core.messages import HumanMessage, AIMessage, BaseMessageprint("核心库导入成功")3.2 配置 DeepSeek-v3.2 模型
在本教程中，我们使用 DeepSeek 的 deepseek-reasoner 模型作为底层 LLM。这个模型的特点是支持「推理过程」输出，可以让我们看到模型的思考过程。
需要说明的是，由于 DeepSeek-v3.2 模型较新，其推理模式在 LangChain 旧版本中可能存在兼容性问题。如果遇到工具调用报错，建议使用自定义的模型适配器。
报错信息👇
我们编写了一个自定义的 DeepSeekReasonerChatModel 适配器👇，来解决这个问题。（扫码免费领取）
# 添加项目路径（用于导入自定义模型）PROJECT_ROOT = Path.cwd()sys.path.insert(0, str(PROJECT_ROOT))# 导入自定义的 DeepSeek 模型适配器# 假设你已经有了适配器文件 skill_system/models.pyfrom skill_system.models import DeepSeekReasonerChatModel# 检查 API Keyapi_key = os.getenv("DEEPSEEK_API_KEY")ifnot api_key: print("未设置 DEEPSEEK_API_KEY") print("请在 .env 文件中添加: DEEPSEEK_API_KEY=your-key")else: print(f"API Key 已配置 (前8位: {api_key[:8]}...)")# 创建模型实例model = DeepSeekReasonerChatModel( api_key=api_key, model_name="deepseek-reasoner", temperature=0.7)print(f"DeepSeek 模型已创建")3.3 理解 LangChain 1.1 Middleware
LangChain 1.0 引入的 Agent Middleware 遵循拦截器模式：
请求 ──▶ [中间件1] ──▶ [中间件2] ──▶ ... ──▶ [核心处理] ──▶ 响应 │ │ ▼ ▼ 可以修改请求 可以修改请求在 Agent 的上下文中，Middleware 可以：
- 拦截模型调用请求：在模型被调用之前获取请求信息。
- 修改请求参数：比如修改工具列表、系统提示等。
- 传递给下一个处理器：调用 handler(request) 继续执行。
ModelRequest 是 Middleware 中最重要的对象，它封装了模型调用的所有信息，并提供了 override 方法：
class ModelRequest: messages: List[BaseMessage] # 消息历史 tools: List[BaseTool] # 可用工具列表 state: Dict[str, Any] # 当前状态（关键！） def override(self, **kwargs) -> ModelRequest: """创建一个修改了指定参数的新 Request""" # 返回新的 ModelRequest，保持不可变性request.override() 是实现动态过滤的核心方法。它允许我们创建一个新的请求对象，其中某些参数被修改了，而其他参数保持不变。
3.4 实现一个简单的日志 Middleware
在实现复杂的工具过滤之前，我们先写一个简单的日志 Middleware，帮助理解整个流程：
class LoggingMiddleware(AgentMiddleware): """ 日志中间件 - 记录每次模型调用的信息 """ def __init__(self, name: str = "Logger"): super().__init__() self.name = name self.call_count = 0 def wrap_model_call( self, request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse] ) -> ModelResponse: """ 拦截模型调用，打印日志信息 """ self.call_count += 1 # 1. 调用前：记录请求信息 print(f"\n{'='*60}") print(f"[{self.name}] 第 {self.call_count} 次模型调用") print(f"{'='*60}") # 打印工具信息 if hasattr(request, 'tools') and request.tools: tool_names = [t.name for t in request.tools] print(f"可用工具 ({len(tool_names)}个): {tool_names}") # 打印状态信息 if hasattr(request, 'state') and request.state: print(f"当前状态: {request.state}") # 2. 调用下一个处理器（这里是实际的模型调用） response = handler(request) # 3. 调用后：可以处理响应 print(f"模型调用完成") print(f"{'='*60}\n") return response3.5 定义状态 Schema
在 LangChain 1.1 中，State Schema 用于定义 Agent 运行时需要追踪的状态信息。对于 Claude Skills，我们需要追踪一个关键状态：skills_loaded（当前已加载的技能列表）。
from langgraph.graph import MessagesStatefrom typing import Annotated, List# 定义 reducer 函数：累积模式def skill_list_accumulator(current: List[str], new: List[str]) -> List[str]: """ 累积模式：合并已加载的 Skills 保持所有已加载的技能，而不是替换 """ ifnot current: return new # 合并并去重，保持顺序 combined = current + [s for s in new if s notin current] return combined# 使用 MessagesState 作为基类class SkillState(MessagesState): """ Skill 状态 Schema """ skills_loaded: Annotated[List[str], skill_list_accumulator] = []3.6 定义外部工具
我们会创建三类工具来演示：
- Loader 工具（始终可见）：用于加载技能。
- 数据分析工具：只有加载了 data_analysis 技能后才可见。
- 文本处理工具：只有加载了 text_processing 技能后才可见。
from langgraph.types import Commandfrom langchain_core.messages import ToolMessage# ==================== Loader 工具 ====================@tooldef skill_data_analysis(runtime) -> Command: """加载数据分析技能。""" instructions = """数据分析技能已成功加载！ 现在你可以使用以下工具： • calculate_statistics(numbers): 计算一组数字的统计信息 • generate_chart(data, chart_type): 生成数据图表 """ return Command( update={ "messages": [ToolMessage( content=instructions, tool_call_id=runtime.tool_call_id )], "skills_loaded": ["data_analysis"] # 关键：直接更新状态 } )@tooldef skill_text_processing(runtime) -> Command: """加载文本处理技能。""" instructions = """文本处理技能已成功加载！ 现在你可以使用以下工具： • summarize_text(text, max_length): 生成文本摘要 • extract_keywords(text, num_keywords): 提取关键词 """ return Command( update={ "messages": [ToolMessage( content=instructions, tool_call_id=runtime.tool_call_id )], "skills_loaded": ["text_processing"] # 关键：直接更新状态 } )# ==================== 数据分析工具 ====================@tooldef calculate_statistics(numbers: List[float]) -> str: """计算一组数字的统计信息。""" import statistics ifnot numbers: return"错误: 数字列表为空" returnf"统计结果: mean={statistics.mean(numbers)}, max={max(numbers)}"@tooldef generate_chart(data: List[float], chart_type: str = "bar") -> str: """根据数据生成图表（模拟）。""" returnf"已生成 {chart_type} 图表，包含 {len(data)} 个数据点"# ==================== 文本处理工具 ====================@tooldef summarize_text(text: str, max_length: int = 100) -> str: """生成文本摘要。""" returnf"摘要: {text[:max_length]}..."@tooldef extract_keywords(text: str, num_keywords: int = 5) -> str: """从文本中提取关键词。""" words = text.split()[:num_keywords] returnf"关键词: {', '.join(words)}"# 组织工具LOADER_TOOLS = [skill_data_analysis, skill_text_processing]DATA_ANALYSIS_TOOLS = [calculate_statistics, generate_chart]TEXT_PROCESSING_TOOLS = [summarize_text, extract_keywords]ALL_TOOLS = LOADER_TOOLS + DATA_ANALYSIS_TOOLS + TEXT_PROCESSING_TOOLS3.7 定义工具映射与过滤逻辑
为了实现动态过滤，我们需要定义一个映射关系：哪些工具属于哪个技能。
# 技能到工具的映射SKILL_TOOL_MAPPING = { "data_analysis": DATA_ANALYSIS_TOOLS, "text_processing": TEXT_PROCESSING_TOOLS,}def get_tools_for_skills(skills_loaded: List[str]) -> List[BaseTool]: """ 根据已加载的技能列表，返回应该暴露给模型的工具 """ # 始终包含 Loader 工具 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 tools3.8 实现 SkillMiddleware（核心）
现在我们来实现整个系统的核心组件：SkillMiddleware。
class SkillMiddleware(AgentMiddleware): """ Skill 中间件 - 实现动态工具过滤 工作原理： 1. 在每次模型调用前拦截请求 2. 从 request.state 中读取 skills_loaded 列表 3. 根据 skills_loaded 过滤工具列表 4. 使用 request.override() 替换工具列表 5. 传递给下一个 handler """ 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]: """从请求状态中提取 skills_loaded""" skills_loaded = [] if hasattr(request, 'state') and request.state isnotNone: if isinstance(request.state, dict): skills_loaded = request.state.get("skills_loaded", []) else: skills_loaded = getattr(request.state, "skills_loaded", []) return skills_loaded def wrap_model_call( self, request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse] ) -> ModelResponse: """【核心方法】拦截模型调用，动态过滤工具""" self.call_count += 1 # Step 1: 从状态中获取已加载的 Skills skills_loaded = self._get_skills_from_state(request) # Step 2: 获取过滤后的工具 filtered_tools = get_tools_for_skills(skills_loaded) # Step 3: 打印日志 if self.verbose: print(f"\n{'─'*60}") print(f"[SkillMiddleware] 第 {self.call_count} 次模型调用") print(f"skills_loaded: {skills_loaded}") print(f"过滤后工具: {[t.name for t in filtered_tools]}") # Step 4: 【关键】使用 request.override() 替换工具列表 filtered_request = request.override(tools=filtered_tools) # Step 5: 调用下一个 handler return handler(filtered_request) asyncdef awrap_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) filtered_request = request.override(tools=filtered_tools) returnawait handler(filtered_request)3.9 创建 Agent
最后，使用 create_agent 将所有组件组装起来。
# 创建 SkillMiddleware 实例skill_middleware = SkillMiddleware(verbose=True)# 定义系统提示SYSTEM_PROMPT = """你是一个智能助手，可以使用各种技能来帮助用户完成任务。1. 你有两类工具：Skill Loader（技能加载器）和功能工具。2. 当用户请求某个功能时，如果当前没有对应的功能工具，请先调用相应的 Skill Loader 加载技能。3. 加载后，使用新获得的工具完成任务。"""# 创建 Agentagent = create_agent( model=model, tools=ALL_TOOLS, # 注册所有工具（但 Middleware 会动态过滤） middleware=(skill_middleware,), # 关键：添加 SkillMiddleware state_schema=SkillState, # 使用我们定义的状态 Schema system_prompt=SYSTEM_PROMPT,)print("Agent 创建成功！")- LangChain Skills Agent 运行测试
测试场景：动态加载数据分析技能
让我们测试核心功能：当用户请求数据分析时，观察工具的动态加载过程。
# 构造输入test_input = { "messages": [HumanMessage(content="我有一组销售数据 [150, 200, 180, 220, 190]，请帮我计算统计信息")], "skills_loaded": [] # 初始状态}# 调用 Agentresult = agent.invoke(test_input)运行结果分析：
────────────────────────────────────────────────────────────[SkillMiddleware] 第 1 次模型调用skills_loaded: []过滤后工具 (2个): ['skill_data_analysis', 'skill_text_processing']────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────[SkillMiddleware] 第 2 次模型调用skills_loaded: ['data_analysis']过滤后工具 (4个): ['skill_data_analysis', 'skill_text_processing', 'calculate_statistics', 'generate_chart']────────────────────────────────────────────────────────────AI 响应：
过程解析：
- 第一次模型调用：skills_loaded 为空，模型只能看到 2 个 Loader 工具。
- AI 决策：发现需要数据分析功能，调用 skill_data_analysis。
- 技能加载：skills_loaded 更新为 [‘data_analysis’]。
- 第二次模型调用：模型看到了 4 个工具（新增了数据分析工具）。
- 任务完成：使用 calculate_statistics 计算统计信息。
这就是 Claude Skills 的核心价值：模型在每次调用时只看到相关的工具，大大减少了 token 消耗和错误率。
- 总结与最佳实践
本文介绍了如何使用 LangChain 1.1 的 Middleware 机制实现 Claude Skills 动态工具过滤。
核心知识点：
- Middleware 机制：
- 作用：在 Agent 执行流程中插入自定义逻辑。
- 核心方法：wrap_model_call(request, handler)。
- 关键操作：request.override(tools=filtered_tools)。
- State Schema：
- 作用：定义 Agent 运行时需要追踪的状态。
- 实现：使用 MessagesState 或 TypedDict。
- 动态工具过滤：
- 原理：根据当前状态（skills_loaded）决定暴露哪些工具。
- 优势：减少 token 消耗、降低错误率、提升响应速度。
通过这种架构，我们可以构建出能力无限扩展、但推理依然高效的 AI Agent。
如果你对 AI Agent、RAG、MCP、大模型微调、企业项目实战 等前沿技术感兴趣，欢迎关注我们！
我们提供系统的课程体系，帮助你从零开始掌握：
- AI Agent 开发：深入理解 Agent 架构与实战，打造智能体。
- RAG 技术：构建高性能的企业级知识库问答系统。
- MCP 协议：掌握下一代 AI 连接标准，连接万物。
- 大模型微调：掌握 Fine-tuning 技术，打造专属垂直领域模型。
- 企业项目实战：15+ 项目实战（多模态RAG、实时语音助手、文档审核、智能客服系统等），将理论知识应用到实际项目中，解决真实业务问题。
最后
我在一线科技企业深耕十二载，见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事，早已在效率与薪资上形成代际优势，我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。
我整理出这套 AI 大模型突围资料包：
- ✅AI大模型学习路线图
- ✅Agent行业报告
- ✅100集大模型视频教程
- ✅大模型书籍PDF
- ✅DeepSeek教程
- ✅AI产品经理入门资料
完整的大模型学习和面试资料已经上传带到CSDN的官方了，有需要的朋友可以扫描下方二维码免费领取【保证100%免费】👇👇
为什么说现在普通人就业/升职加薪的首选是AI大模型？
人工智能技术的爆发式增长，正以不可逆转之势重塑就业市场版图。从DeepSeek等国产大模型引发的科技圈热议，到全国两会关于AI产业发展的政策聚焦，再到招聘会上排起的长队，AI的热度已从技术领域渗透到就业市场的每一个角落。
智联招聘的最新数据给出了最直观的印证：2025年2月，AI领域求职人数同比增幅突破200%，远超其他行业平均水平；整个人工智能行业的求职增速达到33.4%，位居各行业榜首，其中人工智能工程师岗位的求职热度更是飙升69.6%。
AI产业的快速扩张，也让人才供需矛盾愈发突出。麦肯锡报告明确预测，到2030年中国AI专业人才需求将达600万人，人才缺口可能高达400万人，这一缺口不仅存在于核心技术领域，更蔓延至产业应用的各个环节。
资料包有什么？
①从入门到精通的全套视频教程⑤⑥
包含提示词工程、RAG、Agent等技术点
② AI大模型学习路线图（还有视频解说）
全过程AI大模型学习路线
③学习电子书籍和技术文档
市面上的大模型书籍确实太多了，这些是我精选出来的
④各大厂大模型面试题目详解
⑤ 这些资料真的有用吗?
这份资料由我和鲁为民博士共同整理，鲁为民博士先后获得了北京清华大学学士和美国加州理工学院博士学位，在包括IEEE Transactions等学术期刊和诸多国际会议上发表了超过50篇学术论文、取得了多项美国和中国发明专利，同时还斩获了吴文俊人工智能科学技术奖。目前我正在和鲁博士共同进行人工智能的研究。
所有的视频教程由智泊AI老师录制，且资料与智泊AI共享，相互补充。这份学习大礼包应该算是现在最全面的大模型学习资料了。
资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。
智泊AI始终秉持着“让每个人平等享受到优质教育资源”的育人理念，通过动态追踪大模型开发、数据标注伦理等前沿技术趋势，构建起"前沿课程+智能实训+精准就业"的高效培养体系。
课堂上不光教理论，还带着学员做了十多个真实项目。学员要亲自上手搞数据清洗、模型调优这些硬核操作，把课本知识变成真本事！
如果说你是以下人群中的其中一类，都可以来智泊AI学习人工智能，找到高薪工作，一次小小的“投资”换来的是终身受益！
应届毕业生：无工作经验但想要系统学习AI大模型技术，期待通过实战项目掌握核心技术。
零基础转型：非技术背景但关注AI应用场景，计划通过低代码工具实现“AI+行业”跨界。
业务赋能 突破瓶颈：传统开发者（Java/前端等）学习Transformer架构与LangChain框架，向AI全栈工程师转型。
👉获取方式：
😝有需要的小伙伴，可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓**
