LangChain Deep Agents 智能体编排：子智能体委托、任务规划与人工审核中间件

deep-agents-orchestration by langchain-ai/langchain-skills

2,400 周安装量

423 GitHub Stars

GitHub

安装命令

npx skills add https://github.com/langchain-ai/langchain-skills --skill deep-agents-orchestration

AI/机器学习开发开发运维生产力

🇨🇳中文介绍

SubAgentMiddleware：通过 task 工具将工作委托给专门的智能体
TodoListMiddleware：通过 write_todos 工具规划和跟踪任务
HumanInTheLoopMiddleware：在执行敏感操作前需要获得批准

这三个中间件都自动包含在 create_deep_agent() 中。

子智能体（任务委托）

何时使用子智能体	何时使用主智能体
任务需要专门的工具	通用工具足够
希望隔离复杂工作	单步操作
需要为主智能体保持清晰的上下文	可以接受上下文膨胀

默认子智能体："general-purpose" - 自动可用，拥有与主智能体相同的工具/配置。

@tool def search_papers(query: str) -> str: """搜索学术论文。""" return f"Found 10 papers about {query}"

agent = create_deep_agent( subagents=[ { "name": "researcher", "description": "进行网络研究并整理发现", "system_prompt": "彻底搜索，返回简洁摘要", "tools": [search_papers], } ] )

广告位招租

在这里展示您的产品或服务

触达数万 AI 开发者，精准高效

联系我们

主智能体委托任务：task(agent="researcher", instruction="研究 AI 趋势")

</python>
<typescript>
创建一个自定义的 "researcher" 子智能体，配备用于学术论文搜索的专门工具。
```typescript
import { createDeepAgent } from "deepagents";
import { tool } from "@langchain/core/tools";
import { z } from "zod";

const searchPapers = tool(
  async ({ query }) => `Found 10 papers about ${query}`,
  { name: "search_papers", description: "搜索论文", schema: z.object({ query: z.string() }) }
);

const agent = await createDeepAgent({
  subagents: [
    {
      name: "researcher",
      description: "进行网络研究并整理发现",
      systemPrompt: "彻底搜索，返回简洁摘要",
      tools: [searchPapers],
    }
  ]
});

// 主智能体委托任务：task(agent="researcher", instruction="研究 AI 趋势")

agent = create_deep_agent( subagents=[ { "name": "code-deployer", "description": "将代码部署到生产环境", "system_prompt": "你在测试通过后部署代码。", "tools": [run_tests, deploy_to_prod], "interrupt_on": {"deploy_to_prod": True}, # 需要批准 } ], checkpointer=MemorySaver() # 中断功能必需 )

</python>
</ex-subagent-with-hitl>

<fix-subagents-are-stateless>
<python>
子智能体是无状态的 - 在单次调用中提供完整的指令。
```python
# 错误：子智能体不记得之前的调用
# task(agent='research', instruction='查找数据')
# task(agent='research', instruction='你找到了什么？')  # 重新开始！

# 正确：预先提供完整指令
# task(agent='research', instruction='查找关于 AI 的数据，保存到 /research/，返回摘要')

// 正确：预先提供完整指令 // task research: 查找关于 AI 的数据，保存到 /research/，返回摘要

</typescript>
</fix-subagents-are-stateless>

<fix-custom-subagents-dont-inherit-skills>
<python>
自定义子智能体不会继承主智能体的技能。
```python
# 错误：自定义子智能体不会有主智能体的技能
agent = create_deep_agent(
    skills=["/main-skills/"],
    subagents=[{"name": "helper", ...}]  # 没有继承技能
)

# 正确：明确提供技能（通用子智能体确实会继承）
agent = create_deep_agent(
    skills=["/main-skills/"],
    subagents=[{"name": "helper", "skills": ["/helper-skills/"], ...}]
)

待办事项列表（任务规划）

何时使用待办事项列表	何时跳过待办事项列表
复杂的多步骤任务	简单的单动作任务
长时间运行的操作	快速操作（< 3 步）

每个待办事项包含：

content：任务描述
status："pending"、"in_progress"、"completed" 之一

agent = create_deep_agent() # TodoListMiddleware 默认包含

result = agent.invoke({ "messages": [{"role": "user", "content": "创建 REST API：设计模型、实现 CRUD、添加认证、编写测试"}] }, config={"configurable": {"thread_id": "session-1"}})

智能体通过 write_todos 进行规划：

{"content": "设计数据模型", "status": "in_progress"},

{"content": "实现 CRUD 端点", "status": "pending"},

{"content": "添加认证", "status": "pending"},

{"content": "编写测试", "status": "pending"}

</python>
<typescript>
调用一个为多步骤任务自动创建待办事项列表的智能体。
```typescript
import { createDeepAgent } from "deepagents";

