DeepAgent demo解析：构建具备子智能体、人工审批与持久记忆的 AI Agent

一、项目概览

该项目演示了如何使用 deepagents构建一个生产级 AI Agent，核心能力包括：

技术依赖：

from deepagents import create_deep_agent, SubAgent
from langchain.agents.middleware import AgentMiddleware
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command, interrupt
from langchain_openai import ChatOpenAI

模型层使用阿里云 DashScope 提供的 Qwen-Plus，通过 OpenAI 兼容接口接入。

二、模型层：统一共享的 LLM 实例

QWEN_MODEL_NAME = "qwen-plus"
DASHSCOPE_API_KEY = '你的key'
DASHSCOPE_BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
​
shared_llm = ChatOpenAI(
   model=QWEN_MODEL_NAME,
   openai_api_key=DASHSCOPE_API_KEY,
   openai_api_base=DASHSCOPE_BASE_URL,
   temperature=0,
)

设计要点：

• temperature=0 保证输出确定性，适合需要稳定推理的 Agent 场景。

• 所有组件（主 Agent、子 Agent、中间件）共用同一个 shared_llm 实例，避免重复初始化，节省资源。

• 使用 OpenAI 兼容协议访问 DashScope，这意味着你可以无缝替换为任何兼容的模型服务（如 DeepSeek、Moonshot 等）。

三、工具层：三类工具的完整实现

3.1 普通工具：天气查询

@tool
def get_current_weather(city: str) -> str:
   """获取指定城市的当前天气情况"""
   weather_data = {
       "北京": "晴朗，气温25°C，湿度40%",
       "上海": "多云，气温22°C，湿度65%",
       "深圳": "阵雨，气温28°C，湿度80%",
       "成都": "阴天，气温20°C，湿度70%",
  }
   result = weather_data.get(city, f"{city}的天气数据暂不可用")
   return f"🌤️ {city} 天气: {result}"

• 使用 @tool 装饰器将普通函数注册为 LangChain 工具，Agent 可直接调用。

• 函数文档字符串 """获取指定城市的当前天气情况""" 会被 Agent 用作工具描述，指导何时调用该工具。

• 这里使用静态字典模拟 API 调用，实际生产环境可替换为真实的天气 API。

3.2 安全计算工具

@tool
def calculate(expression: str) -> str:
   """执行基本的数学计算"""
   try:
       safe_dict = {
           "__builtins__": {},
           "abs": abs,
           "round": round,
           "min": min,
           "max": max,
           "sum": sum,
      }
       result = eval(expression, safe_dict)
       return f"📊 计算结果: {result}"
   except Exception as e:
       return f"计算表达式 '{expression}' 时出错: {e}"

安全设计是关键：

• 覆盖 __builtins__ 为空字典，阻断 eval 执行任意代码的能力。

• 仅暴露五个无害的数学函数（abs、round、min、max、sum），防止代码注入攻击。

• 使用 try/except 包裹，避免非法表达式导致 Agent 崩溃。

3.3 高风险工具：文件删除

@tool
def delete_file(file_name: str) -> str:
   """删除指定的文件 (高风险操作，需要人工审批)"""
   return f"待删除文件: {file_name}"

• 该工具本身不执行真正的删除操作，仅返回一个待确认的提示。

• 真正的删除逻辑由外部代码在审批通过后执行，形成"工具声明 → 中断 → 审批 → 执行"的安全链路。

四、子智能体：Research Assistant

research_agent = SubAgent(
   name='research_assistant',
   description='专门负责深度研究和信息收集的助手。擅长网络搜索、资料整理和报告生成。',
   system_prompt='你是专业的研究助理。你会使用搜索工具查找最新信息…',
   tools=[calculate],
   model=shared_llm
)

SubAgent 的关键参数解析：

委托机制： 主 Agent 在遇到复杂研究任务时，会自动调用内置的 task 工具，将任务分派给 research_assistant 执行。这种"主-从"架构使得 Agent 系统可以按专业领域进行任务分流。

五、自定义中间件：LoggingMiddleware

class LogggingMiddleware(AgentMiddleware):
   """自定义中间件，记录每个步骤的输入和输出"""
   def __init__(self):
       self.step_count = 0
