LangChain 中文文档

外观

通过实现智能体执行流程中特定节点运行的钩子来构建自定义中间件。

钩子 ​

中间件提供两种风格的钩子来拦截智能体执行：

节点式钩子 ​

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

选择你的中间件所需的钩子。你可以在节点式钩子和包裹式钩子之间选择。

节点式钩子在特定执行节点运行：

HookWhen it runs
before_agent在智能体开始前（每次调用一次）
before_model在每次模型调用前
after_model在每次模型响应后
after_agent在智能体完成后（每次调用一次）
HookWhen it runs
beforeAgent在智能体开始前（每次调用一次）
beforeModel在每次模型调用前
afterModel在每次模型响应后
afterAgent在智能体完成后（每次调用一次）

包裹式钩子围绕每次调用运行，让你能够控制执行：

HookWhen it runs
wrap_model_call围绕每次模型调用
wrap_tool_call围绕每次工具调用
HookWhen it runs
wrapModelCall围绕每次模型调用
wrapToolCall围绕每次工具调用

示例：

装饰器 ​

python
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("Conversation limit reached.")],
            "jump_to": "end"
        }
    return None

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

类 ​

python
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("Conversation limit reached.")],
                "jump_to": "end"
            }
        return None

    def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        print(f"Model returned: {state['messages'][-1].content}")
        return None
typescript
import { createMiddleware, AIMessage } from "langchain";

const createMessageLimitMiddleware = (maxMessages: number = 50) => {
  return createMiddleware({
    name: "MessageLimitMiddleware",
    beforeModel: {
      canJumpTo: ["end"],
      hook: (state) => {
        if (state.messages.length === maxMessages) {
          return {
            messages: [new AIMessage("Conversation limit reached.")],
            jumpTo: "end",
          };
        }
        return;
      }
    },
    afterModel: (state) => {
      const lastMessage = state.messages[state.messages.length - 1];
      console.log(`Model returned: ${lastMessage.content}`);
      return;
    },
  });
};

包裹式钩子 ​

拦截执行并控制处理函数被调用的时机。用于重试、缓存和转换。

由你决定处理函数被调用零次（短路）、一次（正常流程）还是多次（重试逻辑）。

可用的钩子：

  • wrap_model_call - 围绕每次模型调用

  • wrap_tool_call - 围绕每次工具调用

  • wrapModelCall - 围绕每次模型调用

  • wrapToolCall - 围绕每次工具调用

示例：

装饰器 ​

python
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"Retry {attempt + 1}/3 after error: {e}")

类 ​

python
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}")
typescript
import { createMiddleware } from "langchain";

const createRetryMiddleware = (maxRetries: number = 3) => {
  return createMiddleware({
    name: "RetryMiddleware",
    wrapModelCall: (request, handler) => {
      for (let attempt = 0; attempt < maxRetries; attempt++) {
        try {
          return handler(request);
        } catch (e) {
          if (attempt === maxRetries - 1) {
            throw e;
          }
          console.log(`Retry ${attempt + 1}/${maxRetries} after error: ${e}`);
        }
      }
      throw new Error("Unreachable");
    },
  });
};

状态更新 ​

节点式钩子和包裹式钩子都可以更新智能体状态。机制有所不同：

  • 节点式钩子（before_agent、before_model、after_model、after_agent）：直接返回一个字典。该字典通过图的 reducer 应用到智能体状态。

  • 包裹式钩子（wrap_model_call、wrap_tool_call）：对于模型调用，返回带 Command 的 ExtendedModelResponse，以便随模型响应注入状态更新。对于工具调用，直接返回 Command。当你需要根据模型或工具调用期间运行的逻辑来追踪或更新状态时（例如摘要触发点、用量元数据，或从请求或响应计算出的自定义字段），请使用这些钩子。

  • 节点式钩子（beforeAgent、beforeModel、afterModel、afterAgent）：直接返回一个字典。该字典通过图的 reducer 应用到智能体状态。

  • 包裹式钩子（wrapModelCall、wrapToolCall）：对于模型调用，返回带 Command 的 ExtendedModelResponse，以便随模型响应注入状态更新。对于工具调用，直接返回 Command。当你需要根据模型或工具调用期间运行的逻辑来追踪或更新状态时（例如摘要触发点、用量元数据，或从请求或响应计算出的自定义字段），请使用这些钩子。

