Jean's Blog

一个专注软件测试开发技术的个人博客

0%

核心概念（详解）

发表于 分类于
学习DeepAgents学习笔记第二讲

create_deep_agent 函数签名

1
2
3
4
5
6
7
8
9
10
11
12
13
def create_deep_agent(
    model: str | BaseChatModel | None = None,           # AI 模型
    tools: Sequence[BaseTool] | None = None,            # 工具列表
    system_prompt: str | None = None,                   # 系统提示词
    middleware: Sequence[AgentMiddleware] = (),         # 中间件列表
    subagents: list[SubAgent] | None = None,            # 子智能体列表
    store: BaseStore | None = None,                     # 长期记忆存储
    backend: BackendProtocol | None = None,             # 文件存储后端
    interrupt_on: dict[str, bool | InterruptOnConfig] | None = None,  # 人机交互配置
    checkpointer: Checkpointer | None = None,           # 会话检查点
    debug: bool = False,                                # 调试模式
) -> CompiledStateGraph:
    """创建一个深度智能体"""

默认内置功能：

创建 Deep Agent 时，框架会自动添加以下中间件（按顺序）：

  1. TodoListMiddleware - 任务列表管理
  2. FilesystemMiddleware - 文件系统操作（ls, read_file, write_file, edit_file, glob, grep, execute）
  3. SubAgentMiddleware - 子智能体调度（task 工具）
  4. SummarizationMiddleware - 对话历史摘要（防止上下文溢出）
  5. AnthropicPromptCachingMiddleware - 提示词缓存（节省成本）
  6. PatchToolCallsMiddleware - 修复悬空工具调用
  7. HumanInTheLoopMiddleware - 人机交互（如果配置了 interrupt_on）

Middleware 中间件详解

Middleware（中间件）就像是给 AI 加装的功能模块，每个中间件负责一个特定功能。

FilesystemMiddleware - 文件系统中间件

功能说明

让 AI 能够操作文件系统，提供 7 个核心工具：

工具 功能 示例
ls 列出目录文件 ls("/data/")
read_file 读取文件内容 read_file("/data/config.json")
write_file 创建新文件 write_file("/data/new.txt", "内容")
edit_file 编辑现有文件 edit_file("/data/config.json", "old", "new")
glob 模式匹配查找文件 glob("**/*.py")
grep 搜索文件内容 grep("TODO", glob="*.py")
execute 执行 shell 命令 execute("pytest tests/")

代码示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from deepagents import create_deep_agent, FilesystemMiddleware
from deepagents.backends import StateBackend

# 示例 1: 使用默认配置（临时存储）
agent = create_deep_agent(
    system_prompt="你是一个文件管理助手"
)

# AI 可以执行的操作：
result = agent.invoke({
    "messages": [{"role": "user", "content": "创建一个 /data/test.txt 文件，内容是 'Hello World'"}]
})
# AI 会调用: write_file("/data/test.txt", "Hello World")

# 示例 2: 自定义 backend
agent = create_deep_agent(
    backend=lambda rt: StateBackend(rt),  # 使用工厂函数
    system_prompt="你是一个代码分析助手"
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "找出所有 Python 文件中的 TODO 注释"}]
})
# AI 会调用: grep("TODO", glob="*.py", output_mode="content")

工作原理图

sequenceDiagram
    participant User
    participant Agent
    participant FilesystemMiddleware
    participant Backend

    User->>Agent: "创建文件 /data/test.txt"
    Agent->>FilesystemMiddleware: 调用 write_file 工具
    FilesystemMiddleware->>Backend: backend.write("/data/test.txt", content)
    Backend-->>FilesystemMiddleware: WriteResult(path="/data/test.txt")
    FilesystemMiddleware-->>Agent: "文件创建成功"
    Agent-->>User: "已创建文件 /data/test.txt"

SubAgentMiddleware - 子智能体中间件

功能说明

让主 Agent 能够委派任务给专业的子 Agent，实现：

  • 🎯 任务隔离：每个子任务有独立的上下文
  • ⚡ 并行处理：多个子 Agent 可以同时工作
  • 💰 节省 Token：子任务完成后只返回摘要，不保留详细过程

核心概念

