Skip to content
/ 4-deepagent Public

This project is the fourth deepagent companion project in the Cufel-X project, and the content is a deepagent framework adapted to MCP adapter

0 stars 0 forks Branches Tags Activity
Star
Notifications You must be signed in to change notification settings

CUFEL-X/4-deepagent

BranchesTags

Repository files navigation

DeepAgent

基于 LangChain DeepAgents 的金融智能 Agent 框架

Python LangChain DeepAgents MCP License

轻量级、模块化的金融领域 AI Agent 框架，基于 LangChain DeepAgents 构建，开箱即用，实现主流 MCP Host 级别的便捷性，易于扩展。

注：此文档为项目介绍文档，教学练习文档详见 docs/deepagent_tutorial.md

📌 基于 LangChain DeepAgents 深度定制的金融智能 Agent 框架

支持多步骤规划、虚拟文件系统、长期记忆、对话隔离，集成 MCP 服务实现 Claude Desktop 级别的无缝体验。

1. 介绍（Introduction）

1.1 项目定位

cufel_deepagent 是一个建立在 LangChain DeepAgents 之上的定制化框架，专门为金融领域的 AI 应用设计。它提供了一套简洁、强大的工具集，让开发者能够快速构建具有多步骤规划能力的智能 Agent，同时保持代码的模块化和可维护性。

CUFEL-DEEPAGENT介绍.png

1.2 核心优势

1.2.1 🎯 模块化架构

  • 热插拔设计：Memory、Skill、Tool、MCP、SubAgent 等模块完全独立，支持动态加载和卸载
  • 自动注册机制：无需手动配置，新增模块自动被框架发现和注册
  • 清晰的职责分离：每个模块有明确的用途和接口，易于理解和维护

1.2.2 🚀 开箱即用

  • 内置工具集：网页搜索、文件操作、数据查询等常用工具已预装
  • MCP 主机支持：原生支持 Model Context Protocol，实现 Claude Desktop 等MCP Client的使用体验

1.2.3 🧠 高级 Agent 能力

  • 多步骤规划：Agent 能够自动分解复杂任务为多个步骤
  • 文件系统访问：虚拟文件系统支持，安全的文件读写操作
  • 对话隔离：支持多个独立的对话会话，每个会话有独立的历史记录
  • 长期记忆：持久化的记忆存储，跨会话保留用户信息和上下文

1.2.4 💼 金融领域优化

  • 专业工具集：包含公司信息查询、宏观经济分析等金融特定工具
  • 数据安全：凭证管理、安全的 API 调用
  • 可审计性：完整的日志记录和工具调用的 Human in Loop 审核机制

1.3 功能示例

本项目集成了基于 DeepAgent 框架构建的三个简单应用示例，演示如何组合工具、SubAgent、MCP 和 Skill 来完成复杂任务。

1.3.1 撰写行业概览

场景：使用 AI 撰写一篇图文并茂的行业概览并生成配图，。

核心组件：

组件类型 名称 说明
工具 web_search 搜索网络资料，获取写作素材
SubAgent writer-agent 专业的文案写作 Agent
SubAgent image-generator-agent AI 图像生成 Agent
Skill industry-research-with-image 行业研究报告（含配图）技能

工作流程：

1. 规划阶段
   └─ 使用 write_todos 规划文章结构

2. 写作阶段
   └─ 调用 writer-agent 撰写博客内容（Markdown 格式）

3. 绘图阶段
   └─ 调用 image-generator-agent，根据文章内容生成配图

4. 整合阶段
   └─ 将图片插入文章适当位置，生成最终博客

使用示例：

uv run main.py --run "帮我写一篇关于 AI 行业的文章，配上合适的图片"

相关文件：

1.3.2 宏观经济分析与黄金交易模拟

场景：基于世界银行数据分析宏观经济形势，并进行黄金投资模拟交易。

核心组件：

组件类型 名称 说明
MCP worldbank 世界银行宏观经济数据（GDP、通胀、失业率等）
MCP fetch 网页内容抓取，获取上海黄金交易所实时金价
MCP gold_server_cny 自定义黄金交易模拟服务器
Skill macro-economic-analysis 宏观经济分析技能

MCP 服务器配置：

点击展开/收起 MCP 服务器配置代码 
# src/cufel_deepagent/mcp/mcp.py
MCP_SERVERS_CONFIG = {
    "worldbank": {
        "command": "npx",
        "args": ["worldbank-mcp"],
        "transport": "stdio",
    },
    "fetch": {
        "command": "uvx",
        "args": ["mcp-server-fetch"],
        "transport": "stdio",
    },
    "gold_server_cny": {
        "command": os.sys.executable,
        "args": [str(Path(__file__).parent / "servers" / "gold_server_cny.py")],
        "transport": "stdio",
    },
}

