LangChain Agent 中间件详解：提升可靠性与可控性

配置：

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model="gpt-4.1",
    tools=[read_email_tool, send_email_tool],
    checkpointer=InMemorySaver(),
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "send_email_tool": {"allowed_decisions": ["approve", "edit", "reject"]},
                "read_email_tool": False  # 无需人工介入
            }
        )
    ]
)

2.3 模型调用限流（ModelCallLimitMiddleware）

1. 场景：防止 Agent 无限循环或 API 费用失控。
   • thread_limit：整个会话的大模型最大调用次数
   • run_limit：单次用户请求内的最大调用次数
   • exit_behavior：超出限制时，是"end"（优雅终止）还是"error"（抛异常）

配置：

from langchain.agents.middleware import ModelCallLimitMiddleware

agent = create_agent(
    model="gpt-4.1",
    checkpointer=InMemorySaver(),  # 跨多轮对话限制需要 checkpointer
    tools=[],
    middleware=[
        ModelCallLimitMiddleware(
            thread_limit=10,  # 整个对话最多调 10 次模型
            run_limit=5,  # 单次请求最多调 5 次
            exit_behavior="end"
        )
    ]
)

2.4 工具调用限流（ToolCallLimitMiddleware）

1. 场景：针对特定工具设置调用上限（如搜索引擎、数据库查询）。
   • tool_name：指定工具名称（不指定则全局限制）
   • thread_limit / run_limit：同模型限流
   • exit_behavior：支持"continue"（返回错误让 Agent 继续）、"error"、"end"

配置：

from langchain.agents.middleware import ToolCallLimitMiddleware

agent = create_agent(
    model="gpt-4.1",
    tools=[search_tool, db_tool],
    middleware=[
        ToolCallLimitMiddleware(thread_limit=20, run_limit=10),  # 全局限制
        ToolCallLimitMiddleware(tool_name="search", thread_limit=5, run_limit=3)  # 特定工具
    ]
)

2.5 模型降级（ModelFallbackMiddleware）

场景：主模型故障时自动切换到备用模型，提升可靠性。配置的备用模型支持字符串或 BaseChatModel 实例。

from langchain.agents.middleware import ModelFallbackMiddleware

agent = create_agent(
    model="gpt-4.1",
    tools=[],
    middleware=[
        ModelFallbackMiddleware(
            "gpt-4.1-mini",  # 第一备用
            "claude-3-5-sonnet-20241022"  # 第二备用
        )
    ]
)

2.6 工具重试（ToolRetryMiddleware）

1. 场景：工具调用因网络波动失败时，自动指数退避重试。
   • max_retries：最大重试次数
   • retry_on：指定哪些异常需重试（如 ConnectionError）
   • on_failure：重试耗尽后，"return_message"（返回错误消息）或"raise"（抛异常）

配置：

from langchain.agents.middleware import ToolRetryMiddleware

agent = create_agent(
    model="gpt-4.1",
    tools=[api_tool],
    middleware=[
        ToolRetryMiddleware(
            max_retries=3,
            backoff_factor=2.0,  # 指数退避：1s, 2s, 4s
            retry_on=(ConnectionError, TimeoutError),
            on_failure="return_message"
        )
    ]
)

2.7 模型重试（ModelRetryMiddleware）

1. 场景：模型 API 调用失败（如限流、超时）时自动重试。 配置：与工具重试类似，支持自定义异常过滤和失败处理。

from langchain.agents.middleware import ModelRetryMiddleware

agent = create_agent(
    model="gpt-4.1",
    tools=[],
    middleware=[
        ModelRetryMiddleware(
            max_retries=4,
            retry_on=lambda e: hasattr(e, "status_code") and e.status_code == 429,  # 仅重试限流
            on_failure="continue"  # 返回错误消息让 Agent 继续
        )
    ]
)

三、自定义中间件

上述是 langchain 官方提供的一些通用的中间件，这些中间件会在固定的流程节点执行固定的程序。我们也可以定义开发自己的中间件，无非确定两个点：在哪个节点（位置），执行什么程序。langchain 提供了两种实现方式：基于装饰器（快速简洁）和基于类（功能强大）的方式定义我们的中间件。

3.1 核心概念：钩子（Hook）与执行点

1. 钩子（Hook）：代理执行流程中预定义的位置，钩子就是执行点，中间件可以在这里挂载代码（钩子函数）。LangChain 中间件主要提供两类钩子：
   • 节点式钩子（Node-style）：在固定的顺序点运行，如'模型调用前'、'模型调用后'。多个中间件的同类型节点式钩子会按添加顺序依次执行。
   • 包装式钩子（Wrap-style）：包裹整个调用过程（如一次模型调用），可以完全控制是否执行、执行几次以及如何处理结果。
2. 状态（AgentState）：一个字典，存储当前代理运行的全部上下文，包括消息列表、中间变量等。中间件可以通过读取和修改状态来影响后续流程。