1
2
3
4
5
6
7
8
9
# 子智能体定义
SubAgent = {
    "name": "研究分析师",                    # 子智能体名称
    "description": "专门进行深度研究和分析",  # 用途描述（主 Agent 根据此决定是否调用）
    "system_prompt": "你是一个研究专家...",   # 子智能体的系统提示词
    "tools": [web_search, fetch_url],        # 子智能体可用的工具
    "model": "gpt-4",                        # 可选：使用不同的模型
    "middleware": [CustomMiddleware()],      # 可选：额外的中间件
}

代码示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
from deepagents import create_deep_agent

# 示例 1: 使用默认通用子智能体
agent = create_deep_agent(
    system_prompt="你是一个项目经理"
)

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "研究 LangChain、LlamaIndex 和 AutoGPT 三个框架，并比较它们的优缺点"
    }]
})

# AI 会自动：
# 1. 启动 3 个并行的 task 子智能体
# 2. 每个子智能体独立研究一个框架
# 3. 收集所有子智能体的研究结果
# 4. 综合分析并返回比较报告

# 示例 2: 自定义专业子智能体
from langchain_core.tools import tool

@tool
def analyze_code(code: str) -> str:
    """分析代码质量"""
    return f"代码分析结果: {code}"

# 定义专业子智能体
code_reviewer = {
    "name": "code-reviewer",
    "description": "代码审查专家，用于审查代码质量、安全性和最佳实践",
    "system_prompt": """你是一个资深代码审查专家。
    审查代码时关注：
    1. 代码质量和可读性
    2. 安全漏洞
    3. 性能问题
    4. 最佳实践
    返回详细的审查报告。""",
    "tools": [analyze_code],
}

research_analyst = {
    "name": "research-analyst",
    "description": "研究分析专家，用于深度研究复杂主题",
    "system_prompt": "你是一个研究分析专家，擅长收集信息、分析数据并生成报告。",
    "tools": [],  # 使用默认工具
}

agent = create_deep_agent(
    subagents=[code_reviewer, research_analyst],
    system_prompt="你是一个智能助手，可以调用专业子智能体完成任务"
)

# 使用场景 1: 代码审查
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "请审查 /src/main.py 文件的代码质量"
    }]
})
# AI 会：
# 1. 读取文件内容
# 2. 调用 code-reviewer 子智能体
# 3. 返回审查报告

# 使用场景 2: 研究任务
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "研究量子计算的最新进展"
    }]
})
# AI 会调用 research-analyst 子智能体进行深度研究

工作流程图

graph LR
    %% 节点定义（换行用|委派任务| B
    A -->|委派任务| C
    A -->|委派任务| D
    B -->|返回摘要| E
    C -->|返回摘要| E
    D -->|返回摘要| E
    E -->|最终报告| F

    %% 样式定义（适配11.9.0的class语法）
    classDef mainAgent fill:#2ecc71,stroke:#27ae60,stroke-width:2px,rounded:true
    classDef subAgent1 fill:#3498db,stroke:#2980b9,stroke-width:2px,rounded:true
    classDef subAgent2 fill:#6c5ce7,stroke:#4834d4,stroke-width:2px,rounded:true
    classDef subAgent3 fill:#9b59b6,stroke:#8e44ad,stroke-width:2px,rounded:true
    classDef userAgent fill:#f3e5f5,stroke:#ce93d8,stroke-width:2px,rounded:true

    %% 分配样式（分开写，避免11.9.0解析错误）
    class A mainAgent
    class E mainAgent
    class B subAgent1
    class C subAgent2
    class D subAgent3
    class F userAgent

何时使用子智能体？

✅ 应该使用：

  • 任务复杂且多步骤
  • 任务可以独立完成
  • 需要并行处理多个独立任务
  • 需要隔离上下文（避免主线程被大量细节淹没）

❌ 不应该使用：

  • 简单的单步操作
  • 需要看到中间推理过程
  • 任务之间有强依赖关系

SummarizationMiddleware - 对话摘要中间件

功能说明

当对话历史太长时，自动压缩旧消息，防止超出模型上下文限制。

1
2
3
4
5
6
# 默认配置
SummarizationMiddleware(
    model=model,
    max_tokens_before_summary=50000,  # 超过 5 万 token 时触发摘要
    messages_to_keep=6,                # 保留最近 6 条消息不压缩
)

