自定义中间件

通过实现在代理执行流特定点运行的钩子来构建自定义中间件。

钩子

中间件提供两种风格的钩子来拦截代理执行：

节点式钩子

在特定执行点顺序运行。

包装式钩子

围绕每次模型或工具调用运行。

节点式钩子

在特定执行点顺序运行。用于日志记录、验证和状态更新。选择中间件所需的钩子。您可以选择节点式钩子或包装式钩子。 节点式钩子在特定执行点运行：

钩子	运行时机
`before_agent`	代理启动前（每次调用一次）
`before_model`	每次模型调用前
`after_model`	每次模型响应后
`after_agent`	代理完成后（每次调用一次）

包装式钩子围绕每次调用运行，让您控制执行：

钩子	运行时机
`wrap_model_call`	围绕每次模型调用
`wrap_tool_call`	围绕每次工具调用

示例：

装饰器
类

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


@before_model(can_jump_to=["end"])
def check_message_limit(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    if len(state["messages"]) >= 50:
        return {
            "messages": [AIMessage("已达到对话限制。")],
            "jump_to": "end"
        }
    return None

@after_model
def log_response(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    print(f"模型返回了: {state['messages'][-1].content}")
    return None

from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config
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

    @hook_config(can_jump_to=["end"])
    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        if len(state["messages"]) >= self.max_messages:
            return {
                "messages": [AIMessage("已达到对话限制。")],
                "jump_to": "end"
            }
        return None

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

包装式钩子

拦截执行并控制处理器的调用时机。用于重试、缓存和转换。您可以决定处理器是调用零次（短路）、一次（正常流程）还是多次（重试逻辑）。 可用钩子：

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

示例：

装饰器
类

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


@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"重试 {attempt + 1}/3，发生错误: {e}")

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"重试 {attempt + 1}/{self.max_retries}，发生错误: {e}")

状态更新

节点式和包装式钩子都可以更新代理状态。其机制不同：

节点式钩子 (before_agent, before_model, after_model, after_agent): 直接返回一个字典。字典通过图的归约器应用到代理状态。
包装式钩子 (wrap_model_call, wrap_tool_call): 对于模型调用，返回一个包含 Command 的 ExtendedModelResponse，以在模型响应旁注入状态更新。对于工具调用，直接返回一个 Command。当您需要根据在模型或工具调用期间运行的逻辑来跟踪或更新状态时（例如摘要触发点、使用元数据或根据请求或响应计算的自定义字段），请使用这些钩子。

节点式钩子

从节点式钩子返回一个字典，将更新合并到代理状态中。字典键映射到状态字段。

from langchain.agents.middleware import after_model, AgentState
from langgraph.runtime import Runtime
from typing import Any
from typing_extensions import NotRequired


class TrackingState(AgentState):
    model_call_count: NotRequired[int]


@after_model(state_schema=TrackingState)
def increment_after_model(state: TrackingState, runtime: Runtime) -> dict[str, Any] | None:
    return {"model_call_count": state.get("model_call_count", 0) + 1}

包装式钩子

从 wrap_model_call 返回一个包含 Command 的 ExtendedModelResponse，以从模型调用层注入状态更新：

from typing import Callable
from langchain.agents.middleware import (
    wrap_model_call,
    ModelRequest,
    ModelResponse,
    AgentState,
    ExtendedModelResponse
)
from langgraph.types import Command
from typing_extensions import NotRequired

class UsageTrackingState(AgentState):
    """包含令牌使用跟踪的代理状态。"""

    last_model_call_tokens: NotRequired[int]


