中间件

中间件提供了一种更精细地控制代理内部行为的方式。核心代理循环包括调用 model，让其选择要执行的 tools，然后在不再调用任何工具时结束。

中间件提供对这些步骤之前和之后发生事件的控制。每个中间件可以添加三种不同类型的修饰器：

Middleware.before_model：在模型执行前运行。可更新状态或跳转到不同节点（model, tools, __end__）
Middleware.modify_model_request：在模型执行前运行，用于准备模型请求对象。只能修改当前模型请求对象（不能永久更新状态），且不能跳转到不同节点。
Middleware.after_model：在模型执行后、工具执行前运行。可更新状态或跳转到不同节点（model, tools, END）

代理可以包含 before_model、modify_model_request 或 after_model 中间件。无需实现全部三种。

在代理中使用

可以通过将中间件传递给 create_agent 来在代理中使用：

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

agent = create_agent(
    ...,
    middleware=[SummarizationMiddleware(), HumanInTheLoopMiddleware()],
    ...
)

中间件非常灵活，取代了代理中的某些其他功能。因此，当使用中间件时，创建代理的参数会有一些限制：

model 必须是字符串或 BaseChatModel。若传入函数则会报错。如需动态控制模型，请使用 AgentMiddleware.modify_model_request
prompt 必须是字符串或 None。若传入函数则会报错。如需动态控制提示词，请使用 AgentMiddleware.modify_model_request
不得提供 pre_model_hook。请改用 AgentMiddleware.before_model
不得提供 post_model_hook。请改用 AgentMiddleware.after_model

内置中间件

LangChain 提供了若干开箱即用的内置中间件：

摘要
人在回路
Anthropic 提示缓存

摘要

summarizationMiddleware 在接近令牌限制时自动通过摘要旧消息来管理对话历史。该中间件监控消息的总令牌数，并创建简洁摘要以在保持上下文的同时不超出模型限制。 主要特性：

自动令牌计数和阈值监控
智能消息分区，保留 AI/工具消息对
可自定义的摘要提示和令牌限制

使用场景：

超过令牌限制的长时对话
具有广泛上下文的多轮对话

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,  # 在 4000 令牌时触发摘要
            messages_to_keep=20,  # 摘要后保留最后 20 条消息
            summary_prompt="自定义摘要提示...",  # 可选
        ),
    ],
)

配置选项：

model：用于生成摘要的语言模型（必需）
maxTokensBeforeSummary：触发摘要的令牌阈值
messagesToKeep：保留的最近消息数量（默认：20）
tokenCounter：自定义令牌计数函数（默认为基于字符的近似值）
summaryPrompt：摘要生成的自定义提示模板
summaryPrefix：添加到包含摘要的系统消息前的前缀（默认：”## Previous conversation summary:”）

该中间件通过以下方式确保工具调用完整性：

从不拆分 AI 消息与其对应的工具响应
保留最新消息以保持连续性
在新摘要周期中包含之前的摘要

人在回路

humanInTheLoopMiddleware 为 AI 代理执行的工具调用启用人工监督和干预。该中间件拦截工具执行，允许人工操作员在执行前批准、修改、拒绝或手动响应工具调用。 主要特性：

基于配置的工具选择性批准
多种响应类型（接受、编辑、忽略、响应）
使用 LangGraph 中断的异步审批工作流
带上下文信息的自定义审批消息

使用场景：

需要人工批准的高风险操作（数据库写入、文件系统更改）
AI 行为的质量控制和安全检查
需要审计跟踪的合规场景
代理行为的开发和测试

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=[write_file_tool, execute_sql_tool, read_data_tool],
    middleware=[
        HumanInTheLoopMiddleware(
            tool_configs={
                "write_file": {
                    "require_approval": True,
                    "description": "⚠️ 文件写入操作需要批准",
                },
                "execute_sql": {
                    "require_approval": True,
                    "description": "🚨 SQL 执行需要 DBA 批准",
                },
                "read_data": {
                    "require_approval": False,  # 安全操作，无需批准
                },
            },
            message_prefix="工具执行待批准",
        ),
    ],
    checkpointer=InMemorySaver(),  # 中断必需
)

处理审批请求： 当工具需要批准时，代理执行暂停并等待人工输入：

from langchain_core.messages import HumanMessage
from langgraph.types import Command

# 初始调用
result = agent.invoke(
    {
        "messages": [HumanMessage("从数据库中删除旧记录")],
    },
    config
)

