中间件 - Docs by LangChain

中间件提供了一种更紧密控制智能体内部发生事情的方法。核心智能体循环涉及调用模型，让它选择执行工具，然后在其不再调用更多工具时结束。

中间件在每个步骤之前和之后都暴露了钩子：

中间件能做什么？

Monitor

使用日志记录、分析和调试跟踪智能体行为

Modify

转换提示、工具选择和输出格式

Control

添加重试、回退和提前终止逻辑

Enforce

应用速率限制、安全护栏和PII检测

通过传递给 create_agent 添加中间件：

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


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

内置中间件

LangChain 为常见用例提供预构建的中间件：

摘要

当接近令牌限制时，自动总结对话历史。

适用场景：

超出上下文窗口的长对话
历史记录丰富的多轮对话
需要保留完整对话上下文的应用程序

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


agent = create_agent(
    model="openai:gpt-4o",
    tools=[weather_tool, calculator_tool],
    middleware=[
        SummarizationMiddleware(
            model="openai:gpt-4o-mini",
            max_tokens_before_summary=4000,  # Trigger summarization at 4000 tokens
            messages_to_keep=20,  # Keep last 20 messages after summary
            summary_prompt="Custom prompt for summarization...",  # Optional
        ),
    ],
)

Configuration options

model

string

required

生成摘要的模型

max_tokens_before_summary

number

触发摘要的令牌阈值

messages_to_keep

number

default:"20"

保留最近的消息

token_counter

function

自定义令牌计数函数。默认为基于字符的计数。

summary_prompt

string

自定义提示模板。如未指定，则使用内置模板。

summary_prefix

string

default:"## Previous conversation summary:"

摘要消息的前缀

人工增强循环

在智能体执行工具调用之前，暂停智能体执行以供人工审批、编辑或拒绝。

适用场景：

需要人工批准的高风险操作（数据库写入、金融交易）
需要人工监督的合规工作流程
长时间进行的对话，其中使用人工反馈来指导智能体

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


agent = create_agent(
    model="openai:gpt-4o",
    tools=[read_email_tool, send_email_tool],
    checkpointer=InMemorySaver(),
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                # Require approval, editing, or rejection for sending emails
                "send_email_tool": {
                    "allowed_decisions": ["approve", "edit", "reject"],
                },
                # Auto-approve reading emails
                "read_email_tool": False,
            }
        ),
    ],
)

Configuration options

interrupt_on

dict

required

工具名称到审批配置的映射。值可以是 True（使用默认配置中断），False（自动批准），或一个 InterruptOnConfig 对象。

description_prefix

string

default:"Tool execution requires approval"

动作请求描述的前缀

InterruptOnConfig 选项：

allowed_decisions

list[string]

允许的决定列表："approve"、"edit" 或 "reject"

description

string | callable

自定义描述的静态字符串或可调用函数

重要： 人机交互中间件需要一个检查点来在中断期间维护状态。查看人机交互文档以获取完整示例和集成模式。

Anthropic 提示缓存

通过使用Anthropic模型缓存重复的提示前缀来降低成本。

适用场景：

需要长、重复系统提示的应用
在调用之间重用相同上下文的智能体
降低高量部署的API成本

了解有关Anthropic Prompt Caching策略和限制的更多信息。

from langchain_anthropic import ChatAnthropic
from langchain_anthropic.middleware import AnthropicPromptCachingMiddleware
from langchain.agents import create_agent


LONG_PROMPT = """
Please be a helpful assistant.

<Lots more context ...>
"""

agent = create_agent(
    model=ChatAnthropic(model="claude-sonnet-4-latest"),
    system_prompt=LONG_PROMPT,
    middleware=[AnthropicPromptCachingMiddleware(ttl="5m")],
)

# cache store
agent.invoke({"messages": [HumanMessage("Hi, my name is Bob")]})

# cache hit, system prompt is cached
agent.invoke({"messages": [HumanMessage("What's my name?")]})

Configuration options

type

string

default:"ephemeral"

缓存类型。目前仅支持 "ephemeral"。

ttl

string

default:"5m"

缓存内容的存活时间。有效值："5m" 或 "1h"

min_messages_to_cache

number

default:"0"

缓存开始前的最小消息数