工作原理

graph LR
    A[消息历史
60000 tokens] -->|超过阈值| B[SummarizationMiddleware]
    B -->|保留| C[最近 6 条消息]
    B -->|压缩| D[旧消息摘要
5000 tokens]
    C --> E[新的消息历史
15000 tokens]
    D --> E

    style A fill:#f44336,color:#fff
    style E fill:#4CAF50,color:#fff

HumanInTheLoopMiddleware - 人机交互中间件

功能说明

在 AI 执行敏感操作前，暂停并等待人类批准。

1
2
3
4
5
6
7
8
9
10
11
12
from langchain.agents.middleware import InterruptOnConfig

agent = create_deep_agent(
    interrupt_on={
        "execute": True,  # 执行命令前需要批准
        "write_file": InterruptOnConfig(
            interrupt_before=True,   # 执行前中断
            interrupt_after=False,   # 执行后不中断
        ),
        "edit_file": True,
    }
)

交互流程

sequenceDiagram
    participant User
    participant Agent
    participant HumanInTheLoop
    participant Tool

    User->>Agent: "删除所有日志文件"
    Agent->>HumanInTheLoop: 准备执行 execute("rm *.log")
    HumanInTheLoop-->>User: ⚠️ 需要批准：execute("rm *.log")
    User->>HumanInTheLoop: ✅ 批准 / ❌ 拒绝
    alt 批准
        HumanInTheLoop->>Tool: 执行命令
        Tool-->>Agent: 执行结果
    else 拒绝
        HumanInTheLoop-->>Agent: 操作已取消
    end
    Agent-->>User: 返回结果

Backend 后端存储详解

Backend 决定了文件存储在哪里，就像给 AI 配置不同的”硬盘”。

Backend 类型对比

Backend 类型 存储位置 持久化 跨会话 执行命令 适用场景
StateBackend Agent 状态 ✅ (Checkpointer) ❌ ❌ 临时文件、单会话
StoreBackend LangGraph Store ✅ ✅ ❌ 长期记忆、跨会话
SandboxBackend 沙箱环境 ✅ ✅ ✅ 代码执行、测试
CompositeBackend 混合存储 ✅ ✅ ✅ 企业级应用

StateBackend - 临时存储

特点

  • 文件存储在 Agent 的状态中
  • 通过 Checkpointer 持久化
  • 不跨会话：每个对话线程独立
  • 不支持命令执行

代码示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from deepagents import create_deep_agent
from deepagents.backends import StateBackend

# 方式 1: 使用默认配置（自动使用 StateBackend）
agent = create_deep_agent()

# 方式 2: 显式指定 StateBackend
agent = create_deep_agent(
    backend=lambda rt: StateBackend(rt)  # 使用工厂函数
)

# 使用示例
result = agent.invoke({
    "messages": [{"role": "user", "content": "创建文件 /notes.txt，内容是 'Hello'"}]
})

# 文件存储在 state["files"] 中：
# state = {
#     "files": {
#         "/notes.txt": {
#             "content": ["Hello"],
#             "created_at": "2024-01-01T00:00:00",
#             "modified_at": "2024-01-01T00:00:00"
#         }
#     }
# }

StoreBackend - 持久存储

特点

  • 文件存储在 LangGraph Store
  • 跨会话持久化：所有对话共享
  • 支持命名空间隔离
  • 不支持命令执行

代码示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from deepagents import create_deep_agent
from deepagents.backends import StoreBackend
from langgraph.store.memory import InMemoryStore

# 创建 Store
store = InMemoryStore()

# 创建 Agent
agent = create_deep_agent(
    backend=lambda rt: StoreBackend(rt),
    store=store,  # 必须提供 store
)

# 会话 1: 创建文件
result1 = agent.invoke(
    {"messages": [{"role": "user", "content": "创建 /memories/user_prefs.json"}]},
    config={"configurable": {"thread_id": "thread-1"}}
)

# 会话 2: 读取文件（不同线程，但能访问相同文件）
result2 = agent.invoke(
    {"messages": [{"role": "user", "content": "读取 /memories/user_prefs.json"}]},
    config={"configurable": {"thread_id": "thread-2"}}  # 不同线程
)
# ✅ 能够读取到会话 1 创建的文件