​
   def before_model(self, state: dict, runtime: Any):
       self.step_count += 1
       print(f'步骤 {self.step_count}: agent正在处理请求…')
       return state
​
   def after_model(self, state: dict, runtime: Any):
       print(f'步骤 {self.step_count}: agent处理完成…')
       return state

中间件的生命周期：

用户输入 → before_model() → LLM 推理 → after_model() → 返回结果
              ↑                             ↑
          钩子1：记录开始                 钩子2：记录结束

• 继承 AgentMiddleware 基类，重写 before_model 和 after_model 两个钩子方法。

• before_model 在每次模型调用前触发，after_model 在模型返回后触发。

• 这里通过 step_count 计数器追踪 Agent 的执行步数，方便调试和性能分析。

• runtime: Any 参数提供运行时的上下文信息，可以扩展为更复杂的日志记录（如耗时统计、token 用量等）。

六、长期记忆系统

MEMORY_FILE = './agent_memory.md'
​
if not os.path.exists(MEMORY_FILE):
   with open(MEMORY_FILE, 'w', encoding='utf-8') as f:
       f.write("""# 智能体长期记忆
​
## 用户偏好
– 用户更喜欢简洁的回答
– 用户对天气信息比较感兴趣
…
​
## 历史任务记录
– 2025-05-20: 帮助用户规划了一次商务旅行
…
​
## 学习到的知识
– 用户常用的城市：北京、上海、深圳
– 用户工作领域：技术研发
""")

工作机制：

• create_deep_agent 的 memory=[MEMORY_FILE] 参数将文件路径注入 Agent，Agent 在每次对话时自动读取该文件作为上下文。

• Markdown 格式便于人工编辑，也方便 Agent 解析结构化信息。

• 记忆内容按用户偏好、历史记录、学到的知识三个维度组织，形成持续进化的用户画像。

• 跨会话持久化：即使程序重启，记忆也不会丢失。

七、核心组装：create_enhanced_deep_agent()

def create_enhanced_deep_agent():
   checkpointer = MemorySaver()
​
   middleware_list = [
       LogggingMiddleware()
  ]
​
   agent = create_deep_agent(
       model=shared_llm,
       tools=[get_current_weather, calculate, delete_file],
       system_prompt=DEEPAGENT_SYSTEMP_PROMPT,
       subagents=[research_agent],
       checkpointer=checkpointer,
       interrupt_on={'delete_file': True},
       memory=[MEMORY_FILE],
       debug=True,
       middleware=middleware_list
  )
   return agent, checkpointer

create_deep_agent 参数逐项解读：

interrupt_on 的工作原理： 当 Agent 调用 delete_file 工具时，deepagents 会在工具执行前自动暂停，生成一个中断点。开发者通过 agent.get_state(config) 检查 snapshot.next 来判断是否有待审批的中断。

八、中断与审批流程

def run_with_interrupt_demo(agent, checkpointer):
   config = {
       'configurable': {
           "thread_id": "demo_thread_001",
           "user_id": "test_user"
      }
  }
​
   user_query = """请帮我完成以下三个任务:
      1. 查询北京的天气
      2. 计算 123 * 456 / 2 的结果
      3. 删除文件 "old_report.txt"
      对于最后一个任务，如果用户需要确认，请等待批准。"""
​
   result = agent.invoke(
      {"messages": [{"role": "user", "content": user_query}]},
       config=config
  )
​
   snapshot = agent.get_state(config)
   if snapshot.next:
       user_decision = input("\\n⚠️ 是否批准删除操作？(approve/reject): ")
       if user_decision.lower() == "approve":
           resume_value = {"decisions": [{"type": "approve"}]}
           result = agent.invoke(Command(resume=resume_value), config=config)
       else:
           resume_value = {"decisions": [{"type": "reject", "message": "用户拒绝了删除操作"}]}
           result = agent.invoke(Command(resume=resume_value), config=config)

审批流程的状态机：

Agent 执行任务1（天气查询）→ 成功
  ↓
Agent 执行任务2（数学计算）→ 成功
  ↓
Agent 执行任务3（删除文件）→ 触发中断
  ↓
agent.get_state() → snapshot.next 非空
  ↓