节点式钩子 ​

从节点式钩子返回一个字典，将更新合并到智能体状态。字典的键对应状态字段。

python
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}
typescript
import { createMiddleware } from "langchain";
import * as z from "zod";

const trackingStateSchema = z.object({
  modelCallCount: z.number().default(0),
});

const incrementAfterModel = createMiddleware({
  name: "incrementAfterModel",
  stateSchema: trackingStateSchema,
  afterModel: (state) => {
    return { modelCallCount: state.modelCallCount + 1 };
  },
});

包裹式钩子 ​

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

python
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):
    """Agent state with token usage tracking."""

    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}),
    )

直接从 wrapModelCall 返回 Command，以便从模型调用层注入状态更新：

typescript
import * as z from "zod";
import { createMiddleware } from "langchain";
import { Command } from "@langchain/langgraph";

const usageTrackingStateSchema = z.object({
  lastModelCallTokens: z.number().optional(),
});

const trackUsage = createMiddleware({
  name: "trackUsage",
  stateSchema: usageTrackingStateSchema,
  wrapModelCall: async (request, handler) => {
    const response = await handler(request);
    return new Command({ update: { lastModelCallTokens: 150 } });
  },
});

Command 会流经图的 reducer，因此更新会被正确应用，消息是累加的，而不是替换现有状态。

多个中间件的组合 ​

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

  • Commands 通过 reducer 应用： 每个 Command 成为一次独立的状态更新。对于消息而言，这意味着它们是累加的。
  • 冲突时外层优先： 对于非 reducer 状态字段，command 先应用内层，再应用外层。在键冲突时，最外层中间件的值优先。
  • 重试安全： 如果外层中间件实现了可能导致多次调用 handler() 的逻辑（例如重试逻辑），则较早调用的 command 会被丢弃。

当多个中间件层返回响应时，框架会传递最终生成的 AIMessage：

  • AIMessage 逐层传递： 每个中间件的 handler() 接收来自前一层的 AIMessage。当某个中间件返回 AIMessage 时，它就成为下一个中间件 handler 的输入。
  • 不更新消息的 Command 是透传： 如果中间件返回的 Command 其状态更新不涉及 messages，框架会将它在消息流中视为无操作。下一个中间件的 handler 接收来自返回该 Command 的中间件之前的那个中间件的 AIMessage。
  • Reducer 行为与重试安全： Commands 仍然通过 reducer 应用（消息累加，冲突时外层优先）。重试逻辑会丢弃较早调用的 command。
python
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:
    """Reducer: last writer wins (outer overwrites inner)."""
    return b

class CustomMiddlewareState(AgentState):
    """Agent state: trace_layer uses last-wins (outer wins), messages use additive reducer."""

    # 非 reducer 字段，采用后写覆盖：两个中间件都会写入；最外层的值胜出
    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="[Outer ran]")],
            }),
        )

class InnerMiddleware(AgentMiddleware):
    """Adds trace_layer and message. Outer adds to same keys; trace_layer: outer wins, messages: additive."""

    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="[Inner ran]")],
            }),
        )
typescript
import * as z from "zod";
import { createMiddleware } from "langchain";
import { Command, StateSchema, ReducedValue } from "@langchain/langgraph";
import { AIMessage, SystemMessage } from "@langchain/core/messages";

/** 后写覆盖的 reducer：两个中间件都写入时，外层覆盖内层。 */
const customMiddlewareStateSchema = new StateSchema({
  traceLayer: new ReducedValue(
    z.string().optional(),
    { reducer: (a, b) => b },
  ),
});

const outerMiddleware = createMiddleware({
  name: "OuterMiddleware",
  stateSchema: customMiddlewareStateSchema,
  wrapModelCall: async (_request, handler) => {
    await handler(_request);
    return new Command({
      update: {
        traceLayer: "outer",
        messages: [new SystemMessage({ content: "[Outer ran]" })],
      },
    });
  },
});

