LangChain1.0智能体开发：中间件（Middleware）

1、智能体中间件

中间件能够控制和定制化智能体每一步的执行过程 中间件提供了一种更严格地控制智能体内部运作的方式。 核心的智能体循环包括调用模型、让模型选择要执行的工具，以及当模型不再调用工具时结束循环：

2、中间件能做什么

• 运行监控
  通过日志记录、数据分析和调试来跟踪智能体行为
• 任务适配
  提示词转换、工具选择和输出格式化
• 流程控制
  添加重试、备用方案和提前终止逻辑
• 强制规范
  应用速率限制、安全防护和个人身份信息（PII）检测

3、基本用法

在创建Agent时，将中间件列表传递给create_agent接口的middleware参数：

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware, HumanInTheLoopMiddleware

agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[SummarizationMiddleware(), HumanInTheLoopMiddleware()],
)

4、内置中间件

• 任务适配
  4.1 SummarizationMiddleware（摘要中间件）
  4.2 AnthropicPromptCachingMiddleware（Anthropic提示词缓存中间件）
  4.3 LLMToolSelectorMiddleware（LLM工具选择中间件）
  4.4 ContextEditingMiddleware（上下文编辑中间件）
  
• 流程控制
  4.2 HumanInTheLoopMiddleware（人工介入中间件）
  4.6 ModelFallbackMiddleware（模型回退中间件）
  4.8 TodoListMiddleware（待办事项中间件）
  4.10 ToolRetryMiddleware（工具重试中间件）
  4.11 LLMToolEmulator（LLM工具模拟器）
  
• 强制规范
  4.4 ModelCallLimitMiddleware（模型调用限制中间件）
  4.5 ToolCallLimitMiddleware（工具调用限制中间件）
  4.7 PIIMiddleware（PII检测中间件）
  

5、自定义中间件

有两种方式创建中间件：

• 基于装饰器（Decorator-based）—— 适用于单钩子中间件，快捷且简单
• 基于类（Class-based）—— 适用于包含多个钩子的复杂中间件，功能更强大

5.1 基于装饰器创建中间件

对于仅需单个钩子函数（hook）的简单中间件而言，装饰器（decorator）是添加功能的最快方式。

from langchain.agents.middleware import before_model, after_model, wrap_model_call
from langchain.agents.middleware import AgentState, ModelRequest, ModelResponse, dynamic_prompt
from langchain.messages import AIMessage
from langchain.agents import create_agent
from langgraph.runtime import Runtime
from typing import Any, Callable


# Node-style: logging before model calls
@before_model
def log_before_model(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    print(f"About to call model with {len(state['messages'])} messages")
    return None

# Node-style: validation after model calls
@after_model(can_jump_to=["end"])
def validate_output(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    last_message = state["messages"][-1]
    if "BLOCKED" in last_message.content:
        return {
            "messages": [AIMessage("I cannot respond to that request.")],
            "jump_to": "end"
        }
    return None

# Wrap-style: retry logic
@wrap_model_call
def retry_model(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    for attempt in range(3):
        try:
            return handler(request)
        except Exception as e:
            if attempt == 2:
                raise
            print(f"Retry {attempt + 1}/3 after error: {e}")

# Wrap-style: dynamic prompts
@dynamic_prompt
def personalized_prompt(request: ModelRequest) -> str:
    user_id = request.runtime.context.get("user_id", "guest")
    return f"You are a helpful assistant for user {user_id}. Be concise and friendly."

# Use decorators in agent
agent = create_agent(
    model="gpt-4o",
    middleware=[log_before_model, validate_output, retry_model, personalized_prompt],
    tools=[...],
)

可用的装饰器：

• 节点式（在特定执行节点运行）：
  @before_agent—— 智能体启动前（每次调用执行一次）
  @before_model—— 每次调用模型前
  @after_model—— 每次接收模型响应后
  @after_agent—— 智能体执行完成后（每次调用执行一次）
  
• 封装式（拦截并控制执行流程）：
  @wrap_model_call—— 围绕每次模型调用（全程拦截）
  @wrap_tool_call—— 围绕每次工具调用（全程拦截）
  
• 便捷装饰器：
  @dynamic_prompt—— 生成动态系统提示词（等同于能修改提示词的@wrap_model_call装饰器）

5.2 基于类创建中间件

以下情况适合使用基于类创建中间件：

• 需要多个钩子函数（hooks）
• 复杂配置
• 跨项目复用（初始化时配置）

两种类型的钩子：

• 节点式钩子：在执行流程的特定节点运行。
  • before_agent—— 智能体启动前（每次调用执行一次）
  • before_model—— 每次调用模型前
  • after_model—— 每次接收模型响应后
  • after_agent—— 智能体执行完成后（每次调用最多执行一次）
• 封装式钩子：当handler被调用时，拦截执行并控制。
  • wrap_model_call—— 围绕每次模型调用（全程拦截）
  • wrap_tool_call—— 围绕每次工具调用（全程拦截） 你可以决定handler的调用次数：零次（短路）、一次（正常流程）或多次（重试逻辑）。

from langchain.agents.middleware import AgentMiddleware, AgentState
from langgraph.runtime import Runtime
from typing import Any

class LoggingMiddleware(AgentMiddleware):
    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        print(f"About to call model with {len(state['messages'])} messages")
        return None

    def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        print(f"Model returned: {state['messages'][-1].content}")
        return None

6、中间件执行顺序

6.1 关键规则

• before_* 钩子函数：按顺序执行（从第一个到最后一个）
  
• after_* 钩子函数：按逆序执行（从最后一个到第一个）
  
• wrap_* 钩子函数：嵌套执行（第一个中间件包裹所有其他中间件）
  

6.2 跳转功能

若要从中间件中提前退出，需返回一个包含jump_to的字典。
可用的跳转目标：

• "end"：跳转到智能体执行流程的末尾
• "tools"：跳转到工具节点
• "model"：跳转到模型节点（或第一个 before_model 钩子函数）

注意：当从before_model或after_model钩子函数中发起跳转时，跳转到"model"会导致所有before_model中间件重新运行。
若要启用跳转功能，需用 @hook_config(can_jump_to=[...]) 装饰你的钩子函数。

from langchain.agents.middleware import AgentMiddleware, hook_config
from typing import Any

class ConditionalMiddleware(AgentMiddleware):
    @hook_config(can_jump_to=["end", "tools"])
    def after_model(self, state: AgentState, runtime) -> dict[str, Any] | None:
        if some_condition(state):
            return {"jump_to": "end"}
        return None

7、最佳实践经验总结

• 保持中间件专注性 —— 每个中间件应专注做好一件事
• 优雅地处理错误 —— 避免中间件错误导致智能体崩溃
• 使用合适的钩子类型：
  • 节点式（Node-style）适用于顺序逻辑（如日志记录、验证）
  • 封装式（Wrap-style）适用于控制流（如重试、备用方案、缓存）
• 清晰地文档化所有自定义状态属性
• 集成前独立对中间件进行单元测试
• 考虑执行顺序 —— 将关键中间件放在列表首位
• 尽可能使用内置中间件，避免重复造轮子。