DeepAgents 完全指南

┌─────────────────────────────────┐
│         LangSmith               │  ← 可观测性、评测、监控
├─────────────────────────────────┤
│         Deep Agents             │  ← 完整 harness（本文主角）
├─────────────────────────────────┤
│    LangChain create_agent       │  ← 轻量 agent（无内置文件系统/子agent）
├─────────────────────────────────┤
│         LangGraph               │  ← 图运行时（状态、节点、边）
└─────────────────────────────────┘

# 推荐
uv add deepagents

# 或者
pip install deepagents

┌─ 基础栈（Base stack）────────────────────────────────┐
│ 1. SkillsMiddleware          （仅当传入 skills）      │
│ 2. FilesystemMiddleware      （始终存在）             │
│ 3. SubAgentMiddleware        （有子智能体时存在）      │
│ 4. SummarizationMiddleware   （始终存在）             │
│ 5. PatchToolCallsMiddleware  （始终存在）             │
│ 6. AsyncSubAgentMiddleware   （有异步子智能体时存在）   │
│ 7. 你的自定义 middleware       （插入到这里）           │
├─ 尾部栈（Tail stack）────────────────────────────────┤
│ 8. Harness profile 额外中间件                          │
│ 9. 工具排除过滤（excluded_tools）                      │
│ 10. AnthropicPromptCachingMiddleware（非 Anthropic noop）│
│ 11. BedrockPromptCachingMiddleware（未装则跳过）        │
│ 12. FireworksPromptCachingMiddleware（未装则跳过）      │
│ 13. MemoryMiddleware         （仅当传入 memory）       │
│ 14. HumanInTheLoopMiddleware （仅当传入 interrupt_on）  │
└──────────────────────────────────────────────────────┘

system_prompt = USER → BASE → SUFFIX

from deepagents import create_deep_agent

agent = create_deep_agent(
    model=...,           # 模型
    tools=...,           # 自定义工具
    system_prompt=...,   # 系统提示词
    middleware=...,      # 额外中间件
    subagents=...,       # 子智能体配置
    skills=...,          # 技能目录路径
    memory=...,          # AGENTS.md 记忆文件路径
    permissions=...,     # 文件系统权限规则
    backend=...,         # 文件系统后端
    interrupt_on=...,    # 人工审批的工具名映射
    response_format=..., # 结构化输出 schema
    state_schema=...,    # 自定义图状态 schema
    context_schema=...,  # 运行时上下文 schema
    checkpointer=...,    # LangGraph checkpointer（持久化）
    store=...,           # LangGraph store（跨会话存储）
    debug=...,           # 调试模式
    name=...,            # 图的名称
    cache=...,           # 缓存
)

# 1. 字符串（推荐——使用 provider:model 格式）
agent = create_deep_agent(model="openai:gpt-5.5")

# 2. 预初始化的 LangChain ChatModel 实例
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-5.5", temperature=0)
agent = create_deep_agent(model=model)

# 3. 传入 None（已弃用，默认是 claude-sonnet-4-6）
# 从 0.5.3 开始弃用，将在 1.0.0 中移除

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-5.5", use_responses_api=False)
agent = create_deep_agent(model=model)

model = init_chat_model(
    "openai:gpt-5.5",
    use_responses_api=True,
    store=False,
    include=["reasoning.encrypted_content"],
)
agent = create_deep_agent(model=model)

from langchain_openai import ChatOpenAI
from deepagents import create_deep_agent

