跳到主要内容
中间件提供了一种更严格控制代理内部行为的方法。 核心代理循环包括调用模型,让它选择要执行的工具,然后当它不再调用工具时结束:
Core agent loop diagram
中间件在每个步骤之前和之后都暴露了钩子
Middleware flow diagram

中间件能做什么?

监控

使用日志、分析和调试跟踪代理行为

修改

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

控制

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

强制执行

应用速率限制、防护栏和 PII 检测
通过将中间件传递给 create_agent 来添加
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware, HumanInTheLoopMiddleware


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

内置中间件

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

总结

在接近令牌限制时自动总结对话历史。
适用于
  • 超出上下文窗口的长时间对话
  • 具有大量历史记录的多轮对话
  • 需要保留完整对话上下文的应用程序
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware


agent = create_agent(
    model="gpt-4o",
    tools=[weather_tool, calculator_tool],
    middleware=[
        SummarizationMiddleware(
            model="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
        ),
    ],
)
model
字符串
必填
用于生成摘要的模型
max_tokens_before_summary
数字
触发摘要的令牌阈值
messages_to_keep
数字
默认:"20"
要保留的最新消息数量
token_counter
函数
自定义令牌计数函数。默认为基于字符的计数。
summary_prompt
字符串
自定义提示模板。如果未指定,则使用内置模板。
summary_prefix
字符串
默认:"## 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="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,
            }
        ),
    ],
)
interrupt_on
字典
必填
工具名称到批准配置的映射。值可以是 True(使用默认配置中断)、False(自动批准)或 InterruptOnConfig 对象。
description_prefix
字符串
默认:"Tool execution requires approval"
操作请求描述的前缀
InterruptOnConfig 选项
allowed_decisions
list[string]
允许的决策列表:"approve""edit""reject"
description
string | callable
自定义描述的静态字符串或可调用函数
重要提示: 人工干预中间件需要一个 检查点 来在中断之间维护状态。有关完整的示例和集成模式,请参阅 人工干预文档

Anthropic 提示缓存

通过 Anthropic 模型缓存重复的提示前缀来降低成本。
适用于
  • 具有长而重复的系统提示的应用程序
  • 在多次调用中重用相同上下文的代理
  • 为高吞吐量部署降低 API 成本
了解更多关于 Anthropic 提示缓存 策略和限制。
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-5-20250929"),
    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?")]})
type
字符串
默认:"ephemeral"
缓存类型。目前仅支持 "ephemeral"
ttl
字符串
默认:"5m"
缓存内容的生存时间。有效值:"5m""1h"
min_messages_to_cache
数字
默认:"0"
开始缓存前的最小消息数
unsupported_model_behavior
字符串
默认:"warn"
使用非 Anthropic 模型时的行为。选项:"ignore""warn""raise"

模型调用限制

限制模型调用次数,以防止无限循环或过高成本。
适用于
  • 防止失控代理进行过多的 API 调用
  • 在生产部署上强制执行成本控制
  • 在特定调用预算内测试代理行为
from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware


agent = create_agent(
    model="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
        ),
    ],
)
thread_limit
数字
线程中所有运行的最大模型调用次数。默认为无限制。
run_limit
数字
每次单独调用最大模型调用次数。默认为无限制。
exit_behavior
字符串
默认:"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="gpt-4o",
    tools=[...],
    middleware=[global_limiter, search_limiter],
)
tool_name
字符串
要限制的特定工具。如果未提供,则限制适用于所有工具。
thread_limit
数字
线程中所有运行的最大工具调用次数。默认为无限制。
run_limit
数字
每次单独调用最大工具调用次数。默认为无限制。
exit_behavior
字符串
默认:"end"
达到限制时的行为。选项:"end"(优雅终止)或 "error"(抛出异常)

模型回退

当主模型失败时,自动回退到备用模型。
适用于
  • 构建能够处理模型中断的弹性代理
  • 通过回退到更便宜的模型进行成本优化
  • OpenAI、Anthropic 等提供商之间的冗余
from langchain.agents import create_agent
from langchain.agents.middleware import ModelFallbackMiddleware


agent = create_agent(
    model="gpt-4o",  # Primary model
    tools=[...],
    middleware=[
        ModelFallbackMiddleware(
            "gpt-4o-mini",  # Try first on error
            "claude-3-5-sonnet-20241022",  # Then this
        ),
    ],
)
first_model
string | BaseChatModel
必填
当主模型失败时尝试的第一个回退模型。可以是模型字符串(例如,"openai:gpt-4o-mini")或 BaseChatModel 实例。
*additional_models
string | BaseChatModel
如果以前的模型失败,则按顺序尝试的其他回退模型

