第三章：DeepAgents 核心概念

发表于 2026-05-08 更新于 2026-05-09 分类于 python ， rd Changyan：本文字数： 7.5k 阅读时长 ≈ 27 分钟

DeepAgents 是一个 Agent Harness（智能体 harness），在 LangChain 框架和 LangGraph 运行时之上，提供了内置工具、上下文管理、子 Agent 委派、人类审批等开箱即用的能力。

本章系统讲解 DeepAgents 的 12 大核心能力，每个都配有代码示例和 UML 图。

3.1 Models（模型配置）

DeepAgents 支持任何实现了 LangChain Chat Model 接口且支持工具调用的模型。模型以 provider:model 格式指定。

三种模型配置方式

from deepagents import create_deep_agent

# 方式1：简单字符串（最常用）
agent = create_deep_agent(model="openai:gpt-4o")

# 方式2：init_chat_model（支持 provider 特定参数）
from langchain.chat_models import init_chat_model
model = init_chat_model(model="anthropic:claude-sonnet-4-6", thinking_level="medium")
agent = create_deep_agent(model=model)

# 方式3：直接实例化（完全控制）
from langchain_anthropic import ChatAnthropic
model = ChatAnthropic(model="claude-sonnet-4-6", thinking_level="medium")
agent = create_deep_agent(model=model)

已验证的模型（通过官方 Eval 测试）

提供商	推荐模型	文件操作	检索	工具使用	记忆	对话	摘要
Google	`gemini-3.1-pro-preview`	100%	100%	25%	54%	48%	80%
OpenAI	`gpt-5.4`	100%	100%	18%	51%	38%	100%
Anthropic	`claude-sonnet-4-6`	高	高	中	高	中	高
GLM-5	`baseten:zai-org/GLM-5`	92%	100%	87%	44%	29%	60%

关键发现：文件操作和检索是多数模型的强项（90-100%），但工具使用是普遍弱点，GLM-5 以 87% 领先。

运行时动态切换模型

通过中间件实现在运行时切换模型，无需重建 Agent：

from dataclasses import dataclass
from langchain.chat_models import init_chat_model
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable

@dataclass
class Context:
    model: str

@wrap_model_call
def configurable_model(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    model_name = request.runtime.context.model
    model = init_chat_model(model_name)
    return handler(request.override(model=model))

agent = create_deep_agent(
    model="openai:gpt-4o",
    middleware=[configurable_model],
    context_schema=Context,
)

# 用户在运行时选择模型
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Hello!"}]},
    context=Context(model="anthropic:claude-sonnet-4-6"),
)

3.2 Context Engineering（上下文工程）

上下文工程是 在正确的时间、以正确的格式提供正确的信息，让 Agent 可靠地完成任务。这是 DeepAgents 最强大的特性之一。

五类上下文总览

flowchart TD
A["上下文工程"] --> B["输入上下文"]
A --> C["运行时上下文"]
A --> D["上下文压缩"]
A --> E["上下文隔离"]
A --> F["长期记忆"]

B --> B1["系统提示词 + 记忆 + 技能<br/>启动时加载"]
C --> C1["用户元数据 + API Key<br/>每次 invoke 传入"]
D --> D1["大内容卸载 + 历史摘要<br/>到达阈值时触发"]
E --> E1["子 Agent 独立上下文<br/>不污染主 Agent"]
F --> F1["跨会话持久化<br/>StoreBackend"]

style B fill:#e3f2fd
style C fill:#fff9c4
style D fill:#ffebee
style E fill:#f3e5f5
style F fill:#e8f5e9

输入上下文（4个来源）

Agent 启动时的系统提示词由4个来源按固定顺序组装：

顺序	来源	说明
1	自定义 `system_prompt`	你的自定义指令，优先级最高
2	Base Agent Prompt	SDK 内置的行为指导（规划、文件系统、子 Agent）
3	Memory（AGENTS.md）	持久记忆，启动时全部加载
4	Skills（SKILL.md frontmatter）	技能描述，仅加载 frontmatter，完整内容按需加载

系统提示词完整组装顺序：

1. 自定义 system_prompt
2. Base agent prompt
3. 待办列表提示词（write_todos 使用说明）
4. 记忆提示词（AGENTS.md + 使用指南）
5. 技能提示词（技能位置 + frontmatter + 使用说明）
6. 虚拟文件系统提示词（文件系统 + execute 工具文档）
7. 子 Agent 提示词（task 工具使用说明）
8. 用户自定义中间件提示词
9. Human-in-the-loop 提示词（当 interrupt_on 设置时）

运行时上下文

运行时上下文是每次调用时传入的配置，不会自动出现在模型提示词中，除非工具/中间件主动读取：

from dataclasses import dataclass
from langchain.tools import tool, ToolRuntime

@dataclass
class Context:
    user_id: str
    api_key: str

@tool
def fetch_user_data(query: str, runtime: ToolRuntime[Context]) -> str:
    user_id = runtime.context.user_id
    return f"Data for user {user_id}: {query}"

agent = create_deep_agent(
    model="openai:gpt-4o",
    tools=[fetch_user_data],
    context_schema=Context,
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "查询我的最近活动"}]},
    context=Context(user_id="user-123", api_key="sk-..."),
)

关键行为：运行时上下文自动传播给所有子 Agent。

上下文压缩

