langchain-middleware

面向生产环境的 LangChain Agent 中间件模式：

• HumanInTheLoopMiddleware / humanInTheLoopMiddleware：在执行危险的工具调用前暂停，等待人工审批
• 自定义中间件：拦截工具调用，用于错误处理、日志记录和重试逻辑
• Command 恢复执行：在人工决策（批准、编辑、拒绝）后继续执行

要求： 所有 HITL 工作流都需要 Checkpointer 和 thread_id 配置。

---

人工介入

设置一个带有 HITL 中间件的 Agent，在发送邮件前暂停并等待审批。 ```python from langchain.agents import create_agent from langchain.agents.middleware import HumanInTheLoopMiddleware from langgraph.checkpoint.memory import MemorySaver from langchain.tools import tool

@tool def send_email(to: str, subject: str, body: str) -> str: """Send an email.""" return f"Email sent to {to}"

agent = create_agent( model="gpt-4.1", tools=[send_email], checkpointer=MemorySaver(), # Required for HITL middleware=[ HumanInTheLoopMiddleware( interrupt_on={ "send_email": {"allowed_decisions": ["approve", "edit", "reject"]}, } ) ], )

</python>
<typescript>
设置一个带有 HITL 的 Agent，在发送邮件前暂停并等待人工审批。
```typescript
import { createAgent, humanInTheLoopMiddleware } from "langchain";
import { MemorySaver } from "@langchain/langgraph";
import { tool } from "@langchain/core/tools";
import { z } from "zod";

const sendEmail = tool(
  async ({ to, subject, body }) => `Email sent to ${to}`,
  {
    name: "send_email",
    description: "Send an email",
    schema: z.object({ to: z.string(), subject: z.string(), body: z.string() }),
  }
);

const agent = createAgent({
  model: "anthropic:claude-sonnet-4-5",
  tools: [sendEmail],
  checkpointer: new MemorySaver(),
  middleware: [
    humanInTheLoopMiddleware({
      interruptOn: { send_email: { allowedDecisions: ["approve", "edit", "reject"] } },
    }),
  ],
});

运行 Agent，检测中断，然后在获得人工审批后恢复执行。 ```python from langgraph.types import Command

config = {"configurable": {"thread_id": "session-1"}}

Step 1: Agent runs until it needs to call tool

result1 = agent.invoke({ "messages": [{"role": "user", "content": "Send email to [email protected]"}] }, config=config)

Check for interrupt

if "interrupt" in result1: print(f"Waiting for approval: {result1['interrupt']}")

Step 2: Human approves

result2 = agent.invoke( Command(resume={"decisions": [{"type": "approve"}]}), config=config )

</python>
<typescript>
运行 Agent，检测中断，然后在获得人工审批后恢复执行。
```typescript
import { Command } from "@langchain/langgraph";

const config = { configurable: { thread_id: "session-1" } };

// Step 1: Agent runs until it needs to call tool
const result1 = await agent.invoke({
  messages: [{ role: "user", content: "Send email to [email protected]" }]
}, config);

// Check for interrupt
if (result1.__interrupt__) {
  console.log(`Waiting for approval: ${result1.__interrupt__}`);
}

// Step 2: Human approves
const result2 = await agent.invoke(
  new Command({ resume: { decisions: [{ type: "approve" }] } }),
  config
);

如果原始值需要更正，可在批准前编辑工具参数。 ```python # Human edits the arguments — edited_action must include name + args result2 = agent.invoke( Command(resume={ "decisions": [{ "type": "edit", "edited_action": { "name": "send_email", "args": { "to": "[email protected]", # Fixed email "subject": "Project Meeting - Updated", "body": "...", }, }, }] }), config=config ) ``` 如果原始值需要更正，可在批准前编辑工具参数。 ```typescript // Human edits the arguments — editedAction must include name + args const result2 = await agent.invoke( new Command({ resume: { decisions: [{ type: "edit", editedAction: { name: "send_email", args: { to: "[email protected]", // Fixed email subject: "Project Meeting - Updated", body: "...", }, }, }] } }), config ); ``` 拒绝工具调用，并提供说明拒绝原因的反馈。 ```python # Human rejects result2 = agent.invoke( Command(resume={ "decisions": [{ "type": "reject", "feedback": "Cannot delete customer data without manager approval", }] }), config=config ) ``` 根据风险级别，为每个工具配置不同的 HITL 策略。 ```python agent = create_agent( model="gpt-4.1", tools=[send_email, read_email, delete_email], checkpointer=MemorySaver(), middleware=[ HumanInTheLoopMiddleware( interrupt_on={ "send_email": {"allowed_decisions": ["approve", "edit", "reject"]}, "delete_email": {"allowed_decisions": ["approve", "reject"]}, # No edit "read_email": False, # No HITL for reading } ) ], ) ``` ### 可配置项