# 检查是否因审批暂停
state = agent.get_state(config)
if state.next:
    request = state.tasks[0].interrupts[0].value[0]["action_request"]

    # 向人工显示工具详情
    print("工具:", request["action"])
    print("参数:", request["args"])

    # 使用审批决定继续执行
    agent.invoke(
        Command(
            resume=[{"type": "accept"}]  # 或 "edit", "ignore", "response"
        ),
        config=config
    )

响应类型：

accept：使用原始参数执行工具
edit：修改参数后再执行 - { type: "edit", args: { action: "tool_name", args: { modified: "args" } } }
ignore：跳过工具执行并终止代理
response：提供手动响应而非执行工具 - { type: "response", args: "手动响应文本" }

配置：

toolConfigs：工具名称到其审批设置的映射
- requireApproval：该工具是否需要人工批准
- description：审批请求时显示的自定义消息
messagePrefix：审批消息的默认前缀

该中间件按顺序处理工具调用，将多个审批请求捆绑到单个中断中以提高效率。不需要批准的工具会立即执行，不会中断。

Anthropic 提示缓存

AnthropicPromptCachingMiddleware 是一个中间件，可启用 Anthropic 的原生提示缓存功能。提示缓存通过允许从提示中的特定前缀恢复来优化 API 使用。这对于具有重复提示或冗余信息提示的任务特别有用。

了解更多关于 Anthropic 提示缓存（策略、限制等）请点击此处。

使用提示缓存时，你可能希望使用检查点来存储跨调用的对话历史。

from langchain_anthropic import ChatAnthropic
from langchain.agents.middleware.prompt_caching import AnthropicPromptCachingMiddleware
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

LONG_PROMPT = """
请做一个乐于助人的助手。

<更多上下文...>
"""

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

common_config = {"configurable": {"thread_id": "1"}}

# 缓存存储
agent.invoke({"messages": [HumanMessage("嗨，我叫鲍勃")]}, common_config)

# 缓存命中
agent.invoke({"messages": [HumanMessage("我叫什么名字？")]}, common_config)

自定义中间件

代理的中间件是 AgentMiddleware 的子类，实现其一个或多个钩子。 AgentMiddleware 目前提供三种修改核心代理循环的方式：

before_model：在模型运行前执行。可更新状态或通过跳转提前退出。
modify_model_request：在模型运行前执行。不能更新状态或通过跳转提前退出。
after_model：在模型运行后执行。可更新状态或通过跳转提前退出。

为了提前退出，你可以在状态更新中添加一个 jump_to 键，其值为以下之一：

"model"：跳转到模型节点
"tools"：跳转到工具节点
"__end__"：跳转到结束节点

如果指定了此值，所有后续中间件将不会运行。在代理跳转部分了解更多关于提前退出的信息。

`before_model`

在模型运行前执行。可通过返回新状态对象或状态更新来修改状态。签名：

from langchain.agents.middleware import AgentMiddleware, AgentState
from langchain_core.messages import AIMessage

class MyMiddleware(AgentMiddleware):
    def before_model(self, state: AgentState) -> dict[str, Any] | None:
        # 如果对话过长则提前终止
        if len(state["messages"]) > 50:
            return {
                "messages": [AIMessage("抱歉，对话已终止。")],
                "jump_to": "__end__"
            }
        return state

`modify_model_request`

在模型运行前执行，但在所有 before_model 调用之后。这些函数不能修改永久状态或提前退出。它们旨在以无状态*方式修改模型调用。如果你想以有状态**方式修改模型调用，则需要使用 before_model 修改模型请求。模型请求有几个关键属性：

model (BaseChatModel)：要使用的模型。注意：这需要是基础聊天模型，而不是字符串。
system_prompt (str)：要使用的系统提示。将被前置到 messages
messages (消息列表)：消息列表。不应包含系统提示。
tool_choice (任意)：要使用的工具选择
tools (BaseTool 列表)：此模型调用要使用的工具
response_format (ResponseFormat)：用于结构化输出的响应格式

签名：

from langchain.agents.middleware import AgentState, ModelRequest, AgentMiddleware

class MyMiddleware(AgentMiddleware):
    def modify_model_request(self, request: ModelRequest, state: AgentState) -> ModelRequest:
        if len(state["messages"]) > 10:
            request.model = "gpt-5"
        else:
            request.model = "gpt-5-nano"
        return request

`after_model`

在模型运行后执行。可通过返回新状态对象或状态更新来修改状态。签名：

from langchain.agents.middleware import AgentState, AgentUpdate, AgentMiddleware

class MyMiddleware(AgentMiddleware):
    def after_model(self, state: AgentState) -> dict[str, Any] | None:
        ...