工作流程：

1. 数据获取
   ├─ 使用 worldbank MCP 查询各国经济指标
   └─ 使用 fetch MCP 获取上海黄金交易所实时金价

2. 宏观经济分析
   └─ 使用 macro-economic-analysis 技能分析数据

3. 投资模拟
   ├─ 查看黄金账户状态（通过 gold_server_cny Resource）
   └─ 执行买入/卖出操作（通过 gold_server_cny Tool）

使用示例：

uv run main.py --run "分析中美两国近5年的GDP增长趋势，然后查询当前黄金价格，并模拟买入100克黄金"

相关文件：

1.3.3 证券持仓管理与止盈止损模拟

场景：智能体基于止盈止损策略管理证券持仓，模拟真实交易场景中的风险控制。

核心组件：

组件类型 名称 说明
MCP stock_server 证券持仓管理系统（止盈止损模拟服务器）
Resource portfolio://overview 查看持仓总览（当前价、成本、止盈止损线）
Tool check_positions 检查所有持仓是否触发止盈/止损条件
Tool update_price 更新股票当前价，判断是否触发止盈止损
Tool clear_position 清仓卖出全部股票（用于止盈/止损触发时）
Prompt instruction_manual 证券投资专家提示词（含止盈止损策略逻辑）

MCP 服务器配置：

点击展开/收起 MCP 服务器配置代码 
# src/cufel_deepagent/mcp/mcp.py
MCP_SERVERS_CONFIG = {
    # ... 其他服务器配置 ...
    "stock_portfolio": {
        "command": os.sys.executable,
        "args": [str(Path(__file__).parent / "servers" / "stock_server.py")],
        "transport": "stdio",
    },
}

工作流程：

1. 查看持仓
   └─ 使用 portfolio://overview 查看所有持仓状态

2. 价格监控
   └─ 使用 update_price 更新股票当前价

3. 调仓检查
   └─ 使用 check_positions 检查是否触发止盈/止损

4. 执行清仓
   └─ 对触发条件的股票调用 clear_position 清仓

使用示例：

uv run main.py --run "检查我的证券持仓，告诉我哪些达到了清仓条件,并帮我执行"
uv run main.py --run "将深空智能的价格更新为75元，然后检查是否需要调仓"
uv run main.py --run "虚研科技涨到了60元，触发止盈了吗？需要清仓吗？"

相关文件：

1.5 项目文件结构

点击展开/收起项目文件结构 
cufel_deepagent/
├── src/cufel_deepagent/  # 核心包目录
│   ├── core/                 # Agent 核心逻辑
│   │   └── agent.py              # Agent 初始化和配置
│   ├── config.py                 # 统一配置管理
│   ├── tools/                # 工具模块
│   │   ├── tools.py              # 原生工具定义（可编辑）
│   │   ├── mcp_tools.py          # MCP 工具适配器
│   │   └── registry.py           # 工具自动注册
│   ├── mcp/                  # MCP 集成模块
│   │   ├── mcp.py                # MCP 服务器配置（可编辑）
│   │   ├── manager.py            # MCP 工具、资源、提示词管理
│   │   ├── mcp_resources_prompts.py  # Resources/Prompts 便捷函数
│   │   ├── adapter.py            # MCP 适配器
│   │   ├── interceptors.py       # 拦截器和中间件
│   │   └── servers/              # MCP 服务器实现目录（可编辑）
│   │       └── gold_server_cny.py    # 黄金交易教学服务器（示例）
│   ├── subagents/            # 子 Agent 模块
│   │   ├── subagents.py          # SubAgent 定义（可编辑）
│   │   ├── registry.py           # SubAgent 注册
│   │   └── templates.py          # SubAgent 提示词模板
│   ├── repl/                 # 交互式界面
│   │   ├── interactive.py        # 交互模式入口
│   │   ├── processor.py          # 消息处理
│   │   ├── display.py            # 输出格式化
│   │   ├── logging.py            # 统一日志接口
│   │   ├── interrupt_handler.py  # 工具调用审核
│   │   └── tool_display.py       # 工具信息展示
│   └── utils/                # 工具函数
│       └── history.py            # 对话历史管理
├── skills/                   # Skill 脚本目录（可编辑）
│   ├── tell-a-joke/
│   ├── write-fairy-tale/
│   └── ...
├── memories/                 # 长期记忆存储
│   └── *.txt                     # 用户信息、偏好等
├── workspaces/               # 工作区（临时文件）
├── .env                      # 环境变量配置
├── main.py                   # 程序入口
└── README.md                 # 本文件