• 哪些工具需要审批（按工具配置策略）
• 每个工具允许的决策（批准、编辑、拒绝）
• 自定义中间件钩子：before_model、after_model、wrap_tool_call、before_agent、after_agent
• 工具专属中间件（仅应用于特定工具）

---

自定义中间件钩子

共有六个装饰器钩子，分为两种模式：

• 包装钩子（wrap_tool_call、wrap_model_call）：参数为 (request, handler)；调用 handler(request) 继续执行，或提前返回以短路执行。
• 前置/后置钩子（before_model、after_model、before_agent、after_agent）：参数为 (state, runtime)；检查或修改状态。返回 None 或包含状态更新的字典。

`@wrap_tool_call` 会拦截工具执行。**不要使用 `yield`**——它会创建生成器并导致 `NotImplementedError`。

from langchain.agents.middleware import wrap_tool_call

@wrap_tool_call
def retry_middleware(request, handler):
    for attempt in range(3):
        try:
            return handler(request)
        except Exception:
            if attempt == 2:
                raise

@wrap_tool_call
def guard_middleware(request, handler):
    if request.tool_call["name"] == "dangerous_tool":
        return "This tool is disabled"  # short-circuit
    return handler(request)

`createMiddleware({ wrapToolCall })` 会拦截工具执行。

import { createMiddleware } from "langchain";

const retryMiddleware = createMiddleware({
  wrapToolCall: async (request, handler) => {
    for (let attempt = 0; attempt < 3; attempt++) {
      try { return await handler(request); }
      catch (e) { if (attempt === 2) throw e; }
    }
  },
});

`before_model` / `after_model` / `before_agent` / `after_agent` 都使用相同的 `(state, runtime)` 签名。

from langchain.agents.middleware import before_model, after_model

@before_model
def log_calls(state, runtime):
    print(f"Calling model with {len(state['messages'])} messages")

@after_model
def check_output(state, runtime):
    print(f"Model responded")

通过 `createMiddleware` 创建的所有前置/后置钩子都使用相同的 `(state, runtime)` 签名。

import { createMiddleware } from "langchain";

const loggingMiddleware = createMiddleware({
  beforeModel: (state, runtime) => {
    console.log(`Calling model with ${state.messages.length} messages`);
  },
  afterModel: (state, runtime) => {
    console.log("Model responded");
  },
});

### 无法配置的项

• 在工具执行后中断（必须在执行前）
• 跳过 HITL 所需的 Checkpointer

HITL 中间件需要 Checkpointer 来持久化状态。 ```python # WRONG agent = create_agent(model="gpt-4.1", tools=[send_email], middleware=[HumanInTheLoopMiddleware({...})])

CORRECT

agent = create_agent( model="gpt-4.1", tools=[send_email], checkpointer=MemorySaver(), # Required middleware=[HumanInTheLoopMiddleware({...})] )

</python>
<typescript>
HITL 需要 Checkpointer 来持久化状态。
```typescript
// WRONG: No checkpointer
const agent = createAgent({
  model: "anthropic:claude-sonnet-4-5", tools: [sendEmail],
  middleware: [humanInTheLoopMiddleware({ interruptOn: { send_email: true } })],
});

// CORRECT: Add checkpointer
const agent = createAgent({
  model: "anthropic:claude-sonnet-4-5", tools: [sendEmail],
  checkpointer: new MemorySaver(),
  middleware: [humanInTheLoopMiddleware({ interruptOn: { send_email: true } })],
});

使用 HITL 时，始终提供 thread_id 来跟踪对话状态。 ```python # WRONG agent.invoke(input) # No config!

CORRECT

agent.invoke(input, config={"configurable": {"thread_id": "user-123"}})

</python>
</fix-no-thread-id>

<fix-wrong-resume-syntax>
<python>
使用 Command 类在中断后恢复执行。
```python
# WRONG
agent.invoke({"resume": {"decisions": [...]}})

# CORRECT
from langgraph.types import Command
agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config=config)

使用 Command 类在中断后恢复执行。 ```typescript // WRONG await agent.invoke({ resume: { decisions: [...] } });

// CORRECT import { Command } from "@langchain/langgraph"; await agent.invoke(new Command({ resume: { decisions: [{ type: "approve" }] } }), config);

</typescript>
</fix-wrong-resume-syntax>