新状态键

中间件可以通过自定义属性扩展代理的状态，实现中间件组件之间的丰富数据流，并确保代理执行过程中的类型安全。

状态扩展

中间件可以定义在代理执行过程中持续存在的额外状态属性。这些属性成为代理状态的一部分，并可供该中间件的所有钩子使用。

from langchain.agents.middleware import AgentState, AgentMiddleware

class MyState(AgentState):
    model_call_count: int

class MyMiddleware(AgentMiddleware[MyState]):
    state_schema: MyState

    def before_model(self, state: AgentState) -> dict[str, Any] | None:
        # 如果模型调用次数过多则提前终止
        if state["model_call_count"] > 10:
            return {"jump_to": "__end__"}
        return state

    def after_model(self, state: AgentState) -> dict[str, Any] | None:
        return {"model_call_count": state["model_call_count"] + 1}

上下文扩展

目前仅在 JavaScript 中可用。

上下文属性是通过可运行配置传递的配置值。与状态不同，上下文是只读的，通常用于执行过程中不改变的配置。

组合多个中间件

使用多个中间件时，它们的状态和上下文模式会被合并。必须满足所有中间件的所有必需属性：

from langchain.agents.middleware import AgentMiddleware
from langchain_core.messages import HumanMessage
from typing import Any, Dict

class Middleware1State(AgentState):
    prop_1: str
    shared_prop: int

class Middleware2State(AgentState):
    prop_2: bool
    shared_prop: int

class Middleware1(AgentMiddleware):
    def before_model(self, state: Dict[str, Any]) -> Dict[str, Any] | None:
        # 从状态访问 prop1 和 sharedProp
        print(f"Middleware1: prop1={state.get('prop_1')}, sharedProp={state.get('shared_prop')}")
        return None

class Middleware2(AgentMiddleware):
    def before_model(self, state: Dict[str, Any]) -> Dict[str, Any] | None:
        # 从状态访问 prop2 和 sharedProp
        print(f"Middleware2: prop2={state.get('prop_2')}, sharedProp={state.get('shared_prop')}")
        return None

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

代理级上下文模式

代理也可以定义自己的上下文要求，这些要求与中间件要求相结合：

# ...

最佳实践

使用状态存储动态数据：执行过程中变化的属性（用户会话、累积数据）
使用上下文存储配置：静态配置值（API 密钥、功能标志、限制）
尽可能提供默认值：在 Zod 模式中使用 .default() 使属性可选
记录要求：清晰记录中间件所需的状态和上下文属性

中间件执行顺序

你可以提供多个中间件。它们按以下逻辑执行： before_model：按传入顺序执行。如果较早的中间件提前退出，则后续中间件不会运行 modify_model_request：按传入顺序执行。 after_model：按传入顺序的反向执行。如果较早的中间件提前退出，则后续中间件不会运行

代理跳转

为了提前退出，你可以在状态更新中添加一个 jump_to 键，其值为以下之一：

"model"：跳转到模型节点
"tools"：跳转到工具节点
"__end__"：跳转到结束节点

如果指定了此值，所有后续中间件将不会运行。如果你跳转到 model 节点，所有 before_model 中间件都会运行。禁止从现有的 before_model 中间件跳转到 model。使用示例：

from langchain.agents.types import AgentState, AgentUpdate, AgentJump
from langchain.agents.middleware import AgentMiddleware

class MyMiddleware(AgentMiddleware):
    def after_model(self, state: AgentState) -> AgentUpdate | AgentJump | None:
        return {
        "messages": ...,
        "jump_to": "model"
    }

快速开始

核心组件

进阶使用

在生产环境使用

在代理中使用

内置中间件

摘要

人在回路

Anthropic 提示缓存

自定义中间件

`before_model`

`modify_model_request`

`after_model`

新状态键

状态扩展

上下文扩展

组合多个中间件

代理级上下文模式

最佳实践

中间件执行顺序

代理跳转

​在代理中使用

​内置中间件

​摘要

​人在回路

​Anthropic 提示缓存

​自定义中间件

​before_model

​modify_model_request

​after_model

​新状态键

​状态扩展

​上下文扩展

​组合多个中间件

​代理级上下文模式

​最佳实践

​中间件执行顺序

​代理跳转

在代理中使用

内置中间件

摘要

人在回路

Anthropic 提示缓存

自定义中间件

`before_model`

`modify_model_request`

`after_model`

新状态键

状态扩展

上下文扩展

组合多个中间件

代理级上下文模式

最佳实践

中间件执行顺序

代理跳转