const agent = await createDeepAgent();  // 包含 TodoListMiddleware

const result = await agent.invoke({
  messages: [{ role: "user", content: "创建 REST API：设计模型、实现 CRUD、添加认证、编写测试" }]
}, { configurable: { thread_id: "session-1" } });

从最终状态访问待办事项列表

todos = result.get("todos", []) for todo in todos: print(f"[{todo['status']}] {todo['content']}")

</python>
</ex-access-todo-state>

<fix-todolist-requires-thread-id>
<python>
待办事项列表状态需要 thread_id 才能在多次调用间持久化。
```python
# 错误：没有 thread_id 时每次都是新状态
agent.invoke({"messages": [...]})

# 正确：使用 thread_id
config = {"configurable": {"thread_id": "user-session"}}
agent.invoke({"messages": [...]}, config=config)  # 待办事项保留

人在回路（审批工作流）

何时使用人在回路	何时跳过人在回路
高风险操作（数据库写入、部署）	只读操作
合规要求人工监督	完全自动化的工作流

agent = create_deep_agent( interrupt_on={ "write_file": True, # 允许所有决策 "execute_sql": {"allowed_decisions": ["approve", "reject"]}, "read_file": False, # 无中断 }, checkpointer=MemorySaver() # 中断功能必需 )

</python>
<typescript>
配置哪些工具在执行前需要人工批准。
```typescript
import { createDeepAgent } from "deepagents";
import { MemorySaver } from "@langchain/langgraph";

const agent = await createDeepAgent({
  interruptOn: {
    write_file: true,
    execute_sql: { allowedDecisions: ["approve", "reject"] },
    read_file: false,
  },
  checkpointer: new MemorySaver()  // 必需
});

agent = create_deep_agent( interrupt_on={"write_file": True}, checkpointer=MemorySaver() )

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

步骤 1：智能体提议 write_file - 执行暂停

result = agent.invoke({ "messages": [{"role": "user", "content": "将配置写入 /prod.yaml"}] }, config=config)

步骤 2：检查中断

state = agent.get_state(config) if state.next: print(f"待处理操作")

步骤 3：批准并恢复执行

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

</python>
<typescript>
完整工作流：触发中断、检查状态、批准操作、恢复执行。
```typescript
import { createDeepAgent } from "deepagents";
import { MemorySaver, Command } from "@langchain/langgraph";

const agent = await createDeepAgent({
  interruptOn: { write_file: true },
  checkpointer: new MemorySaver()
});

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

// 步骤 1：智能体提议 write_file - 执行暂停
let result = await agent.invoke({
  messages: [{ role: "user", content: "将配置写入 /prod.yaml" }]
}, config);

// 步骤 2：检查中断
const state = await agent.getState(config);
if (state.next) {
  console.log("待处理操作");
}

// 步骤 3：批准并恢复执行
result = await agent.invoke(
  new Command({ resume: { decisions: [{ type: "approve" }] } }), config
);

子智能体名称、工具、模型、系统提示
哪些工具需要批准
每个工具允许的决策类型
待办事项列表的内容和结构

智能体无法配置的内容

工具名称（task、write_todos）
人在回路协议（批准/编辑/拒绝结构）
中断功能的检查点要求
使子智能体有状态（它们是临时的）

agent = create_deep_agent(interrupt_on={"write_file": True}, checkpointer=MemorySaver())

</python>
<typescript>
当使用 interruptOn 进行人在回路工作流时，检查点是必需的。
```typescript
// 错误
const agent = await createDeepAgent({ interruptOn: { write_file: true } });

// 正确
const agent = await createDeepAgent({ interruptOn: { write_file: true }, checkpointer: new MemorySaver() });

config = {"configurable": {"thread_id": "session-1"}} agent.invoke({...}, config=config)

使用相同配置通过 Command 恢复

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

</python>
<typescript>
恢复中断的工作流需要一致的 thread_id。
```typescript
// 错误：没有 thread_id 无法恢复
await agent.invoke({ messages: [...] });

// 正确
const config = { configurable: { thread_id: "session-1" } };
await agent.invoke({ messages: [...] }, config);
// 使用相同配置通过 Command 恢复
await agent.invoke(new Command({ resume: { decisions: [{ type: "approve" }] } }), config);

🇺🇸English

SubAgentMiddleware : Delegate work via task tool to specialized agents
TodoListMiddleware : Plan and track tasks via write_todos tool
HumanInTheLoopMiddleware : Require approval before sensitive operations

All three are automatically included in create_deep_agent().

Subagents (Task Delegation)

Use Subagents When	Use Main Agent When
Task needs specialized tools	General-purpose tools sufficient
Want to isolate complex work	Single-step operation
Need clean context for main agent	Context bloat acceptable

Default subagent : "general-purpose" - automatically available with same tools/config as main agent.

@tool def search_papers(query: str) -> str: """Search academic papers.""" return f"Found 10 papers about {query}"

agent = create_deep_agent( subagents=[ { "name": "researcher", "description": "Conduct web research and compile findings", "system_prompt": "Search thoroughly, return concise summary", "tools": [search_papers], } ] )

Main agent delegates: task(agent="researcher", instruction="Research AI trends")

</python>
<typescript>
Create a custom "researcher" subagent with specialized tools for academic paper search.
```typescript
import { createDeepAgent } from "deepagents";
import { tool } from "@langchain/core/tools";
import { z } from "zod";