const innerMiddleware = createMiddleware({
  name: "InnerMiddleware",
  stateSchema: customMiddlewareStateSchema,
  wrapModelCall: async (_request, handler) => {
    await handler(_request);
    return new Command({
      update: {
        traceLayer: "inner",
        messages: [new SystemMessage({ content: "[Inner ran]" })],
      },
    });
  },
});

创建中间件 ​

你可以用两种方式创建中间件：

基于装饰器的中间件 ​

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

可用的装饰器：

节点式：

  • @before_agent - 在智能体开始前运行（每次调用一次）
  • @before_model - 在每次模型调用前运行
  • @after_model - 在每次模型响应后运行
  • @after_agent - 在智能体完成后运行（每次调用一次）

包裹式：

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

便捷：

  • @dynamic_prompt - 生成动态系统提示词

示例：

python
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"About to call model with {len(state['messages'])} 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"Retry {attempt + 1}/3 after error: {e}")

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

何时使用装饰器：

  • 只需要单个钩子
  • 不需要复杂的配置
  • 快速原型开发

基于类的中间件 ​

对于包含多个钩子或配置的复杂中间件更强大。当你需要为同一个钩子同时定义同步和异步实现，或者想在单个中间件中组合多个钩子时，请使用类。

AgentMiddleware 子类可以声明三个类属性，智能体工厂会在编译时获取：

  • state_schema — 用自定义字段扩展智能体状态。参见 自定义状态 schema。
  • tools — 注册随中间件附带的其他工具（例如待办事项列表中间件上的 write_todos）。
  • transformers — 注册感知作用域的流式转换器工厂。参见 自定义流式转换器。

createMiddleware 接受三个配置字段，智能体工厂会在编译时获取：

  • stateSchema — 用自定义字段扩展智能体状态。参见 自定义状态 schema。
  • tools — 注册随中间件附带的其他工具。
  • streamTransformers — 注册感知作用域的流式转换器工厂。参见 自定义流式转换器。

示例：

python
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"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

    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"Model returned: {state['messages'][-1].content}")
        return None

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

何时使用类：

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

使用 createMiddleware 函数定义自定义中间件：

typescript
import { createMiddleware } from "langchain";

const loggingMiddleware = createMiddleware({
  name: "LoggingMiddleware",
  beforeModel: (state) => {
    console.log(`About to call model with ${state.messages.length} messages`);
    return;
  },
  afterModel: (state) => {
    const lastMessage = state.messages[state.messages.length - 1];
    console.log(`Model returned: ${lastMessage.content}`);
    return;
  },
});

自定义状态 schema ​

如果你的中间件需要跨钩子追踪状态，中间件可以用自定义属性扩展智能体的状态。这使中间件能够：

  • 跨执行追踪状态：维护计数器、标志位或其他在智能体整个执行生命周期中持续存在的值
  • 在钩子之间共享数据：将信息从 before_model 传递给 after_model，或在不同的中间件实例之间传递
  • 在钩子之间共享数据：将信息从 beforeModel 传递给 afterModel，或在不同的中间件实例之间传递
  • 实现横切关注点：在不修改核心智能体逻辑的情况下添加限流、用量追踪、用户上下文或审计日志等功能
  • 做出条件决策：使用累积的状态来决定是继续执行、跳转到不同的节点，还是动态修改行为

装饰器 ​

python
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",
})

类 ​

python
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",
})
typescript
import { createMiddleware, createAgent, HumanMessage } from "langchain";
import { StateSchema } from "@langchain/langgraph";
import * as z from "zod";

const CustomState = new StateSchema({
  modelCallCount: z.number().default(0),
  userId: z.string().optional(),
});

const callCounterMiddleware = createMiddleware({
  name: "CallCounterMiddleware",
  stateSchema: CustomState,
  beforeModel: {
    canJumpTo: ["end"],
    hook: (state) => {
      if (state.modelCallCount > 10) {
        return { jumpTo: "end" };
      }

      return;
    },
  },
  afterModel: (state) => {
    return { modelCallCount: state.modelCallCount + 1 };
  },
});

const agent = createAgent({
  model: "gpt-5.5",
  tools: [...],
  middleware: [callCounterMiddleware],
});

