Deep Agents Orchestration

Deep Agents 包含三种编排能力：

1. SubAgentMiddleware：通过 task 工具将工作委派给专用 Agent
2. TodoListMiddleware：通过 write_todos 工具规划和跟踪任务
3. HumanInTheLoopMiddleware：在执行敏感操作前要求获得批准

这三种能力都会自动包含在 create_deep_agent() 中。

---

子 Agent（任务委派）

适合使用子 Agent 的情况	适合使用主 Agent 的情况
任务需要专用工具	通用工具已足够
希望隔离复杂工作	单步操作
需要为主 Agent 提供干净的上下文	可以接受上下文膨胀

主 Agent 使用 `task` 工具 -> 创建全新的子 Agent -> 子 Agent 自主执行 -> 返回最终报告。

默认子 Agent：“general-purpose”——自动提供，并使用与主 Agent 相同的工具/配置。

创建一个名为“researcher”的自定义子 Agent，并为其配备专门用于搜索学术论文的工具。 ```python from deepagents import create_deep_agent from langchain.tools import tool

@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>
创建一个名为“researcher”的自定义子 Agent，并为其配备专门用于搜索学术论文的工具。
```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 配置 HITL 审批，以便执行敏感操作时需要获得批准。 ```python from deepagents import create_deep_agent from langgraph.checkpoint.memory import MemorySaver

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>
子 Agent 是无状态的——请在单次调用中提供完整指令。
```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')

子 Agent 是无状态的——请在单次调用中提供完整指令。 ```typescript // WRONG: Subagents don't remember previous calls // task research: Find data // task research: What did you find? // Starts fresh!

// 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>
自定义子 Agent 不会继承主 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（任务规划）

适合使用 TodoList 的情况	可以跳过 TodoList 的情况
复杂的多步骤任务	简单的单操作任务
长时间运行的操作	快速操作（< 3 步）

``` write_todos(todos: list[dict]) -> None ```

每个待办事项包含：

• content：任务描述
• status：可以是 "pending"、"in_progress" 或 "completed" 之一

调用 Agent，让它自动为多步骤任务创建待办列表。 ```python from deepagents import create_deep_agent

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>
调用 Agent，让它自动为多步骤任务创建待办列表。
```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" } });

调用后，从 Agent 的最终状态中获取待办列表。 ```python result = agent.invoke({...}, config={"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>
待办列表状态需要 `thread_id`，才能在多次调用之间持久保存。
```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

---

人工介入（审批工作流）

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

配置执行前需要人工审批的工具。 ```python from deepagents import create_deep_agent from langgraph.checkpoint.memory import MemorySaver

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>
配置执行前需要人工审批的工具。
```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
});

完整工作流：触发中断、检查状态、批准操作并恢复执行。 ```python from deepagents import create_deep_agent from langgraph.checkpoint.memory import MemorySaver from langgraph.types import Command

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>
完整工作流：触发中断、检查状态、批准操作并恢复执行。
```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
);

拒绝待处理的操作并提供反馈，提示 Agent 尝试其他方法。 ```python result = agent.invoke( Command(resume={"decisions": [{"type": "reject", "message": "Run tests first"}]}), config=config, ) ``` 拒绝待处理的操作并提供反馈，提示 Agent 尝试其他方法。 ```typescript const result = await agent.invoke( new Command({ resume: { decisions: [{ type: "reject", message: "Run tests first" }] } }), config, ); ``` 修改拟执行操作的参数，然后再允许执行。 ```python result = agent.invoke( Command(resume={"decisions": [{ "type": "edit", "edited_action": { "name": "execute_sql", "args": {"query": "DELETE FROM users WHERE last_login < '2020-01-01' LIMIT 100"}, }, }]}), config=config, ) ``` ### Agent 可以配置的内容

• 子 Agent 名称、工具、模型、系统提示词
• 哪些工具需要审批
• 每个工具允许的决策类型
• TodoList 内容和结构

Agent 无法配置的内容

• 工具名称（task、write_todos）
• HITL 协议（approve/edit/reject 结构）
• 跳过中断所需的 checkpointer 要求
• 将子 Agent 设为有状态（它们是临时的）

使用 interrupt_on 实现 HITL 工作流时必须配置 Checkpointer。 ```python # WRONG agent = create_deep_agent(interrupt_on={"write_file": True})

CORRECT

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

</python>
<typescript>
使用 interruptOn 实现 HITL 工作流时必须配置 Checkpointer。
```typescript
// WRONG
const agent = await createDeepAgent({ interruptOn: { write_file: true } });

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

恢复中断的工作流时必须使用一致的 thread_id。 ```python # WRONG: Can't resume without thread_id agent.invoke({"messages": [...]})

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>
恢复中断的工作流时必须使用一致的 thread_id。
```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);

中断发生在 invoke() 调用之间，而非执行过程中。 ```python result = agent.invoke({...}, config=config) # Step 1: triggers interrupt if "__interrupt__" in result: # Step 2: check for interrupt result = agent.invoke( # Step 3: resume Command(resume={"decisions": [{"type": "approve"}]}), config=config, ) ```