长任务产生大量工具输出和对话历史。两种内置压缩机制：

卸载（Offloading）：当工具调用输入/结果超过 20,000 tokens 时触发。

摘要（Summarization）：当上下文占满 85% 时，自动压缩旧消息。

flowchart TD
A["Agent 接收工具响应"] --> B{"工具结果 > 20K tokens?"}
B -->|是| C["卸载到虚拟文件系统"]
C --> D["替换为文件路径 + 前10行预览"]
D --> E["继续对话"]
B -->|否| F{"上下文占用 >= 85%?"}
F -->|是| G["自动摘要旧消息"]
G --> H["保留摘要 + 最近10% tokens"]
H --> E
F -->|否| I{"上下文溢出?"}
I -->|是| J["立即摘要 + 重试"]
J --> E
I -->|否| E

摘要双重保障：

组件	作用
上下文内摘要	LLM 生成结构化摘要（包含会话意图、已创建产物、下一步），替换完整历史
文件系统保存	完整的原始对话消息写入文件系统，可随时恢复

可选的摘要工具中间件（deepagents>=1.6.0），让 Agent 在合适的时机（如任务间隙）主动触发摘要：

from deepagents.middleware.summarization import create_summarization_tool_middleware

agent = create_deep_agent(
    model="openai:gpt-4o",
    middleware=[create_summarization_tool_middleware(model, backend)],
)

上下文隔离

子 Agent 是上下文隔离的核心手段 — 子 Agent 在独立上下文中运行，主 Agent 只收到最终摘要。详见 3.5 子 Agent 委派。

3.3 Backends（后端系统）

DeepAgents 给 Agent 提供虚拟文件系统（ls、read_file、write_file、edit_file、glob、grep），底层通过可插拔后端实现。

六种内置后端

flowchart TD
A["虚拟文件系统"] --> B["StateBackend"]
A --> C["FilesystemBackend"]
A --> D["LocalShellBackend"]
A --> E["StoreBackend"]
A --> F["SandboxBackend"]
A --> G["CompositeBackend"]

B --> B1["内存状态<br/>单次会话"]
C --> C1["本地磁盘<br/>virtual_mode 安全"]
D --> D1["本地磁盘 + Shell<br/>⚠️ 无沙箱"]
E --> E1["LangGraph Store<br/>跨会话持久化"]
F --> F1["隔离沙箱<br/>Modal / Daytona / AgentCore"]
G --> G1["路由组合<br/>按路径前缀分发"]

style B fill:#e3f2fd
style C fill:#fff9c4
style D fill:#ffebee
style E fill:#e8f5e9
style F fill:#f3e5f5
style G fill:#e8eaf6

后端	持久性	Shell 访问	隔离性	适用场景
StateBackend	单次会话（检查点）	❌	进程级	临时草稿、中间结果
FilesystemBackend	本地磁盘	❌	`virtual_mode` 路径限制	本地开发、CI、挂载卷
LocalShellBackend	本地磁盘 + Shell	✅	⚠️ 无	仅限可信本地开发
StoreBackend	跨会话（LangGraph Store）	❌	命名空间隔离	长期记忆、LangSmith 部署
SandboxBackend	隔离环境	✅	完全沙箱	生产级代码执行
CompositeBackend	按路由混合	按路由	按路由	混合持久化需求

StoreBackend 命名空间隔离

命名空间控制数据的访问范围，是实现多租户的关键：

from deepagents.backends import StoreBackend

# 用户级别隔离 — 每个用户独立存储
StoreBackend(namespace=lambda rt: (rt.server_info.user.identity,))

# Agent 级别共享 — 同一 Agent 所有用户共享
StoreBackend(namespace=lambda rt: (rt.server_info.assistant_id,))

# 组织级别共享 — 同一组织所有用户共享
StoreBackend(namespace=lambda rt: (rt.context.org_id,))

# 多维度组合 — Agent + 用户双重隔离
StoreBackend(namespace=lambda rt: (
    rt.server_info.assistant_id,
    rt.server_info.user.identity,
))

CompositeBackend 路由组合

最灵活的后端 — 按路径前缀路由到不同后端，长前缀优先：

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

agent = create_deep_agent(
    model="openai:gpt-4o",
    backend=CompositeBackend(
        default=StateBackend(),                      # 未匹配路径 → 内存
        routes={
            "/memories/": StoreBackend(               # /memories/ → 持久化存储
                namespace=lambda rt: (rt.server_info.user.identity,),
            ),
            "/policies/": StoreBackend(               # /policies/ → 组织级存储
                namespace=lambda rt: (rt.context.org_id,),
            ),
        },
    ),
    store=InMemoryStore(),
)

路由行为：/workspace/plan.md → StateBackend（临时），/memories/agent.md → StoreBackend（持久化）。

自定义后端

实现 BackendProtocol 对接任何存储（S3、Postgres 等）：

from deepagents.backends.protocol import (
    BackendProtocol, WriteResult, EditResult,
    LsResult, ReadResult, GrepResult, GlobResult,
)