@wrap_model_call(state_schema=UsageTrackingState)
def track_usage(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ExtendedModelResponse:
    response = handler(request)
    return ExtendedModelResponse(
        model_response=response,
        command=Command(update={"last_model_call_tokens": 150}),
    )

Command 通过图的归约器流动，因此更新被正确应用，消息是累加的，而不是替换现有状态。

多个中间件的组合

当多个中间件层返回 ExtendedModelResponse 时，它们的命令会组合：

命令通过归约器应用： 每个 Command 成为一个独立的状态更新。对于消息，这意味着它们是累加的。
外层在冲突中获胜： 对于非归约器状态字段，命令按内层优先应用，然后是外层。对于有冲突的键，最外层中间件的值优先。
重试安全： 如果外层中间件实现了可能导致再次调用 handler() 的逻辑（例如重试逻辑），则之前调用中的命令会被丢弃。

from typing import Annotated, Callable

from langchain.agents.middleware import (
    AgentMiddleware,
    AgentState,
    ExtendedModelResponse,
    ModelRequest,
    ModelResponse,
)
from langchain.messages import SystemMessage
from langgraph.types import Command
from typing_extensions import NotRequired


def _last_wins(_a: str, b: str) -> str:
    """归约器：最后写入者获胜（外层覆盖内层）。"""
    return b


class CustomMiddlewareState(AgentState):
    """代理状态：trace_layer 使用 last-wins（外层获胜），messages 使用累加归约器。"""

    # 使用 last-wins 的非归约器字段：两个中间件都写入；最外层值获胜
    trace_layer: NotRequired[Annotated[str, _last_wins]]


class OuterMiddleware(AgentMiddleware):
    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ExtendedModelResponse:
        response = handler(request)
        return ExtendedModelResponse(
            model_response=response,
            command=Command(update={
                "trace_layer": "outer",
                "messages": [SystemMessage(content="[外层已运行]")],
            }),
        )


class InnerMiddleware(AgentMiddleware):
    """添加 trace_layer 和 message。外层添加到相同的键；trace_layer：外层获胜，messages：累加。"""

    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ):
        response = handler(request)
        return ExtendedModelResponse(
            model_response=response,
            command=Command(update={
                "trace_layer": "inner",
                "messages": [SystemMessage(content="[内层已运行]")],
            }),
        )

创建中间件

您可以通过两种方式创建中间件：

基于装饰器的中间件

对于单钩子中间件，快速简单。使用装饰器包装单个函数。

基于类的中间件

对于具有多个钩子或配置的复杂中间件更强大。

基于装饰器的中间件

对于单钩子中间件，快速简单。使用装饰器包装单个函数。 可用装饰器： 节点式：

@before_agent - 代理启动前运行（每次调用一次）
@before_model - 每次模型调用前运行
@after_model - 每次模型响应后运行
@after_agent - 代理完成后运行（每次调用一次）

包装式：

@wrap_model_call - 用自定义逻辑包装每次模型调用
@wrap_tool_call - 用自定义逻辑包装每次工具调用

便捷：

@dynamic_prompt - 生成动态系统提示

示例：

from langchain.agents.middleware import (
    before_model,
    wrap_model_call,
    AgentState,
    ModelRequest,
    ModelResponse,
)
from langchain.agents import create_agent
from langgraph.runtime import Runtime
from typing import Any, Callable


@before_model
def log_before_model(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    print(f"即将使用 {len(state['messages'])} 条消息调用模型")
    return None

@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"重试 {attempt + 1}/3，发生错误: {e}")

agent = create_agent(
    model="gpt-5.5",
    middleware=[log_before_model, retry_model],
    tools=[...],
)

何时使用装饰器：

需要单个钩子
无复杂配置
快速原型设计

基于类的中间件

对于具有多个钩子或配置的复杂中间件更强大。当您需要为同一个钩子定义同步和异步实现，或者想要将多个钩子组合在一个中间件中时，请使用类。 python 一个 AgentMiddleware 子类可以声明三个类属性，代理工厂在编译时会拾取这些属性：

state_schema — 使用自定义字段扩展代理状态。参见自定义状态模式。
tools — 注册随中间件一起提供的额外工具（例如，待办事项列表中间件上的 write_todos）。
transformers — 注册作用域感知的流转换器工厂。参见自定义流转换器。 :::

示例：

from langchain.agents.middleware import (
    AgentMiddleware,
    AgentState,
    ModelRequest,
    ModelResponse,
)
from langgraph.runtime import Runtime
from typing import Any, Callable

