---

关于作者

• 深耕领域：大语言模型开发 / RAG 知识库 / AI Agent 落地 / 模型微调
• 技术栈：Python | RAG (LangChain / Dify + Milvus) | FastAPI + Docker
• 工程能力：专注模型工程化部署、知识库构建与优化，擅长全流程解决方案

「让 AI 交互更智能，让技术落地更高效」
欢迎技术探讨与项目合作，解锁大模型与智能交互的无限可能！

---

LangChain 1.0 中间件Hook机制与实战应用

摘要：本文深入解析LangChain 1.0中间件的6个Hook点（before_agent、before_model、wrap_model_call、wrap_tool_call、after_model、after_agent），并通过SummarizationMiddleware等实战案例，帮助开发者掌握中间件的具体应用方法。

一、中间件Hook机制概览

1.1 什么是Hook

Hook（钩子）是中间件系统中用于在特定时机插入自定义逻辑的接口。LangChain 1.0提供了6个核心Hook点，覆盖Agent执行的完整生命周期。

1.2 6个核心Hook点

Hook点	执行时机	核心作用	典型应用场景
before_agent	Agent执行开始前	全局初始化、环境检查	资源初始化、配置加载
before_model	模型调用前	输入预处理、数据验证	上下文压缩、参数校验
wrap_model_call	模型调用包装	拦截和控制模型调用	缓存、重试、熔断
wrap_tool_call	工具调用包装	拦截和控制工具执行	权限检查、工具审计
after_model	模型调用后	输出处理、结果验证	格式转换、质量检查
after_agent	Agent执行结束后	资源清理、状态记录	日志记录、资源释放

1.3 Hook执行流程

二、Hook详解与实战

2.1 before_agent - Agent启动前

执行时机： Agent执行的最开始

核心作用： 提供全局初始化的机会，通常用于设置全局状态、检查环境配置、初始化资源等。

应用场景：

• 全局配置加载
• 环境变量检查
• 资源初始化（数据库连接、缓存连接等）
• 用户身份验证

代码示例：

from langchain.agents.middleware import Middleware

class InitMiddleware(Middleware):
    def before_agent(self, context):
        # 加载全局配置
        context.config = load_config()
        # 检查必要的环境变量
        if not os.getenv("API_KEY"):
            raise ValueError("API_KEY not set")
        return context

2.2 before_model - 模型调用前

执行时机： 模型调用前，输入预处理阶段

核心作用： 对输入数据进行预处理、验证、清洗等操作。这是确保数据质量的第一道防线。

应用场景：

• 上下文压缩（SummarizationMiddleware）
• 输入参数校验
• 敏感信息脱敏
• Prompt动态注入

实战案例：SummarizationMiddleware上下文压缩

SummarizationMiddleware是LangChain 1.0官方提供的before_model中间件，用于自动压缩历史会话，减少token使用，提高响应速度。

核心特性：

1. 官方中间件集成：使用 from langchain.agents.middleware import SummarizationMiddleware
2. 自动压缩：在 create_agent中通过 middleware参数集成
3. 智能保留：自动压缩历史消息，保留最近的对话
4. 无需手动管理：中间件自动处理压缩逻辑

工作原理：

当历史消息的token数量超过阈值（500）且消息数量超过保留数量（5条）时，中间件会自动：

1. 将旧消息发送给摘要模型进行压缩
2. 保留最近的N条消息
3. 将摘要结果作为上下文传递给Agent

代码示例：

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langchain_deepseek import ChatDeepSeek

# 创建摘要模型
summary_model = ChatDeepSeek(model="deepseek-chat")

# 创建SummarizationMiddleware
summarization_middleware = SummarizationMiddleware(
    model=summary_model,
    max_tokens=500,      # 触发摘要的token阈值
    keep_messages=5      # 保留最近的消息数量
)

# 创建主模型
main_model = ChatDeepSeek(model="deepseek-chat")

# 创建Agent时集成中间件
agent = create_agent(
    model=main_model,
    tools=tools,
    system_prompt=prompt,
    middleware=[summarization_middleware]  # 集成中间件
)

预期效果：

• 压缩前：20条消息，约1000+ tokens
• 压缩后：5-6条消息（保留最近5条 + 摘要），约300-500 tokens

2.3 wrap_model_call - 模型调用包装

执行时机： 包装实际的模型调用过程

核心作用： 决定如何与底层模型交互，是实现高级功能（如缓存、重试、熔断等）的关键位置。

应用场景：

• 结果缓存
• 模型降级（主模型失败时切换到备用模型）
• 重试机制
• 限流控制

代码示例：

class CacheMiddleware(Middleware):
    def __init__(self):
        self.cache = {}
  
    def wrap_model_call(self, model_call, context):
        # 生成缓存key
        cache_key = hash(context.messages)
      
        # 检查缓存
        if cache_key in self.cache:
            return self.cache[cache_key]
      
        # 调用模型
        result = model_call(context)
      
        # 存入缓存
        self.cache[cache_key] = result
        return result

2.4 wrap_tool_call - 工具调用包装

执行时机： 每次工具调用时

核心作用： 拦截和控制工具的实际执行过程，可用于权限检查、日志记录、重试等。

应用场景：

• 工具调用权限检查
• 工具调用审计日志
• 工具调用重试
• 工具调用熔断

代码示例：

class ToolAuditMiddleware(Middleware):
    def wrap_tool_call(self, tool_call, context, tool_name, tool_args):
        # 记录工具调用日志
        logger.info(f"Tool called: {tool_name}, args: {tool_args}")
      
        # 检查权限
        if not self.has_permission(context.user, tool_name):
            raise PermissionError(f"No permission to use tool: {tool_name}")
      
        # 执行工具调用
        result = tool_call(tool_name, tool_args)
      
        # 记录结果
        logger.info(f"Tool result: {result}")
        return result

