DeepAgents 深入学习指南：生产级 AI 智能体开发

+----------------------------------+
|  Deep Agents                     |  <-- 开箱即用的完整框架，包含所有生产级特性
|  +----------------------------+  |
|  |  LangChain create_agent    |  |  <-- 轻量级智能体框架，最小工具循环
|  |  +----------------------+  |  |
|  |  |  LangGraph           |  |  |  <-- 图运行时引擎，提供状态管理和流式输出
|  |  +----------------------+  |  |
|  +----------------------------+  |
+----------------------------------+

deepagents/
+-- libs/
|   +-- deepagents/     # 核心 SDK（本教程重点）——所有智能体逻辑
|   +-- cli/            # CLI 部署工具——init/dev/deploy 命令
|   +-- code/           # Deep Agents Code 终端编码助手
|   +-- acp/            # Agent Context Protocol 支持
|   +-- evals/          # 评估套件——回归测试和性能基准
|   +-- partners/       # 第三方集成（Daytona、Modal、QuickJS 等）
+-- examples/           # 示例项目——学习实战用法的最佳起点
+-- .github/            # CI/CD 工作流

libs/deepagents/deepagents/
+-- graph.py            # 核心入口：create_deep_agent 函数
+-- _models.py          # 模型字符串解析逻辑
+-- _tools.py           # 工具描述覆写机制
+-- _messages_reducer.py # 增量消息归约器（DeltaChannel 实现）
+-- backends/           # 后端实现——存储抽象层
|   +-- protocol.py     # BackendProtocol 接口定义
|   +-- state.py        # StateBackend——内存/临时存储
|   +-- filesystem.py   # FilesystemBackend——本地文件系统
|   +-- local_shell.py  # LocalShellBackend——本地 Shell + 文件系统
|   +-- composite.py    # CompositeBackend——路径路由组合
|   +-- store.py        # StoreBackend——LangGraph BaseStore
|   +-- sandbox.py      # BaseSandbox——沙箱基类
|   +-- langsmith.py    # LangSmithSandbox——远程安全执行
|   +-- context_hub.py  # ContextHubBackend——LangSmith Hub
+-- middleware/          # 中间件——扩展机制核心
|   +-- filesystem.py   # 文件系统中间件 + 内置文件工具
|   +-- subagents.py    # 同步子代理中间件
|   +-- async_subagents.py # 异步子代理中间件
|   +-- memory.py       # 记忆中间件
|   +-- skills.py       # 技能中间件
|   +-- summarization.py # 对话摘要中间件
|   +-- rubric.py       # 自我评估迭代中间件
|   +-- permissions.py  # 权限模块（向后兼容重导出）
|   +-- patch_tool_calls.py # 修补悬空工具调用
+-- profiles/           # 配置档案
    +-- harness/        # Harness Profile——运行时行为调优
    +-- provider/       # Provider Profile——模型构造参数

# 使用 uv（推荐——依赖解析速度远优于 pip）
uv add deepagents

# 或使用 pip
pip install deepagents

import deepagents
print(deepagents.__version__)  # 应输出 '0.6.8' 或更高

from deepagents import create_deep_agent

# 创建智能体——使用模型字符串格式
agent = create_deep_agent(
    model="openai:gpt-5.5",
)

# 调用智能体
result = agent.invoke({"messages": "你好，请介绍一下你自己"})
print(result["messages"][-1].content)

