langchain-middleware

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

<overview> Middleware patterns for production LangChain agents:

HumanInTheLoopMiddleware / humanInTheLoopMiddleware: Pause before dangerous tool calls for human approval
Custom middleware: Intercept tool calls for error handling, logging, retry logic
Command resume: Continue execution after human decisions (approve, edit, reject)

Requirements: Checkpointer + thread_id config for all HITL workflows. </overview>

<overview> 生产环境下LangChain Agent的中间件模式：

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

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

Human-in-the-Loop

人机协同审批（Human-in-the-Loop）

<ex-basic-hitl-setup> <python> Set up an agent with HITL middleware that pauses before sending emails for approval. ```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>
Set up an agent with HITL that pauses before sending emails for human approval.
```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"] } },
    }),
  ],
});

</typescript> </ex-basic-hitl-setup> <ex-running-with-interrupts> <python> Run the agent, detect an interrupt, then resume execution after human approval. ```python from langgraph.types import Command

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

<ex-basic-hitl-setup> <python> 配置带有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}"

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

</typescript> </ex-basic-hitl-setup> <ex-running-with-interrupts> <python> 运行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 john@example.com"}] }, 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>
Run the agent, detect an interrupt, then resume execution after human approval.
```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 john@example.com" }]
}, 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
);

</typescript> </ex-running-with-interrupts> <ex-editing-tool-arguments> <python> Edit the tool arguments before approving when the original values need correction. ```python

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 john@example.com" }]
}, 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
);

</typescript> </ex-running-with-interrupts> <ex-editing-tool-arguments> <python> 当原始参数需要修正时，在批准前编辑工具参数。 ```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": "alice@company.com", # Fixed email "subject": "Project Meeting - Updated", "body": "...", }, }, }] }), config=config )

</python>
<typescript>
Edit the tool arguments before approving when the original values need correction.
```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: "alice@company.com",  // Fixed email
            subject: "Project Meeting - Updated",
            body: "...",
          },
        },
      }]
    }
  }),
  config
);

</typescript> </ex-editing-tool-arguments> <ex-rejecting-with-feedback> <python> Reject a tool call and provide feedback explaining why it was rejected. ```python

</python>
<typescript>
当原始参数需要修正时，在批准前编辑工具参数。
```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: "alice@company.com",  # Fixed email
            subject: "Project Meeting - Updated",
            body: "...",
          },
        },
      }]
    }
  }),
  config
);

</typescript> </ex-editing-tool-arguments> <ex-rejecting-with-feedback> <python> 拒绝工具调用并提供反馈说明拒绝原因。 ```python

Human rejects

result2 = agent.invoke( Command(resume={ "decisions": [{ "type": "reject", "feedback": "Cannot delete customer data without manager approval", }] }), config=config )

</python>
</ex-rejecting-with-feedback>

<ex-multiple-tools-different-policies>
<python>
Configure different HITL policies for each tool based on risk level.
```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
            }
        )
    ],
)

</python> </ex-multiple-tools-different-policies> <boundaries>

result2 = agent.invoke( Command(resume={ "decisions": [{ "type": "reject", "feedback": "Cannot delete customer data without manager approval", }] }), config=config )

</python>
</ex-rejecting-with-feedback>

<ex-multiple-tools-different-policies>
<python>
根据风险级别为每个工具配置不同的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
            }
        )
    ],
)

</python> </ex-multiple-tools-different-policies> <boundaries>

What You CAN Configure

可配置项

Which tools require approval (per-tool policies)
Allowed decisions per tool (approve, edit, reject)

Custom middleware hooks:

before_model

after_model

wrap_tool_call

before_agent

after_agent

Tool-specific middleware (apply only to certain tools)

哪些工具需要审批（按工具配置策略）
每个工具允许的决策类型（批准、编辑、拒绝）

自定义中间件钩子：

before_model

after_model

wrap_tool_call

before_agent

after_agent

工具专属中间件（仅应用于特定工具）

What You CANNOT Configure

不可配置项

Interrupt after tool execution (must be before)
Skip checkpointer requirement for HITL </boundaries>

<fix-missing-checkpointer> <python> HITL middleware requires a checkpointer to persist state. ```python

工具执行后中断（仅能在执行前中断）
跳过HITL对Checkpointer的要求 </boundaries>

<fix-missing-checkpointer> <python> 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 requires a checkpointer to persist state.
```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 } })],
});

</typescript> </fix-missing-checkpointer> <fix-no-thread-id> <python> Always provide thread_id when using HITL to track conversation state. ```python

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

</typescript> </fix-missing-checkpointer> <fix-no-thread-id> <python> 使用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>
Use Command class to resume execution after an interrupt.
```python

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)

</python>
<typescript>
Use Command class to resume execution after an interrupt.
```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>

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

</python>
<typescript>
使用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>