deep-agents-orchestration（深度 Agent 编排）

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

All three are automatically included in create_deep_agent().

Subagents (Task Delegation)

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.

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.

# 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.

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.

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.

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.

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.

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.

// 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.

// 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);