const result = await agent.invoke({
  messages: [new HumanMessage("Hello")],
  modelCallCount: 0,
  userId: "user-123",
});

状态字段可以是公开的或私有的。以下划线（_）开头的字段被视为私有字段，不会包含在智能体的结果中。只返回公开字段（不带前导下划线的字段）。

这对于存储不应暴露给调用方的内部中间件状态很有用，例如临时追踪变量或内部标志位：

typescript
import { StateSchema } from "@langchain/langgraph";
import * as z from "zod";

const PrivateState = new StateSchema({
  // 公共字段 - 包含在调用结果中
  publicCounter: z.number().default(0),
  // 私有字段 - 从调用结果中排除
  _internalFlag: z.boolean().default(false),
});

const middleware = createMiddleware({
  name: "ExampleMiddleware",
  stateSchema: PrivateState,
  afterModel: (state) => {
    // 两个字段在执行期间都可以访问
    if (state._internalFlag) {
      return { publicCounter: state.publicCounter + 1 };
    }
    return { _internalFlag: true };
  },
});

const result = await agent.invoke({
  messages: [new HumanMessage("Hello")],
  publicCounter: 0
});

// 结果只包含 publicCounter，不包含 _internalFlag
console.log(result.publicCounter); // 1
console.log(result._internalFlag); // undefined

自定义流式转换器 ​

INFO

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

INFO

中间件注册的转换器需要 langchain@1.4.3 或更高版本。

中间件可以注册流式转换器工厂，将实时智能体流中的事件映射到类型化的扩展通道上。这对于在不耦合框架内置映射的情况下呈现计数器、侧信道产物、部分输出或线级脱敏非常有用。

在编译时，中间件注册的工厂会与调用方直接传给智能体工厂的任何内容合并。最终排序规则使内置的 ToolCallTransformer 保持在前面，并让调用方提供的条目排在最后。

将 transformers 类属性设置为工厂可调用对象的元组。每个工厂的形式为 Callable[[tuple[str, ...]], StreamTransformer]，并以 factory(scope) 的方式调用，其中 scope 是 mini-mux 作用域元组（根为 ()，子图为非空）；每次调用返回新的转换器可保持每个子图相互隔离。

python
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()],
)

将 streamTransformers 以工厂元组的形式传给 createMiddleware。每个工厂的形式为 () => StreamTransformer<any>（零参数），并在每个作用域调用一次；每次调用返回新的转换器可保持每个子图相互隔离。

typescript
import { createAgent, createMiddleware } from "langchain";

const toolActivityMiddleware = createMiddleware({
  name: "ToolActivityMiddleware",
  streamTransformers: [toolActivityTransformer],
});

const agent = createAgent({
  model: "gpt-5-nano",
  tools: [...],
  middleware: [toolActivityMiddleware],
});

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

自定义上下文 ​

中间件可以定义自定义上下文 schema 来访问每次调用的元数据。与状态不同，上下文是只读的，不会在调用之间持久化。因此它非常适合用于：

  • 用户信息：传递在执行过程中不会变化的用户 ID、角色或偏好
  • 配置覆盖：提供限流或功能开关等每次调用的设置
  • 租户/工作区上下文：为多租户应用包含组织特定数据
  • 请求元数据：传递中间件所需的请求 ID、API 密钥或其他元数据

使用 Zod 定义上下文 schema，并在中间件钩子中通过 runtime.context 访问它。上下文 schema 中的必填字段会在 TypeScript 层面强制执行，确保你在调用 agent.invoke() 时必须提供它们。

typescript
import { createAgent, createMiddleware, HumanMessage } from "langchain";
import * as z from "zod";

const contextSchema = z.object({
  userId: z.string(),
  tenantId: z.string(),
  apiKey: z.string().optional(),
});

const userContextMiddleware = createMiddleware({
  name: "UserContextMiddleware",
  contextSchema,
  wrapModelCall: (request, handler) => {
    // 从运行时访问上下文
    const { userId, tenantId } = request.runtime.context;

    // 将用户上下文添加到系统消息中
    const contextText = `User ID: ${userId}, Tenant: ${tenantId}`;
    const newSystemMessage = request.systemMessage.concat(contextText);

    return handler({
      ...request,
      systemMessage: newSystemMessage,
    });
  },
});

