> ## Documentation Index
> Fetch the complete documentation index at: https://cndoc-langchain.site/llms.txt
> Use this file to discover all available pages before exploring further.
# 自定义中间件
通过实现在代理执行流程特定点运行的钩子来构建自定义中间件。
## 钩子
中间件提供两种风格的钩子来拦截代理执行：
在特定执行点按顺序运行。
在每次模型或工具调用前后运行。
### 节点式钩子
在特定执行点按顺序运行。用于日志记录、验证和状态更新。
选择你的中间件所需的钩子。你可以在节点式钩子和包装式钩子之间进行选择。
**节点式钩子**在特定执行点运行：
| 钩子 | 运行时机 |
| -------------- | ------------- |
| `before_agent` | 代理启动前（每次调用一次） |
| `before_model` | 每次模型调用前 |
| `after_model` | 每次模型响应后 |
| `after_agent` | 代理完成后（每次调用一次） |
**包装式钩子**在每次调用前后运行，让你控制执行：
| 钩子 | 运行时机 |
| ----------------- | --------- |
| `wrap_model_call` | 在每次模型调用前后 |
| `wrap_tool_call` | 在每次工具调用前后 |
**示例：**
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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
```
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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` - 在每次工具调用前后
**示例：**
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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}")
```
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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`](https://reference.langchain.com/python/langgraph/types/Command) 的 [`ExtendedModelResponse`](https://reference.langchain.com/python/langchain/agents/middleware/types/ExtendedModelResponse)，以便在模型响应旁注入状态更新。对于工具调用，直接返回一个 [`Command`](https://reference.langchain.com/python/langgraph/types/Command)。当你需要基于模型或工具调用期间运行的逻辑来跟踪或更新状态时使用这些，例如摘要触发点、使用元数据或从请求或响应计算的自定义字段。
### 节点式钩子
从节点式钩子返回一个字典以将更新合并到代理状态。字典键映射到状态字段。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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`](https://reference.langchain.com/python/langgraph/types/Command) 的 [`ExtendedModelResponse`](https://reference.langchain.com/python/langchain/agents/middleware/types/ExtendedModelResponse)，以从模型调用层注入状态更新：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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`](https://reference.langchain.com/python/langgraph/types/Command) 通过图的归约器流动，因此更新被正确应用，并且消息是累加的，而不是替换现有状态。
#### 与多个中间件组合
当多个中间件层返回 `ExtendedModelResponse` 时，它们的命令会组合：
* **命令通过归约器应用：** 每个 `Command` 成为一个单独的状态更新。对于消息，这意味着它们是累加的。
* **外部在冲突时获胜：** 对于非归约器状态字段，命令按从内到外的顺序应用。最外层中间件的值在冲突键上优先。
* **重试安全：** 如果外部中间件实现了可能导致多次调用 `handler()` 的逻辑（例如重试逻辑），则来自早期调用的命令将被丢弃。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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 使用最后获胜（外部获胜），messages 使用累加归约器。"""
# 非归约器字段，最后获胜：两个中间件都写入；最外层值获胜
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 和消息。外部添加到相同的键；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`](https://reference.langchain.com/python/langchain/agents/middleware/types/before_agent) - 在代理启动前运行（每次调用一次）
* [`@before_model`](https://reference.langchain.com/python/langchain/agents/middleware/types/before_model) - 在每次模型调用前运行
* [`@after_model`](https://reference.langchain.com/python/langchain/agents/middleware/types/after_model) - 在每次模型响应后运行
* [`@after_agent`](https://reference.langchain.com/python/langchain/agents/middleware/types/after_agent) - 在代理完成后运行（每次调用一次）
**包装式：**
* [`@wrap_model_call`](https://reference.langchain.com/python/langchain/agents/middleware/types/wrap_model_call) - 用自定义逻辑包装每次模型调用
* [`@wrap_tool_call`](https://reference.langchain.com/python/langchain/agents/middleware/types/wrap_tool_call) - 用自定义逻辑包装每次工具调用
**便捷：**
* [`@dynamic_prompt`](https://reference.langchain.com/python/langchain/agents/middleware/types/dynamic_prompt) - 生成动态系统提示
**示例：**
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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.4",
middleware=[log_before_model, retry_model],
tools=[...],
)
```
**何时使用装饰器：**
* 需要单个钩子
* 无复杂配置
* 快速原型设计
### 基于类的中间件
对于具有多个钩子或配置的复杂中间件更强大。当你需要为同一个钩子定义同步和异步实现，或者想在单个中间件中组合多个钩子时使用类。
**示例：**
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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.4",
middleware=[LoggingMiddleware()],
tools=[...],
)
```
**何时使用类：**
* 为同一个钩子定义同步和异步实现
* 单个中间件中需要多个钩子
* 需要复杂配置（例如，可配置阈值、自定义模型）
* 通过初始化时配置在项目间复用
## 自定义状态模式
如果你的中间件需要跨钩子跟踪状态，中间件可以用自定义属性扩展代理的状态。这使得中间件能够：
* **跨执行跟踪状态**：维护计数器、标志或其他在整个代理执行生命周期中持久化的值
* **在钩子之间共享数据**：将信息从 `before_model` 传递到 `after_model` 或在不同中间件实例之间传递
* **实现横切关注点**：添加速率限制、使用跟踪、用户上下文或审计日志等功能，而无需修改核心代理逻辑
* **做出条件决策**：使用累积状态来确定是否继续执行、跳转到不同节点或动态修改行为
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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.4",
middleware=[check_call_limit, increment_counter],
tools=[],
)
# 使用自定义状态调用
result = agent.invoke({
"messages": [HumanMessage("你好")],
"model_call_count": 0,
"user_id": "user-123",
})
```
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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.4",
middleware=[CallCounterMiddleware()],
tools=[],
)
# 使用自定义状态调用
result = agent.invoke({
"messages": [HumanMessage("你好")],
"model_call_count": 0,
"user_id": "user-123",
})
```
## 执行顺序
使用多个中间件时，了解它们的执行方式：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
agent = create_agent(
model="gpt-5.4",
middleware=[middleware1, middleware2, middleware3],
tools=[...],
)
```
**前置钩子按顺序运行：**
1. `middleware1.before_agent()`
2. `middleware2.before_agent()`
3. `middleware3.before_agent()`
**代理循环开始**
4. `middleware1.before_model()`
5. `middleware2.before_model()`
6. `middleware3.before_model()`
**包装钩子像函数调用一样嵌套：**
7. `middleware1.wrap_model_call()` → `middleware2.wrap_model_call()` → `middleware3.wrap_model_call()` → 模型
**后置钩子按相反顺序运行：**
8. `middleware3.after_model()`
9. `middleware2.after_model()`
10. `middleware1.after_model()`
**代理循环结束**
11. `middleware3.after_agent()`
12. `middleware2.after_agent()`
13. `middleware1.after_agent()`
**关键规则：**
* `before_*` 钩子：从先到后
* `after_*` 钩子：从后到先（反向）
* `wrap_*` 钩子：嵌套（第一个中间件包装所有其他中间件）
## 代理跳转
要从中间件提前退出，返回一个包含 `jump_to` 的字典：
**可用跳转目标：**
* `'end'`：跳转到代理执行结束（或第一个 `after_agent` 钩子）
* `'tools'`：跳转到工具节点
* `'model'`：跳转到模型节点（或第一个 `before_model` 钩子）
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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
```
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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
```
## 最佳实践
1. 保持中间件专注 - 每个中间件应做好一件事
2. 优雅地处理错误 - 不要让中间件错误导致代理崩溃
3. **使用适当的钩子类型**：
* 节点式用于顺序逻辑（日志记录、验证）
* 包装式用于控制流（重试、回退、缓存）
4. 清晰地记录任何自定义状态属性
5. 在集成前独立进行单元测试
6. 考虑执行顺序 - 将关键中间件放在列表前面
7. 尽可能使用内置中间件
## 示例
### 动态提示
在运行时动态修改系统提示，以便在每次模型调用前注入上下文、用户特定指令或其他信息。这是最常见的中间件用例之一。
使用 `ModelRequest` 上的 `system_message` 字段来读取和修改系统提示。它包含一个 [`SystemMessage`](https://reference.langchain.com/python/langchain-core/messages/system/SystemMessage) 对象（即使代理是用字符串 `system_prompt` 创建的）。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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": "附加上下文。"}
]
new_system_message = SystemMessage(content=new_content)
return handler(request.override(system_message=new_system_message))
```
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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`](https://reference.langchain.com/python/langchain-core/messages/system/SystemMessage) 对象，即使代理是用 `system_prompt="string"` 创建的
* 使用 `SystemMessage.content_blocks` 将内容作为块列表访问，无论原始内容是字符串还是列表
* 修改系统消息时，使用 `content_blocks` 并追加新块以保留现有结构
* 你可以将 [`SystemMessage`](https://reference.langchain.com/python/langchain-core/messages/system/SystemMessage) 对象直接传递给 `create_agent` 的 `system_prompt` 参数，用于缓存控制等高级用例
### 动态模型选择
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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))
```
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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 服务器），请参阅[运行时工具注册](/oss/python/langchain/agents#dynamic-tools)。
**好处：**
* **更短的提示** - 通过仅暴露相关工具来降低复杂性
* **更好的准确性** - 模型从更少的选项中正确选择
* **权限控制** - 基于用户访问动态过滤工具
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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.4",
tools=all_tools, # 所有可用工具需要预先注册
middleware=[select_tools],
)
```
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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.4",
tools=all_tools, # 所有可用工具需要预先注册
middleware=[ToolSelectorMiddleware()],
)
```
### 工具调用监控
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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"正在执行工具: {request.tool_call['name']}")
print(f"参数: {request.tool_call['args']}")
try:
result = handler(request)
print("工具执行成功")
return result
except Exception as e:
print(f"工具执行失败: {e}")
raise
```
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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"执行工具: {request.tool_call['name']}")
print(f"参数: {request.tool_call['args']}")
try:
result = handler(request)
print("工具执行成功")
return result
except Exception as e:
print(f"工具执行失败: {e}")
raise
```
### 提示缓存（Anthropic）
使用 Anthropic 模型时，使用带有缓存控制指令的结构化内容块来缓存大型系统提示：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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": "这是一个需要分析的大型文档：\n\n...",
# 直到此处的内容被缓存
"cache_control": {"type": "ephemeral"}
}
]
new_system_message = SystemMessage(content=new_content)
return handler(request.override(system_message=new_system_message))
```
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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": "这是一个需要分析的大型文档：\n\n...",
"cache_control": {"type": "ephemeral"} # 此内容将被缓存
}
]
new_system_message = SystemMessage(content=new_content)
return handler(request.override(system_message=new_system_message))
```
**注意：**
* `ModelRequest.system_message` 始终是一个 [`SystemMessage`](https://reference.langchain.com/python/langchain-core/messages/system/SystemMessage) 对象，即使代理是用 `system_prompt="string"` 创建的
* 使用 `SystemMessage.content_blocks` 将内容作为块列表访问，无论原始内容是字符串还是列表
* 修改系统消息时，使用 `content_blocks` 并追加新块以保留现有结构
* 你可以将 [`SystemMessage`](https://reference.langchain.com/python/langchain-core/messages/system/SystemMessage) 对象直接传递给 `create_agent` 的 `system_prompt` 参数，用于缓存控制等高级用例
:::
## 附加资源
* [中间件 API 参考](https://reference.langchain.com/python/langchain/middleware/)
* [内置中间件](/oss/python/langchain/middleware/built-in)
* [测试代理](/oss/python/langchain/test/)
***
[通过 MCP 将这些文档](/use-these-docs)连接到 Claude、VSCode 等，以获取实时答案。
[在 GitHub 上编辑此页面](https://github.com/langchain-ai/docs/edit/main/src/oss/langchain/middleware/custom.mdx) 或 [提交问题](https://github.com/langchain-ai/docs/issues/new/choose)。