class S3Backend(BackendProtocol):
    def __init__(self, bucket: str, prefix: str = ""):
        self.bucket = bucket
        self.prefix = prefix.rstrip("/")

    def _key(self, path: str) -> str:
        return f"{self.prefix}{path}"

    def ls(self, path: str) -> LsResult: ...
    def read(self, file_path: str, offset=0, limit=2000) -> ReadResult: ...
    def write(self, file_path: str, content: str) -> WriteResult: ...
    def edit(self, file_path: str, old_string: str, new_string: str, replace_all=False) -> EditResult: ...
    def grep(self, pattern: str, path=None, glob=None) -> GrepResult: ...
    def glob(self, pattern: str, path="/") -> GlobResult: ...

策略钩子（Policy Hooks）

超越路径级别 allow/deny 的高级控制（限流、审计日志、内容检查）：

class GuardedBackend(FilesystemBackend):
    def __init__(self, *, deny_prefixes: list[str], **kwargs):
        super().__init__(**kwargs)
        self.deny_prefixes = [p if p.endswith("/") else p + "/" for p in deny_prefixes]

    def write(self, file_path: str, content: str) -> WriteResult:
        if any(file_path.startswith(p) for p in self.deny_prefixes):
            return WriteResult(error=f"Writes are not allowed under {file_path}")
        return super().write(file_path, content)

3.4 Sandboxes（沙箱执行）

沙箱后端在隔离环境中执行代码，同时提供文件系统工具和 execute Shell 命令工具。是生产级 Agent 的安全基础。

为什么需要沙箱？

风险	说明
任意代码执行	Agent 可能运行危险的 Shell 命令
凭证泄露	本地环境变量中的 API Key 可能被读取
文件系统破坏	Agent 可能修改/删除本地文件
网络外传	Agent 可能通过网络泄露数据

5种沙箱提供商

提供商	包名	特点
Modal	`langchain-modal`	`modal.Sandbox.create()`
Runloop	`langchain-runloop`	RunloopSDK devbox
Daytona	`langchain-daytona`	标签、TTL、快照
LangSmith	`langsmith[sandbox]`	私有 Beta，`SandboxClient`
AgentCore	`langchain-agentcore-codeinterpreter`	AWS Bedrock 代码解释器

基本使用模式

from daytona import Daytona
from langchain_daytona import DaytonaSandbox

client = Daytona()
sandbox = client.create()

backend = DaytonaSandbox(sandbox=sandbox)

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    system_prompt="你是一个 Python 编程助手，拥有沙箱执行权限。",
    backend=backend,
)

try:
    result = agent.invoke({"messages": [{"role": "user", "content": "写一个排序算法并测试"}]})
finally:
    sandbox.stop()  # 清理沙箱

两种作用域

flowchart LR
A["沙箱作用域"] --> B["Thread 级别<br/>每次对话一个沙箱"]
A --> C["Assistant 级别<br/>所有对话共享沙箱"]

B --> B1["数据隔离<br/>TTL 自动清理"]
C --> C1["状态累积<br/>适合编码助手"]

style B fill:#e8f5e9
style C fill:#e3f2fd

作用域	生命周期	适用场景
Thread 级别（默认）	每次对话独立沙箱，TTL 到期自动清理	数据分析机器人，每次对话干净启动
Assistant 级别	所有对话共享沙箱，文件和安装的包持久化	编码助手，维护长期工作空间

两种集成架构

架构	优点	缺点
Agent in Sandbox	接近本地开发体验	API Key 必须放入沙箱（安全风险），更新需重建镜像
Sandbox as Tool（推荐）	API Key 在沙箱外，迭代快，可并行多沙箱	每次执行有网络延迟

安全关键原则

⛔ 永远不要把密钥放入沙箱 — 即使是短生命周期的凭证，上下文注入攻击也能读取并外传。

安全处理密钥的方式：

在沙箱外的工具中处理密钥（推荐）— Agent 调用工具名但永远看不到凭证
使用网络代理注入凭证 — 代理拦截 HTTP 请求自动附加认证头
如必须注入：启用 HITL 审批所有工具调用 + 阻止沙箱网络访问

3.5 Subagents（子 Agent 委派）

子 Agent 解决上下文膨胀问题 — 主 Agent 委派专门的工作给子 Agent，子 Agent 在独立上下文中执行，只返回最终摘要给主 Agent。

同步子 Agent 配置

research_subagent = {
    "name": "researcher",
    "description": "深入研究特定主题，返回结构化发现",
    "system_prompt": "你是专业研究员。返回简洁摘要，不超过 500 字。",
    "tools": [internet_search],       # 可选，默认继承主 Agent 工具
    "model": "anthropic:claude-sonnet-4-6",  # 可选，默认继承主 Agent 模型
    "skills": ["/skills/research/"],  # 不继承主 Agent 技能，需单独指定
}

writing_subagent = {
    "name": "writer",
    "description": "撰写精炼的分析报告",
    "system_prompt": "你是专业作家。语言简洁有力，结构清晰。",
}

agent = create_deep_agent(
    model="openai:gpt-4o",
    tools=[internet_search],
    subagents=[research_subagent, writing_subagent],
    system_prompt="你是研究协调员。用 researcher 做调研，用 writer 写报告。",
)

子 Agent 属性继承规则

属性	是否继承自主 Agent	说明
`tools`	✅ 继承	指定时完全覆盖
`model`	✅ 继承	可用字符串或模型对象
`interrupt_on`	✅ 继承	子 Agent 值覆盖
`permissions`	✅ 继承	指定时完全替换
`system_prompt`	❌ 不继承	必须指定
`middleware`	❌ 不继承	需单独配置
`skills`	❌ 不继承	仅通用子 Agent 继承

