deepagents-implementation

Deep Agents の実装

コアコンセプト

Deep Agents は、LangGraph 上に構築された、必要な機能が揃ったエージェントハーネスを提供します。

• create_deep_agent：設定済みのエージェントを作成するファクトリー関数
• ミドルウェア：注入される機能（ファイルシステム、TODO、サブエージェント、要約）
• バックエンド：プラグイン可能なファイルストレージ（state、filesystem、store、composite）
• サブエージェント：task ツールによる分離されたタスク実行

返されるエージェントはコンパイル済みの LangGraph StateGraph で、ストリーミング、チェックポイント、LangGraph Studio と互換性があります。

必須のインポート

# Core
from deepagents import create_deep_agent

# Subagents
from deepagents import CompiledSubAgent

# Backends
from deepagents.backends import (
    StateBackend,       # Ephemeral (default)
    FilesystemBackend,  # Real disk
    StoreBackend,       # Persistent cross-thread
    CompositeBackend,   # Route paths to backends
)

# LangGraph (for checkpointing, store, streaming)
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.memory import InMemoryStore

# LangChain (for custom models, tools)
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

基本的な使い方

最小限のエージェント

from deepagents import create_deep_agent

# Uses Claude Sonnet 4 by default
agent = create_deep_agent()

result = agent.invoke({"messages": [{"role": "user", "content": "Hello!"}]})

カスタムツールを使用する場合

from langchain_core.tools import tool
from deepagents import create_deep_agent

@tool
def web_search(query: str) -> str:
    """Search the web for information."""
    return tavily_client.search(query)

agent = create_deep_agent(
    tools=[web_search],
    system_prompt="You are a research assistant. Search the web to answer questions.",
)

result = agent.invoke({"messages": [{"role": "user", "content": "What is LangGraph?"}]})

カスタムモデルを使用する場合

from langchain.chat_models import init_chat_model
from deepagents import create_deep_agent

# OpenAI
model = init_chat_model("openai:gpt-4o")

# Or Anthropic with custom settings
from langchain_anthropic import ChatAnthropic
model = ChatAnthropic(model_name="claude-sonnet-4-5-20250929", max_tokens=8192)

agent = create_deep_agent(model=model)

チェックポイントを使用する場合（永続化）

from langgraph.checkpoint.memory import InMemorySaver
from deepagents import create_deep_agent

agent = create_deep_agent(checkpointer=InMemorySaver())

# Must provide thread_id with checkpointer
config = {"configurable": {"thread_id": "user-123"}}
result = agent.invoke({"messages": [...]}, config)

# Resume conversation
result = agent.invoke({"messages": [{"role": "user", "content": "Follow up"}]}, config)

ストリーミング

エージェントは、すべての LangGraph ストリームモードに対応しています。

更新をストリーミングする

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Write a report"}]},
    stream_mode="updates"
):
    print(chunk)  # {"node_name": {"key": "value"}}

メッセージをストリーミングする（トークン単位）

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Explain quantum computing"}]},
    stream_mode="messages"
):
    # Real-time token streaming
    print(chunk.content, end="", flush=True)

非同期ストリーミング

async for chunk in agent.astream(
    {"messages": [...]},
    stream_mode="updates"
):
    print(chunk)

複数のストリームモード

for mode, chunk in agent.stream(
    {"messages": [...]},
    stream_mode=["updates", "messages"]
):
    if mode == "messages":
        print("Token:", chunk.content)
    else:
        print("Update:", chunk)

バックエンドの設定

StateBackend（デフォルト - 一時的）

ファイルはエージェントの state に保存され、スレッド内でのみ保持されます。

# Implicit - this is the default
agent = create_deep_agent()

# Explicit
from deepagents.backends import StateBackend
agent = create_deep_agent(backend=lambda rt: StateBackend(rt))

FilesystemBackend（実ディスク）

ディスク上の実ファイルを読み書きします。シェルコマンド用の execute ツールを使用できるようになります。

from deepagents.backends import FilesystemBackend

agent = create_deep_agent(
    backend=FilesystemBackend(root_dir="/path/to/project"),
)

StoreBackend（スレッドをまたいで永続化）

LangGraph Store を使用して、会話をまたいでデータを永続化します。