class LoggingMiddleware(AgentMiddleware):
    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        print(f"即将使用 {len(state['messages'])} 条消息调用模型")
        return None

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

    async def abefore_model(
        self, state: AgentState, runtime: Runtime
    ) -> dict[str, Any] | None:
        # before_model 的异步版本
        return None

    async def aafter_model(
        self, state: AgentState, runtime: Runtime
    ) -> dict[str, Any] | None:
        # after_model 的异步版本
        print(f"模型返回了: {state['messages'][-1].content}")
        return None


agent = create_agent(
    model="gpt-5.5",
    middleware=[LoggingMiddleware()],
    tools=[...],
)

何时使用类：

为同一个钩子定义同步和异步实现
单个中间件需要多个钩子
需要复杂配置（例如，可配置阈值、自定义模型）
通过初始化时配置在项目间重用

:::

自定义状态模式

如果您的中间件需要在钩子之间跟踪状态，中间件可以使用自定义属性扩展代理的状态。这使得中间件能够：

跨执行跟踪状态：维护计数器、标志或其他在代理执行生命周期内持续存在的值
在钩子之间共享数据：将信息从 before_model 传递到 after_model 或在不同的中间件实例之间传递
实现横切关注点：添加速率限制、使用跟踪、用户上下文或审计日志等功能，而无需修改核心代理逻辑
做出条件决策：使用累积的状态来确定是否继续执行、跳转到不同的节点或动态修改行为

装饰器
类

from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langchain.agents.middleware import AgentState, before_model, after_model
from typing_extensions import NotRequired
from typing import Any
from langgraph.runtime import Runtime


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


@before_model(state_schema=CustomState, can_jump_to=["end"])
def check_call_limit(state: CustomState, runtime: Runtime) -> dict[str, Any] | None:
    count = state.get("model_call_count", 0)
    if count > 10:
        return {"jump_to": "end"}
    return None


@after_model(state_schema=CustomState)
def increment_counter(state: CustomState, runtime: Runtime) -> dict[str, Any] | None:
    return {"model_call_count": state.get("model_call_count", 0) + 1}


agent = create_agent(
    model="gpt-5.5",
    middleware=[check_call_limit, increment_counter],
    tools=[],
)

# 使用自定义状态调用
result = agent.invoke({
    "messages": [HumanMessage("Hello")],
    "model_call_count": 0,
    "user_id": "user-123",
})

from langchain.agents import create_agent
from langchain.messages import HumanMessage
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:
        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:
        return {"model_call_count": state.get("model_call_count", 0) + 1}


agent = create_agent(
    model="gpt-5.5",
    middleware=[CallCounterMiddleware()],
    tools=[],
)

# 使用自定义状态调用
result = agent.invoke({
    "messages": [HumanMessage("Hello")],
    "model_call_count": 0,
    "user_id": "user-123",
})

自定义流转换器

中间件注册的转换器需要 langchain>=1.3.2。

中间件可以注册流转换器工厂，将实时代理流中的事件投影到类型化的扩展通道上。这对于暴露计数器、侧信道工件、部分输出或线路级脱敏非常有用，而无需耦合到框架的内置投影。在编译时，中间件注册的工厂与调用者直接传递给代理工厂的任何内容合并。最终排序规则将内置的 ToolCallTransformer 保持在前面，并让调用者提供的条目落在最后。将 transformers 类属性设置为工厂可调用对象的元组。每个工厂的形状为 Callable[[tuple[str, ...]], StreamTransformer]，并作为 factory(scope) 调用，其中 scope 是 mini-mux 作用域元组（() 表示根，非空表示子图）；每次调用返回一个新的转换器可保持每个子图隔离。

from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware


class ToolActivityMiddleware(AgentMiddleware):
    transformers = (ToolActivityTransformer,)


agent = create_agent(
    model="gpt-5-nano",
    tools=[...],
    middleware=[ToolActivityMiddleware()],
)

有关完整的排序规则和 PII 脱敏示例，请参阅在中间件上注册转换器。

执行顺序

使用多个中间件时，了解它们的执行方式：

agent = create_agent(
    model="gpt-5.5",
    middleware=[middleware1, middleware2, middleware3],
    tools=[...],
)

执行流程

前置钩子按顺序运行：

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 的字典： 可用的跳转目标：