关键说明：

  • tools/tools.py、mcp/mcp.py、subagents/subagents.py 是可编辑的，用于定义你的自定义工具、MCP 服务和子 Agent
  • mcp/servers/ 目录用于存放自定义 MCP 服务器实现（推荐使用 FastMCP）
  • skills/ 目录中的 SKILL.md 文件定义了 Agent 可用的技能
  • memories/ 中的文件用于存储跨会话的用户信息
  • workspaces/ 用于临时文件和项目文件

2. 快速开始（Quickstart）

2.1 环境配置

安装 Node.js 和 uv（MCP 依赖）

MCP 服务需要 Node.js（npx）和 uv/uvx 环境：

系统 安装命令
Windows powershell -ExecutionPolicy Bypass -c "irm https://gitee.com/wangnov/uv-custom/releases/download/0.10.0/uv-installer-custom.ps1 | iex"
macOS curl -LsSf https://gitee.com/wangnov/uv-custom/releases/download/0.10.0/uv-installer-custom.sh | sh

nodejs 推荐通过官方渠道下载 https://nodejs.org/en/download

安装 Python 依赖并配置凭证：

# 1. 安装依赖
uv sync

# 2. 配置 .env 文件（必需：QWEN_API_KEY）

必需凭证：QWEN_MODEL、QWEN_API、QWEN_URL

2.2 三种使用方式

2.2.1 方式 1：默认单轮交互

最简单的方式，运行一个预设的问题：

uv run main.py

输出示例：

============================================================
运行默认单轮交互示例 (你好)
============================================================
Deep Agent 单次交互
提示词: 你好

[Agent 处理中...]

2.2.2 方式 2：自定义单轮交互

运行一个自定义的问题，Agent 处理后返回结果：

uv run main.py --run "分析一下当前的经济形势"

或简写：

uv run main.py -r "分析一下当前的经济形势"

适用场景：

  • 快速测试 Agent 的能力
  • 集成到其他系统中
  • 批量处理多个问题

2.2.3 方式 3：交互模式（多轮对话）

启动交互式 REPL，进行多轮对话：

uv run main.py --interactive

或简写：

uv run main.py -i

交互模式下的命令：

  • 输入问题并按 Enter 发送
  • 输入 exit 或 quit 退出
  • 输入 help 查看帮助
  • 输入 node 切换节点显示（调试模式）

适用场景：

  • 与 Agent 进行多轮对话
  • 实时调试和测试
  • 交互式数据分析

2.3 对话隔离（多个独立会话）

所有三种方式都支持通过 -c 或 --conversation 参数指定对话名称，实现对话隔离：

# 单轮交互 + 对话隔离
uv run main.py --run "你好" -c user_alice

# 交互模式 + 对话隔离
uv run main.py --interactive -c user_bob

# 默认单轮 + 对话隔离
uv run main.py -c user_charlie

对话隔离的优势：

  • 每个对话有独立的历史记录文件
  • 不同用户的对话完全隔离
  • 支持并行处理多个对话
  • 便于多用户场景

文件位置：

memories/
├── conversation_history.json           # 默认对话历史
├── conversation_history_user_alice.json
├── conversation_history_user_bob.json
└── conversation_history_user_charlie.json

📖 示例对话体验

项目提供了一个完整的示例对话，展示 Agent 的多轮交互能力：

# 进入示例对话（会自动加载预置的对话历史）
uv run main.py -i -c "example"

示例对话内容：

  • 中美宏观经济分析报告撰写（含配图生成）
  • 股票账户与黄金账户管理（清仓、买入操作）
  • 深空智能公司投资分析报告

💡 提示：使用 -c example 会加载 memories/conversation_history_example.json 作为初始对话历史，你可以从中途继续对话，体验完整的多轮交互流程。

2.4 调试模式

所有运行方式都支持 --debug 参数启用调试模式：

uv run main.py --debug
uv run main.py --run "问题" --debug
uv run main.py --interactive --debug

调试模式会显示：

  • langgraph 中 Agent 执行的每个节点
  • 工具调用的详细信息
  • 中间步骤的结果

2.5 Python 代码调用

除了命令行方式，你也可以在 Python 代码中直接导入并调用 cufel_deepagent：

2.5.1 导入与基本使用

点击展开/收起导入与基本使用代码 
from cufel_deepagent import run_interactive_mode, run_single_chat

# 方式 1: 运行单轮对话
run_single_chat(
    prompt="你好，请介绍一下自己",
    debug=False,
    conversation_name=None
)