from deepagents import create_deep_agent
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息。"""
    # 实际使用中替换为真实 API 调用
    return f"{city}今天晴，温度25度"

agent = create_deep_agent(
    model="openai:gpt-5.5",
    tools=[get_weather],
    system_prompt="你是一个天气助手，帮助用户查询天气信息。",
)

result = agent.invoke({"messages": "北京今天天气怎么样？"})
print(result["messages"][-1].content)

from deepagents import create_deep_agent

agent = create_deep_agent(model="openai:gpt-5.5")

# 流式输出
for chunk in agent.stream(
    {"messages": "请写一首关于春天的诗"},
    config={"configurable": {"thread_id": "poem-session"}},
    stream_mode="values",
):
    if "messages" in chunk:
        messages = chunk["messages"]
        last_msg = messages[-1]
        if hasattr(last_msg, "content") and last_msg.content:
            print(last_msg.content, end="", flush=True)

import asyncio
from deepagents import create_deep_agent

async def main():
    agent = create_deep_agent(model="openai:gpt-5.5")
    result = await agent.ainvoke({"messages": "你好"})
    print(result["messages"][-1].content)

asyncio.run(main())

def create_deep_agent(
    # === 核心配置 ===
    model: str | BaseChatModel | None = None,       # 智能体的"大脑"
    tools: Sequence[BaseTool | Callable | dict] | None = None,  # 自定义工具
    system_prompt: str | SystemMessage | None = None,  # 自定义系统提示词
    *,
    # === 高级特性 ===
    middleware: Sequence[AgentMiddleware] = (),         # 额外中间件
    subagents: Sequence[SubAgent | CompiledSubAgent | AsyncSubAgent] | None = None,
    skills: list[str] | None = None,                   # 技能路径
    memory: list[str] | None = None,                   # 记忆文件路径
    permissions: list[FilesystemPermission] | None = None,
    # === 基础设施 ===
    backend: BackendProtocol | BackendFactory | None = None,
    checkpointer: Checkpointer | None = None,
    store: BaseStore | None = None,
    cache: BaseCache | None = None,
    # === 行为控制 ===
    interrupt_on: dict[str, bool | InterruptOnConfig] | None = None,
    response_format: ResponseFormat | type | dict | None = None,
    # === 高级定制 ===
    state_schema: type[DeepAgentState] | None = None,
    context_schema: type[ContextT] | None = None,
    debug: bool = False,
    name: str | None = None,
) -> CompiledStateGraph

USER（用户 system_prompt）—— 最前面，获得最高注意力
    |
BASE 或 CUSTOM（默认或自定义基础提示词）
    |
SUFFIX（HarnessProfile 尾部提示词）—— 最后面，最接近对话

Base Stack（基础栈——从外到内）：
  1. TodoListMiddleware          -- 任务规划，最先执行
  2. SkillsMiddleware            -- 技能加载
  3. FilesystemMiddleware        -- 文件系统（必需，不可排除）
  4. SubAgentMiddleware          -- 同步子代理（有子代理时必需）
  5. SummarizationMiddleware     -- 对话摘要
  6. PatchToolCallsMiddleware    -- 修补悬空工具调用
  7. AsyncSubAgentMiddleware     -- 异步子代理

  <-- 用户自定义 middleware 插入位置 -->

Tail Stack（尾部栈——从外到内）：
  8. Harness Profile extra_middleware
  9. _ToolExclusionMiddleware    -- 工具排除
  10. AnthropicPromptCachingMiddleware
  11. MemoryMiddleware           -- 记忆加载
  12. HumanInTheLoopMiddleware   -- 人机协作（最外层防线）

class DeepAgentState(AgentState):
    messages: Required[Annotated[
        list[AnyMessage],
        DeltaChannel(_messages_delta_reducer, snapshot_frequency=50)
    ]]

# 方式 1：字符串格式（强烈推荐——自动应用 Profile）
agent = create_deep_agent(model="openai:gpt-5.5")

# 方式 2：预初始化实例（跳过 Profile 自动应用）
from langchain.chat_models import init_chat_model
model = init_chat_model("anthropic:claude-sonnet-4-6", temperature=0.0)
agent = create_deep_agent(model=model)

# 方式 3：None（已弃用——会触发 DeprecationWarning）
agent = create_deep_agent(model=None)

用户传入 "openai:gpt-5.5"
    ↓
init_chat_model("openai:gpt-5.5", **provider_profile_kwargs)
    ↓
ProviderProfile 查找（优先级递减）：
  第1优先级：精确匹配模型全名 "openai:gpt-5.5"
  第2优先级：精确匹配 provider "openai"（含别名匹配）
  第3优先级：无匹配 → 使用默认参数
    ↓
返回 BaseChatModel 实例

from deepagents import register_provider_profile, ProviderProfile

register_provider_profile(
    "my-provider",
    ProviderProfile(
        init_kwargs={"temperature": 0.0, "max_tokens": 4096},
    ),
)

# 切换回 Chat Completions API
model = init_chat_model("openai:gpt-5.5", use_responses_api=False)

# 零数据留存（ZDR）
model = init_chat_model(
    "openai:gpt-5.5",
    use_responses_api=True,
    store=False,
    include=["reasoning.encrypted_content"],
)

@dataclass(frozen=True)
class HarnessProfile:
    base_system_prompt: str | None = None       # 完全替换 BASE 提示词
    system_prompt_suffix: str | None = None     # 追加到末尾（最常用！）
    excluded_tools: list[str] | None = None     # 排除内置工具
    excluded_middleware: list[type | str] | None = None  # 排除中间件
    extra_middleware: list[AgentMiddleware] | None = None  # 额外中间件
    general_purpose_subagent: GeneralPurposeSubagentProfile | None = None
    tool_description_overrides: dict[str, str] | None = None  # 覆写工具描述

from deepagents import register_harness_profile, HarnessProfile, GeneralPurposeSubagentProfile

register_harness_profile(
    "anthropic:claude-haiku-4-5",
    HarnessProfile(
        system_prompt_suffix="使用简洁的回答风格。不超过3句话，除非用户要求详细说明。",
        general_purpose_subagent=GeneralPurposeSubagentProfile(
            description="轻量级研究助手。用于搜索和收集信息的简单任务。",
        ),
        excluded_tools=["execute"],  # Haiku 不需要 Shell
    ),
)

profile = HarnessProfile(
    general_purpose_subagent=GeneralPurposeSubagentProfile(enabled=False),
)

class LsSchema(BaseModel):
    path: str  # 绝对路径

class ReadFileSchema(BaseModel):
    file_path: str   # 绝对路径
    offset: int = 0  # 起始行号（0索引）
    limit: int = 100 # 默认100行

class WriteFileSchema(BaseModel):
    file_path: str
    content: str

class EditFileSchema(BaseModel):
    file_path: str      # 绝对路径
    old_string: str     # 精确匹配文本
    new_string: str     # 替换文本
    replace_all: bool = False

class GlobSchema(BaseModel):
    pattern: str          # glob 模式
    path: str | None = None  # 搜索根目录

class GrepSchema(BaseModel):
    pattern: str           # 字面量文本，非正则
    path: str | None = None
    glob: str | None = None  # 文件过滤
    output_mode: Literal["files_with_matches", "content", "count"] = "files_with_matches"

class ExecuteSchema(BaseModel):
    command: str
    timeout: int | None = None

class BackendProtocol:
    def ls(self, path: str) -> LsResult | list[FileInfo]: ...
    def read(self, path: str) -> ReadResult: ...
    def write(self, path: str, content: str | bytes) -> WriteResult: ...
    def edit(self, path: str, old: str, new: str, replace_all: bool) -> EditResult: ...
    def glob(self, pattern: str, path: str | None) -> GlobResult: ...
    def grep(self, pattern: str, path: str | None, ...) -> GrepResult: ...
    def download_files(self, paths: list[str]) -> list[FileDownloadResponse]: ...
    def upload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]: ...

class SandboxBackendProtocol(BackendProtocol):
    def execute(self, command: str, timeout: int | None) -> ExecuteResponse: ...

from deepagents.backends import StateBackend
agent = create_deep_agent(model="openai:gpt-5.5", backend=StateBackend())
# 通过 files 参数预加载文件
result = agent.invoke(
    {"messages": "读取 /data/task.md"},
    files={"/data/task.md": FileData(content="# 任务\n...", encoding="utf-8")},
)

from deepagents.backends import FilesystemBackend
agent = create_deep_agent(
    model="openai:gpt-5.5",
    backend=FilesystemBackend(root_dir="/path/to/project"),
)

from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
composite = CompositeBackend(
    default=StateBackend(),          # 临时工作文件
    routes={"/memories/": StoreBackend()},  # 记忆持久化
)

from deepagents.backends import StoreBackend
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
agent = create_deep_agent(
    model="openai:gpt-5.5",
    backend=StoreBackend(store=store),
    store=store,  # 需要是同一个实例
)

请求方向 ——>
    +------------------------------------------+
    |  中间件A  before（先执行）                 |
    |    +----------------------------------+  |
    |    |  中间件B  before                 |  |
    |    |    +------------------------+   |  |
    |    |    |    核心（模型调用）      |   |  |
    |    |    +------------------------+   |  |
    |    |  中间件B  after（后执行）         |  |
    |    +----------------------------------+  |
    |  中间件A  after（后执行）                  |
    +------------------------------------------+
                                    <—— 响应方向

from langchain.agents.middleware.types import AgentMiddleware, ModelRequest

class LoggingMiddleware(AgentMiddleware):
    def wrap_model_call(self, request: ModelRequest, handler):
        messages = request.state.get("messages", [])
        print(f"[日志] 消息数: {len(messages)}")
        response = handler(request)  # 转交给内层
        if hasattr(response, 'tool_calls') and response.tool_calls:
            print(f"[日志] 工具调用: {[tc['name'] for tc in response.tool_calls]}")
        return response

agent = create_deep_agent(model="openai:gpt-5.5", middleware=[LoggingMiddleware()])

from langchain_core.messages import ToolMessage

class RetryMiddleware(AgentMiddleware):
    def __init__(self, max_retries: int = 3): self.max_retries = max_retries

    def after_tool_call(self, result, *, tool_call_id, tool_name, state, runtime):
        if "error" in str(result).lower():
            state["messages"].append(ToolMessage(
                content=f"工具调用失败: {result}\n请重试。",
                tool_call_id=tool_call_id,
            ))
        return result

register_harness_profile("lightweight", HarnessProfile(
    excluded_middleware=["SummarizationMiddleware", "TodoListMiddleware"],
))

from deepagents import create_deep_agent, SubAgent, FilesystemPermission

researcher: SubAgent = {
    "name": "researcher",
    # description 是最重要的字段——主智能体据此决定何时委托
    "description": (
        "研究特定主题并返回结构化发现。"
        "用于需要深度搜索和分析的任务。"
        "不要用于简单问答——主智能体直接回答即可。"
    ),
    "system_prompt": "你是研究助手。搜索信息后返回结构化结果。",
    "model": "anthropic:claude-haiku-4-5",
    "tools": [web_search],
    "skills": ["/skills/research/"],
    "permissions": [FilesystemPermission(operations=["read"], paths=["/research/**"], mode="allow")],
}

agent = create_deep_agent(model="openai:gpt-5.5", subagents=[researcher])

from deepagents import create_deep_agent, AsyncSubAgent

processor: AsyncSubAgent = {
    "name": "data-processor",
    "description": "在远程处理大型数据集。",
    "graph_id": "data-processing-graph",
    "url": "https://my-server.example.com",
    "headers": {"Authorization": "Bearer xxx"},
}

start_async_task → task_id → 继续工作 → check_async_task → 完成/调整

/skills/user/web-research/
+-- SKILL.md          # 必需：YAML 元数据 + Markdown 指令
+-- helper.py         # 可选：辅助文件

---
name: web-research
description: 结构化网络研究方法论
allowed-tools: web_search read_file
---

# Web Research Skill

## 何时使用
- 用户要求研究主题
- 需要收集综合多个信息源

## 工作流程
1. 明确研究目标
2. 使用 web_search 初步搜索
3. 进行 2-3 次针对性搜索
4. 综合发现，标注来源
5. 返回结构化结果

agent = create_deep_agent(
    model="openai:gpt-5.5",
    skills=["/skills/user/", "/skills/project/"],  # 后加载覆盖先加载
    backend=FilesystemBackend(root_dir="/my/project"),
)

# 项目概述
电商后端服务，FastAPI + SQLAlchemy。

## 构建/测试
- 构建：make build
- 测试：make test

## 代码风格
- Google 风格 docstring
- 行宽 120 字符

agent = create_deep_agent(
    model="openai:gpt-5.5",
    memory=["./AGENTS.md", "~/.deepagents/AGENTS.md"],
    backend=FilesystemBackend(root_dir="/"),
)

from deepagents import FilesystemPermission

permissions = [
    FilesystemPermission(operations=["read"], paths=["/project/**"], mode="allow"),
    FilesystemPermission(operations=["write"], paths=["/secrets/**"], mode="deny"),
    FilesystemPermission(operations=["write"], paths=["/production/**"], mode="interrupt"),
]

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="openai:gpt-5.5",
    interrupt_on={
        "write_file": True,     # 写文件前审批
        "edit_file": True,      # 编辑文件前审批
        "execute": True,        # 执行命令前审批
    },
    checkpointer=checkpointer,  # HITL 需要 checkpointer
)

from langchain.agents.middleware import InterruptOnConfig

agent = create_deep_agent(
    model="openai:gpt-5.5",
    interrupt_on={
        "execute": InterruptOnConfig(
            description="Shell 命令执行审批",
            approve_label="允许执行",
            reject_label="拒绝执行",
        ),
    },
)

智能体请求执行工具 → HITL 拦截 → 用户审批/编辑/拒绝 → 继续/终止

from deepagents.middleware.summarization import SummarizationMiddleware
from deepagents.backends import FilesystemBackend

summ = SummarizationMiddleware(
    model="openai:gpt-5.5-mini",
    backend=FilesystemBackend(root_dir="/data"),
    trigger=("fraction", 0.85),    # 窗口 85% 时触发
    keep=("fraction", 0.10),       # 保留最近 10%
)

agent = create_deep_agent(model="openai:gpt-5.5", middleware=[summ])

trigger=("tokens", 8000)      # 按 token 数
trigger=("messages", 50)       # 按消息数
trigger=("fraction", 0.85)     # 按窗口比例

from deepagents.middleware.summarization import TruncateArgsSettings

summ = SummarizationMiddleware(
    model="openai:gpt-5.5-mini",
    trigger=("fraction", 0.85),
    keep=("fraction", 0.10),
    truncate_args=TruncateArgsSettings(
        trigger=("fraction", 0.60),  # 60% 时开始截断
        keep=("fraction", 0.20),     # 保留最近 20% 不截断
        max_length=500,               # 参数最大长度
    ),
)

智能体完成任务 → 评审子代理评估（对照 rubric）
    ↓
satisfied → 返回结果
needs_revision → 注入反馈，继续迭代
failed/max_iterations → 终止

from deepagents import create_deep_agent, RubricMiddleware

rubric = RubricMiddleware(
    rubric="""
    - 代码通过所有测试
    - 符合项目风格指南
    - 包含错误处理
    - 不引入新 lint 警告
    """,
    model="openai:gpt-5.5-mini",
    max_iterations=3,
)

agent = create_deep_agent(model="openai:gpt-5.5", middleware=[rubric])

# stream_mode="values" -- 每步返回完整状态
for chunk in agent.stream({"messages": "你好"}, stream_mode="values"):
    print(chunk["messages"][-1].content)

# stream_mode="updates" -- 仅增量更新
for chunk in agent.stream({"messages": "你好"}, stream_mode="updates"):
    print(chunk)

from langgraph.checkpoint.memory import MemorySaver

agent = create_deep_agent(
    model="openai:gpt-5.5",
    checkpointer=MemorySaver(),
)

# 第一次对话
agent.invoke({"messages": "我叫张三"}, config={"configurable": {"thread_id": "user-1"}})
# 同一会话继续
agent.invoke({"messages": "我叫什么名字？"}, config={"configurable": {"thread_id": "user-1"}})

from langgraph.store.memory import InMemoryStore
agent = create_deep_agent(model="openai:gpt-5.5", store=InMemoryStore())

content-builder-agent/
+-- AGENTS.md              # 品牌声音和写作标准
+-- content_writer.py      # 主程序
+-- subagents.yaml         # 子代理配置
+-- skills/
    +-- blog-post/SKILL.md
    +-- social-media/SKILL.md

from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend

agent = create_deep_agent(
    memory=["./AGENTS.md"],
    skills=["./skills/"],
    tools=[generate_cover, generate_social_image],
    subagents=load_subagents("subagents.yaml"),
    backend=FilesystemBackend(root_dir=EXAMPLE_DIR),
)

agent = create_deep_agent(
    model=init_chat_model("anthropic:claude-sonnet-4-5", temperature=0.0),
    tools=[tavily_search, think_tool],
    system_prompt=INSTRUCTIONS,
    subagents=[{
        "name": "research-agent",
        "description": "研究主题并返回发现。每次只给一个主题。",
        "system_prompt": RESEARCHER_INSTRUCTIONS,
        "tools": [tavily_search, think_tool],
    }],
)

# Coding Agent

## 工作流
### 阶段1：规划 → 探索仓库 → 定位文件 → write_todos
### 阶段2：实施 → 逐步修改 → 运行测试
### 阶段3：审查 → 完整测试 → Linter → 端到端审查
### 阶段4：交付 → 提交 → 总结

用户请求 → LangGraph Platform → Deep Agent → Backend（沙箱/远程）→ LangSmith（追踪监控）

deepagents init my-agent
deepagents dev          # 本地开发
deepagents deploy       # 部署到 LangGraph Platform

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.store.postgres.aio import AsyncPostgresStore

agent = create_deep_agent(
    model="openai:gpt-5.5",
    checkpointer=AsyncPostgresSaver(conn_string="postgresql://..."),
    store=AsyncPostgresStore(conn_string="postgresql://..."),
)

os.environ["LANGSMITH_API_KEY"] = "your-key"
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "my-agent"

+----------------------------------------------------+
|                  create_deep_agent                  |
|                                                    |
|  +-----------+  +--------------+  +----------+    |
|  |   Model   |  | System Prompt|  |   Tools  |    |
|  | (resolved)|  | (assembled)  |  | (merged) |    |
|  +-----+-----+  +------+------+  +-----+----+    |
|        |               |               |           |
|  +-----+---------------+---------------+------+   |
|  |            Middleware Stack（12层）         |   |
|  +--------------------------------------------+   |
|                                                    |
|  +--------------------------------------------+   |
|  |                  Backend                    |   |
|  +--------------------------------------------+   |
|                                                    |
|  +--------------------------------------------+   |
|  |       LangGraph Runtime                    |   |
|  |  Checkpointer / Store / Cache / Streaming  |   |
|  +--------------------------------------------+   |
+----------------------------------------------------+

from deepagents.backends.protocol import BackendProtocol, ReadResult, WriteResult

class RemoteAPIBackend(BackendProtocol):
    def __init__(self, api_url: str, api_key: str): ...
    def ls(self, path: str): ...
    def read(self, path: str): ...
    def write(self, path: str, content: str | bytes): ...
    def edit(self, path: str, old: str, new: str, replace_all: bool): ...
    def glob(self, pattern: str, path: str | None): ...
    def grep(self, pattern: str, path: str | None, ...): ...
    def download_files(self, paths: list[str]): ...
    def upload_files(self, files: list[tuple[str, bytes]]): ...

[project.entry-points."deepagents.provider_profiles"]
my_profile = "my_package:register_my_provider"

[project.entry-points."deepagents.harness_profiles"]
my_harness = "my_package:register_my_harness"

from deepagents import (
    create_deep_agent, DeepAgentState,
    SubAgent, CompiledSubAgent, AsyncSubAgent,
    SubAgentMiddleware, AsyncSubAgentMiddleware,
    FilesystemMiddleware, FilesystemPermission,
    MemoryMiddleware, RubricMiddleware,
    HarnessProfile, HarnessProfileConfig,
    GeneralPurposeSubagentProfile,
    ProviderProfile,
    register_harness_profile, register_provider_profile,
    __version__,
)

from deepagents.backends import (
    StateBackend, FilesystemBackend, LocalShellBackend,
    CompositeBackend, StoreBackend,
    LangSmithSandbox, ContextHubBackend,
    BackendProtocol,
)

入门（1-2天）：第二章（快速上手）→ 第三章（API详解）→ 第六章（工具）
进阶（3-5天）：第七章（Backend）→ 第九章（子代理）→ 第十一/十二章（Skills/Memory）
高级（5-7天）：第八章（中间件）→ 第十三/十四章（安全/HITL）→ 第十五章（压缩）
专家（持续深入）：第四/五章（Profile）→ 第十六章（Rubric）→ 第十九/二十章（部署/架构）