'end': 跳转到代理执行结束（或第一个 after_agent 钩子）
'tools': 跳转到工具节点
'model': 跳转到模型节点（或第一个 before_model 钩子）

装饰器
类

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


@after_model
@hook_config(can_jump_to=["end"])
def check_for_blocked(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    last_message = state["messages"][-1]
    if "BLOCKED" in last_message.content:
        return {
            "messages": [AIMessage("我无法回应那个请求。")],
            "jump_to": "end"
        }
    return None

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

class BlockedContentMiddleware(AgentMiddleware):
    @hook_config(can_jump_to=["end"])
    def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        last_message = state["messages"][-1]
        if "BLOCKED" in last_message.content:
            return {
                "messages": [AIMessage("我无法回应那个请求。")],
                "jump_to": "end"
            }
        return None

最佳实践

保持中间件专注 - 每个中间件应做好一件事
优雅地处理错误 - 不要让中间件错误导致代理崩溃
使用合适的钩子类型：
- 节点式用于顺序逻辑（日志记录、验证）
- 包装式用于控制流（重试、回退、缓存）
清晰地记录任何自定义状态属性
在集成之前独立测试中间件
考虑执行顺序 - 将关键中间件放在列表的前面
尽可能使用内置中间件

示例

动态提示

在运行时动态修改系统提示，以便在每次模型调用前注入上下文、用户特定指令或其他信息。这是最常见的中间件用例之一。使用 ModelRequest 上的 system_message 字段来读取和修改系统提示。它包含一个 SystemMessage 对象（即使代理是使用字符串 system_prompt 创建的）。

装饰器
类

from collections.abc import Callable

from langchain.agents.middleware import ModelRequest, ModelResponse, wrap_model_call
from langchain.messages import SystemMessage


@wrap_model_call
def add_context(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    new_content = list(request.system_message.content_blocks) + [
        {"type": "text", "text": "Additional context."}
    ]
    new_system_message = SystemMessage(content=new_content)
    return handler(request.override(system_message=new_system_message))

from collections.abc import Callable

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


class ContextMiddleware(AgentMiddleware):
    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse:
        new_content = list(request.system_message.content_blocks) + [
            {"type": "text", "text": "Additional context."}
        ]
        new_system_message = SystemMessage(content=new_content)
        return handler(request.override(system_message=new_system_message))

ModelRequest.system_message 始终是一个 SystemMessage 对象，即使代理是使用 system_prompt="string" 创建的
使用 SystemMessage.content_blocks 以块列表的形式访问内容，无论原始内容是字符串还是列表
修改系统消息时，使用 content_blocks 并附加新块以保留现有结构
您可以直接将 SystemMessage 对象传递给 create_agent 的 system_prompt 参数，用于高级用例，如缓存控制

动态模型选择

装饰器
类

from collections.abc import Callable

from langchain.agents.middleware import ModelRequest, ModelResponse, wrap_model_call
from langchain.chat_models import init_chat_model

complex_model = init_chat_model("claude-sonnet-4-6")
simple_model = init_chat_model("claude-haiku-4-5-20251001")


@wrap_model_call
def dynamic_model(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    if len(request.messages) > 10:
        model = complex_model
    else:
        model = simple_model
    return handler(request.override(model=model))

from collections.abc import Callable

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

complex_model = init_chat_model("claude-sonnet-4-6")
simple_model = init_chat_model("claude-haiku-4-5-20251001")


class DynamicModelMiddleware(AgentMiddleware):
    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse:
        if len(request.messages) > 10:
            model = complex_model
        else:
            model = simple_model
        return handler(request.override(model=model))

动态选择工具

在运行时选择相关工具以提高性能和准确性。本节涵盖过滤预注册工具。有关注册在运行时发现的工具（例如来自 MCP 服务器），请参阅运行时工具注册。 好处：

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

装饰器
类

from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable


@wrap_model_call
def select_tools(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    """基于状态/上下文选择相关工具的中间件。"""
    # 基于状态/上下文选择一小部分相关的工具
    relevant_tools = select_relevant_tools(request.state, request.runtime)
    return handler(request.override(tools=relevant_tools))

agent = create_agent(
    model="gpt-5.5",
    tools=all_tools,  # 所有可用工具需要预先注册
    middleware=[select_tools],
)

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


class ToolSelectorMiddleware(AgentMiddleware):
    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse:
        """基于状态/上下文选择相关工具的中间件。"""
        # 基于状态/上下文选择一小部分相关的工具
        relevant_tools = select_relevant_tools(request.state, request.runtime)
        return handler(request.override(tools=relevant_tools))

agent = create_agent(
    model="gpt-5.5",
    tools=all_tools,  # 所有可用工具需要预先注册
    middleware=[ToolSelectorMiddleware()],
)

工具调用监控

装饰器
类

from collections.abc import Callable

from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage
from langchain.tools.tool_node import ToolCallRequest
from langgraph.types import Command


@wrap_tool_call
def monitor_tool(
    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("Tool completed successfully")
        return result
    except Exception as e:
        print(f"Tool failed: {e}")
        raise

from collections.abc import Callable

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


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("Tool completed successfully")
            return result
        except Exception as e:
            print(f"Tool failed: {e}")
            raise

提示缓存 (Anthropic)

使用 Anthropic 模型时，使用带有缓存控制指令的结构化内容块来缓存大型系统提示：

装饰器
类

from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from langchain.messages import SystemMessage
from typing import Callable


@wrap_model_call
def add_cached_context(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    # 始终使用内容块
    new_content = list(request.system_message.content_blocks) + [
        {
            "type": "text",
            "text": "Here is a large document to analyze:\n\n<document>...</document>",
            # 此点之前的内容将被缓存
            "cache_control": {"type": "ephemeral"}
        }
    ]

    new_system_message = SystemMessage(content=new_content)
    return handler(request.override(system_message=new_system_message))

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


class CachedContextMiddleware(AgentMiddleware):
    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse:
        # 始终使用内容块
        new_content = list(request.system_message.content_blocks) + [
            {
                "type": "text",
                "text": "Here is a large document to analyze:\n\n<document>...</document>",
                "cache_control": {"type": "ephemeral"}  # 此内容将被缓存
            }
        ]

        new_system_message = SystemMessage(content=new_content)
        return handler(request.override(system_message=new_system_message))

注意事项：

ModelRequest.system_message 始终是一个 SystemMessage 对象，即使代理是使用 system_prompt="string" 创建的
使用 SystemMessage.content_blocks 以块列表的形式访问内容，无论原始内容是字符串还是列表
修改系统消息时，使用 content_blocks 并附加新块以保留现有结构
您可以直接将 SystemMessage 对象传递给 create_agent 的 system_prompt 参数，用于高级用例，如缓存控制

:::

其他资源

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

Get started

Core components

Middleware

Frontend

Advanced usage

Agent development

Production

钩子

节点式钩子

包装式钩子

节点式钩子

包装式钩子

状态更新

节点式钩子

包装式钩子

多个中间件的组合

创建中间件

基于装饰器的中间件

基于类的中间件

基于装饰器的中间件

基于类的中间件

自定义状态模式

自定义流转换器

执行顺序

代理跳转

最佳实践

示例

动态提示

动态模型选择

动态选择工具

工具调用监控

提示缓存 (Anthropic)

其他资源

​钩子

节点式钩子

包装式钩子

​节点式钩子

​包装式钩子

​状态更新

​节点式钩子

​包装式钩子

​多个中间件的组合

​创建中间件

基于装饰器的中间件

基于类的中间件

​基于装饰器的中间件

​基于类的中间件

​自定义状态模式

​自定义流转换器

​执行顺序

​代理跳转

​最佳实践

​示例

​动态提示

​动态模型选择

​动态选择工具

​工具调用监控

​提示缓存 (Anthropic)

​其他资源

钩子

节点式钩子

包装式钩子

状态更新

节点式钩子

包装式钩子

多个中间件的组合

创建中间件

基于装饰器的中间件

基于类的中间件

自定义状态模式

自定义流转换器

执行顺序

代理跳转

最佳实践

示例

动态提示

动态模型选择

动态选择工具

工具调用监控

提示缓存 (Anthropic)

其他资源