deep-agents-orchestration

Deep Agents には、次の3つのオーケストレーション機能が含まれています。

1. SubAgentMiddleware：task ツールを使って、作業を専門エージェントに委任します
2. TodoListMiddleware：write_todos ツールを使って、タスクを計画・追跡します
3. HumanInTheLoopMiddleware：機密性の高い操作の前に承認を必須にします

3つすべてが create_deep_agent() に自動的に含まれます。

---

サブエージェント（タスクの委任）

サブエージェントを使う場合	メインエージェントを使う場合
タスクに専門的なツールが必要な場合	汎用ツールで十分な場合
複雑な作業を分離したい場合	1つのステップで完了する操作の場合
メインエージェント用にクリーンなコンテキストが必要な場合	コンテキストの肥大化を許容できる場合

メインエージェントが `task` ツールを持ちます -> 新しいサブエージェントを作成します -> サブエージェントが自律的に実行します -> 最終レポートを返します。

デフォルトのサブエージェント：「general-purpose」— メインエージェントと同じツール／設定で自動的に利用できます。

学術論文の検索に特化したツールを備える、カスタムの「researcher」サブエージェントを作成します。 ```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」サブエージェントを作成します。
```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")

CompiledSubAgent を使って、事前構築済みの LangGraph グラフをサブエージェントとしてラップします。 ```python from deepagents import create_deep_agent, CompiledSubAgent from langgraph.graph import StateGraph, MessagesState

builder = StateGraph(MessagesState)

... add nodes/edges ...

graph = builder.compile()

agent = create_deep_agent( subagents=[ CompiledSubAgent( name="my-graph", description="Run the custom LangGraph workflow", runnable=graph, # Must be a compiled graph; state must have "messages" key ) ] )

</python>
</ex-compiledsubagent>

<ex-subagent-with-hitl>
<python>
機密性の高い操作について 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
)

サブエージェントはステートレスです。1回の呼び出しで完全な指示を渡してください。 ```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')

</python>
<typescript>
サブエージェントはステートレスです。1回の呼び出しで完全な指示を渡してください。
```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

カスタムサブエージェントは、メインエージェントからスキルを継承しません。 ```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/"], ...}] )

</python>
</fix-custom-subagents-dont-inherit-skills>

---

## TodoList（タスクの計画）

<when-to-use-todolist>

| TodoList を使う場合 | TodoList を省略する場合 |
|------------------|-------------------|
| 複雑で複数のステップからなるタスク | 1つの簡単な操作で済むタスク |
| 長時間実行される操作 | すぐに終わる操作（3ステップ未満） |

</when-to-use-todolist>

<todolist-tool>

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

各Todo項目には次の内容があります。
- `content`：タスクの説明
- `status`：`"pending"`、`"in_progress"`、`"completed"` のいずれか
</todolist-tool>

<ex-todolist-usage>
<python>
複数のステップからなるタスクに対して、Todoリストを自動作成するエージェントを呼び出します。
```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"}
# ]

複数のステップからなるタスクに対して、Todoリストを自動作成するエージェントを呼び出します。 ```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" } });

</typescript>
</ex-todolist-usage>

<ex-access-todo-state>
<python>
呼び出し後、エージェントの最終状態からTodoリストにアクセスします。
```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']}")

Todoリストの状態を呼び出し間で保持するには、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

</python>
</fix-todolist-requires-thread-id>

---
## ヒューマン・イン・ザ・ループ（承認ワークフロー）

<when-to-use-hitl>

| HITL を使用する場合 | HITL を使用しない場合 |
|--------------|---------------|
| 重大な操作（DB の書き込み、デプロイ） | 読み取り専用の操作 |
| コンプライアンス上、人間による監督が必要な場合 | 完全に自動化されたワークフロー |

</when-to-use-hitl>

<ex-hitl-setup>
<python>
実行前に人間の承認を必要とするツールを設定します。
```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
)

実行前に人間の承認を必要とするツールを設定します。 ```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 });

</typescript>
</ex-hitl-setup>

<ex-approval-workflow>
<python>
割り込みを発生させ、状態を確認し、操作を承認して、実行を再開するまでの完全なワークフローです。
```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)

割り込みを発生させ、状態を確認し、操作を承認して、実行を再開するまでの完全なワークフローです。 ```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 );

</typescript>
</ex-approval-workflow>

<ex-reject-with-feedback>
<python>
保留中の操作をフィードバック付きで拒否し、エージェントに別の方法を試すよう促します。
```python
result = agent.invoke(
    Command(resume={"decisions": [{"type": "reject", "message": "Run tests first"}]}),
    config=config,
)

保留中の操作をフィードバック付きで拒否し、エージェントに別の方法を試すよう促します。 ```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, ) ``` ### エージェントが設定できるもの

• サブエージェントの名前、ツール、モデル、システムプロンプト
• 承認が必要なツール
• ツールごとに許可される判断タイプ
• TodoList の内容と構成

エージェントが設定できないもの

• ツール名（task、write_todos）
• HITL プロトコル（承認／編集／拒否の構造）
• 割り込みに必要なチェックポインター要件の省略
• サブエージェントをステートフルにすること（サブエージェントは一時的なものです）

HITL ワークフローで interrupt_on を使用する場合、チェックポインターが必要です。 ```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>
HITL ワークフローで interruptOn を使用する場合、チェックポインターが必要です。
```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, ) ```