通用子 Agent

每个 Deep Agent 自动获得一个 general-purpose 子 Agent，属性与主 Agent 相同，且是唯一自动继承技能的子 Agent：

# 覆盖通用子 Agent
subagents=[{
    "name": "general-purpose",
    "description": "通用研究助手",
    "system_prompt": "你是通用助手。",
    "tools": [internet_search],
    "model": "openai:gpt-4o",
}]

# 禁用通用子 Agent（通过 Harness Profile）
from deepagents import GeneralPurposeSubagentProfile
register_harness_profile(
    "openai:gpt-4o",
    HarnessProfile(general_purpose_subagent=GeneralPurposeSubagentProfile(enabled=False)),
)

结构化输出

子 Agent 支持 response_format 返回结构化 JSON：

from pydantic import BaseModel, Field

class ResearchFindings(BaseModel):
    summary: str = Field(description="研究发现摘要")
    confidence: float = Field(description="置信度 0-1")
    sources: list[str] = Field(description="来源 URL 列表")

research_subagent = {
    "name": "researcher",
    "description": "研究主题并返回结构化发现",
    "system_prompt": "深入研究给定主题。",
    "tools": [web_search],
    "response_format": ResearchFindings,
}

子 Agent 委派流程

sequenceDiagram
participant User as 用户
participant Main as 主 Agent
participant Sub as 子 Agent
participant Tool as 工具

User->>Main: 提交复杂任务
Main->>Main: write_todos 规划任务
Main->>Sub: task(name="researcher", task="研究X")
Note over Sub: 独立上下文窗口<br/>不共享主Agent历史
Sub->>Tool: 调用工具执行子任务
Tool-->>Sub: 返回工具结果
Sub->>Sub: 完成子任务
Sub-->>Main: 返回浓缩摘要（或结构化JSON）
Note over Main: 只收到摘要<br/>上下文不被污染
Main->>Main: 继续主任务
Main-->>User: 返回最终结果

3.6 Async Subagents（异步子 Agent）

异步子 Agent 允许主管 Agent 启动后台任务后立即返回，继续与用户交互，同时子 Agent 在后台工作。

Preview 特性：deepagents 0.5.0+，API 可能变化。

同步 vs 异步子 Agent 对比

维度	同步子 Agent	异步子 Agent
执行模式	主管阻塞等待完成	立即返回 Job ID，主管继续
并发性	并行但阻塞	并行且非阻塞
中途更新	❌ 不可能	✅ `update_async_task`
取消	❌ 不可能	✅ `cancel_async_task`
有状态	无状态	有状态（跨交互维护线程状态）
适用场景	需要等待结果再继续	长时间复杂任务，交互式管理

5个核心工具

工具	用途	返回值
`start_async_task`	启动后台任务	任务 ID（立即返回）
`check_async_task`	查询任务状态和结果	状态 + 结果（如完成）
`update_async_task`	向运行中的任务发送新指令	确认 + 更新后状态
`cancel_async_task`	取消运行中的任务	确认
`list_async_tasks`	列出所有任务及状态	所有任务摘要

配置与使用

from deepagents import AsyncSubAgent, create_deep_agent

async_subagents = [
    AsyncSubAgent(
        name="researcher",
        description="信息收集和综合研究代理",
        graph_id="researcher",
        # 不设 url → ASGI 传输（同部署）
    ),
    AsyncSubAgent(
        name="coder",
        description="代码生成和审查代理",
        graph_id="coder",
        # url="https://coder-deployment.langsmith.dev"  # 远程 HTTP 传输
    ),
]

agent = create_deep_agent(
    model="openai:gpt-4o",
    subagents=async_subagents,
    system_prompt="""启动异步任务后，始终将控制权还给用户。
    绝不在启动任务后立即调用 check_async_task。""",
)

异步子 Agent 生命周期

stateDiagram-v2
[*] --> PENDING: start_async_task
PENDING --> RUNNING: 后台开始执行
RUNNING --> SUCCESS: 任务完成
RUNNING --> ERROR: 执行失败
RUNNING --> UPDATED: update_async_task
UPDATED --> RUNNING: 用新指令重启
RUNNING --> CANCELLED: cancel_async_task
SUCCESS --> [*]
ERROR --> [*]
CANCELLED --> [*]

两种传输方式

传输方式	触发条件	特点
ASGI（推荐）	省略 `url`	进程内调用，零网络延迟，无需额外认证
HTTP	设置 `url`	远程 Agent Protocol 服务器，独立扩展，适合跨团队

状态管理

任务元数据存储在主管 Agent 的专用状态通道（async_tasks）中，与消息历史分离。这确保即使消息历史被压缩摘要，任务 ID 也不会丢失。

3.7 Human-in-the-Loop（人类审批）

对敏感操作配置人类审批，Agent 在执行前暂停等待人工确认。

`interrupt_on` 配置

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

agent = create_deep_agent(
    model="openai:gpt-4o",
    tools=[delete_file, send_email, write_file, read_file],
    interrupt_on={
        "delete_file": True,                                          # 默认：所有决策允许
        "send_email": {"allowed_decisions": ["approve", "edit", "reject"]},  # 自定义
        "write_file": {"allowed_decisions": ["approve", "reject"]},   # 不允许编辑
        "read_file": False,                                           # 不需要审批
    },
    checkpointer=MemorySaver(),  # 人类审批必须配置 checkpointer！
)

