LangChain 1.1 实现 Claude Skills 动态工具过滤：从原理到实践

在搭建复杂的 Agent 时，我们往往需要接入几十甚至上百个工具。如果把所有工具一次性丢给大模型（LLM），不仅会迅速耗尽上下文窗口（Context Window），还会加剧“选择困难”，让模型在冗长的工具描述中迷失方向，表现出明显的“降智”现象。

本文将带你从零开始，深入理解并通过 LangChain 1.1 实现 Claude Skills 的核心功能——动态工具过滤，教你如何让 AI Agent 在运行时智能地选择真正需要的工具，从而在面对大量工具时，既避免无效信息占满上下文窗口，又缓解“选择困难”，显著降低 Token 消耗，并提升模型在复杂任务中的 推理效率与准确性。

完整代码免费赠送👇

---

1. Claude Code Skills 背景介绍

Vibe Coding (氛围编程) 的兴起标志着从传统的代码补全（如 GitHub Copilot）向 代理式编程（Agentic Coding） 的范式转变。开发者不再受困于语法细节，而是转向更高层次的意图表达。

Skill：特定领域的“操作手册”

在 Claude Code 的架构中，Skill（技能） 扮演着至关重要的角色。如果说 Claude 模型是大脑，MCP (Model Context Protocol) 是连接外部世界的手脚，那么 Skill 就是存储特定领域专业知识的操作手册。

虽然 Claude Opus/Sonnet 拥有广泛的编程知识，但它并不了解某家特定初创公司的内部部署脚本、某种冷门框架的特殊配置，或者某个团队特定的代码审查规范。传统的解决方案是将这些信息全部塞入系统提示词（System Prompt）或上下文窗口中，但这会导致两个严重问题：

1. 上下文窗口迅速耗尽：增加了推理成本（Token Economics）。
2. 注意力分散：过多的无关信息会干扰模型的注意力，导致“迷失中间”（Lost in the Middle）现象。

Skill 通过引入 动态加载（Dynamic Loading）和渐进式披露（Progressive Disclosure） 机制，优雅地解决了这一难题。它允许开发者将海量的程序性知识（Procedural Knowledge）封装在本地文件系统中，Agent 仅在识别到用户意图与某个 Skill 匹配时，才会按需加载相关的指令和脚本。

官方介绍：What are Skills?

---

2. Claude Skills 给 Agent 开发带来的启发

2.1 传统大模型的工具调用困境

传统的 AI Agent 在处理任务时，通常会将 所有可用的工具（Tools）一次性暴露给大语言模型。想象一下，如果你有 50 个工具，每次模型调用都需要处理这 50 个工具的描述信息：

传统架构：
┌─────────────────────────────────────────────────────────────┐
│  Agent 启动时加载所有 50 个工具                              │
│                                                             │
│  用户: "你好"                                                │
│  模型收到: 50 个工具描述 + 用户消息    ← 浪费大量 Token！    │
│                                                             │
│  用户: "计算 1+1"                                           │
│  模型收到: 50 个工具描述 + 用户消息    ← 还是 50 个工具！    │
└─────────────────────────────────────────────────────────────┘

这会带来几个严重问题：

1. Token 消耗巨大：每个工具的描述可能有几百个 token，50 个工具就是上万个 token。
2. 大模型困惑：面对过多选择，模型容易选错工具或产生幻觉。
3. 响应延迟：处理大量工具描述需要更长时间。
4. 成本高昂：API 调用按 token 计费，浪费严重。

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）动态过滤工具列表，然后将过滤后的请求传递给模型。

---

3. 从零复现 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 os
import sys
from pathlib import Path
from typing import List, Callable, Any, Optional
from typing_extensions import TypedDict

# 加载环境变量
from dotenv import load_dotenv
load_dotenv(override=True)

# LangChain 1.0 核心导入
from langchain.agents import create_agent
from langchain.agents.middleware import (
    AgentMiddleware,
    ModelRequest,
    ModelResponse,
)
from langchain_core.tools import BaseTool, tool
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage

print("核心库导入成功")

3.2 配置 DeepSeek-v3.2 模型

在本教程中，我们使用 DeepSeek 的 deepseek-reasoner 模型作为底层 LLM。这个模型的特点是支持「推理过程」输出，可以让我们看到模型的思考过程。

提示：你也可以替换为 OpenAI、Anthropic 或其他兼容 LangChain 的模型。核心的 Middleware 机制是通用的。

需要说明的是，由于 DeepSeek-v3.2 模型较新，其推理模式在 LangChain 旧版本中可能存在兼容性问题。如果遇到工具调用报错，建议使用自定义的模型适配器。

报错信息👇

我们编写了一个自定义的 DeepSeekReasonerChatModel 适配器👇，来解决这个问题。（免费领取）

# 添加项目路径（用于导入自定义模型）
PROJECT_ROOT = Path.cwd()
sys.path.insert(0, str(PROJECT_ROOT))

# 导入自定义的 DeepSeek 模型适配器
# 假设你已经有了适配器文件 skill_system/models.py
from skill_system.models import DeepSeekReasonerChatModel

# 检查 API Key
api_key = os.getenv("DEEPSEEK_API_KEY")
if not 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 可以：

1. 拦截模型调用请求：在模型被调用之前获取请求信息。
2. 修改请求参数：比如修改工具列表、系统提示等。
3. 传递给下一个处理器：调用 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 response

3.5 定义状态 Schema