from langgraph.store.memory import InMemoryStore
from deepagents.backends import StoreBackend

store = InMemoryStore()

agent = create_deep_agent(
    backend=lambda rt: StoreBackend(rt),
    store=store,  # Required for StoreBackend
)

CompositeBackend（ハイブリッドルーティング）

異なるパスを異なるバックエンドにルーティングします。

from langgraph.store.memory import InMemoryStore
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend

store = InMemoryStore()

agent = create_deep_agent(
    backend=CompositeBackend(
        default=StateBackend(),           # /workspace/* → ephemeral
        routes={
            "/memories/": StoreBackend(store=store),     # persistent
            "/preferences/": StoreBackend(store=store), # persistent
        },
    ),
    store=store,
)

# Files under /memories/ persist across all conversations
# Files under /workspace/ are ephemeral per-thread

サブエージェント

デフォルトの汎用エージェントを使用する

デフォルトでは、メインエージェントのすべてのツールを利用できる general-purpose サブエージェントが使用可能です。

agent = create_deep_agent(tools=[web_search])

# The agent can now delegate via the `task` tool:
# task(subagent_type="general-purpose", prompt="Research topic X in depth")

カスタムサブエージェントを定義する

from deepagents import create_deep_agent

research_agent = {
    "name": "researcher",
    "description": "Conducts deep research on complex topics with web search",
    "system_prompt": """You are an expert researcher.
    Search thoroughly, cross-reference sources, and synthesize findings.""",
    "tools": [web_search, document_reader],
}

code_agent = {
    "name": "coder",
    "description": "Writes, reviews, and debugs code",
    "system_prompt": "You are an expert programmer. Write clean, tested code.",
    "tools": [code_executor, linter],
    "model": "openai:gpt-4o",  # Optional: different model per subagent
}

agent = create_deep_agent(
    subagents=[research_agent, code_agent],
    system_prompt="Delegate research to the researcher and coding to the coder.",
)

コンパイル済みの LangGraph サブエージェント

既存の LangGraph グラフをサブエージェントとして使用します。

from deepagents import CompiledSubAgent, create_deep_agent
from langgraph.prebuilt import create_react_agent

# Existing graph
custom_graph = create_react_agent(
    model="anthropic:claude-sonnet-4-5-20250929",
    tools=[specialized_tool],
    prompt="Custom workflow instructions",
)

agent = create_deep_agent(
    subagents=[CompiledSubAgent(
        name="custom-workflow",
        description="Runs my specialized analysis workflow",
        runnable=custom_graph,
    )]
)

カスタムミドルウェアを使用するサブエージェント

from langchain.agents.middleware import AgentMiddleware

class LoggingMiddleware(AgentMiddleware):
    def transform_response(self, response):
        print(f"Subagent response: {response}")
        return response

agent_spec = {
    "name": "logged-agent",
    "description": "Agent with extra logging",
    "system_prompt": "You are helpful.",
    "tools": [],
    "middleware": [LoggingMiddleware()],  # Added after default middleware
}

Human-in-the-Loop

基本的な割り込み設定

人間の承認を得るため、特定のツールを実行する前に処理を一時停止します。

from deepagents import create_deep_agent

agent = create_deep_agent(
    tools=[send_email, delete_file, web_search],
    interrupt_on={
        "send_email": True,      # Simple interrupt
        "delete_file": True,     # Require approval before delete
        # web_search not listed - runs without approval
    },
    checkpointer=checkpointer,   # Required for interrupts
)

オプションを指定した割り込み

agent = create_deep_agent(
    tools=[send_email],
    interrupt_on={
        "send_email": {
            "allowed_decisions": ["approve", "edit", "reject"]
        },
    },
    checkpointer=checkpointer,
)

# Invoke - will pause at send_email
config = {"configurable": {"thread_id": "user-123"}}
result = agent.invoke({"messages": [...]}, config)

# Check state
state = agent.get_state(config)
if state.next:  # Has pending interrupt
    # Resume with approval
    from langgraph.types import Command
    agent.invoke(Command(resume={"approved": True}), config)

    # Or resume with edit
    agent.invoke(Command(resume={"edited_args": {"to": "[email protected]"}}), config)

    # Or reject
    agent.invoke(Command(resume={"rejected": True}), config)

サブエージェントのツールで割り込む