四种审批决策

决策	操作	典型用途
`approve`	原样执行工具调用	确认安全操作
`edit`	修改工具参数后执行	修正错误参数
`reject`	跳过执行，返回拒绝信息	阻止危险操作
`respond`	返回用户反馈给 LLM（不执行）	“ask user” 式工具

处理中断的完整流程

from langgraph.types import Command

config = {"configurable": {"thread_id": "my-thread"}}

# 1. 调用 Agent — 遇到中断时暂停
result = agent.invoke(
    {"messages": [{"role": "user", "content": "删除文件 temp.txt"}]},
    config=config,
    version="v2",
)

# 2. 检查中断
if result.interrupts:
    interrupt_value = result.interrupts[0].value
    action_requests = interrupt_value["action_requests"]

    # 3. 提供决策
    decisions = [{"type": "approve"}]

    # 4. 恢复执行（必须使用相同的 thread_id！）
    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config,  # 同一个 config
        version="v2",
    )

编辑决策 — 修改工具参数

decisions = [{
    "type": "edit",
    "edited_action": {
        "name": action_request["name"],  # 必须包含工具名
        "args": {"to": "team@company.com", "subject": "...", "body": "..."}
    }
}]

result = agent.invoke(
    Command(resume={"decisions": decisions}),
    config=config,
    version="v2",
)

批量审批 — 多个工具调用

Agent 可能一次调用多个需要审批的工具，所有中断会批量合并，必须按顺序逐一提供决策：

if result.interrupts:
    action_requests = interrupt_value["action_requests"]
    assert len(action_requests) == 2  # 两个工具需要审批

    decisions = [
        {"type": "approve"},  # 第一个工具：delete_file
        {"type": "reject"}    # 第二个工具：send_email
    ]

    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config,
        version="v2",
    )

子 Agent 中的中断

子 Agent 可以有自己的 interrupt_on 配置，覆盖主 Agent 设置。子 Agent 工具也可以直接调用 interrupt() 暂停：

from langgraph.types import interrupt

@tool
def request_approval(action_description: str) -> str:
    """请求人类审批"""
    approval = interrupt({
        "type": "approval_request",
        "action": action_description,
    })
    if approval.get("approved"):
        return f"操作 '{action_description}' 已批准"
    else:
        return f"操作 '{action_description}' 已拒绝"

人类审批流程

sequenceDiagram
participant Agent as Agent
participant Check as 检查点
participant Human as 人类审批者

Agent->>Agent: 调用受保护工具(如 delete_file)
Agent->>Check: 保存当前状态到 Checkpointer
Agent->>Agent: interrupt — 暂停执行
Note over Agent,Human: 等待人类介入

Human->>Check: 读取当前状态
Human->>Check: 选择审批操作

alt 批准(approve)
    Check->>Agent: 原样执行工具调用
else 编辑(edit)
    Check->>Agent: 修改参数后执行
else 拒绝(reject)
    Check->>Agent: 不执行，返回拒绝信息
else 反馈(response)
    Check->>Agent: 返回用户反馈给 LLM
end

Agent->>Agent: 继续执行

3.8 Permissions（权限控制）

声明式的、基于路径的文件系统访问控制，决定 Agent 可以读写哪些文件。

范围：仅覆盖内置文件系统工具（ls、read_file、write_file、edit_file、glob、grep），不包括自定义工具、MCP 工具和沙箱 execute。

规则结构

from deepagents import FilesystemPermission, create_deep_agent

FilesystemPermission(
    operations=["read", "write"],   # 覆盖的操作：read=ls/read_file/glob/grep，write=write_file/edit_file
    paths=["/workspace/**"],        # Glob 匹配路径（支持 ** 递归，{a,b} 交替）
    mode="allow",                   # allow 或 deny
)

评估逻辑

先匹配先生效：按声明顺序评估，第一条匹配的规则决定结果
默认允许：如果没有任何规则匹配，操作被允许

常见模式

# 1. 只读 Agent（禁止所有写入）
permissions=[
    FilesystemPermission(operations=["write"], paths=["/**"], mode="deny"),
]

# 2. 限制在工作目录
permissions=[
    FilesystemPermission(operations=["read", "write"], paths=["/workspace/**"], mode="allow"),
    FilesystemPermission(operations=["read", "write"], paths=["/**"], mode="deny"),
]

# 3. 保护敏感文件 + 允许工作区
permissions=[
    FilesystemPermission(operations=["read", "write"], paths=["/workspace/.env"], mode="deny"),
    FilesystemPermission(operations=["read", "write"], paths=["/workspace/**"], mode="allow"),
    FilesystemPermission(operations=["read", "write"], paths=["/**"], mode="deny"),
]

# 4. 只读记忆（禁止修改共享记忆/策略文件）
permissions=[
    FilesystemPermission(operations=["write"], paths=["/memories/**", "/policies/**"], mode="deny"),
]

⚠️ 规则顺序至关重要

# ✅ 正确：先拒绝 .env，再允许 workspace，最后拒绝其他
correct = [
    FilesystemPermission(operations=["read", "write"], paths=["/workspace/.env"], mode="deny"),
    FilesystemPermission(operations=["read", "write"], paths=["/workspace/**"], mode="allow"),
    FilesystemPermission(operations=["read", "write"], paths=["/**"], mode="deny"),
]