# 方式 2: 运行交互模式
run_interactive_mode(
    debug=False,
    conversation_name=None
)

2.5.2 方法属性详解

run_single_chat() - 单轮对话
点击展开/收起 run_single_chat 代码 
def run_single_chat(
    prompt: str,                          # 用户提示词（必需）
    debug: bool = False,                  # 是否启用调试模式，显示每个节点的执行情况
    conversation_name: str | None = None  # 对话名称，用于隔离不同对话的历史记录
) -> None:
    """运行单次 agent 对话"""

参数说明：

  • prompt（字符串）：用户输入的问题或指令
  • debug（布尔值）：
    • True：显示 langgraph 中每个节点的执行过程和中间结果
    • False：（默认）只显示最终结果，用于生产环境
  • conversation_name（字符串或 None）：
    • 指定对话名称时，会在 memories/ 中创建独立的历史文件
    • None（默认）：使用默认的对话历史文件

使用示例：

# 基础用法
run_single_chat("分析苹果公司的财务状况")

# 带调试模式
run_single_chat("分析苹果公司的财务状况", debug=True)

# 指定对话名称，支持多用户场景
run_single_chat("分析苹果公司的财务状况", conversation_name="user_alice")

# 完整参数
run_single_chat(
    prompt="分析苹果公司的财务状况",
    debug=True,
    conversation_name="financial_analyst_session_1"
)
run_interactive_mode() - 交互模式
点击展开/收起 run_interactive_mode 代码 
def run_interactive_mode(
    debug: bool = False,                  # 是否启用调试模式，显示节点执行情况
    conversation_name: str | None = None  # 对话名称，用于隔离不同对话的历史记录
) -> None:
    """以交互式 REPL 模式启动多轮对话"""

参数说明：

  • debug（布尔值）：
    • True：显示节点执行情况，支持输入 node 命令切换显示
    • False（默认）：标准模式，只显示 Agent 输出
  • conversation_name（字符串或 None）：
    • 指定对话名称时，创建独立的对话会话
    • None（默认）：使用默认对话

使用示例：

# 基础用法 - 启动交互式对话
run_interactive_mode()

# 启用调试模式
run_interactive_mode(debug=True)

# 指定对话名称 - 为不同的用户创建独立会话
run_interactive_mode(conversation_name="user_bob")

# 完整参数
run_interactive_mode(
    debug=True,
    conversation_name="data_analysis_session"
)

交互模式中的命令：

  • 输入问题 → Agent 处理并返回结果
  • exit 或 quit → 退出程序
  • help → 显示帮助信息
  • node → 在调试模式下切换节点显示（调试模式专用）

2.5.3 参数对比

参数 run_single_chat run_interactive_mode 说明
prompt ✅ 必需 ❌ 不需要 用户问题
debug ✅ 可选 ✅ 可选 调试模式
conversation_name ✅ 可选 ✅ 可选 对话隔离
返回值 None None 两个方法都不返回值
使用场景 单个问题、集成、自动化 多轮交互、实时调试 -

3. 模块管理（Module Management）

本节介绍如何为 Agent 添加和管理各种模块。

3.1 长期记忆（Memory）

长期记忆用于存储跨会话的用户信息、偏好和知识。

3.1.1 存储位置

点击展开/收起记忆文件结构 
memories/
├── conversation_history.json           # 对话历史（自动生成）
├── user_profile.txt                    # 用户信息
├── preferences.txt                     # 用户偏好
└── domain_knowledge.txt                # 领域知识

3.1.2 添加记忆

在 Agent 运行过程中，可以通过工具将信息写入记忆：

# 在 Agent 的工具中或通过 write_file 工具
# 写入用户信息
write_file(
    file_path="/memories/user_profile.txt",
    content="用户姓名: 张三\n公司: ABC 金融公司\n职位: 分析师"
)

# 写入用户偏好
write_file(
    file_path="/memories/preferences.txt",
    content="偏好语言: 中文\n报告格式: 详细分析\n更新频率: 每日"
)

3.1.3 读取记忆

Agent 会在启动时自动加载 memories/ 目录中的所有 .txt 文件作为上下文。

3.1.4 最佳实践

  • 使用清晰的文件名和结构化内容
  • 定期更新过期的信息
  • 敏感信息应加密存储
  • 为每个用户创建独立的记忆文件

3.2 工具（Tools）

工具是 Agent 可以调用的函数，用于执行特定的操作。

3.2.1 添加自定义工具

编辑 src/cufel_deepagent/tools/tools.py，使用 @tool 装饰器定义新工具：

点击展开/收起自定义工具代码示例 
from langchain.tools import tool