unsupported_model_behavior

string

default:"warn"

使用非Anthropic模型时的行为。选项："ignore"、"warn"或"raise"

模型调用限制

限制模型调用的次数，以防止无限循环或过高的成本。

适用场景：

防止智能体过多调用API
在生产部署中实施成本控制
在特定的调用预算内测试智能体行为

from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware


agent = create_agent(
    model="openai:gpt-4o",
    tools=[...],
    middleware=[
        ModelCallLimitMiddleware(
            thread_limit=10,  # Max 10 calls per thread (across runs)
            run_limit=5,  # Max 5 calls per run (single invocation)
            exit_behavior="end",  # Or "error" to raise exception
        ),
    ],
)

Configuration options

thread_limit

number

线程中所有运行的最大模型调用次数。默认无限制。

run_limit

number

单次调用中模型的最大调用次数。默认无限制。

exit_behavior

string

default:"end"

达到限制时的行为。选项："end"（优雅终止）或"error"（抛出异常）

工具调用限制

限制对特定工具或所有工具的调用次数。

适用场景：

防止过度调用昂贵的第三方API
限制网络搜索或数据库查询
对特定工具的使用实施速率限制

from langchain.agents import create_agent
from langchain.agents.middleware import ToolCallLimitMiddleware


# Limit all tool calls
global_limiter = ToolCallLimitMiddleware(thread_limit=20, run_limit=10)

# Limit specific tool
search_limiter = ToolCallLimitMiddleware(
    tool_name="search",
    thread_limit=5,
    run_limit=3,
)

agent = create_agent(
    model="openai:gpt-4o",
    tools=[...],
    middleware=[global_limiter, search_limiter],
)

Configuration options

tool_name

string

限制的特定工具。如果未提供，限制将应用于所有工具。

thread_limit

number

线程中所有运行的最大工具调用次数。默认无限制。

run_limit

number

单次调用中工具调用的最大次数。默认无限制。

exit_behavior

string

default:"end"

达到限制时的行为。选项："end"（优雅终止）或"error"（抛出异常）

模型回退

当主模型失败时自动回退到备用模型。

适用场景：

构建能够处理模型故障的弹性智能体
通过回退到更便宜的模型进行成本优化
在 OpenAI、Anthropic 等提供商之间实现冗余

from langchain.agents import create_agent
from langchain.agents.middleware import ModelFallbackMiddleware


agent = create_agent(
    model="openai:gpt-4o",  # Primary model
    tools=[...],
    middleware=[
        ModelFallbackMiddleware(
            "openai:gpt-4o-mini",  # Try first on error
            "anthropic:claude-3-5-sonnet-20241022",  # Then this
        ),
    ],
)

Configuration options

first_model

string | BaseChatModel

required

当主模型失败时首先尝试的回退模型。可以是模型字符串（例如，"openai:gpt-4o-mini"）或一个 BaseChatModel 实例。

*additional_models

string | BaseChatModel

尝试的额外后备模型，如果之前的模型失败

个人信息识别

检测和处理对话中的个人身份信息。

适用场景：

需要符合规定的医疗和金融应用
需要清理日志的客户服务智能体
处理敏感用户数据的任何应用

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware


agent = create_agent(
    model="openai:gpt-4o",
    tools=[...],
    middleware=[
        # Redact emails in user input
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        # Mask credit cards (show last 4 digits)
        PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
        # Custom PII type with regex
        PIIMiddleware(
            "api_key",
            detector=r"sk-[a-zA-Z0-9]{32}",
            strategy="block",  # Raise error if detected
        ),
    ],
)

Configuration options

pii_type

string

required

要检测的PII类型。可以是内置类型（email、credit_card、ip、mac_address、url）或自定义类型名称。

strategy

string

default:"redact"

如何处理检测到的个人身份信息（PII）。选项：

"block" - 检测到时抛出异常
"redact" - 替换为 [REDACTED_TYPE]
"mask" - 部分掩码（例如，****-****-****-1234）
"hash" - 替换为确定性哈希

detector

function | regex

自定义检测函数或正则表达式模式。如果未提供，则使用内置的 PII 类型检测器。

apply_to_input

boolean

default:"True"

在模型调用前检查用户消息