const agent = createAgent({
  model: "gpt-5.5",
  middleware: [userContextMiddleware],
  tools: [],
  contextSchema,
});

const result = await agent.invoke(
  { messages: [new HumanMessage("Hello")] },
  // 必须提供必填字段（userId、tenantId）
  {
    context: {
      userId: "user-123",
      tenantId: "acme-corp",
    },
  }
);

必填上下文字段：当你在 contextSchema 中定义必填字段（没有 .optional() 或 .default() 的字段）时，TypeScript 会强制要求在 agent.invoke() 调用期间必须提供这些字段。这确保了类型安全，并防止因缺少必填上下文而出现运行时错误。

typescript
// 如果 userId 或 tenantId 缺失，这会导致 TypeScript 错误
const result = await agent.invoke(
  { messages: [new HumanMessage("Hello")] },
  { context: { userId: "user-123" } } // 错误：需要 tenantId
);

执行顺序 ​

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

python
agent = create_agent(
    model="gpt-5.5",
    middleware=[middleware1, middleware2, middleware3],
    tools=[...],
)
typescript
const agent = createAgent({
  model: "gpt-5.5",
  middleware: [middleware1, middleware2, middleware3],
  tools: [...],
});

执行流程 ​

Before 钩子按顺序运行：

  1. middleware1.before_agent()
  2. middleware2.before_agent()
  3. middleware3.before_agent()

智能体循环开始

  1. middleware1.before_model()
  2. middleware2.before_model()
  3. middleware3.before_model()

Wrap 钩子像函数调用一样嵌套：

  1. middleware1.wrap_model_call() → middleware2.wrap_model_call() → middleware3.wrap_model_call() → model

After 钩子按相反顺序运行：

  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 的字典：

可用的跳转目标：

  • 'end'：跳到智能体执行的末尾（或第一个 after_agent 钩子）
  • 'tools'：跳到工具节点
  • 'model'：跳到模型节点（或第一个 before_model 钩子）

装饰器 ​

python
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("I cannot respond to that request.")],
            "jump_to": "end"
        }
    return None

类 ​

python
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("I cannot respond to that request.")],
                "jump_to": "end"
            }
        return None
typescript
import { createAgent, createMiddleware, AIMessage } from "langchain";

const agent = createAgent({
  model: "gpt-5.5",
  middleware: [
    createMiddleware({
      name: "BlockedContentMiddleware",
      beforeModel: {
        canJumpTo: ["end"],
        hook: (state) => {
          if (state.messages.at(-1)?.content.includes("BLOCKED")) {
            return {
              messages: [new AIMessage("I cannot respond to that request.")],
              jumpTo: "end" as const,
            };
          }
          return;
        },
      },
    }),
  ],
});

const result = await agent.invoke({
    messages: "Hello, world! BLOCKED"
});

/**
 * 预期输出：
 * I cannot respond to that request.
 */
console.log(result.messages.at(-1)?.content);

最佳实践 ​

  1. 让中间件保持专注——每个中间件做好一件事
  2. 妥善处理错误——不要让中间件错误导致智能体崩溃
  3. 使用合适的钩子类型：
    • 顺序逻辑使用节点式钩子（日志记录、验证）
    • 控制流使用包裹式钩子（重试、回退、缓存）
  4. 清晰地记录任何自定义状态属性
  5. 在集成之前对中间件进行独立的单元测试
  6. 考虑执行顺序——将关键中间件放在列表前面
  7. 尽可能使用内置中间件

示例 ​

动态提示词 ​

在运行时动态修改系统提示词，以便在每次模型调用前注入上下文、针对用户的指令或其他信息。这是最常见的中间件用例之一。

使用 ModelRequest 上的 system_message 字段来读取和修改系统提示词。它包含一个 SystemMessage 对象（即使智能体是用字符串 system_prompt 创建的）。

装饰器 ​

python
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))

类 ​

python
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))

INFO

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