@tool
def analyze_stock_price(symbol: str, period: str = "1y") -> str:
    """分析股票价格趋势

    Args:
        symbol: 股票代码（如 AAPL）
        period: 分析周期（1d, 1w, 1m, 1y）

    Returns:
        分析结果
    """
    # 实现你的逻辑
    return f"分析 {symbol} 在 {period} 内的价格趋势..."

@tool
def get_financial_report(company_id: str) -> str:
    """获取公司财务报告

    Args:
        company_id: 公司 ID

    Returns:
        财务报告内容
    """
    # 实现你的逻辑
    return f"公司 {company_id} 的财务报告..."

3.2.2 工具规范

  • 使用 @tool 装饰器标记
  • 函数名使用下划线命名法（snake_case）
  • 所有参数必须有类型注解
  • 返回类型必须是 str
  • 提供详细的 docstring

3.2.3 工具自动注册

新增的工具会自动被框架发现和注册，无需额外配置。

3.2.4 工具调用审核

某些工具（如 write_file、web_search）会在执行前请求用户确认。可以在 src/cufel_deepagent/core/agent.py 中的 interrupt_on 配置修改审核规则。

3.3 技能（Skills）

技能是 Agent 可以使用的操作指南，通常用于复杂的多步骤任务。

3.3.1 创建新技能

在 skills/ 目录下创建新文件夹，并添加 SKILL.md 文件：

mkdir skills/my-skill
touch skills/my-skill/SKILL.md

3.3.2 技能文件格式

SKILL.md 使用 YAML 前置元数据 + Markdown 内容：

点击展开/收起技能文件格式示例 
---
name: analyze-company
description: 分析一个公司的基本信息、财务状况和市场前景
---

## 分析步骤

1. **获取公司基本信息**
   - 公司名称、成立时间、主营业务
   - 行业分类、市场地位

2. **分析财务状况**
   - 收入、利润、现金流
   - 负债率、ROE 等关键指标

3. **评估市场前景**
   - 行业趋势
   - 竞争对手分析
   - 增长潜力

## 输出格式

提供结构化的分析报告，包括：
- 公司概览
- 财务分析
- 市场评估
- 投资建议

3.3.3 使用技能

Agent 会自动识别和使用相关的技能。用户可以直接请求：

"帮我分析一下 Apple 公司"

Agent 会自动应用 analyze-company 技能来完成任务。

3.3.4 最佳实践

  • 技能名称应清晰描述其功能
  • 提供详细的步骤说明
  • 包含预期的输出格式
  • 列出所需的工具和资源

3.4 MCP 服务（Model Context Protocol）

MCP（Model Context Protocol）允许 Agent 接入外部服务和工具，如 Akshare、世界银行、飞书等。本框架支持 MCP 的 三元一体自动解析：Tools（工具）、Resources（资源）、Prompts（提示词）。

3.4.1 MCP 服务器目录结构

src/cufel_deepagent/mcp/
├── mcp.py                    # MCP 服务器配置入口
├── servers/                  # MCP 服务器实现（存放位置）
│   ├── gold_server_cny.py   # 黄金交易教学服务器（示例）
│   └── ...
├── manager.py                # MCP 管理器（自动解析三元一体）
├── mcp_resources_prompts.py  # Resources/Prompts 便捷函数
└── interceptors.py          # 拦截器和中间件

3.4.2 配置 MCP 服务器

编辑 src/cufel_deepagent/mcp/mcp.py，定义 MCP 服务器：

点击展开/收起 MCP 服务器配置代码 
import os
from pathlib import Path

MCP_SERVERS_CONFIG = {
    "fetch": {
        "command": "uvx",
        "args": ["mcp-server-fetch"],
        "transport": "stdio",
    },
    "worldbank": {
        "command": "npx",
        "args": ["worldbank-mcp"],
        "transport": "stdio",
    },
    # 自定义 MCP 服务器（推荐使用 Python FastMCP）
    "my_server": {
        "command": os.sys.executable,
        "args": [str(Path(__file__).parent / "servers" / "my_server.py")],
        "transport": "stdio",
    },
}

3.4.3 三元一体自动解析机制

框架会自动从 MCP 服务器解析三类信息并集成到系统提示词：

类型 解析内容 系统提示词中的作用 用途说明
Tools 工具函数签名和描述 直接作为 Agent 可调用的工具 执行操作（买入/卖出/查询等）
Resources URI、名称、描述（元数据） 列出可用资源，Agent 可主动查询 数据源（账户状态/文档等）
Prompts 名称、描述、内容（完整模板） 整合到系统提示词 模板化的专家指导

示例 - gold_server_cny.py：