# ❌ 错误：/workspace/** 先匹配了 .env，deny 规则永远不会到达
incorrect = [
    FilesystemPermission(operations=["read", "write"], paths=["/workspace/**"], mode="allow"),
    FilesystemPermission(operations=["read", "write"], paths=["/workspace/.env"], mode="deny"),  # 永远不会到达！
]

子 Agent 权限

子 Agent 默认继承主 Agent 的权限。设置子 Agent 的 permissions 会完全替换（不是合并）主 Agent 的规则：

subagents=[{
    "name": "auditor",
    "description": "只读代码审查员",
    "system_prompt": "审查代码问题。",
    "permissions": [           # 完全替换主 Agent 权限
        FilesystemPermission(operations=["write"], paths=["/**"], mode="deny"),
        FilesystemPermission(operations=["read"], paths=["/workspace/**"], mode="allow"),
        FilesystemPermission(operations=["read"], paths=["/**"], mode="deny"),
    ],
}]

3.9 Memory（长期记忆）

让 Agent 在不同会话/线程之间持久化信息。默认的 Agent 状态仅在单次线程内持久化，长期记忆通过 StoreBackend 实现跨线程持久化。

三层记忆作用域

flowchart TD
A["记忆作用域"] --> B["Agent 级别<br/>所有用户共享"]
A --> C["用户级别<br/>每个用户独立"]
A --> D["组织级别<br/>同一组织共享"]

B --> B1["namespace: (assistant_id,)"]
C --> C1["namespace: (user_id,)"]
D --> D1["namespace: (org_id,)"]

style B fill:#e3f2fd
style C fill:#e8f5e9
style D fill:#fff9c4

作用域	命名空间	效果	典型用途
Agent 级别	`(rt.server_info.assistant_id,)`	所有用户共享	Agent 人格、累积知识
用户级别	`(rt.server_info.user.identity,)`	每个用户独立	用户偏好、个人记忆
组织级别	`(rt.context.org_id,)`	组织内共享	合规策略、组织知识
多维度	`(assistant_id, user_id)`	Agent + 用户双重隔离	每个用户的每个 Agent 独立

AGENTS.md 文件

AGENTS.md 是 Agent 的主记忆文件，存储事实、偏好和学习到的行为：

## 响应风格
- 保持回复简洁
- 尽量使用代码示例

## 用户偏好
- 偏好 Python 语言
- 使用中文回答

Agent 在启动时读取，通过 edit_file 更新。

记忆读写控制

权限	用途	实现方式
读写（默认）	用户偏好、Agent 自我改进	Agent 通过 `edit_file` 更新
只读	组织策略、合规规则、共享知识库	通过 Permissions 禁止写入，或通过应用代码预填充

后台记忆整合（Sleep-Time Compute）

默认记忆在对话中写入（热路径），也可以在对话之间通过后台任务整合：

方式	优点	缺点
热路径（对话中）	立即可用	增加延迟
后台整合（对话间）	无用户延迟，可跨对话综合	下次对话才可用

# 后台整合 Agent — 定期回顾对话并更新记忆
consolidation_agent = create_deep_agent(
    model="openai:gpt-4o",
    system_prompt="""回顾最近的对话并更新用户记忆文件。
    合并新事实，移除过时信息，保持简洁。""",
    tools=[search_recent_conversations],
)

# Cron 定时执行（每6小时）
# schedule="0 */6 * * *"

记忆 vs 技能

	记忆（AGENTS.md）	技能（SKILL.md）
加载时机	启动时全部加载	按需加载（渐进式披露）
适用内容	始终相关的上下文（偏好、约定）	特定任务的知识和流程
更新频率	经常变化	相对稳定
Token 影响	每次都消耗	只在需要时消耗
层叠规则	用户+项目合并	用户+项目后者覆盖

flowchart LR
subgraph 记忆加载
    M1["Agent 启动"] --> M2["加载 AGENTS.md"]
    M2 --> M3["全部注入系统提示词"]
    M3 --> M4["每次调用都消耗 Token"]
end
subgraph 技能加载
    S1["Agent 启动"] --> S2["仅读取 Frontmatter"]
    S2 --> S3["用户提问匹配技能"]
    S3 --> S4["按需加载完整 SKILL.md"]
    S4 --> S5["只在匹配时消耗 Token"]
end

3.10 Skills（技能系统）

技能是可复用的 Agent 能力，遵循 Agent Skills 规范，通过渐进式披露减少 Token 消耗。

SKILL.md 格式

每个技能是一个目录，包含 SKILL.md 入口文件和可选的支撑文件：

---
name: langgraph-docs
description: 用于获取 LangGraph 文档以提供准确指导的技能
license: MIT
compatibility: 需要网络访问
metadata:
  author: langchain
  version: "1.0"
allowed-tools: fetch_url
---

# langgraph-docs

## 说明

### 1. 获取文档索引
使用 fetch_url 工具读取: https://docs.langchain.com/llms.txt

### 2. 选择相关文档
从索引中识别 2-4 个最相关的 URL

### 3. 获取选中文档
使用 fetch_url 工具读取选中的 URL

### 4. 提供准确指导
阅读文档后，完成用户的请求

渐进式披露三阶段