const searchPapers = tool(
  async ({ query }) => `Found 10 papers about ${query}`,
  { name: "search_papers", description: "Search papers", schema: z.object({ query: z.string() }) }
);

const agent = await createDeepAgent({
  subagents: [
    {
      name: "researcher",
      description: "Conduct web research and compile findings",
      systemPrompt: "Search thoroughly, return concise summary",
      tools: [searchPapers],
    }
  ]
});

// Main agent delegates: task(agent="researcher", instruction="Research AI trends")

agent = create_deep_agent( subagents=[ { "name": "code-deployer", "description": "Deploy code to production", "system_prompt": "You deploy code after tests pass.", "tools": [run_tests, deploy_to_prod], "interrupt_on": {"deploy_to_prod": True}, # Require approval } ], checkpointer=MemorySaver() # Required for interrupts )

</python>
</ex-subagent-with-hitl>

<fix-subagents-are-stateless>
<python>
Subagents are stateless - provide complete instructions in a single call.
```python
# WRONG: Subagents don't remember previous calls
# task(agent='research', instruction='Find data')
# task(agent='research', instruction='What did you find?')  # Starts fresh!

# CORRECT: Complete instructions upfront
# task(agent='research', instruction='Find data on AI, save to /research/, return summary')

// CORRECT: Complete instructions upfront // task research: Find data on AI, save to /research/, return summary

</typescript>
</fix-subagents-are-stateless>

<fix-custom-subagents-dont-inherit-skills>
<python>
Custom subagents don't inherit skills from the main agent.
```python
# WRONG: Custom subagent won't have main agent's skills
agent = create_deep_agent(
    skills=["/main-skills/"],
    subagents=[{"name": "helper", ...}]  # No skills inherited
)

# CORRECT: Provide skills explicitly (general-purpose subagent DOES inherit)
agent = create_deep_agent(
    skills=["/main-skills/"],
    subagents=[{"name": "helper", "skills": ["/helper-skills/"], ...}]
)

TodoList (Task Planning)

Use TodoList When	Skip TodoList When
Complex multi-step tasks	Simple single-action tasks
Long-running operations	Quick operations (< 3 steps)

Each todo item has:

content: Description of the task
status: One of "pending", "in_progress", "completed"

agent = create_deep_agent() # TodoListMiddleware included by default

result = agent.invoke({ "messages": [{"role": "user", "content": "Create a REST API: design models, implement CRUD, add auth, write tests"}] }, config={"configurable": {"thread_id": "session-1"}})

Agent's planning via write_todos:

[

{"content": "Design data models", "status": "in_progress"},

{"content": "Implement CRUD endpoints", "status": "pending"},

{"content": "Add authentication", "status": "pending"},

{"content": "Write tests", "status": "pending"}

]

</python>
<typescript>
Invoke an agent that automatically creates a todo list for a multi-step task.
```typescript
import { createDeepAgent } from "deepagents";

const agent = await createDeepAgent();  // TodoListMiddleware included

const result = await agent.invoke({
  messages: [{ role: "user", content: "Create a REST API: design models, implement CRUD, add auth, write tests" }]
}, { configurable: { thread_id: "session-1" } });

Access todo list from final state

todos = result.get("todos", []) for todo in todos: print(f"[{todo['status']}] {todo['content']}")

</python>
</ex-access-todo-state>

<fix-todolist-requires-thread-id>
<python>
Todo list state requires a thread_id for persistence across invocations.
```python
# WRONG: Fresh state each time without thread_id
agent.invoke({"messages": [...]})

# CORRECT: Use thread_id
config = {"configurable": {"thread_id": "user-session"}}
agent.invoke({"messages": [...]}, config=config)  # Todos preserved

Human-in-the-Loop (Approval Workflows)

Use HITL When	Skip HITL When
High-stakes operations (DB writes, deployments)	Read-only operations
Compliance requires human oversight	Fully automated workflows

agent = create_deep_agent( interrupt_on={ "write_file": True, # All decisions allowed "execute_sql": {"allowed_decisions": ["approve", "reject"]}, "read_file": False, # No interrupts }, checkpointer=MemorySaver() # REQUIRED for interrupts )

</python>
<typescript>
Configure which tools require human approval before execution.
```typescript
import { createDeepAgent } from "deepagents";
import { MemorySaver } from "@langchain/langgraph";

const agent = await createDeepAgent({
  interruptOn: {
    write_file: true,
    execute_sql: { allowedDecisions: ["approve", "reject"] },
    read_file: false,
  },
  checkpointer: new MemorySaver()  // REQUIRED
});

agent = create_deep_agent( interrupt_on={"write_file": True}, checkpointer=MemorySaver() )

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

Step 1: Agent proposes write_file - execution pauses

result = agent.invoke({ "messages": [{"role": "user", "content": "Write config to /prod.yaml"}] }, config=config)

Step 2: Check for interrupts

state = agent.get_state(config) if state.next: print(f"Pending action")

Step 3: Approve and resume

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

</python>
<typescript>
Complete workflow: trigger an interrupt, check state, approve action, and resume execution.
```typescript
import { createDeepAgent } from "deepagents";
import { MemorySaver, Command } from "@langchain/langgraph";

const agent = await createDeepAgent({
  interruptOn: { write_file: true },
  checkpointer: new MemorySaver()
});

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

// Step 1: Agent proposes write_file - execution pauses
let result = await agent.invoke({
  messages: [{ role: "user", content: "Write config to /prod.yaml" }]
}, config);

// Step 2: Check for interrupts
const state = await agent.getState(config);
if (state.next) {
  console.log("Pending action");
}

// Step 3: Approve and resume
result = await agent.invoke(
  new Command({ resume: { decisions: [{ type: "approve" }] } }), config
);

Subagent names, tools, models, system prompts
Which tools require approval
Allowed decision types per tool
TodoList content and structure

What Agents CANNOT Configure

Tool names (task, write_todos)
HITL protocol (approve/edit/reject structure)
Skip checkpointer requirement for interrupts
Make subagents stateful (they're ephemeral)

CORRECT

agent = create_deep_agent(interrupt_on={"write_file": True}, checkpointer=MemorySaver())

</python>
<typescript>
Checkpointer is required when using interruptOn for HITL workflows.
```typescript
// WRONG
const agent = await createDeepAgent({ interruptOn: { write_file: true } });

// CORRECT
const agent = await createDeepAgent({ interruptOn: { write_file: true }, checkpointer: new MemorySaver() });

CORRECT

config = {"configurable": {"thread_id": "session-1"}} agent.invoke({...}, config=config)

Resume with Command using same config

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

</python>
<typescript>
A consistent thread_id is required to resume interrupted workflows.
```typescript
// WRONG: Can't resume without thread_id
await agent.invoke({ messages: [...] });

// CORRECT
const config = { configurable: { thread_id: "session-1" } };
await agent.invoke({ messages: [...] }, config);
// Resume with Command using same config
await agent.invoke(new Command({ resume: { decisions: [{ type: "approve" }] } }), config);

Weekly Installs

2.4K

Repository

langchain-ai/la…n-skills

GitHub Stars

423

First Seen

Mar 3, 2026

Security Audits

Gen Agent Trust HubPass SocketPass SnykPass

Installed on

claude-code2.0K

codex1.9K

cursor1.9K

github-copilot1.8K

opencode1.8K

gemini-cli1.8K

React 组合模式指南：Vercel 组件架构最佳实践，提升代码可维护性

102,200 周安装

LangChain Deep Agents 智能体编排：子智能体委托、任务规划与人工审核中间件

🇨🇳中文介绍

子智能体（任务委托）

相关 Skills

主智能体委托任务：task(agent="researcher", instruction="研究 AI 趋势")

待办事项列表（任务规划）

智能体通过 write_todos 进行规划：

[

{"content": "设计数据模型", "status": "in_progress"},

{"content": "实现 CRUD 端点", "status": "pending"},

{"content": "添加认证", "status": "pending"},

{"content": "编写测试", "status": "pending"}

]

从最终状态访问待办事项列表

人在回路（审批工作流）

步骤 1：智能体提议 write_file - 执行暂停

步骤 2：检查中断

步骤 3：批准并恢复执行

智能体无法配置的内容

正确

正确

使用相同配置通过 Command 恢复

🇺🇸English

Subagents (Task Delegation)

Main agent delegates: task(agent="researcher", instruction="Research AI trends")

TodoList (Task Planning)

Agent's planning via write_todos:

[

{"content": "Design data models", "status": "in_progress"},

{"content": "Implement CRUD endpoints", "status": "pending"},

{"content": "Add authentication", "status": "pending"},

{"content": "Write tests", "status": "pending"}

]

Access todo list from final state

Human-in-the-Loop (Approval Workflows)

Step 1: Agent proposes write_file - execution pauses

Step 2: Check for interrupts

Step 3: Approve and resume

What Agents CANNOT Configure

CORRECT

CORRECT

Resume with Command using same config

最新 Skills