SandboxBackend - 沙箱执行

特点

  • 文件存储在隔离的沙箱环境（如 Docker 容器）
  • 支持命令执行（execute 工具）
  • 安全隔离
  • 适合代码执行、测试

代码示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
from deepagents import create_deep_agent
from deepagents.backends.sandbox import BaseSandbox
from deepagents.backends.protocol import ExecuteResponse

# 自定义沙箱实现
class DockerSandbox(BaseSandbox):
    def __init__(self, container_id: str):
        self.container_id = container_id

    def execute(self, command: str) -> ExecuteResponse:
        """在 Docker 容器中执行命令"""
        import subprocess
        result = subprocess.run(
            ["docker", "exec", self.container_id, "sh", "-c", command],
            capture_output=True,
            text=True
        )
        return ExecuteResponse(
            output=result.stdout + result.stderr,
            exit_code=result.returncode
        )

    @property
    def id(self) -> str:
        return self.container_id

# 创建 Agent
sandbox = DockerSandbox(container_id="my-sandbox")
agent = create_deep_agent(
    backend=sandbox,
    system_prompt="你是一个代码执行助手"
)

# 使用示例
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "创建一个 Python 脚本 test.py，然后运行它"
    }]
})

# AI 会：
# 1. write_file("/test.py", "print('Hello')")
# 2. execute("python /test.py")
# 3. 返回执行结果

CompositeBackend - 混合存储（企业级）

特点

  • 路由机制：根据路径前缀选择不同的 Backend
  • 组合多个 Backend 的优势
  • 最灵活、最强大

架构图

graph TB
    Agent[Agent] --> Composite[CompositeBackend]

    Composite -->|路径: /| Default[Default Backend
SandboxBackend]
    Composite -->|路径: /memories/| Route1[StoreBackend
长期记忆]
    Composite -->|路径: /cache/| Route2[StateBackend
临时缓存]

    Default --> Docker[Docker 容器]
    Route1 --> Store[LangGraph Store]
    Route2 --> State[Agent State]

    style Composite fill:#4CAF50,color:#fff
    style Default fill:#2196F3,color:#fff
    style Route1 fill:#FF9800,color:#fff
    style Route2 fill:#9C27B0,color:#fff

代码示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
from deepagents import create_deep_agent
from deepagents.backends import StateBackend, StoreBackend, CompositeBackend
from langgraph.store.memory import InMemoryStore

# 创建 Store
store = InMemoryStore()

# 创建混合 Backend
composite_backend = CompositeBackend(
    default=DockerSandbox(container_id="my-sandbox"),  # 默认使用沙箱
    routes={
        "/memories/": lambda rt: StoreBackend(rt),     # /memories/ 路径使用持久存储
        "/cache/": lambda rt: StateBackend(rt),        # /cache/ 路径使用临时存储
    }
)

# 创建 Agent
agent = create_deep_agent(
    backend=composite_backend,
    store=store,
    system_prompt="你是一个智能助手"
)