apply_to_output

boolean

default:"False"

检查模型调用后的AI消息

apply_to_tool_results

boolean

default:"False"

执行后检查工具结果消息

规划

添加复杂多步骤任务的待办事项列表管理功能。

此中间件自动为智能体提供write_todos工具和系统提示，以指导有效的任务规划。

from langchain.agents import create_agent
from langchain.agents.middleware import TodoListMiddleware
from langchain.messages import HumanMessage


agent = create_agent(
    model="openai:gpt-4o",
    tools=[...],
    middleware=[TodoListMiddleware()],
)

result = agent.invoke({"messages": [HumanMessage("Help me refactor my codebase")]})
print(result["todos"])  # Array of todo items with status tracking

Configuration options

system_prompt

string

定制系统提示，用于指导todo的使用。如未指定，则使用内置提示。

tool_description

string

为 write_todos 工具自定义描述。如未指定，则使用内置描述。

LLM 工具选择器

使用LLM智能选择相关工具，然后再调用主模型。

适用场景：

拥有众多工具（10+）的智能体，其中大多数工具在每个查询中并不相关
通过过滤无关工具来减少令牌使用
提高模型关注度和准确性

from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolSelectorMiddleware


agent = create_agent(
    model="openai:gpt-4o",
    tools=[tool1, tool2, tool3, tool4, tool5, ...],  # Many tools
    middleware=[
        LLMToolSelectorMiddleware(
            model="openai:gpt-4o-mini",  # Use cheaper model for selection
            max_tools=3,  # Limit to 3 most relevant tools
            always_include=["search"],  # Always include certain tools
        ),
    ],
)

Configuration options

model

string | BaseChatModel

工具选择模型。可以是模型字符串或 BaseChatModel 实例。默认为智能体的主模型。

system_prompt

string

选择模型的说明。如未指定，则使用内置提示。

max_tools

number

最大选择工具的数量。默认无限制。

always_include

list[string]

始终包含在选择中的工具名称列表

工具重试

自动重试失败的工具调用，并具有可配置的指数退避。

适用场景：

处理外部API调用中的短暂失败
提高网络依赖工具的可靠性
构建能够优雅处理临时错误的弹性智能体

from langchain.agents import create_agent
from langchain.agents.middleware import ToolRetryMiddleware


agent = create_agent(
    model="openai:gpt-4o",
    tools=[search_tool, database_tool],
    middleware=[
        ToolRetryMiddleware(
            max_retries=3,  # Retry up to 3 times
            backoff_factor=2.0,  # Exponential backoff multiplier
            initial_delay=1.0,  # Start with 1 second delay
            max_delay=60.0,  # Cap delays at 60 seconds
            jitter=True,  # Add random jitter to avoid thundering herd
        ),
    ],
)

Configuration options

max_retries

number

default:"2"

初始调用后的最大重试次数（默认总共3次尝试）

tools

list[BaseTool | str]

可选的工具或工具名称列表，用于应用重试逻辑。如果 None，则应用于所有工具。

retry_on

tuple[type[Exception], ...] | callable

default:"(Exception,)"

要重试的异常类型元组，或者一个接受异常并返回True以指示是否应该重试的可调用对象。

on_failure

string | callable

default:"return_message"

所有重试耗尽后的行为。选项：

"return_message" - 返回包含错误详情的 ToolMessage（允许 LLM 处理失败）
"raise" - 重新抛出异常（停止智能体执行）
自定义可调用函数 - 接收异常并返回 ToolMessage 内容字符串的函数

backoff_factor

number

default:"2.0"

指数退避的乘数。每次重试等待 initial_delay * (backoff_factor ** retry_number) 秒。设置为 0.0 以实现恒定延迟。

initial_delay

number

default:"1.0"

首次重试前的初始延迟（秒）

max_delay

number

default:"60.0"

重试之间最大延迟时间（秒）（指数退避增长）

jitter

boolean

default:"true"

是否向延迟添加随机抖动（±25%）以避免群体效应

LLM 工具模拟器

使用LLM模拟工具执行以进行测试，用AI生成的响应替换实际工具调用。

适用场景：

在不执行真实工具的情况下测试智能体行为
当外部工具不可用或昂贵时开发智能体
在实施实际工具之前原型化智能体工作流程

