自定义中间件 - Docs by LangChain

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

钩子

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

节点式钩子

在特定执行点顺序运行。

包裹式钩子

在每次模型或工具调用前后运行。

节点式钩子

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

钩子	运行时机
`beforeAgent`	智能体开始之前（每次调用一次）
`beforeModel`	每次模型调用之前
`afterModel`	每次模型响应之后
`afterAgent`	智能体完成之后（每次调用一次）

包裹式钩子在每次调用前后运行，让你控制执行：

钩子	运行时机
`wrapModelCall`	在每次模型调用前后
`wrapToolCall`	在每次工具调用前后

示例：

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

包裹式钩子

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

wrapModelCall - 在每次模型调用前后
wrapToolCall - 在每次工具调用前后

示例：

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

状态更新

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

节点式钩子 (beforeAgent, beforeModel, afterModel, afterAgent)：直接返回一个字典。该字典使用图的归约器应用到智能体状态。
包裹式钩子 (wrapModelCall, wrapToolCall)：对于模型调用，直接返回一个 Command 以在模型响应旁注入状态更新。对于工具调用，直接返回一个 Command。当你需要基于模型或工具调用期间运行的逻辑来跟踪或更新状态时使用这些，例如摘要触发点、使用元数据或从请求或响应计算的自定义字段。

节点式钩子

从节点式钩子返回一个字典以将更新合并到智能体状态。字典键映射到状态字段。

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

包裹式钩子

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

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 通过图的归约器流动，因此更新被正确应用，消息是累加的而不是替换现有状态。

多个中间件的组合

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

AIMessage 流动： 每个中间件的 handler() 接收来自前一层的 AIMessage。当中间件返回一个 AIMessage 时，该消息成为下一个中间件处理器的输入。
不带消息更新的 Command 是透传的： 如果中间件返回的 Command 的状态更新不涉及 messages，框架将其视为消息流的无操作。下一个中间件的处理器接收来自返回 Command 的中间件之前的中间件的 AIMessage。
归约器行为和重试安全性： Command 仍然通过归约器应用（消息累加，外部在冲突时胜出）。重试逻辑会丢弃来自早期调用的 Command。

import * as z from "zod";
import { createMiddleware } from "langchain";
import { Command, StateSchema, ReducedValue } from "@langchain/langgraph";
import { AIMessage, SystemMessage } from "@langchain/core/messages";

/** 最后胜出归约器：当两个中间件都写入时，外部覆盖内部。 */
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]" })],
      },
    });
  },
});

创建中间件

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

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

自定义状态模式

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

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

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.4",
  tools: [...],
  middleware: [callCounterMiddleware],
});

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

状态字段可以是公共的或私有的。以下划线 (_) 开头的字段被视为私有，不会包含在智能体的结果中。只有公共字段（没有前导下划线的字段）才会被返回。这对于存储不应暴露给调用者的内部中间件状态很有用，例如临时跟踪变量或内部标志：

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

const PrivateState = new StateSchema({
  // 公共字段 - 包含在 invoke 结果中
  publicCounter: z.number().default(0),
  // 私有字段 - 从 invoke 结果中排除
  _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
});

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

自定义上下文

中间件可以定义自定义上下文模式以访问每次调用的元数据。与状态不同，上下文是只读的，不会在调用间持久化。这使其非常适合：

用户信息：传递在执行期间不会改变的用户 ID、角色或偏好
配置覆盖：提供每次调用的设置，如速率限制或功能标志
租户/工作区上下文：包含多租户应用程序的组织特定数据
请求元数据：传递中间件所需的请求 ID、API 密钥或其他元数据

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

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) => {
    // 从 runtime 访问上下文
    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.4",
  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() 调用期间必须提供这些字段。这确保了类型安全并防止因缺少必需上下文而导致的运行时错误。

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

执行顺序

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

const agent = createAgent({
  model: "gpt-5.4",
  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 钩子）

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

const agent = createAgent({
  model: "gpt-5.4",
  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);

最佳实践

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

示例

动态提示

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

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

const addContextMiddleware = createMiddleware({
  name: "AddContextMiddleware",
  wrapModelCall: async (request, handler) => {
    return handler({
      ...request,
      systemMessage: request.systemMessage.concat(`附加上下文。`),
    });
  },
});

const agent = createAgent({
  model: "google-genai:gemini-3.1-pro-preview",
  systemPrompt: "你是一个有用的助手。",
  middleware: [addContextMiddleware],
});

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

动态模型选择

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 服务器），请参阅运行时工具注册。 好处：

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

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.4",
  tools: allTools,
  middleware: [toolSelectorMiddleware],
});

工具调用监控

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 模型时，使用带有缓存控制指令的结构化内容块来缓存大型系统提示：

装饰器
类

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 参数，用于缓存控制等高级用例

::: 在中间件中使用 ModelRequest 中的 systemMessage 字段修改系统消息。它包含一个 SystemMessage 对象（即使智能体是用字符串 systemPrompt 创建的）。 示例：链接中间件 - 不同的中间件可以使用不同的方法：

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.1-pro-preview",
  systemPrompt: "You are a helpful assistant.",
  middleware: [myMiddleware, myOtherMiddleware],
});

生成的系统消息将是：

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 来保留缓存控制元数据或其他中间件创建的结构化内容块。

附加资源

将这些文档连接到 Claude、VSCode 等，通过 MCP 获取实时答案。

在 GitHub 上编辑此页面或提交问题。

​钩子

节点式钩子

包裹式钩子

​节点式钩子

​包裹式钩子

​状态更新

​节点式钩子

​包裹式钩子

​多个中间件的组合

​创建中间件

​自定义状态模式

​自定义上下文

​执行顺序

​智能体跳转

​最佳实践

​示例

​动态提示

​动态模型选择

​动态选择工具

​工具调用监控

​提示缓存 (Anthropic)

​附加资源

钩子

节点式钩子

包裹式钩子

状态更新

节点式钩子

包裹式钩子

多个中间件的组合

创建中间件

自定义状态模式

自定义上下文

执行顺序

智能体跳转

最佳实践

示例

动态提示

动态模型选择

动态选择工具

工具调用监控

提示缓存 (Anthropic)

附加资源