deepagents-implementation

What this skill does

# Deep Agents Implementation

## Core Concepts

Deep Agents provides a batteries-included agent harness built on LangGraph:

- **`create_deep_agent`**: Factory function that creates a configured agent
- **Middleware**: Injected capabilities (filesystem, todos, subagents, summarization)
- **Backends**: Pluggable file storage (state, filesystem, store, composite)
- **Subagents**: Isolated task execution via the `task` tool

The agent returned is a compiled LangGraph `StateGraph`, compatible with streaming, checkpointing, and LangGraph Studio.

## Essential Imports

```python
# 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
```

## Basic Usage

### Minimal Agent

```python
from deepagents import create_deep_agent

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

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

### With Custom Tools

```python
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?"}]})
```

### With Custom Model

```python
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)
```

### With Checkpointing (Persistence)

```python
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)
```

## Streaming

The agent supports all LangGraph stream modes.

### Stream Updates

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

### Stream Messages (Token-by-Token)

```python
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 Streaming

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

### Multiple Stream Modes

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

## Backend Configuration

### StateBackend (Default - Ephemeral)

Files stored in agent state, persist within thread only.

```python
# 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 (Real Disk)

Read/write actual files on disk. Enables `execute` tool for shell commands.

```python
from deepagents.backends import FilesystemBackend

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

### StoreBackend (Persistent Cross-Thread)

Uses LangGraph Store for persistence across conversations.

```python
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 (Hybrid Routing)

Route different paths to different backends.

```python
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
```

## Subagents

### Using the Default General-Purpose Agent

By default, a `general-purpose` subagent is available with all main agent tools.

```python
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")
```

### Defining Custom Subagents

```python
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.",
)
```

### Pre-compiled LangGraph Subagents

Use existing LangGraph graphs as subagents.

```python
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,
    )]
)
```

### Subagent with Custom Middleware

```python
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

### Basic Interrupt Configuration

Pause execution before specific tools for human approval.

```python
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