PII 检测

检测和处理对话中的个人身份信息。
适用于
  • 具有合规性要求的医疗保健和金融应用程序
  • 需要清理日志的客户服务代理
  • 任何处理敏感用户数据的应用程序
from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware


agent = create_agent(
    model="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
        ),
    ],
)
pii_type
字符串
必填
要检测的 PII 类型。可以是内置类型(emailcredit_cardipmac_addressurl)或自定义类型名称。
strategy
字符串
默认:"redact"
如何处理检测到的 PII。选项:
  • "block" - 检测到时抛出异常
  • "redact" - 替换为 [REDACTED_TYPE]
  • "mask" - 部分遮盖(例如,****-****-****-1234
  • "hash" - 替换为确定性哈希
detector
function | regex
自定义检测器函数或正则表达式模式。如果未提供,则使用 PII 类型的内置检测器。
apply_to_input
布尔值
默认:"True"
在模型调用前检查用户消息
apply_to_output
布尔值
默认:"False"
在模型调用后检查 AI 消息
apply_to_tool_results
布尔值
默认:"False"
执行后检查工具结果消息

规划

为复杂的、多步骤任务添加待办事项列表管理功能。
此中间件自动为代理提供 write_todos 工具和系统提示,以指导有效的任务规划。
from langchain.agents import create_agent
from langchain.agents.middleware import TodoListMiddleware
from langchain.messages import HumanMessage


agent = create_agent(
    model="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
system_prompt
字符串
用于指导待办事项使用的自定义系统提示。如果未指定,则使用内置提示。
tool_description
字符串
write_todos 工具的自定义描述。如果未指定,则使用内置描述。

LLM 工具选择器

在调用主模型之前,使用 LLM 智能选择相关工具。
适用于
  • 拥有许多工具(10+)但每次查询中大多数不相关的代理
  • 通过筛选不相关的工具来减少令牌使用
  • 提高模型焦点和准确性
from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolSelectorMiddleware


agent = create_agent(
    model="gpt-4o",
    tools=[tool1, tool2, tool3, tool4, tool5, ...],  # Many tools
    middleware=[
        LLMToolSelectorMiddleware(
            model="gpt-4o-mini",  # Use cheaper model for selection
            max_tools=3,  # Limit to 3 most relevant tools
            always_include=["search"],  # Always include certain tools
        ),
    ],
)
model
string | BaseChatModel
用于工具选择的模型。可以是模型字符串或 BaseChatModel 实例。默认为代理的主模型。
system_prompt
字符串
选择模型的指令。如果未指定,则使用内置提示。
max_tools
数字
要选择的最大工具数。默认为无限制。
always_include
list[string]
始终包含在选择中的工具名称列表

工具重试

自动重试失败的工具调用,并可配置指数退避。
适用于
  • 处理外部 API 调用中的瞬时故障
  • 提高依赖网络的工具的可靠性
  • 构建能够优雅处理临时错误的弹性代理
from langchain.agents import create_agent
from langchain.agents.middleware import ToolRetryMiddleware


agent = create_agent(
    model="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
        ),
    ],
)
max_retries
数字
默认:"2"
初始调用后最大重试次数(默认情况下总共尝试 3 次)
tools
list[BaseTool | str]
可选的工具或工具名称列表,用于应用重试逻辑。如果为 None,则应用于所有工具。
retry_on
tuple[type[Exception], ...] | callable
默认:"(Exception,)"
要么是要重试的异常类型元组,要么是一个可调用函数,该函数接受一个异常并返回 True 表示应该重试。
on_failure
string | callable
默认:"return_message"
所有重试耗尽时的行为。选项:
  • "return_message" - 返回包含错误详细信息的 ToolMessage(允许 LLM 处理失败)
  • "raise" - 重新抛出异常(停止代理执行)
  • 自定义可调用函数 - 接受异常并返回 ToolMessage 内容字符串的函数
backoff_factor
数字
默认:"2.0"
指数退避的乘数。每次重试等待 initial_delay * (backoff_factor ** retry_number) 秒。设置为 0.0 表示固定延迟。
initial_delay
数字
默认:"1.0"
首次重试前的初始延迟(秒)
max_delay
数字
默认:"60.0"
重试之间的最大延迟(秒)(限制指数退避增长)
jitter
布尔值
默认:"true"
是否为延迟添加随机抖动(±25%)以避免拥堵

LLM 工具模拟器

使用 LLM 模拟工具执行进行测试,用 AI 生成的响应替换实际的工具调用。
适用于
  • 在不执行真实工具的情况下测试代理行为
  • 在外部工具不可用或昂贵时开发代理
  • 在实施实际工具之前对代理工作流进行原型设计