在 LangChain 1.1 中，State Schema 用于定义 Agent 运行时需要追踪的状态信息。对于 Claude Skills，我们需要追踪一个关键状态：skills_loaded（当前已加载的技能列表）。

from langgraph.graph import MessagesState
from typing import Annotated, List

# 定义 reducer 函数：累积模式
def skill_list_accumulator(current: List[str], new: List[str]) -> List[str]:
    """
    累积模式：合并已加载的 Skills
    保持所有已加载的技能，而不是替换
    """
    if not current:
        return new
    # 合并并去重，保持顺序
    combined = current + [s for s in new if s not in current]
    return combined

# 使用 MessagesState 作为基类
class SkillState(MessagesState):
    """
    Skill 状态 Schema
    """
    skills_loaded: Annotated[List[str], skill_list_accumulator] = []

3.6 定义外部工具

我们会创建三类工具来演示：

1. Loader 工具（始终可见）：用于加载技能。
2. 数据分析工具：只有加载了 data_analysis 技能后才可见。
3. 文本处理工具：只有加载了 text_processing 技能后才可见。

from langgraph.types import Command
from langchain_core.messages import ToolMessage

# ==================== Loader 工具 ====================
@tool
def 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"]  # 关键：直接更新状态
        }
    )

@tool
def 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"]  # 关键：直接更新状态
        }
    )

# ==================== 数据分析工具 ====================
@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)} 个数据点"

# ==================== 文本处理工具 ====================
@tool
def summarize_text(text: str, max_length: int = 100) -> str:
    """生成文本摘要。"""
    return f"摘要: {text[:max_length]}..."

@tool
def extract_keywords(text: str, num_keywords: int = 5) -> str:
    """从文本中提取关键词。"""
    words = text.split()[:num_keywords]
    return f"关键词: {', '.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_TOOLS

3.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 tools

3.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 is not None:
            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)
    
    async def 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)
        return await handler(filtered_request)

3.9 创建 Agent

最后，使用 create_agent 将所有组件组装起来。

# 创建 SkillMiddleware 实例
skill_middleware = SkillMiddleware(verbose=True)

# 定义系统提示
SYSTEM_PROMPT = """
你是一个智能助手，可以使用各种技能来帮助用户完成任务。
1. 你有两类工具：Skill Loader（技能加载器）和功能工具。
2. 当用户请求某个功能时，如果当前没有对应的功能工具，请先调用相应的 Skill Loader 加载技能。
3. 加载后，使用新获得的工具完成任务。
"""

# 创建 Agent
agent = create_agent(
    model=model,
    tools=ALL_TOOLS,  # 注册所有工具（但 Middleware 会动态过滤）
    middleware=(skill_middleware,),  # 关键：添加 SkillMiddleware
    state_schema=SkillState,  # 使用我们定义的状态 Schema
    system_prompt=SYSTEM_PROMPT,
)

print("Agent 创建成功！")

---

4. LangChain Skills Agent 运行测试

测试场景：动态加载数据分析技能

让我们测试核心功能：当用户请求数据分析时，观察工具的动态加载过程。

# 构造输入
test_input = {
    "messages": [HumanMessage(content="我有一组销售数据 [150, 200, 180, 220, 190]，请帮我计算统计信息")],
    "skills_loaded": []  # 初始状态
}

# 调用 Agent
result = 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 响应：

根据您提供的销售数据 [150, 200, 180, 220, 190]，统计结果如下：平均值 188.0，最大值 220…

过程解析：

1. 第一次模型调用：skills_loaded 为空，模型只能看到 2 个 Loader 工具。
2. AI 决策：发现需要数据分析功能，调用 skill_data_analysis。
3. 技能加载：skills_loaded 更新为 ['data_analysis']。
4. 第二次模型调用：模型看到了 4 个工具（新增了数据分析工具）。
5. 任务完成：使用 calculate_statistics 计算统计信息。

这就是 Claude Skills 的核心价值：模型在每次调用时只看到相关的工具，大大减少了 token 消耗和错误率。

---

5. 总结与最佳实践

本文介绍了如何使用 LangChain 1.1 的 Middleware 机制实现 Claude Skills 动态工具过滤。

核心知识点：

1. Middleware 机制：

   • 作用：在 Agent 执行流程中插入自定义逻辑。
   • 核心方法：wrap_model_call(request, handler)。
   • 关键操作：request.override(tools=filtered_tools)。

2. State Schema：

   • 作用：定义 Agent 运行时需要追踪的状态。
   • 实现：使用 MessagesState 或 TypedDict。

3. 动态工具过滤：

   • 原理：根据当前状态（skills_loaded）决定暴露哪些工具。
   • 优势：减少 token 消耗、降低错误率、提升响应速度。

通过这种架构，我们可以构建出能力无限扩展、但推理依然高效的 AI Agent。

如果你对 AI Agent、RAG、MCP、大模型微调、企业项目实战 等前沿技术感兴趣，欢迎关注我们！

我们提供系统的课程体系，帮助你从零开始掌握：

• AI Agent 开发：深入理解 Agent 架构与实战，打造智能体。
• RAG 技术：构建高性能的企业级知识库问答系统。
• MCP 协议：掌握下一代 AI 连接标准，连接万物。
• 大模型微调：掌握 Fine-tuning 技术，打造专属垂直领域模型。
• 企业项目实战：15+ 项目实战（多模态RAG、实时语音助手、文档审核、智能客服系统等），将理论知识应用到实际项目中，解决真实业务问题。

立即加入👉 赋范空间，开启你的 AI 进阶之旅！