# Interrupts apply to subagents too
agent = create_deep_agent(
    subagents=[research_agent],
    interrupt_on={
        "web_search": True,  # Interrupt even when subagent calls it
    },
    checkpointer=checkpointer,
)

カスタムミドルウェア

ミドルウェアの構造

from langchain.agents.middleware.types import (
    AgentMiddleware,
    ModelRequest,
    ModelResponse,
)
from langchain_core.tools import tool

class MyMiddleware(AgentMiddleware):
    # Tools to inject
    tools = []

    # System prompt content to inject
    system_prompt = ""

    def transform_request(self, request: ModelRequest) -> ModelRequest:
        """Modify request before sending to model."""
        return request

    def transform_response(self, response: ModelResponse) -> ModelResponse:
        """Modify response after receiving from model."""
        return response

ミドルウェアを介してツールを注入する

from langchain_core.tools import tool

@tool
def get_current_time() -> str:
    """Get the current time."""
    from datetime import datetime
    return datetime.now().isoformat()

class TimeMiddleware(AgentMiddleware):
    tools = [get_current_time]
    system_prompt = "You have access to get_current_time for time-sensitive tasks."

agent = create_deep_agent(middleware=[TimeMiddleware()])

コンテキスト注入ミドルウェア

class UserContextMiddleware(AgentMiddleware):
    def __init__(self, user_preferences: dict):
        self.user_preferences = user_preferences

    @property
    def system_prompt(self):
        return f"User preferences: {self.user_preferences}"

agent = create_deep_agent(
    middleware=[UserContextMiddleware({"theme": "dark", "language": "en"})]
)

応答ログミドルウェア

import logging

class LoggingMiddleware(AgentMiddleware):
    def transform_response(self, response: ModelResponse) -> ModelResponse:
        logging.info(f"Agent response: {response.messages[-1].content[:100]}...")
        return response

agent = create_deep_agent(middleware=[LoggingMiddleware()])

MCP ツールの統合

MCP（Model Context Protocol）サーバーに接続し、追加のツールを利用できるようにします。

from langchain_mcp_adapters.client import MultiServerMCPClient
from deepagents import create_deep_agent

async def main():
    mcp_client = MultiServerMCPClient({
        "filesystem": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path"],
        },
        "github": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-github"],
            "env": {"GITHUB_TOKEN": os.environ["GITHUB_TOKEN"]},
        },
    })

    mcp_tools = await mcp_client.get_tools()

    agent = create_deep_agent(tools=mcp_tools)

    async for chunk in agent.astream(
        {"messages": [{"role": "user", "content": "List my repos"}]}
    ):
        print(chunk)

実装時の確認項目

実ディスク、人間による割り込み、永続化、または MCP サブプロセスを扱うセットアップでは、この順序どおりの表を使用してください（最小限の create_deep_agent() のスモークテストでは不要です）。

手順	合格条件
1	FilesystemBackend または execute： root_dir の範囲が意図的に限定されている（誤ってホームディレクトリやファイルシステムのルートを指定していない）。本番環境のパスを信頼する前に、使い捨てディレクトリでスモークテストを実施してください。
2	interrupt_on / 再開： checkpointer が設定されていること。割り込みが発生する可能性のあるすべての invoke / astream に、configurable["thread_id"] を含む config が指定されていること。一時停止後、Command(resume=...) を実行する前に agent.get_state(config) で割り込み待ちの state を確認できること。また、設定されている場合、再開時のペイロードがツールのオプション（例：allowed_decisions）と一致すること。
3	Store、PostgreSQL saver、MCP： 認証情報はコミット済みのソースではなく、環境変数またはシークレットマネージャーから取得すること。MCP の command / args / 必須の env キーが、デプロイ先ホストで実際に利用できるもの（例：npx、トークン）と一致すること。

参考資料

詳細なリファレンスドキュメントについては、以下を参照してください。

• 組み込みツールのリファレンス - すべてのエージェントで利用可能なツール（ファイルシステム、タスク管理、サブエージェントへの委任）とパス要件の一覧
• 一般的なパターン - メモリを備えたリサーチエージェント、ディスクにアクセスするコードアシスタント、複数の専門家で構成されるチーム、本番環境向け PostgreSQL セットアップなど、実運用に適した例