from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolEmulator


agent = create_agent(
    model="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="claude-sonnet-4-5-20250929"),
    ],
)
tools
list[str | BaseTool]
要模拟的工具名称(str)或 BaseTool 实例列表。如果为 None(默认),将模拟所有工具。如果为空列表,则不模拟任何工具。
model
string | BaseChatModel
默认:"anthropic:claude-3-5-sonnet-latest"
用于生成模拟工具响应的模型。可以是模型标识符字符串或 BaseChatModel 实例。

上下文编辑

通过修剪、总结或清除工具使用来管理对话上下文。
适用于
  • 需要定期清理上下文的长时间对话
  • 从上下文中移除失败的工具尝试
  • 自定义上下文管理策略
from langchain.agents import create_agent
from langchain.agents.middleware import ContextEditingMiddleware, ClearToolUsesEdit


agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[
        ContextEditingMiddleware(
            edits=[
                ClearToolUsesEdit(max_tokens=1000),  # Clear old tool uses
            ],
        ),
    ],
)
edits
list[ContextEdit]
默认:"[ClearToolUsesEdit()]"
要应用的 ContextEdit 策略列表
token_count_method
字符串
默认:"approximate"
令牌计数方法。选项:"approximate""model"
ClearToolUsesEdit 选项
trigger
数字
默认:"100000"
触发编辑的令牌计数
clear_at_least
数字
默认:"0"
最少回收的令牌数
keep
数字
默认:"3"
要保留的最新工具结果数量
clear_tool_inputs
布尔值
默认:"False"
是否清除工具调用参数
exclude_tools
list[string]
默认:"()"
要从清除中排除的工具名称列表
placeholder
字符串
默认:"[cleared]"
清除输出的占位符文本

自定义中间件

通过实现代理执行流程中特定点的钩子来构建自定义中间件。 您可以通过两种方式创建中间件:
  1. 基于装饰器 - 适用于单钩子中间件,快速简单
  2. 基于类 - 适用于具有多个钩子的复杂中间件,功能更强大

基于装饰器的中间件

对于只需要单个钩子的简单中间件,装饰器提供了最快的方式来添加功能
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 - 代理完成时(每次调用一次)
包装式(拦截和控制执行) 便利装饰器:

何时使用装饰器

在以下情况下使用装饰器:

• 您只需要一个钩子
• 没有复杂的配置

在以下情况下使用类:

• 需要多个钩子
• 复杂的配置
• 跨项目重用(初始化时配置)

基于类的中间件

两种钩子风格

节点式钩子

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

包装式钩子

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

节点式钩子

在执行流中的特定点运行
  • 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("gpt-4o")
        else:
            request.model = init_chat_model("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="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="gpt-4o",
    middleware=[middleware1, middleware2, middleware3],
    tools=[...],
)
前置钩子按顺序运行
  1. middleware1.before_agent()
  2. middleware2.before_agent()
  3. middleware3.before_agent()
代理循环开始
  1. middleware1.before_model()
  2. middleware2.before_model()
  3. middleware3.before_model()
包装钩子像函数调用一样嵌套
  1. middleware1.wrap_model_call()middleware2.wrap_model_call()middleware3.wrap_model_call() → 模型
后置钩子按相反顺序运行
  1. middleware3.after_model()
  2. middleware2.after_model()
  3. middleware1.after_model()
代理循环结束
  1. middleware3.after_agent()
  2. middleware2.after_agent()
  3. 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_modelafter_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

最佳实践

  1. 保持中间件的专注性——每个中间件只做好一件事
  2. 优雅地处理错误——不要让中间件错误导致代理崩溃
  3. 使用适当的钩子类型:
    • 节点式用于顺序逻辑(日志记录、验证)
    • 包装式用于控制流(重试、回退、缓存)
  4. 清晰地记录所有自定义状态属性
  5. 在集成之前独立进行中间件单元测试
  6. 考虑执行顺序——将关键中间件放在列表前面
  7. 尽可能使用内置中间件,不要重复造轮子 :)

示例

动态选择工具

在运行时选择相关工具以提高性能和准确性。
优势
  • 更短的提示 - 通过只公开相关工具来降低复杂性
  • 更高的准确性 - 模型从更少的选项中正确选择
  • 权限控制 - 根据用户访问动态筛选工具
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="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()],
)

附加资源


以编程方式连接这些文档到 Claude、VSCode 等,通过 MCP 获取实时答案。
© . This site is unofficial and not affiliated with LangChain, Inc.