2.5 after_model - 模型调用后

执行时机： 模型调用完成后

核心作用： 处理模型返回的原始结果，进行验证、格式转换、提取关键信息等。

应用场景：

• 输出格式验证
• 敏感信息过滤
• 结果质量检查
• 响应数据转换

代码示例：

class OutputFilterMiddleware(Middleware):
    def after_model(self, context, response):
        # 过滤敏感词
        filtered_content = self.filter_sensitive_words(response.content)
        response.content = filtered_content
        return response

2.6 after_agent - Agent执行结束后

执行时机： Agent执行生命周期结束时

核心作用： 清理资源、记录最终状态、生成报告等。

应用场景：

• 资源释放
• 执行日志记录
• 性能指标统计
• 会话状态保存

代码示例：

class CleanupMiddleware(Middleware):
    def after_agent(self, context, final_response):
        # 记录执行时间
        execution_time = time.time() - context.start_time
        logger.info(f"Agent execution time: {execution_time}s")
      
        # 清理临时资源
        self.cleanup_temp_resources()
      
        return final_response

三、数据传递机制

3.1 Context对象

Hook间的数据传递通过共享的Context对象实现。Context对象包含：

属性	说明
messages	对话消息列表
config	配置信息
metadata	元数据（可在Hook间传递）
user	用户信息
start_time	执行开始时间

3.2 元数据传递

元数据传递机制允许中间件在不直接共享状态的情况下传递信息：

def before_model(self, context):
    # 在元数据中标记VIP用户
    if context.user.is_vip:
        context.metadata["priority"] = "high"
    return context

def wrap_model_call(self, model_call, context):
    # 后续中间件可以根据优先级调整处理策略
    if context.metadata.get("priority") == "high":
        # 使用更好的模型
        pass
    return model_call(context)

四、高级使用模式

4.1 条件Hook执行

通过智能的条件判断来决定是否执行特定的Hook：

class ConditionalMiddleware(Middleware):
    def before_model(self, context):
        # 只有VIP用户才启用复杂处理
        if not context.user.is_vip:
            return context  # 直接返回，不执行后续逻辑
      
        # VIP用户的特殊处理
        context.messages = self.enhance_prompt(context.messages)
        return context

4.2 错误恢复机制

在不同Hook层级实现恢复机制：

class RetryMiddleware(Middleware):
    def wrap_model_call(self, model_call, context):
        max_retries = 3
        for attempt in range(max_retries):
            try:
                return model_call(context)
            except Exception as e:
                if attempt == max_retries - 1:
                    raise
                logger.warning(f"Attempt {attempt + 1} failed, retrying...")
                time.sleep(2 ** attempt)  # 指数退避

4.3 性能优化Hook

缓存Hook通过在 wrap_model_call阶段检查缓存：

class SmartCacheMiddleware(Middleware):
    def __init__(self):
        self.cache = LRUCache(maxsize=1000)
        self.ttl = 3600  # 1小时过期
  
    def wrap_model_call(self, model_call, context):
        cache_key = self.generate_key(context.messages)
      
        # 检查缓存
        cached_result = self.cache.get(cache_key)
        if cached_result and not self.is_expired(cached_result):
            return cached_result
      
        # 调用模型
        result = model_call(context)
      
        # 存入缓存
        self.cache[cache_key] = result
        return result

五、中间件组合实战

5.1 企业级中间件栈

一个典型的企业级Agent可能需要以下中间件组合：

from langchain.agents import create_agent
from langchain.agents.middleware import (
    SummarizationMiddleware,
    CostTrackingMiddleware
)

# 创建中间件栈
middleware_stack = [
    InitMiddleware(),              # 初始化
    AuthMiddleware(),              # 身份验证
    RateLimitMiddleware(),         # 限流
    SummarizationMiddleware(       # 上下文压缩
        model=summary_model,
        max_tokens=500
    ),
    PIIMiddleware(),               # 敏感信息脱敏
    CacheMiddleware(),             # 结果缓存
    CostTrackingMiddleware(),      # 成本追踪
    AuditMiddleware()              # 审计日志
]

# 创建Agent
agent = create_agent(
    model=main_model,
    tools=tools,
    system_prompt=prompt,
    middleware=middleware_stack
)

5.2 执行顺序的重要性

中间件的执行顺序会影响最终效果：

顺序原则：

1. 安全和权限检查优先
2. 数据预处理在模型调用前
3. 缓存应在模型调用包装器中
4. 日志和追踪在最后

六、总结

主题	核心内容
6个Hook点	before_agent、before_model、wrap_model_call、wrap_tool_call、after_model、after_agent
数据传递	Context对象、元数据机制
实战案例	SummarizationMiddleware上下文压缩、缓存中间件、审计中间件
高级模式	条件执行、错误恢复、性能优化
最佳实践	中间件组合、执行顺序、企业级中间件栈

通过深入理解和灵活应用中间件的Hook机制，开发者可以构建出功能强大、稳定可靠的AI Agent系统。中间件不仅提供了丰富的扩展点，更重要的是它保持了Agent核心逻辑的简洁性，让复杂的功能增强变得模块化和可维护。

参考资源

• LangChain中间件文档：https://python.langchain.com/docs/concepts/middleware/
• LangGraph生命周期：https://langchain-ai.github.io/langgraph/concepts/agentic_concepts/
• 中间件最佳实践：https://python.langchain.com/docs/guides/productionization/