langchain-middleware by langchain-ai/langchain-skills
npx skills add https://github.com/langchain-ai/langchain-skills --skill langchain-middleware要求: 所有 HITL 工作流都需要配置检查点(Checkpointer)和 thread_id。
@tool def send_email(to: str, subject: str, body: str) -> str: """发送电子邮件。""" return f"Email sent to {to}"
agent = create_agent( model="gpt-4.1", tools=[send_email], checkpointer=MemorySaver(), # HITL 必需 middleware=[ HumanInTheLoopMiddleware( interrupt_on={ "send_email": {"allowed_decisions": ["approve", "edit", "reject"]}, } ) ], )
</python>
<typescript>
设置一个带有 HITL 的智能体,使其在发送电子邮件前暂停等待人工批准。
```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: "发送电子邮件",
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"] } },
}),
],
});
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
config = {"configurable": {"thread_id": "session-1"}}
result1 = agent.invoke({ "messages": [{"role": "user", "content": "发送邮件到 john@example.com"}] }, config=config)
if "interrupt " in result1: print(f"等待批准: {result1['interrupt ']}")
result2 = agent.invoke( Command(resume={"decisions": [{"type": "approve"}]}), config=config )
</python>
<typescript>
运行智能体,检测中断,然后在人工批准后恢复执行。
```typescript
import { Command } from "@langchain/langgraph";
const config = { configurable: { thread_id: "session-1" } };
// 步骤 1: 智能体运行直到需要调用工具
const result1 = await agent.invoke({
messages: [{ role: "user", content: "发送邮件到 john@example.com" }]
}, config);
// 检查是否有中断
if (result1.__interrupt__) {
console.log(`等待批准: ${result1.__interrupt__}`);
}
// 步骤 2: 人工批准
const result2 = await agent.invoke(
new Command({ resume: { decisions: [{ type: "approve" }] } }),
config
);
before_model, after_model, wrap_tool_call, before_agent, after_agentagent = create_agent( model="gpt-4.1", tools=[send_email], checkpointer=MemorySaver(), # 必需 middleware=[HumanInTheLoopMiddleware({...})] )
</python>
<typescript>
HITL 需要一个检查点来持久化状态。
```typescript
// 错误:没有检查点
const agent = createAgent({
model: "anthropic:claude-sonnet-4-5", tools: [sendEmail],
middleware: [humanInTheLoopMiddleware({ interruptOn: { send_email: true } })],
});
// 正确:添加检查点
const agent = createAgent({
model: "anthropic:claude-sonnet-4-5", tools: [sendEmail],
checkpointer: new MemorySaver(),
middleware: [humanInTheLoopMiddleware({ interruptOn: { send_email: true } })],
});
agent.invoke(input, config={"configurable": {"thread_id": "user-123"}})
</python>
</fix-no-thread-id>
<fix-wrong-resume-syntax>
<python>
使用 Command 类在中断后恢复执行。
```python
# 错误
agent.invoke({"resume": {"decisions": [...]}})
# 正确
from langgraph.types import Command
agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config=config)
// 正确示例 import { Command } from "@langchain/langgraph"; await agent.invoke(new Command({ resume: { decisions: [{ type: "approve" }] } }), config);
</typescript>
</fix-wrong-resume-syntax>
每周安装量
2.3K
代码仓库
GitHub 星标数
423
首次出现
2026年3月4日
安全审计
安装于
claude-code1.9K
codex1.8K
cursor1.8K
github-copilot1.8K
opencode1.8K
gemini-cli1.8K
Requirements: Checkpointer + thread_id config for all HITL workflows.
@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"] } },
}),
],
});
config = {"configurable": {"thread_id": "session-1"}}
result1 = agent.invoke({ "messages": [{"role": "user", "content": "Send email to john@example.com"}] }, config=config)
if "interrupt " in result1: print(f"Waiting for approval: {result1['interrupt ']}")
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
);
before_model, after_model, wrap_tool_call, before_agent, after_agentagent = 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 } })],
});
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
# WRONG
agent.invoke({"resume": {"decisions": [...]}})
# CORRECT
from langgraph.types import Command
agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config=config)
// CORRECT import { Command } from "@langchain/langgraph"; await agent.invoke(new Command({ resume: { decisions: [{ type: "approve" }] } }), config);
</typescript>
</fix-wrong-resume-syntax>
Weekly Installs
2.3K
Repository
GitHub Stars
423
First Seen
Mar 4, 2026
Security Audits
Gen Agent Trust HubPassSocketPassSnykPass
Installed on
claude-code1.9K
codex1.8K
cursor1.8K
github-copilot1.8K
opencode1.8K
gemini-cli1.8K
React 组合模式指南:Vercel 组件架构最佳实践,提升代码可维护性
102,200 周安装