from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolEmulator


agent = create_agent(
    model="openai:gpt-4o",
    tools=[get_weather, search_database, send_email],
    middleware=[
        # Emulate all tools by default
        LLMToolEmulator(),

        # Or emulate specific tools
        # LLMToolEmulator(tools=["get_weather", "search_database"]),

        # Or use a custom model for emulation
        # LLMToolEmulator(model="anthropic:claude-3-5-sonnet-latest"),
    ],
)

Configuration options

tools

list[str | BaseTool]

工具名称列表（字符串）或要模拟的BaseTool实例。如果为None（默认），则将模拟所有工具。如果为空列表，则不模拟任何工具。

model

string | BaseChatModel

default:"anthropic:claude-3-5-sonnet-latest"

用于生成模拟工具响应的模型。可以是模型标识符字符串或BaseChatModel实例。

上下文编辑

通过裁剪、总结或清除工具使用来管理对话上下文。

适用场景：

需要定期清理上下文的长时间对话
从上下文中移除失败的工具尝试
自定义上下文管理策略

from langchain.agents import create_agent
from langchain.agents.middleware import ContextEditingMiddleware, ClearToolUsesEdit


agent = create_agent(
    model="openai:gpt-4o",
    tools=[...],
    middleware=[
        ContextEditingMiddleware(
            edits=[
                ClearToolUsesEdit(max_tokens=1000),  # Clear old tool uses
            ],
        ),
    ],
)

Configuration options

edits

list[ContextEdit]

default:"[ClearToolUsesEdit()]"

应用 ContextEdit 策略的列表

token_count_method

string

default:"approximate"

令牌计数方法。选项："approximate" 或 "model"

ClearToolUsesEdit options:

trigger

number

default:"100000"

触发编辑的令牌计数

clear_at_least

number

default:"0"

最小回收令牌数

keep

number

default:"3"

保留最近工具结果的数目

clear_tool_inputs

boolean

default:"False"

是否清除工具调用参数

exclude_tools

list[string]

default:"()"

要排除清理的工具名称列表

placeholder

string

default:"[cleared]"

清除输出占位文本

自定义中间件

通过实现运行在智能体执行流程特定点的钩子来构建自定义中间件。您可以通过两种方式创建中间件：

基于装饰器的 - 适用于单钩子中间件，快速简单
基于类的 - 对于具有多个钩子的复杂中间件来说更强大

基于装饰器的中间件

对于只需要单个钩子的简单中间件，装饰器提供了添加功能的最快方式：

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="openai: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 修改提示的功能）

何时使用装饰器

Use decorators when

• 需要单个钩子
• 无复杂配置

Use classes when

• 需要多个钩子
• 复杂配置
• 在多个项目中重用（初始化时配置）

基于类的中间件

两种钩子样式

Node-style hooks

在特定的执行点顺序运行。用于日志记录、验证和状态更新。

Wrap-style hooks

使用完全控制处理程序调用的方式拦截执行。用于重试、缓存和转换。

节点式钩子

在执行流程的特定点运行：

before_agent - 在智能体启动之前（每次调用一次）
before_model - 在每次模型调用之前
after_model - 在每次模型响应之后
after_agent - 在智能体完成之后（每次调用最多一次）

示例：日志中间件

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

示例：会话长度限制

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

class MessageLimitMiddleware(AgentMiddleware):
    def __init__(self, max_messages: int = 50):
        super().__init__()
        self.max_messages = max_messages

    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        if len(state["messages"]) == self.max_messages:
            return {
                "messages": [AIMessage("Conversation limit reached.")],
                "jump_to": "end"
            }
        return None

包装式钩子

拦截处理程序被调用时的执行和控制：

wrap_model_call - 在每次模型调用周围
wrap_tool_call - 在每次工具调用周围

您决定处理程序被调用零次（短路）、一次（正常流程）还是多次（重试逻辑）。 示例：模型重试中间件

from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse
from typing import Callable

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

    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse:
        for attempt in range(self.max_retries):
            try:
                return handler(request)
            except Exception as e:
                if attempt == self.max_retries - 1:
                    raise
                print(f"Retry {attempt + 1}/{self.max_retries} after error: {e}")