model = ChatOpenAI(
    model="qwen-plus",  # 或 qwen3-max、qwen-max 等
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

agent = create_deep_agent(model=model, tools=[my_tools])

from langchain.tools import tool

@tool
def search_database(query: str) -> str:
    """搜索产品数据库，返回匹配结果"""
    return db.search(query)

@tool
def generate_report(data: dict) -> str:
    """根据数据生成分析报告"""
    return report_gen.run(data)

agent = create_deep_agent(model=model, tools=[search_database, generate_report])

agent = create_deep_agent(
    model=model,
    system_prompt="你是一个数据分析助手。在给出结论之前，先用 Python 工具验证数据。",
)

from pydantic import BaseModel, Field

class ResearchReport(BaseModel):
    """研究报告的结构化输出"""
    title: str = Field(description="报告标题")
    summary: str = Field(description="核心发现摘要")
    key_findings: list[str] = Field(description="关键发现列表")
    confidence: float = Field(description="置信度 0-1")
    sources: list[str] = Field(description="参考来源")

agent = create_deep_agent(
    model=model,
    tools=[web_search],
    response_format=ResearchReport,
)

result = agent.invoke({"messages": [{"role": "user", "content": "调研 LangGraph 的最新特性"}]})

# 访问结构化输出
report = result["structured_response"]
print(report.title)
print(report.confidence)

from langgraph.checkpoint.memory import MemorySaver

agent = create_deep_agent(
    model=model,
    tools=tools,
    checkpointer=MemorySaver(),  # 内存持久化
)

# 使用 thread_id 区分不同会话
config = {"configurable": {"thread_id": "user-123-session-456"}}
result = agent.invoke({"messages": [...]}, config=config)

from langgraph.checkpoint.postgres import PostgresSaver

checkpointer = PostgresSaver.from_conn_string("postgresql://...")
agent = create_deep_agent(model=model, checkpointer=checkpointer)

class BackendProtocol(Protocol):
    def ls(self, path: str) -> list[str]: ...
    def read(self, file_path: str, offset: int = None, limit: int = None) -> str: ...
    def write(self, file_path: str, content: str) -> None: ...
    def edit(self, file_path: str, old_string: str, new_string: str, replace_all: bool = False) -> None: ...
    def delete(self, file_path: str) -> None: ...
    def glob(self, pattern: str, path: str = "/") -> list[str]: ...
    def grep(self, pattern: str, path: str = "/", glob: str = None) -> list[str]: ...

# 不传 backend 参数，默认使用 StateBackend
agent = create_deep_agent(model=model, tools=tools)

# 文件存在 state["files"] 字典中，key 是路径，value 是内容字符串
result = agent.invoke({"messages": [{"role": "user", "content": "写一个 hello.py"}]})
print(result["files"])  # {"/hello.py": "print('hello world')"}

from deepagents.backends import CompositeBackend, StateBackend, StoreBackend, SandboxBackend

backend = CompositeBackend(
    default=StateBackend(),
    routes={
        "/workspace": StateBackend(),
        "/artifacts": StoreBackend(store=my_store, namespace=("artifacts",)),
        "/memory": StoreBackend(store=my_store, namespace=("agent_memory",)),
        "/sandbox": SandboxBackend(root_dir="/tmp/agent_sandbox", timeout=60),
        "/data": CompositeBackend.FilesystemBackend(root_dir="/path/to/shared_data"),
    },
)

agent = create_deep_agent(model=model, backend=backend)

from deepagents import create_deep_agent
from deepagents.backends import SandboxBackend

backend = SandboxBackend(
    root_dir="/tmp/agent_workspace",
    timeout=120,          # 命令超时（秒）
    max_output_size=50000,
)

agent = create_deep_agent(model=model, backend=backend)

# agent 可以调用 execute("pip install requests") 或 execute("python analysis.py")

from deepagents.backends.filesystem import FilesystemBackend

backend = FilesystemBackend(root_dir="/Users/zata/code/my-project")

agent = create_deep_agent(model=model, backend=backend)

# agent 可以 read_file / write_file / edit_file / glob / grep 操作真实文件

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

store = InMemoryStore()
backend = StoreBackend(store=store, namespace=("shared_files",))

agent = create_deep_agent(model=model, backend=backend)

# 文件在不同 thread 之间共享（只要使用同一个 store）

from langgraph.store.postgres import PostgresStore

store = PostgresStore.from_conn_string("postgresql://...")
backend = StoreBackend(store=store, namespace=("knowledge_base",))
agent = create_deep_agent(model=model, backend=backend)

agent = create_deep_agent(
    model=model,
    backend=backend,
    permissions=[
        {"path": "/workspace/*", "operations": ["read", "write", "edit"], "mode": "allow"},
        {"path": "/memory/*", "operations": ["read", "write"], "mode": "allow"},
        {"path": "/system/*", "mode": "deny"},
        {"path": "/*", "operations": ["read"], "mode": "allow"},
        {"path": "/*", "operations": ["write", "edit", "delete"], "mode": "interrupt"},
    ],
)

from deepagents import create_deep_agent
from deepagents.subagents import SubAgent

researcher = SubAgent(
    name="research-agent",
    description="负责搜索和整理信息，输出结构化研究报告",
    system_prompt="你是一个信息研究员。用 web_search 和 read_file 工具完成调研任务。",
    model="openai:gpt-5.5",  # 子智能体可以用不同模型
    tools=[web_search],
)

analyst = SubAgent(
    name="data-analyst",
    description="负责数据分析和可视化",
    system_prompt="你是一个数据分析师。用 Python 分析数据，生成报告。",
    model="anthropic:claude-sonnet-4-6",
    tools=[execute_python, read_file],
)

agent = create_deep_agent(
    model="openai:gpt-5.5",
    subagents=[researcher, analyst],
    system_prompt="你是一个团队负责人。用 task 工具委派任务给子智能体。",
)

{
  "tool": "task",
  "args": {
    "description": "调研 LangGraph 最新特性并输出报告",
    "subagent_type": "research-agent"
  }
}

from langgraph.graph import StateGraph, MessagesState, END

# 构建一个自定义图
workflow = StateGraph(MessagesState)
workflow.add_node("analyze", analyze_node)
workflow.add_node("report", report_node)
workflow.set_entry_point("analyze")
workflow.add_edge("analyze", "report")
workflow.add_edge("report", END)
custom_graph = workflow.compile()

agent = create_deep_agent(
    model=model,
    subagents=[
        {
            "name": "custom-analyst",
            "description": "自定义分析管道",
            "graph": custom_graph,
        }
    ],
)

主 Agent 上下文                          子 Agent 上下文
┌─────────────────────┐                ┌─────────────────────────────┐
│ system prompt       │                │ 子智能体的 system prompt     │
│ user message        │                │ task description (作为输入) │
│ ... 历史消息 ...     │                │ 子智能体的工具和文件系统      │
│ tool: task(...)     │──委派──────────→│                             │
│                     │                │  (子 agent 内部循环)         │
│                     │←───────────────│ 最终输出（文本摘要）         │
│ tool result         │                └─────────────────────────────┘
│ ... 继续 ...        │
└─────────────────────┘

import asyncio

# 同一 agent 的多个独立任务并行执行
results = await asyncio.gather(
    agent.ainvoke({"messages": [("user", "研究主题 A")]}, config),
    agent.ainvoke({"messages": [("user", "研究主题 B")]}, config),
)

deep_researcher = SubAgent(
    name="deep-researcher",
    model="openai:gpt-5.5",
    system_prompt="深入研究并写报告",
    subagents={
        "subagents": [
            SubAgent(
                name="fact-checker",
                model="openai:gpt-5.5",
                system_prompt="验证研究结论中的关键事实",
            ),
        ],
    },
)

压缩前：
[sys] [msg1] [msg2] [msg3] [msg4] [msg5] ... [msgN]
       ↑ 全部保留在上下文中

压缩后：
[sys] [摘要消息（包含历史概要+文件路径）] [msgN-2] [msgN-1] [msgN]
                              ↑ 完整历史在 /conversation_history/

from deepagents import create_deep_agent
from deepagents.middleware.summarization import SummarizationMiddleware

agent = create_deep_agent(
    model=model,
    tools=tools,
    middleware=[
        SummarizationMiddleware(
            token_threshold=100000,   # 超过 100k token 触发压缩
            keep_messages=10,          # 保留最近 10 条消息
        )
    ],
)

# 你可以自定义阈值
from deepagents.middleware.filesystem import FilesystemMiddleware

fs_mw = FilesystemMiddleware(
    large_output_threshold=10000,  # 10000 token 超限
    large_output_dir="/my_tool_results",
)

agent = create_deep_agent(model=model, middleware=[fs_mw])

agent = create_deep_agent(
    model=model,
    tools=tools,
    memory="/path/to/AGENTS.md",  # 启动时注入到系统提示词
)

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

store = InMemoryStore()
memory_backend = StoreBackend(store=store, namespace=("agent_memory",))

agent = create_deep_agent(
    model=model,
    backend=memory_backend,
    system_prompt="你是一个有记忆的助手。在 /memory 目录中维护长期记忆。",
)

my_skills/
├── code-review/
│   └── SKILL.md
├── data-viz/
│   ├── SKILL.md
│   └── templates/
│       └── chart.py
└── web-research/
    └── SKILL.md

---
name: code-review
description: 对代码库进行系统性审查，检查安全性、性能、可维护性
---

## 步骤

1. 读取目标文件列表（用 glob 工具）
2. 逐一读取关键文件
3. 检查以下方面：
   - 安全性：SQL注入、XSS、不安全的反序列化
   - 性能：N+1 查询、内存泄漏、不必要的全表扫描
   - 可维护性：函数长度、重复代码、缺失的类型注解
4. 用 write_file 工具输出审查报告到 /reports/code-review-{date}.md

## 工具

优先使用 grep 搜索特定模式，用 read_file 读取上下文。

agent = create_deep_agent(
    model=model,
    tools=tools,
    skills="/path/to/my_skills",
)

my_skills/
└── report-generator/
    ├── SKILL.md
    ├── templates/
    │   ├── report_template.md
    │   └── chart_template.py
    └── config.yaml

---
name: evaluate-research
description: 用评分标准评估研究报告质量
---

## 评分标准

按以下维度评估研究报告（每项 1-5 分）：

1. **事实准确性**：关键事实是否有可靠来源支撑
2. **逻辑连贯性**：结论是否从证据自然推出
3. **完整性**：是否遗漏了重要角度
4. **可操作性**：读者能否据此做出决策

## 流程

1. 阅读研究报告全文
2. 逐项打分并给出具体依据
3. 汇总总分和改进建议

agent = create_deep_agent(
    model=model,
    tools=[send_email, delete_record, web_search, write_file],
    interrupt_on={
        "send_email": {"allowed_decisions": ["approve", "edit", "reject"]},
        "delete_record": {"allowed_decisions": ["approve", "reject"]},
    },
)

# 第一次 invoke 会触发中断
result = agent.invoke({"messages": [...]})

# 检查是否中断
if result["__interrupt__"]:
    # 获取中断信息
    interrupt_info = result["__interrupt__"][0]
    print(f"工具: {interrupt_info.value['action_request']['action']}")
    print(f"参数: {interrupt_info.value['action_request']['args']}")

    # 人工决策
    human_decision = interrupt_info.value["resume"]
    # 或者用 resume 重新启动
    resumed = agent.invoke(
        {"__interrupt__": [{"resume": {"type": "accept"}}]},
        config=config,
    )

agent = create_deep_agent(
    model=model,
    backend=backend,
    permissions=[
        {"path": "/workspace/*", "mode": "allow"},
        {"path": "/*", "operations": ["write", "edit", "delete"], "mode": "interrupt"},
    ],
)

config = {"configurable": {"thread_id": "session-1"}}

# 第一次调用，触发中断
result1 = agent.invoke({"messages": [...]}, config=config)

# 检查中断并恢复
if "__interrupt__" in result1:
    decision = input("批准/编辑/拒绝？(approve/edit/reject): ")
    result2 = agent.invoke(
        Command(resume={"type": decision}),
        config=config,
    )

from langchain_core.messages import AIMessageChunk

config = {"configurable": {"thread_id": "session-1"}}

for event in agent.stream(
    {"messages": [{"role": "user", "content": "分析这份数据"}]},
    config=config,
    stream_mode="messages",
):
    # event 是 (message_chunk, metadata) 元组
    msg_chunk, metadata = event
    if hasattr(msg_chunk, "content") and isinstance(msg_chunk.content, str):
        print(msg_chunk.content, end="", flush=True)

for event in agent.stream(
    input_data,
    config=config,
    stream_mode="updates",
):
    # 每个节点的更新
    for node_name, node_update in event.items():
        print(f"节点: {node_name}")
        if "messages" in node_update:
            for msg in node_update["messages"]:
                print(f"  {msg.type}: {msg.content[:100] if isinstance(msg.content, str) else msg.content}")

for event in agent.stream(
    input_data,
    config=config,
    stream_mode="updates",
):
    for node_name, node_update in event.items():
        if "files" in node_update:
            print(f"文件变化: {node_update['files'].keys()}")

FROM python:3.12-slim
WORKDIR /app

# 安装依赖
COPY pyproject.toml uv.lock ./
RUN pip install uv && uv sync --frozen

# 复制代码
COPY . .

# 暴露端口
EXPOSE 8000

# 启动（LangGraph serve）
CMD ["uv", "run", "langgraph", "serve", "--host", "0.0.0.0", "--port", "8000"]

# 模型 API Key
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-...

# LangSmith（可选，但强烈推荐）
LANGSMITH_TRACING=true
LANGSMITH_API_KEY=lsv2_...
LANGSMITH_PROJECT=my-agent

# 数据库（生产 checkpointer）
DATABASE_URL=postgresql://user:pass@host:5432/db

export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY=your_key
export LANGSMITH_PROJECT=production-agent

from langsmith import Client

client = Client()

# 创建评测数据集
dataset = client.create_dataset("deep-agents-eval")
client.create_examples(
    dataset_id=dataset.id,
    inputs=[
        {"messages": [{"role": "user", "content": "调研 X 并写报告"}]},
        {"messages": [{"role": "user", "content": "修复 bug #123"}]},
    ],
)

# 运行评测
client.evaluate(
    agent,
    data=dataset.id,
    evaluators=[
        # 自定义评估器
        lambda outputs, reference: {"accuracy": compute_match(outputs, reference)},
    ],
)

from deepagents import create_deep_agent
from deepagents.subagents import SubAgent
from langchain.tools import tool

@tool
def web_search(query: str) -> str:
    """搜索互联网获取信息"""
    # 接入 Tavily / Bocha / 百炼 WebSearch MCP
    ...

@tool
def web_extract(url: str) -> str:
    """抽取网页正文内容"""
    ...

research_subagent = SubAgent(
    name="researcher",
    description="搜索和整理信息",
    system_prompt="你是一个信息研究员。用 web_search 和 web_extract 完成调研。",
    model="openai:gpt-5.5",
    tools=[web_search, web_extract],
)

agent = create_deep_agent(
    model="openai:gpt-5.5",
    subagents=[research_subagent],
    system_prompt="""你是一个研究助手。
    收到调研任务后，用 task 工具委派给 researcher 子智能体。
    基于子智能体的输出，综合并生成最终回答。""",
)

from deepagents.backends.filesystem import FilesystemBackend

backend = FilesystemBackend(root_dir="/path/to/repo")

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    backend=backend,
    system_prompt="""你是一个代码编辑助手。
    先用 glob 和 grep 了解代码库结构。
    修改代码前先 read_file 阅读上下文。
    修改后用 execute 工具运行测试验证。
    修改必须最小化，不要过度工程化。""",
)