点击展开/收起 FastMCP 服务器示例代码 
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("黄金交易教学系统")

# 1. Resource: 账户资产状态（只列元数据，内容需查询）
@mcp.resource("account://status")
def get_account_status() -> str:
    """查看当前黄金账户的现金余额与黄金储量情况"""
    ...

# 2. Tool: 买入黄金（自动注册为 Agent 工具）
@mcp.tool()
def buy_gold(amount_grams: float, current_price_cny: float) -> str:
    """使用现金买入黄金"""
    ...

# 3. Prompt: 功能说明（完整内容整合到系统提示词）
@mcp.prompt()
def instruction_manual() -> str:
    """提供关于此黄金账户的功能介绍和操作规范"""
    return ("你是一位金融教学专家...")

3.4.4 Resources 与 Prompts 的区别

特性 Resources Prompts
内容获取 Agent 主动调用 read_mcp_resource 工具查询 自动整合到系统提示词
存储内容 动态数据（账户状态、实时数据等） 静态模板（使用说明、专家指导等）
适用场景 需要时查询的数据源 始终需要的指导信息

3.4.5 错误处理

框架对 MCP 错误进行了智能分类处理：

  • 业务错误（如服务器不支持的功能）：DEBUG 级别日志
  • 真正异常：WARNING 级别日志
  • ExceptionGroup：DEBUG 级别，避免刷屏

3.5 子 Agent（SubAgents）

SubAgent 是专业化的 Agent，每个都有特定的专长和工具集。

3.5.1 创建新 SubAgent

编辑 src/cufel_deepagent/subagents/subagents.py，使用 @subagent 装饰器定义：

点击展开/收起 SubAgent 定义代码 
from cufel_deepagent.subagents.registry import subagent

@subagent(
    name="financial-analyst",
    description="金融分析专家，擅长财务分析和投资建议"
)
def financial_analyst_agent():
    """金融分析 SubAgent"""
    # SubAgent 的配置和工具集
    return {
        "tools": ["analyze_stock", "get_financial_report", "compare_companies"],
        "system_prompt": "你是一位资深的金融分析师..."
    }

@subagent(
    name="data-scientist",
    description="数据科学家，擅长数据分析和可视化"
)
def data_scientist_agent():
    """数据科学 SubAgent"""
    return {
        "tools": ["query_database", "create_chart", "statistical_analysis"],
        "system_prompt": "你是一位经验丰富的数据科学家..."
    }

3.5.2 调用 SubAgent

在主 Agent 中，可以通过 task() 工具调用 SubAgent：

# 在 Agent 的响应中
task(
    name="financial-analyst",
    task="分析 Apple 公司的财务状况和投资价值"
)

3.5.3 SubAgent 最佳实践

  • 为每个 SubAgent 定义清晰的专长领域
  • 提供专业化的工具集
  • 使用专业的系统提示词
  • 保持 SubAgent 的独立性和可复用性

4. 架构深度解析（Architecture Deep Dive）

本节深入讲解 cufel_deepagent 的内部架构。

4.1 核心组件

4.1.1 Agent 初始化（core/agent.py）

setup_agent() 函数是 Agent 的入口点，负责：

  • 凭证验证：检查必需的 API 密钥和配置
  • 后端配置：设置虚拟文件系统和路径映射
  • 工具注册：自动收集和注册所有可用工具
  • SubAgent 加载：动态加载所有已注册的 SubAgent
  • 中断配置：设置工具调用审核规则
  • LLM 初始化：创建 LLM 实例
  • 检查点保存：配置对话历史保存
点击展开/收起 setup_agent 代码 
def setup_agent(conversation_name: str | None = None):
    """设置并返回配置好的 Deep Agent"""
    # 1. 验证凭证
    cred_result = Config.validate_credentials()
    if not cred_result['valid']:
        raise ValueError(f"Missing credentials: {cred_result['missing']}")

    # 2. 配置后端
    backend = make_backend(runtime)

    # 3. 加载工具和 SubAgent
    native_tools = list(NATIVE_TOOLS)
    subagents = get_all_subagents()

    # 4. 创建 Agent
    agent = create_deep_agent(
        model=llm,
        memory=memory_files,
        skills=[skills_dir],
        tools=[*native_tools, *MCP_TOOLS],
        subagents=subagents,
        interrupt_on=interrupt_on,
        system_prompt=system_prompt
    )

    return agent

4.1.2 配置管理（config.py）

Config 类提供统一的配置接口：

  • 路径管理：标准化所有目录路径
  • 凭证获取：从环境变量读取 API 密钥
  • 凭证验证：检查必需凭证的有效性
  • 系统提示生成：动态生成 Agent 的系统提示词
  • !!注意!!：此处的提示词直接影响Agent本身各方面的工具和todo的执行能力