示例：动态模型选择

from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse
from langchain.chat_models import init_chat_model
from typing import Callable

class DynamicModelMiddleware(AgentMiddleware):
    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse:
        # Use different model based on conversation length
        if len(request.messages) > 10:
            request.model = init_chat_model("openai:gpt-4o")
        else:
            request.model = init_chat_model("openai:gpt-4o-mini")

        return handler(request)

示例：工具调用监控

from langchain.tools.tool_node import ToolCallRequest
from langchain.agents.middleware import AgentMiddleware
from langchain_core.messages import ToolMessage
from langgraph.types import Command
from typing import Callable

class ToolMonitoringMiddleware(AgentMiddleware):
    def wrap_tool_call(
        self,
        request: ToolCallRequest,
        handler: Callable[[ToolCallRequest], ToolMessage | Command],
    ) -> ToolMessage | Command:
        print(f"Executing tool: {request.tool_call['name']}")
        print(f"Arguments: {request.tool_call['args']}")

        try:
            result = handler(request)
            print(f"Tool completed successfully")
            return result
        except Exception as e:
            print(f"Tool failed: {e}")
            raise

自定义状态架构

中间件可以通过自定义属性扩展智能体的状态。定义一个自定义状态类型并将其设置为 state_schema：

from langchain.agents.middleware import AgentState, AgentMiddleware
from typing_extensions import NotRequired
from typing import Any

class CustomState(AgentState):
    model_call_count: NotRequired[int]
    user_id: NotRequired[str]

class CallCounterMiddleware(AgentMiddleware[CustomState]):
    state_schema = CustomState

    def before_model(self, state: CustomState, runtime) -> dict[str, Any] | None:
        # Access custom state properties
        count = state.get("model_call_count", 0)

        if count > 10:
            return {"jump_to": "end"}

        return None

    def after_model(self, state: CustomState, runtime) -> dict[str, Any] | None:
        # Update custom state
        return {"model_call_count": state.get("model_call_count", 0) + 1}

agent = create_agent(
    model="openai:gpt-4o",
    middleware=[CallCounterMiddleware()],
    tools=[...],
)

# Invoke with custom state
result = agent.invoke({
    "messages": [HumanMessage("Hello")],
    "model_call_count": 0,
    "user_id": "user-123",
})

执行顺序

当使用多个中间件时，理解执行顺序很重要：

agent = create_agent(
    model="openai:gpt-4o",
    middleware=[middleware1, middleware2, middleware3],
    tools=[...],
)

Execution flow (click to expand)

在钩子按顺序运行之前：

middleware1.before_agent()
middleware2.before_agent()
middleware3.before_agent()

智能体循环开始

middleware1.before_model()
middleware2.before_model()
middleware3.before_model()

将钩子嵌套包装，就像函数调用一样：

middleware1.wrap_model_call() → middleware2.wrap_model_call() → middleware3.wrap_model_call() → 模型

在钩子按反向顺序执行之后：

middleware3.after_model()
middleware2.after_model()
middleware1.after_model()

智能体循环结束

middleware3.after_agent()
middleware2.after_agent()
middleware1.after_agent()

关键规则：

before_* 钩子：从第一个到最后一个
after_* 钩子：从最后一个到第一个（反向）
wrap_* 钩子：嵌套（第一个中间件包裹所有其他中间件）

智能体跳跃

为了从中间件中提前退出，返回一个包含 jump_to 的字典：

class EarlyExitMiddleware(AgentMiddleware):
    def before_model(self, state: AgentState, runtime) -> dict[str, Any] | None:
        # Check some condition
        if should_exit(state):
            return {
                "messages": [AIMessage("Exiting early due to condition.")],
                "jump_to": "end"
            }
        return None

可用跳转目标：

"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

最佳实践

保持中间件专注 - 每个中间件都应该擅长做一件事
优雅地处理错误 - 不要让中间件错误导致智能体崩溃
使用适当的钩子类型：
- 使用 Node-style 钩子进行顺序逻辑（日志记录、验证）
- 使用 Wrap-style 钩子进行控制流（重试、回退、缓存）
明确记录任何自定义状态属性
在集成前独立对中间件进行单元测试
考虑执行顺序 - 将关键中间件放在列表的第一位
尽可能使用内置中间件，不要重新发明轮子 :)