使用 ModelRequest 中的 systemMessage 字段来读取和修改系统提示词。它包含一个 SystemMessage 对象（即使智能体是用字符串 systemPrompt 创建的）。

ts
import { createMiddleware, SystemMessage, createAgent } from "langchain";

const addContextMiddleware = createMiddleware({
  name: "AddContextMiddleware",
  wrapModelCall: async (request, handler) => {
    return handler({
      ...request,
      systemMessage: request.systemMessage.concat(`Additional context.`),
    });
  },
});

const agent = createAgent({
  model: "google-genai:gemini-3.6-flash",
  systemPrompt: "You are a helpful assistant.",
  middleware: [addContextMiddleware],
});
ts
import { createMiddleware, SystemMessage, createAgent } from "langchain";

const addContextMiddleware = createMiddleware({
  name: "AddContextMiddleware",
  wrapModelCall: async (request, handler) => {
    return handler({
      ...request,
      systemMessage: request.systemMessage.concat(`Additional context.`),
    });
  },
});

const agent = createAgent({
  model: "openai:gpt-5.5",
  systemPrompt: "You are a helpful assistant.",
  middleware: [addContextMiddleware],
});
ts
import { createMiddleware, SystemMessage, createAgent } from "langchain";

const addContextMiddleware = createMiddleware({
  name: "AddContextMiddleware",
  wrapModelCall: async (request, handler) => {
    return handler({
      ...request,
      systemMessage: request.systemMessage.concat(`Additional context.`),
    });
  },
});

const agent = createAgent({
  model: "anthropic:claude-sonnet-4-6",
  systemPrompt: "You are a helpful assistant.",
  middleware: [addContextMiddleware],
});
ts
import { createMiddleware, SystemMessage, createAgent } from "langchain";

const addContextMiddleware = createMiddleware({
  name: "AddContextMiddleware",
  wrapModelCall: async (request, handler) => {
    return handler({
      ...request,
      systemMessage: request.systemMessage.concat(`Additional context.`),
    });
  },
});

const agent = createAgent({
  model: "openrouter:openrouter:z-ai/glm-5.2",
  systemPrompt: "You are a helpful assistant.",
  middleware: [addContextMiddleware],
});
ts
import { createMiddleware, SystemMessage, createAgent } from "langchain";

const addContextMiddleware = createMiddleware({
  name: "AddContextMiddleware",
  wrapModelCall: async (request, handler) => {
    return handler({
      ...request,
      systemMessage: request.systemMessage.concat(`Additional context.`),
    });
  },
});

const agent = createAgent({
  model: "fireworks:accounts/fireworks/models/glm-5p2",
  systemPrompt: "You are a helpful assistant.",
  middleware: [addContextMiddleware],
});
ts
import { createMiddleware, SystemMessage, createAgent } from "langchain";

const addContextMiddleware = createMiddleware({
  name: "AddContextMiddleware",
  wrapModelCall: async (request, handler) => {
    return handler({
      ...request,
      systemMessage: request.systemMessage.concat(`Additional context.`),
    });
  },
});

const agent = createAgent({
  model: "baseten:zai-org/GLM-5.2",
  systemPrompt: "You are a helpful assistant.",
  middleware: [addContextMiddleware],
});
ts
import { createMiddleware, SystemMessage, createAgent } from "langchain";

const addContextMiddleware = createMiddleware({
  name: "AddContextMiddleware",
  wrapModelCall: async (request, handler) => {
    return handler({
      ...request,
      systemMessage: request.systemMessage.concat(`Additional context.`),
    });
  },
});

const agent = createAgent({
  model: "ollama:north-mini-code-1.0",
  systemPrompt: "You are a helpful assistant.",
  middleware: [addContextMiddleware],
});

使用 SystemMessage.concat 保留其他中间件创建的缓存控制元数据或结构化内容块。

动态模型选择 ​

装饰器 ​

python
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
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))
ts
import { createMiddleware, initChatModel } from "langchain";

const models = {
  complex: await initChatModel("claude-sonnet-4-6"),
  simple: await initChatModel("claude-haiku-4-5-20251001"),
};