flowchart TD
A["🚀 启动阶段"] --> B["读取每个 SKILL.md 的 frontmatter<br/>(名称 + 描述)"]
B --> C["几十个技能只消耗几百 tokens"]
C --> D["🔍 匹配阶段"]
D --> E["用户提问后，Agent 检查<br/>哪个技能的 description 匹配"]
E --> F["只选择 1-2 个相关技能"]
F --> G["📖 加载阶段"]
G --> H["Agent 读取匹配技能的<br/>完整 SKILL.md 内容"]
H --> I["按需加载，只消耗需要的 tokens"]
I --> J["⚡ 执行阶段"]
J --> K["Agent 按照技能指令<br/>调用工具、访问参考文档"]

style A fill:#e3f2fd
style D fill:#fff9c4
style G fill:#f3e5f5
style J fill:#e8f5e9

约束与限制

约束	限制
`description` 字段	超过 1024 字符会被截断
`SKILL.md` 文件大小	超过 10 MB 会被跳过

技能注册与层叠

agent = create_deep_agent(
    model="openai:gpt-4o",
    skills=["/skills/user/", "/skills/project/"],  # 后者优先级更高
)

后覆盖前（Last-One-Wins）：当多个技能源包含同名技能时，skills 列表中靠后的源优先。

子 Agent 技能

子 Agent 类型	技能行为
通用子 Agent	自动继承主 Agent 技能
自定义子 Agent	不继承 — 必须通过 `skills` 参数单独指定

技能 vs 工具的选择指南

用技能：上下文量大，需要减少系统提示词 Token；需要捆绑多个工具为一组能力
用工具：Agent 没有文件系统访问权限；单次操作足够

3.11 Profiles（配置包）

Profile 是为特定模型/提供商预打包的配置，在选择对应模型时自动生效。分为 Harness Profile 和 Provider Profile 两种。

Harness Profile（控制 Agent 行为）

影响提示词组装、工具可见性、中间件和通用子 Agent：

from deepagents import (
    HarnessProfile, GeneralPurposeSubagentProfile,
    register_harness_profile,
)

register_harness_profile(
    "openai:gpt-4o",
    HarnessProfile(
        system_prompt_suffix="请在 100 字内回复。",     # 追加到系统提示词末尾
        excluded_tools={"execute"},                     # 移除 execute 工具
        excluded_middleware={"SummarizationMiddleware"}, # 移除摘要中间件
        general_purpose_subagent=GeneralPurposeSubagentProfile(enabled=False),  # 禁用通用子 Agent
    ),
)

Harness Profile 完整字段：

字段	类型	说明
`base_system_prompt`	`str`	替换内置基础系统提示词
`system_prompt_suffix`	`str`	追加到组装后的提示词末尾
`tool_description_overrides`	`dict`	覆盖工具描述（按工具名）
`excluded_tools`	`frozenset`	移除指定工具
`excluded_middleware`	`frozenset`	移除指定中间件（禁止移除 Filesystem/SubAgent/Permission 中间件）
`extra_middleware`	`list`	追加中间件
`general_purpose_subagent`	`GeneralPurposeSubagentProfile`	禁用/重命名/重定义通用子 Agent

Provider Profile（控制模型构建）

仅影响模型初始化参数，只在传入 provider:model 字符串时生效：

from deepagents import ProviderProfile, register_provider_profile

# 提供商级别：所有 OpenAI 模型 temperature=0
register_provider_profile(
    "openai",
    ProviderProfile(init_kwargs={"temperature": 0}),
)

# 模型级别：gpt-4o 额外设置 reasoning_effort（继承上面的 temperature=0）
register_provider_profile(
    "openai:gpt-4o",
    ProviderProfile(init_kwargs={"reasoning_effort": "medium"}),
)

注册键与合并规则

两种 Profile 共享键格式：

键类型	格式	作用范围
提供商级	`"openai"`	该提供商的所有模型
模型级	`"openai:gpt-4o"`	仅该模型，合并覆盖提供商级

合并行为：提供商级和模型级共存时，模型级字段覆盖提供商级，未设置的模型级字段继承提供商级。

从 YAML 加载

# openai.yaml
system_prompt_suffix: 回复简洁。
excluded_tools:
  - execute
  - grep
general_purpose_subagent:
  enabled: false

import yaml
from deepagents import HarnessProfileConfig, register_harness_profile

with open("openai.yaml") as f:
    register_harness_profile("openai", HarnessProfileConfig.from_dict(yaml.safe_load(f)))

3.12 Streaming（流式输出）

DeepAgents 基于 LangGraph 的流式基础设施，支持实时输出 Agent 和子 Agent 的执行过程。

四种流模式

模式	用途	输出内容
`"updates"`	跟踪步骤进度	每个节点完成的更新
`"messages"`	流式输出 Token	主 Agent 和子 Agent 的逐 Token 输出
`"custom"`	自定义进度信号	工具内通过 `get_stream_writer` 发出
`["updates", "messages", "custom"]`	组合模式	以上全部同时

子 Agent 流式输出的关键配置

for chunk in agent.stream(
    input,
    stream_mode="updates",
    subgraphs=True,   # ← 必须！否则收不到子 Agent 事件
    version="v2",     # ← 必须！统一 StreamPart 格式（需 LangGraph >= 1.1）
):
    ...

V2 流式格式

每个 chunk 是统一的 StreamPart 字典：