示例

动态选择工具

在运行时选择相关工具以提高性能和准确性。

优势：

更短的提示 - 通过仅展示相关工具来降低复杂性
更高的准确性 - 模型从更少的选项中选择正确答案
权限控制 - 根据用户访问动态过滤工具

from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, ModelRequest
from typing import Callable


class ToolSelectorMiddleware(AgentMiddleware):
    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse:
        """Middleware to select relevant tools based on state/context."""
        # Select a small, relevant subset of tools based on state/context
        relevant_tools = select_relevant_tools(request.state, request.runtime)
        request.tools = relevant_tools
        return handler(request)

agent = create_agent(
    model="openai:gpt-4o",
    tools=all_tools,  # All available tools need to be registered upfront
    # Middleware can be used to select a smaller subset that's relevant for the given run.
    middleware=[ToolSelectorMiddleware()],
)

Show Extended example: GitHub vs GitLab tool selection

from dataclasses import dataclass
from typing import Literal, Callable

from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse
from langchain_core.tools import tool


@tool
def github_create_issue(repo: str, title: str) -> dict:
    """Create an issue in a GitHub repository."""
    return {"url": f"https://github.com/{repo}/issues/1", "title": title}

@tool
def gitlab_create_issue(project: str, title: str) -> dict:
    """Create an issue in a GitLab project."""
    return {"url": f"https://gitlab.com/{project}/-/issues/1", "title": title}

all_tools = [github_create_issue, gitlab_create_issue]

@dataclass
class Context:
    provider: Literal["github", "gitlab"]

class ToolSelectorMiddleware(AgentMiddleware):
    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse:
        """Select tools based on the VCS provider."""
        provider = request.runtime.context.provider

        if provider == "gitlab":
            selected_tools = [t for t in request.tools if t.name == "gitlab_create_issue"]
        else:
            selected_tools = [t for t in request.tools if t.name == "github_create_issue"]

        request.tools = selected_tools
        return handler(request)

agent = create_agent(
    model="openai:gpt-4o",
    tools=all_tools,
    middleware=[ToolSelectorMiddleware()],
    context_schema=Context,
)

# Invoke with GitHub context
agent.invoke(
    {
        "messages": [{"role": "user", "content": "Open an issue titled 'Bug: where are the cats' in the repository `its-a-cats-game`"}]
    },
    context=Context(provider="github"),
)

关键点：

预先注册所有工具
中间件根据每个请求选择相关子集
使用 context_schema 进行配置要求

其他资源

中间件API参考 - 自定义中间件的完整指南
人机协作 - 为敏感操作添加人工审核
测试智能体 - 测试安全机制的战略

在GitHub上编辑此页面的源代码。

通过MCP将这些文档编程连接到Claude、VSCode等，以获取实时答案。

LangChain v1.0

开始使用

核心组件

高级用法

生产环境使用

​中间件能做什么？

Monitor

Modify

Control

Enforce

​内置中间件

​摘要

​人工增强循环

​Anthropic 提示缓存

​模型调用限制

​工具调用限制

​模型回退

​个人信息识别

​规划

​LLM 工具选择器

​工具重试

​LLM 工具模拟器

​上下文编辑

​自定义中间件

​基于装饰器的中间件

​可用装饰器

​何时使用装饰器

Use decorators when

Use classes when

​基于类的中间件

​两种钩子样式

Node-style hooks

Wrap-style hooks

​节点式钩子

​包装式钩子

​自定义状态架构

​执行顺序

​智能体跳跃

​最佳实践

​示例

​动态选择工具

​其他资源

中间件能做什么？

内置中间件

摘要

人工增强循环

Anthropic 提示缓存

模型调用限制

工具调用限制

模型回退

个人信息识别

规划

LLM 工具选择器

工具重试

LLM 工具模拟器

上下文编辑

自定义中间件

基于装饰器的中间件

可用装饰器

何时使用装饰器

基于类的中间件

两种钩子样式

节点式钩子

包装式钩子

自定义状态架构

执行顺序

智能体跳跃

最佳实践

示例

动态选择工具

其他资源