关键方法：

  • get_root_dir()：获取项目根目录
  • get_memory_dir()、get_skills_dir()、get_workspace_dir()：获取各类目录
  • validate_credentials()：验证凭证并返回详细结果
  • generate_system_prompt()：生成完整的系统提示词

4.1.3 工具系统（tools/）

tools.py：定义原生工具

  • 使用 @tool 装饰器
  • 支持自定义工具的快速添加

registry.py：工具自动注册

  • 使用 functools.lru_cache 缓存工具列表
  • 自动发现 tools.py 中的所有工具
  • 提供 NATIVE_TOOLS 集合供 Agent 使用

mcp_tools.py：MCP 工具适配器

  • 将 MCP 服务的工具转换为 LangChain 工具
  • 处理 MCP 工具的调用和结果转换

4.1.4 MCP 集成（mcp/）

mcp.py：MCP 服务器配置入口

  • 定义 MCP_SERVERS_CONFIG 字典
  • 支持 Python 脚本和命令行工具（uvx/npx）
  • 自动连接和管理多个 MCP 服务器

servers/：MCP 服务器实现目录

  • 存放 MCP 服务器的 Python 实现
  • 使用 FastMCP 框架快速开发
  • 支持 Tools、Resources、Prompts 三元一体

manager.py：MCP 管理器

  • load_mcp_tools()：加载 MCP 工具
  • get_resources_info()：获取资源元数据
  • get_prompts_info()：获取提示词完整信息
  • 5分钟缓存机制，避免重复连接
  • 智能异常处理（业务错误 vs 真正异常）

mcp_resources_prompts.py：Resources/Prompts 便捷函数

  • get_mcp_resources_info_for_prompt()：生成 Resources 提示词
  • get_mcp_prompts_info_for_prompt()：生成 Prompts 提示词
  • 供 config.py 的 generate_system_prompt() 调用

adapter.py：MCP 适配器

  • 将 MCP 工具转换为 LangChain 兼容格式
  • 处理工具调用的参数转换

interceptors.py：拦截器和中间件

  • 认证和错误处理
  • 重试逻辑（指数退避：0.5s, 1s, 2s）
  • 进度回调（DEBUG 级别，避免干扰主流程）

4.1.5 SubAgent 系统（subagents/）

subagents.py：SubAgent 定义

  • 使用 @subagent 装饰器定义新 SubAgent
  • 每个 SubAgent 有独立的工具集和系统提示

registry.py：SubAgent 注册

  • 自动发现和注册所有 SubAgent
  • 提供 get_all_subagents() 函数

templates.py：提示词模板

  • 通用的 SubAgent 系统提示模板
  • 避免重复编写相同的指导内容

4.1.6 交互界面（repl/）

interactive.py：交互模式入口

  • run_interactive_mode()：启动多轮对话
  • run_single_chat()：运行单轮对话
  • 支持对话隔离和调试模式

processor.py：消息处理

  • 处理 Agent 的流式输出
  • 管理工具调用和中断

display.py：输出格式化

  • 格式化 Agent 的响应
  • 美化工具调用信息

logging.py：统一日志接口

  • 提供 info()、warning()、error() 等函数
  • 统一的日志格式和输出

interrupt_handler.py：工具调用审核

  • should_interrupt()：判断是否需要中断
  • get_user_decision()：获取用户对工具调用的决定
  • handle_tool_execution()：处理工具执行决定

tool_display.py：工具信息展示

  • 工具描述和参数信息
  • 格式化工具调用日志

4.2 数据流

点击展开/收起数据流程图 
用户输入
    ↓
main.py (命令行解析)
    ↓
interactive.py (交互模式或单轮模式)
    ↓
setup_agent() (Agent 初始化)
    ├─ Config.validate_credentials() (凭证验证)
    ├─ 加载 tools (tools/registry.py)
    ├─ 加载 MCP tools (mcp/manager.py)
    ├─ 加载 subagents (subagents/registry.py)
    └─ 加载 skills (从 skills/ 目录)
    ↓
Agent 处理
    ├─ 理解用户意图
    ├─ 规划多步骤任务
    ├─ 调用工具或 SubAgent
    └─ 生成响应
    ↓
processor.py (处理流式输出)
    ├─ 工具调用审核 (interrupt_handler.py)
    ├─ 格式化输出 (display.py)
    └─ 记录日志 (logging.py)
    ↓
用户看到结果
    ↓
memories/ (保存对话历史和记忆)

4.3 虚拟文件系统