# 使用示例
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": """
        1. 在 /test.py 创建测试脚本（存储在沙箱）
        2. 在 /memories/user_prefs.json 保存用户偏好（持久存储）
        3. 在 /cache/temp.txt 创建临时文件（临时存储）
        """
    }]
})

# 文件路由：
# /test.py          -> DockerSandbox（可执行）
# /memories/...     -> StoreBackend（跨会话持久化）
# /cache/...        -> StateBackend（会话内临时）

路由规则

1
2
3
4
5
# 路径匹配规则（最长前缀优先）
"/memories/notes.txt"     -> StoreBackend   # 匹配 /memories/
"/cache/temp.txt"         -> StateBackend   # 匹配 /cache/
"/test.py"                -> DockerSandbox  # 不匹配任何路由，使用 default
"/memories/deep/file.txt" -> StoreBackend   # 匹配 /memories/（支持嵌套）

Store 长期记忆详解

Store 是 LangGraph 提供的跨会话持久化存储，用于保存需要长期记住的信息。

Store vs Backend 的区别

特性 Store Backend
用途 存储结构化数据（记忆、偏好） 存储文件内容
访问方式 通过 namespace + key 通过文件路径
跨会话 ✅ 所有会话共享 取决于 Backend 类型
典型用例 用户偏好、历史记录 文档、代码文件

代码示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
from deepagents import create_deep_agent
from deepagents.backends import StoreBackend, CompositeBackend, StateBackend
from langgraph.store.memory import InMemoryStore

# 创建 Store
store = InMemoryStore()

# 方式 1: 仅使用 Store（通过 StoreBackend）
agent = create_deep_agent(
    backend=lambda rt: StoreBackend(rt),
    store=store,
)

# 方式 2: 混合使用（推荐）
composite_backend = CompositeBackend(
    default=lambda rt: StateBackend(rt),      # 临时文件
    routes={
        "/memories/": lambda rt: StoreBackend(rt),  # 长期记忆
    }
)

agent = create_deep_agent(
    backend=composite_backend,
    store=store,
)

# 使用示例：保存用户偏好
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "记住我的偏好：我喜欢简洁的代码风格，使用 4 空格缩进"
    }]
})

# AI 会创建文件：
# write_file("/memories/user_prefs.json", '{"code_style": "简洁", "indent": 4}')

# 下次对话（不同会话）
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "帮我写一个 Python 函数"
    }],
    config={"configurable": {"thread_id": "new-thread"}}
})

# AI 会：
# 1. read_file("/memories/user_prefs.json")  # 读取偏好
# 2. 根据偏好生成代码（4 空格缩进）

Store 的命名空间

1
2
3
4
5
6
7
8
9
10
11
12
13
# StoreBackend 自动使用命名空间
# 格式: (assistant_id, "filesystem") 或 ("filesystem",)

# 如果配置中有 assistant_id
config = {
    "metadata": {"assistant_id": "assistant-123"}
}

# 文件会存储在：
# namespace = ("assistant-123", "filesystem")
# key = "/memories/user_prefs.json"

# 这样不同 assistant 的文件互不干扰

Interrupt_on 人机交互详解

Interrupt_on 让你能够控制 AI 在执行哪些操作前需要人类批准，防止 AI 执行危险操作。

配置方式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from langchain.agents.middleware import InterruptOnConfig

# 方式 1: 简单配置（布尔值）
interrupt_on = {
    "execute": True,      # 执行命令前中断
    "write_file": True,   # 写文件前中断
    "edit_file": True,    # 编辑文件前中断
}

# 方式 2: 详细配置
interrupt_on = {
    "execute": InterruptOnConfig(
        interrupt_before=True,   # 执行前中断
        interrupt_after=False,   # 执行后不中断
    ),
    "write_file": InterruptOnConfig(
        interrupt_before=True,
        interrupt_after=True,    # 执行后也中断（查看结果）
    ),
}

agent = create_deep_agent(
    interrupt_on=interrupt_on
)

代码示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver

# 创建带人机交互的 Agent
checkpointer = MemorySaver()

agent = create_deep_agent(
    checkpointer=checkpointer,
    interrupt_on={
        "execute": True,
        "write_file": True,
        "edit_file": True,
    }
)

# 配置
config = {"configurable": {"thread_id": "thread-1"}}

# 第一步：用户请求
result = agent.invoke({
    "messages": [{"role": "user", "content": "删除所有 .log 文件"}]
}, config=config)

# Agent 会中断，等待批准
# result 包含待执行的操作信息

# 查看待批准的操作
print(result)
# 输出: 准备执行 execute("rm *.log")

# 第二步：人类批准
# 方式 1: 批准
result = agent.invoke(None, config=config)  # 继续执行

# 方式 2: 拒绝（修改状态）
# 需要手动修改 checkpointer 中的状态来拒绝操作

中断流程图

stateDiagram-v2
    [*] --> UserRequest: 用户请求
    UserRequest --> AgentPlanning: Agent 规划
    AgentPlanning --> CheckInterrupt: 检查是否需要中断

    CheckInterrupt --> Interrupt: 需要批准
    CheckInterrupt --> Execute: 不需要批准

    Interrupt --> WaitApproval: 等待人类决策
    WaitApproval --> Approved: ✅ 批准
    WaitApproval --> Rejected: ❌ 拒绝

    Approved --> Execute: 执行操作
    Rejected --> AgentPlanning: 重新规划

    Execute --> [*]: 返回结果