用户输入 approve/reject
  ↓
agent.invoke(Command(resume=…)) → 恢复执行
  ↓
Agent 根据决策继续（执行删除 / 跳过）

关键机制：

• thread_id 是会话的唯一标识，用于跨多次 invoke 调用维护同一会话的状态。

• Command(resume=…) 是 LangGraph 的恢复指令，将用户的审批决定注入回 Agent 的执行流。

• snapshot.next 返回 None 表示没有待处理的中断，Agent 已执行完毕。

九、状态恢复：跨会话持久化

def demo_state_recovery(agent, checkpointer):
   config = {
       "configurable": {
           "thread_id": "demo_thread_001",    # 使用相同的 thread_id
           "user_id": "test_user"
      }
  }
​
   state = agent.get_state(config)
   if state.values:
       print("🔄 从持久化状态恢复会话…")
       print(f"   历史消息数量: {len(state.values.get('messages', []))}")
   else:
       print("📝 未找到历史状态，开始新会话")

• 相同的 thread_id 指向同一个持久化会话。

• agent.get_state(config) 从 MemorySaver 中读取历史状态，包含完整的消息历史。

• 注意：这里的 MemorySaver 是内存存储，程序退出后状态会丢失。生产环境应替换为数据库支持的检查点（如 SqliteSaver 或 PostgresSaver）。

十、架构全景图

┌─────────────────────────────────────────────────────┐
│                     DeepAgent                       │
│                                                     │
│ ┌──────────────┐ ┌──────────────┐                 │
│ │ Logging     │ │ MemorySaver │                 │
│ │ Middleware │ │ (Checkpoint)│                 │
│ └──────┬───────┘ └──────┬───────┘                 │
│         │                 │                         │
│ ┌──────▼─────────────────▼───────┐                 │
│ │         主 Agent               │                 │
│ │   (Qwen-Plus via DashScope)     │                 │
│ │                                 │                 │
│ │ 工具:                         │                 │
│ │ ├── get_current_weather       │                 │
│ │ ├── calculate                 │                 │
│ │ └── delete_file (需审批)       │                 │
│ │                                 │                 │
│ │ 中断机制: interrupt_on         │                 │
│ │ 记忆: agent_memory.md         │                 │
│ └──────────────┬──────────────────┘                 │
│                 │ task 委托                         │
│ ┌──────────────▼──────────────────┐                 │
│ │       子 Agent                 │                 │
│ │     research_assistant           │                 │
│ │                                 │                 │
│ │ 工具: [calculate]               │                 │
│ │ 角色: 深度研究与信息收集       │                 │
│ └─────────────────────────────────┘                 │
└─────────────────────────────────────────────────────┘

十一、最佳实践总结

11.1 安全设计

• 工具安全：calculate 工具通过覆盖 __builtins__ 实现了受限的 eval，防止代码注入。

• 审批门禁：delete_file 工具不直接执行操作，而是通过 interrupt_on 触发人工审批，形成人机协作的安全闭环。

• API 密钥管理：密钥硬编码仅为演示用途，生产环境应使用环境变量或密钥管理服务。

11.2 架构设计

• 单一模型实例：主 Agent 和子 Agent 共享同一个 shared_llm，避免资源浪费。

• 中间件可插拔：AgentMiddleware 的继承机制使得日志、监控、限流等功能可以独立开发和组合。

• 文件级记忆：使用 Markdown 文件存储长期记忆，兼顾可读性和可编辑性。

11.3 可扩展点

• 替换为持久化检查点：将 MemorySaver 替换为 SqliteSaver 或 PostgresSaver，支持程序重启后恢复。

• 增加子智能体：可注册多个不同专业领域的 SubAgent，如代码审查助手、数据分析助手等。

• 增强中间件：可在 before_model / after_model 中添加 token 计数、耗时统计、错误重试等逻辑。

• 丰富记忆系统：可将 agent_memory.md 替换为向量数据库，支持语义检索。

十二、完整执行流程

本文基于 deepagents 示例代码拆解，涵盖了从工具定义到子智能体委托、从中间件到人工审批的完整链路。希望对正在构建 AI Agent 系统的开发者有所启发。