cufel_deepagent 使用虚拟文件系统来隔离不同的目录：

点击展开/收起虚拟文件系统结构 
/                    → 项目根目录
├── /skills/         → skills/ 目录
├── /memories/       → memories/ 目录
├── /workspace/      → workspaces/ 目录
├── /tools/          → src/cufel_deepagent/tools/ 目录
└── /mcp/            → src/cufel_deepagent/mcp/ 目录

这样 Agent 可以安全地访问这些目录，而不会意外修改其他文件。

4.4 凭证验证流程

启动 Agent
    ↓
Config.validate_credentials()
    ├─ 检查 QWEN_MODEL
    ├─ 检查 QWEN_API
    ├─ 检查 QWEN_URL
    ├─ 检查 BOCHA_API_KEY (可选)
    └─ 返回验证结果
    ↓
如果缺少必需凭证
    ├─ 记录详细错误信息
    ├─ 显示缺少的凭证列表
    └─ 抛出异常，停止启动
    ↓
如果所有凭证有效
    └─ 继续 Agent 初始化

4.5 工具调用审核流程

Agent 决定调用工具
    ↓
检查 interrupt_on 配置
    ├─ 如果 False → 直接执行
    ├─ 如果 Dict → 需要审核
    └─ 如果 True → 需要审核
    ↓
显示工具信息给用户
    ├─ 工具名称
    ├─ 工具参数
    └─ 允许的决定选项
    ↓
获取用户决定
    ├─ approve → 执行工具
    ├─ reject → 拒绝执行
    └─ edit → 编辑参数后执行
    ↓
执行决定
    └─ 继续 Agent 处理

5. 高级用法（Advanced Usage）

5.1 自定义系统提示词

编辑 .env 文件，添加自定义系统提示词：

CUSTOM_SYSTEM_PROMPT="你是一位资深的金融分析师，专门为机构投资者提供投资建议。"

5.2 集成外部数据源

通过添加工具来集成外部数据源：

@tool
def query_external_api(endpoint: str, params: dict) -> str:
    """查询外部 API"""
    import requests
    response = requests.get(endpoint, params=params)
    return response.json()

5.3 多 Agent 协作

使用 SubAgent 实现多个专业化 Agent 的协作：

# 主 Agent 可以调用多个 SubAgent
task(name="financial-analyst", task="分析财务数据")
task(name="data-scientist", task="生成可视化报告")

5.4 性能优化

  • 缓存：使用 functools.lru_cache 缓存频繁调用的函数
  • 异步处理：MCP 工具支持异步调用
  • 流式输出：Agent 支持流式响应，提高用户体验

5.5 安全最佳实践

  • 凭证管理：使用环境变量存储敏感信息，不要硬编码
  • 输入验证：在工具中验证用户输入
  • 日志过滤：敏感信息不要记录到日志
  • 工具审核：为危险操作启用审核机制

6. 常见问题（FAQ）

6.1 如何添加新的 LLM 提供商？

编辑 src/cufel_deepagent/core/agent.py，修改 LLM 初始化部分：

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="your-model",
    base_url="your-api-url",
    api_key="your-api-key"
)

6.2 如何调试 Agent 的行为？

使用 --debug 参数启动 Agent，查看详细的执行日志：

uv run main.py --debug

6.3 对话历史存储在哪里？

对话历史存储在 memories/ 目录中的 JSON 文件：

  • 默认对话：memories/conversation_history.json
  • 指定对话：memories/conversation_history_{name}.json

6.4 如何清除对话历史？

直接删除对应的 JSON 文件即可。

6.5 支持哪些 LLM 模型？

任何支持 OpenAI API 兼容接口的模型都可以使用，包括：

  • Qwen（阿里云）
  • OpenAI GPT 系列
  • 其他兼容 OpenAI API 的模型

测试

项目包含完整的单元测试套件，覆盖 Memory、Skill、Tool、MCP 四个核心系统。

# 运行所有测试
uv run pytest tests/ -v

# 运行特定模块
uv run pytest tests/test_memory_system.py -v
uv run pytest tests/test_skill_system.py -v
uv run pytest tests/test_tool_system.py -v
uv run pytest tests/test_mcp_system.py -v

贡献指南

欢迎提交 Issue 和 Pull Request！

致谢与引用 (Credits & Acknowledgments)

本教学项目的实现中参考或集成了以下开源项目：

最后更新：2026-02-07

About

This project is the fourth deepagent companion project in the Cufel-X project, and the content is a deepagent framework adapted to MCP adapter

Topics

langchain deepagent deepagents-langgraph cufel

Resources

Readme
Activity
Custom properties

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

No releases published

Packages

No packages published

Languages