{
    "type": "updates" | "messages" | "custom",   # 事件类型
    "ns":   () or ("tools:abc123",),              # 命名空间（标识来源）
    "data": <payload>                              # 实际数据
}

命名空间识别来源：

命名空间	来源
`()` 空元组	主 Agent
`("tools:abc123",)`	主 Agent 通过 `task` 工具调用的子 Agent
`("tools:abc123", "model_request:def456")`	子 Agent 内部的模型请求节点

流式 Token 输出

current_source = ""
for chunk in agent.stream(
    input,
    stream_mode="messages",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "messages":
        token, metadata = chunk["data"]
        is_subagent = any(s.startswith("tools:") for s in chunk["ns"])

        if is_subagent:
            subagent_ns = next(s for s in chunk["ns"] if s.startswith("tools:"))
            if subagent_ns != current_source:
                print(f"\n--- [子 Agent: {subagent_ns}] ---")
                current_source = subagent_ns
            if token.content:
                print(token.content, end="", flush=True)
        else:
            if "main" != current_source:
                print("\n--- [主 Agent] ---")
                current_source = "main"
            if token.content:
                print(token.content, end="", flush=True)

自定义进度信号

from langgraph.config import get_stream_writer

@tool
def analyze_data(topic: str) -> str:
    writer = get_stream_writer()
    writer({"status": "starting", "topic": topic, "progress": 0})
    # ... 执行工作 ...
    writer({"status": "analyzing", "progress": 50})
    # ... 更多工作 ...
    writer({"status": "complete", "progress": 100})
    return f'"{topic}" 的分析结果...'

过滤摘要 Token

当上下文压缩触发摘要时，可以通过 lc_source 过滤掉摘要 Token：

for chunk in agent.stream(
    {"messages": [...]},
    stream_mode="messages",
    version="v2",
):
    token, metadata = chunk["data"]
    if metadata.get("lc_source") == "summarization":
        continue  # 跳过摘要 Token
    # 处理正常 Token...

3.13 中间件

中间件是横切关注点的处理层，类似 Web 开发中的 Express/Koa 中间件。

工具调用中间件

from langchain.agents.middleware import wrap_tool_call

@wrap_tool_call
def log_tool_calls(request, handler):
    """记录每次工具调用"""
    print(f"工具调用: {request.name}({request.args})")
    result = handler(request)
    print(f"调用完成")
    return result

agent = create_deep_agent(
    model="openai:gpt-4o",
    tools=[search, get_weather],
    middleware=[log_tool_calls],
)

模型调用中间件

from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse

@wrap_model_call
def token_counter(request: ModelRequest, handler) -> ModelResponse:
    """统计每次模型调用的 Token 使用量"""
    response = handler(request)
    usage = response.response.usage_metadata
    print(f"Token 使用: {usage}")
    return response

内置中间件

中间件	作用
`TodoListMiddleware`	待办列表管理
`FilesystemMiddleware`	文件系统操作
`SubAgentMiddleware`	同步子 Agent 调度
`AsyncSubAgentMiddleware`	异步子 Agent 调度
`SummarizationMiddleware`	上下文压缩摘要
`PatchToolCallsMiddleware`	修复中断的工具调用
`PermissionsMiddleware`	文件系统权限控制

中间件管道流程

flowchart LR
A["工具调用请求"] --> B["中间件1: 日志"]
B --> C["中间件2: 权限校验"]
C --> D["中间件3: 上下文压缩"]
D --> E["执行工具"]
E --> F["中间件3: 压缩结果"]
F --> G["中间件2: 审计记录"]
G --> H["中间件1: 调用完成日志"]
H --> I["返回结果"]

style A fill:#e1f5fe
style E fill:#fff9c4
style I fill:#e8f5e9

本章小结

核心能力	关键要点
Models	`provider:model` 格式，3种配置方式，运行时动态切换，Eval 验证的模型列表
Context Engineering	5类上下文（输入/运行时/压缩/隔离/长期），85%摘要阈值，双重保障摘要
Backends	6种后端（State/Filesystem/LocalShell/Store/Sandbox/Composite），命名空间隔离，自定义后端协议
Sandboxes	5种沙箱提供商，Thread/Assistant 作用域，Sandbox as Tool 架构推荐，密钥永远不入沙箱
Subagents	同步子 Agent，上下文隔离，属性继承规则，结构化输出，通用子 Agent
Async Subagents	5个核心工具，非阻塞执行，中途更新/取消，ASGI/HTTP 传输
Human-in-the-Loop	`interrupt_on` 配置，4种审批决策，批量审批，子 Agent 中断
Permissions	先匹配先生效，默认允许，规则顺序关键，子 Agent 完全替换
Memory	三层作用域（Agent/用户/组织），AGENTS.md，只读/读写控制，后台整合
Skills	SKILL.md 格式，渐进式披露三阶段，后覆盖前，技能 vs 工具选择指南
Profiles	Harness Profile（Agent行为）+ Provider Profile（模型参数），提供商/模型两级注册，YAML 加载
Streaming	4种流模式，V2 统一格式，命名空间识别来源，自定义进度信号，摘要 Token 过滤

DeepAgents 的核心哲学：不是让模型更聪明，而是通过 Harness（工具 + 上下文管理 + 安全护栏）让模型可靠地完成复杂任务。先做好上下文工程和安全，再追求性能优化。