from deepagents.backends import SandboxBackend

backend = SandboxBackend(
    root_dir="/tmp/data_sandbox",
    timeout=300,
)

agent = create_deep_agent(
    model="openai:gpt-5.5",
    backend=backend,
    system_prompt="""你是一个数据分析师。
    先用 execute("ls /") 查看数据文件。
    用 execute("python -c \"...\\"") 或 execute("head -n 100 data.csv") 探索数据。
    用 execute("python analysis.py") 执行分析脚本。
    用 write_file 生成分析报告。""",
)

import os
from openai import OpenAI
from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI

# 百炼 Responses API 客户端（用于 web_search）
bailian = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

@tool
def bailian_web_search(query: str) -> str:
    """搜索互联网获取最新信息（使用百炼 web_search）"""
    response = bailian.responses.create(
        model="qwen3-max",
        input=query,
        tools=[{"type": "web_search"}],
    )
    # 提取搜索结果
    results = []
    for item in response.output:
        if getattr(item, "type", None) == "web_search_call":
            results.append(str(item))
        elif getattr(item, "type", None) == "message":
            for content in getattr(item, "content", []):
                if hasattr(content, "text"):
                    results.append(content.text)
    return "\n".join(results)

# DeepAgents 走 Chat Completions
model = ChatOpenAI(
    model="qwen-plus",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

agent = create_deep_agent(
    model=model,
    tools=[bailian_web_search],
    system_prompt="你是一个研究助手。用 bailian_web_search 查询最新信息。",
)

# 一键安装
curl -LsSf https://langch.in/dcode | bash

# 在项目目录中启动
cd your-project
dcode

发送 [system, user, assistant(tool_calls), tool(results)] → 模型返回 tool_calls 或 text → 循环

需要 DeepAgents 的 harness 能力？
├─ 是 → 需要 web_search？
│   ├─ 是 → 自己包装搜索为 function tool（推荐）
│   │       或混合模式（DeepAgents + 百炼 Responses API）
│   └─ 否 → 直接用 DeepAgents + Chat Completions
└─ 否 → 需要服务端内置工具（web_search 等）？
    ├─ 是 → 直接用百炼 Responses API
└─ 否 → LangChain create_agent / LangGraph

from deepagents import create_deep_agent
from deepagents.middleware import RubricMiddleware

rubric_mw = RubricMiddleware(
    rubric=[
        {"criterion": "所有关键论断都有来源引用", "weight": 3},
        {"criterion": "结论与证据逻辑一致", "weight": 2},
        {"criterion": "无事实错误", "weight": 3},
    ],
    grader_model="openai:gpt-5.5",
    max_iterations=3,
)

agent = create_deep_agent(
    model="openai:gpt-5.5",
    middleware=[rubric_mw],
    system_prompt="你是一个研究助手，写完报告后自我检查再提交。",
)