const dynamicModelMiddleware = createMiddleware({
  name: "DynamicModelMiddleware",
  wrapModelCall: (request, handler) => {
    const modifiedRequest = { ...request };
    if (request.messages.length > 10) {
      modifiedRequest.model = models.complex;
    } else {
      modifiedRequest.model = models.simple;
    }
    return handler(modifiedRequest);
  },
});

动态选择工具 ​

在运行时选择相关工具以提高性能和准确性。本节介绍过滤预注册的工具。关于注册在运行时发现的工具（例如来自 MCP 服务器），参见 运行时工具注册。

好处：

  • 更短的提示词 - 只暴露相关工具，降低复杂度
  • 更高的准确性 - 模型从更少的选项中做出正确选择
  • 权限控制 - 根据用户访问权限动态过滤工具

装饰器 ​

python
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:
    """Middleware to select relevant tools based on state/context."""
    # 根据状态/上下文选择一小部分相关工具
    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],
)

类 ​

python
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:
        """Middleware to select relevant tools based on state/context."""
        # 根据状态/上下文选择一小部分相关工具
        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()],
)
typescript
import { createAgent, createMiddleware } from "langchain";

const toolSelectorMiddleware = createMiddleware({
  name: "ToolSelector",
  wrapModelCall: (request, handler) => {
    // 根据状态/上下文选择一小部分相关工具
    const relevantTools = selectRelevantTools(request.state, request.runtime);
    const modifiedRequest = { ...request, tools: relevantTools };
    return handler(modifiedRequest);
  },
});

const agent = createAgent({
  model: "gpt-5.5",
  tools: allTools,
  middleware: [toolSelectorMiddleware],
});

工具调用监控 ​

装饰器 ​

python
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

类 ​

python
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
ts
import { createMiddleware } from "langchain";

const toolMonitoringMiddleware = createMiddleware({
  name: "ToolMonitoringMiddleware",
  wrapToolCall: (request, handler) => {
    console.log(`Executing tool: ${request.toolCall.name}`);
    console.log(`Arguments: ${JSON.stringify(request.toolCall.args)}`);
    try {
      const result = handler(request);
      console.log("Tool completed successfully");
      return result;
    } catch (e) {
      console.log(`Tool failed: ${e}`);
      throw e;
    }
  },
});

提示词缓存（Anthropic） ​

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

装饰器 ​

python
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))

类 ​

python
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 参数

使用 ModelRequest 中的 systemMessage 字段在中间件中修改系统消息。它包含一个 SystemMessage 对象（即使智能体是用字符串 systemPrompt 创建的）。

示例：中间件链 - 不同的中间件可以使用不同的方法：

typescript
import { createMiddleware, SystemMessage, createAgent } from "langchain";

// 中间件 1：使用 systemMessage 进行简单拼接
const myMiddleware = createMiddleware({
  name: "MyMiddleware",
  wrapModelCall: async (request, handler) => {
    return handler({
      ...request,
      systemMessage: request.systemMessage.concat(`Additional context.`),
    });
  },
});

// 中间件 2：使用 systemMessage 并采用结构化内容（保留结构）
const myOtherMiddleware = createMiddleware({
  name: "MyOtherMiddleware",
  wrapModelCall: async (request, handler) => {
    return handler({
      ...request,
      systemMessage: request.systemMessage.concat(
        new SystemMessage({
          content: [
            {
              type: "text",
              text: " More additional context. This will be cached.",
              cache_control: { type: "ephemeral", ttl: "5m" },
            },
          ],
        })
      ),
    });
  },
});

const agent = createAgent({
  model: "google_genai:gemini-3.6-flash",
  systemPrompt: "You are a helpful assistant.",
  middleware: [myMiddleware, myOtherMiddleware],
});

最终的系统消息将是：

typescript
new SystemMessage({
  content: [
    { type: "text", text: "You are a helpful assistant." },
    { type: "text", text: "Additional context." },
    {
        type: "text",
        text: " More additional context. This will be cached.",
        cache_control: { type: "ephemeral", ttl: "5m" },
    },
  ],
});

使用 SystemMessage.concat 保留其他中间件创建的缓存控制元数据或结构化内容块。

其他资源 ​