静态运行时上下文（Runtime）：一个对象，提供代理运行时的环境信息，如当前线程 ID、用户上下文等。

3.2 基于装饰器的中间件

装饰器方式适用于只需要一个钩子的简单中间件。LangChain 提供了多个装饰器，分别对应不同的执行点。

3.2.1 节点式装饰器

• 返回值说明
  • 返回 None：正常流程，继续执行下一个中间件或实际处理器。
  • 返回一个字典：该字典会与当前状态合并（state.update(returned_dict)），可用于添加或修改状态中的字段。特殊键 'jump_to' 可以强制跳转到指定节点（如 'end' 提前结束代理）。

中间件返回的 message 列表，跟工具是一样的，可以兼具同时删除消息，和添加新消息到 messages 列表后面（默认）。

from langchain.agents.middleware import after_model
from langchain.messages import AIMessage

@after_model(can_jump_to=["end"])  # 声明允许跳转到 end 节点
def block_sensitive_output(state, runtime):
    last_msg = state["messages"][-1]
    if "password" in last_msg.content.lower():  # 修改最后一条消息，并跳转到结束节点
        return {"messages": [AIMessage("I cannot reveal that information.")], "jump_to": "end"}
    return None

这些装饰器定义的函数会在特定的执行点（位置、钩子）被调用，不能控制是否调用下游中间件或实际处理器，只能读取/修改状态。

3.2.2 包装式装饰器

包装式装饰器，我们可以包裹实际的调用处理程序（handler），完全控制其执行。

参数说明：

• request：包含调用所需的信息，如 request.messages（消息列表）、request.runtime（运行时）、request.model（当前模型）等。
• handler：下一个中间件或实际处理器的可调用对象。调用 handler(request) 会继续执行包装链，最终返回响应。

from langchain.agents.middleware import wrap_model_call
import time

@wrap_model_call
def retry_with_backoff(request, handler):
    max_retries = 3
    for attempt in range(max_retries):
        try:
            return handler(request)
        except Exception as e:
            if attempt == max_retries - 1:
                raise  # 最后一次失败则向上抛出异常
            wait = (2 ** attempt)  # 指数退避：1, 2, 4 秒
            time.sleep(wait)
            print(f"Retry {attempt+1}/{max_retries} after {wait}s")

使用注意

• 装饰器函数本身会作为中间件对象，直接传入 middleware 列表。
• 多个中间件的执行顺序：节点式钩子按列表顺序执行；包装式钩子则按列表顺序从外到内包装（即列表最后一个包装器最先接触 handler，最内层执行）。

3.3 基于类的中间件

当中间件需要多个钩子、复杂配置或复用性时，应继承 AgentMiddleware 基类。

3.3.1 节点式钩子方法

在子类中重写以下方法即可，那这样一个类中间件实例化对象就包含了多个钩子

from langchain.agents.middleware import AgentMiddleware

class MyMiddleware(AgentMiddleware):
    def before_agent(self, state, runtime):  # 返回 None 或字典
        pass

    def before_model(self, state, runtime):
        pass

    def after_model(self, state, runtime):
        pass

    def after_agent(self, state, runtime):
        pass

3.3.2 包装式钩子方法

重写 wrap_model_call 和 wrap_tool_call 方法，包装式方法中，你必须显式调用 handler(request) 来继续执行链，否则下游中间件和实际调用将被跳过。你可以调用多次（重试），也可以根据条件不调用（短路）

class MyMiddleware(AgentMiddleware):
    def wrap_model_call(self, request, handler):
        # 在调用前可以修改 request
        # 可以选择是否调用 handler，以及调用几次
        response = handler(request)
        # 在调用后可以修改 response
        return response

    def wrap_tool_call(self, request, handler):
        # 类似 wrap_model_call，但 request 是 ToolRequest 类型
        pass

3.3.3 类中间件的初始化

可以通过 init 接收配置参数，然后保存为实例变量，在钩子方法中使用。

class RetryMiddleware(AgentMiddleware):
    def __init__(self, max_retries=3):
        self.max_retries = max_retries
        super().__init__()

四、深入理解 AgentState 和 Runtime

4.1 AgentState

agent 全局维护的状态对象，它是一个 TypedDict，至少包含以下字段：

• messages：List[BaseMessage]，当前对话的所有消息（包括人类消息、AI 消息、工具消息等）。
• intermediate_steps：记录工具执行的历史。
• 还可能包含其他自定义字段（如 todos、jump_to 等）。

中间件可以通过返回字典来更新状态，例如添加新字段或修改现有字段。但需注意：某些字段（如 messages）的修改需要遵循不可变原则或正确追加。

4.2 Runtime

提供运行时的上下文信息，常用属性：

• runtime.context：一个字典，包含了调用 agent.invoke(…, config={"configurable": {...}}) 时传入的自定义上下文数据。
• runtime.thread_id：当前线程 ID，可用于区分不同会话。