DeerFlow
深度完全教程

# 1. 安装 Python 依赖（使用 uv）
curl -LsSf https://astral.sh/uv/install.sh | sh
uv sync

# 2. 安装前端依赖
cd web && npm install && cd ..

# 3. 启动后端服务
uv run python -m deerflow.server

# 4. 启动前端开发服务器（新终端）
cd web && npm run dev

# 5. 启动 Docker 沙箱（代码执行需要）
docker compose up sandbox -d

# DeerFlow 主配置文件
# 位置：项目根目录 config.yaml

# ── LLM 配置 ──────────────────────────────────────────
llm:
  # 主模型：负责 Lead Agent 的规划和决策（推荐使用较强模型）
  default_model: "deepseek-chat"
  
  # 研究模型：Sub-Agent 执行搜索和分析（可使用较轻量模型）
  research_model: "deepseek-chat"
  
  # 代码模型：代码生成和审查
  code_model: "deepseek-coder"
  
  # 最大 Token 限制
  max_tokens: 8192
  
  # 温度参数（0.0-1.0，研究任务推荐 0.3-0.7）
  temperature: 0.5
  
  # 推理模式（仅支持推理模型）
  enable_reasoning: false

# ── Sub-Agent 配置 ─────────────────────────────────────
agents:
  # 最大并行 Sub-Agent 数量
  max_concurrent: 5
  
  # 单个 Sub-Agent 最大执行时长（秒）
  timeout: 300
  
  # Context 压缩阈值（token 数，超过后自动摘要）
  context_compress_threshold: 6000
  
  # 最大递归深度（防止 Agent 无限循环）
  max_recursion_depth: 10

# ── 沙箱配置 ────────────────────────────────────────────
sandbox:
  # 执行模式：docker | local | e2b
  mode: "docker"
  
  # Docker 镜像
  image: "deerflow/sandbox:latest"
  
  # CPU 限制（核心数）
  cpu_limit: 2.0
  
  # 内存限制
  memory_limit: "2g"
  
  # 网络访问（代码执行时是否允许网络）
  network_access: false
  
  # 文件挂载（宿主机路径:容器路径）
  volumes:
    - "./workspace:/workspace"

# ── 记忆系统配置 ─────────────────────────────────────────
memory:
  # 启用持久化记忆
  enabled: true
  
  # 存储后端：sqlite | postgres | redis
  backend: "sqlite"
  
  # 记忆自动更新触发条件（每 N 次会话）
  auto_update_interval: 5
  
  # 最大存储条目数（per user）
  max_entries_per_user: 1000

# ── Skills 配置 ──────────────────────────────────────────
skills:
  # 技能目录（相对路径）
  skills_dir: "./skills"
  
  # 自动发现并加载技能
  auto_discover: true
  
  # 默认启用的内置技能
  builtin_enabled:
    - web_search
    - web_browse
    - code_execute
    - file_read
    - file_write

# ── 服务配置 ─────────────────────────────────────────────
server:
  host: "0.0.0.0"
  port: 8000
  
  # CORS 配置
  cors_origins:
    - "http://localhost:3000"
    - "https://your-domain.com"
  
  # 请求速率限制（每分钟）
  rate_limit: 60
  
  # 最大并发任务数
  max_concurrent_tasks: 20

# ── 日志配置 ─────────────────────────────────────────────
logging:
  level: "INFO"  # DEBUG | INFO | WARNING | ERROR
  format: "json"  # json | text
  file: "./logs/deerflow.log"

# ── LLM API Keys ─────────────────────────────────────
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
DEEPSEEK_API_KEY=sk-...
MOONSHOT_API_KEY=...
ARK_API_KEY=...           # ByteDance Doubao

# ── 自定义 API Base（代理/本地部署）
OPENAI_BASE_URL=https://api.openai.com/v1

# ── 搜索工具 API Keys ─────────────────────────────────
TAVILY_API_KEY=tvly-...
BRAVE_SEARCH_API_KEY=BSA...
SERPAPI_API_KEY=...

# ── IM 集成 ────────────────────────────────────────────
TELEGRAM_BOT_TOKEN=123456789:ABC...
SLACK_BOT_TOKEN=xoxb-...
FEISHU_APP_ID=cli_...
FEISHU_APP_SECRET=...

# ── 数据库（可选，默认使用 SQLite）
DATABASE_URL=postgresql://user:pass@localhost/deerflow

# ── 安全配置 ──────────────────────────────────────────
SECRET_KEY=your-secret-key-here
API_KEY=deerflow-api-key-for-clients

from deerflow.skills import BaseSkill, SkillResult
from deerflow.skills.registry import register_skill
from pydantic import BaseModel, Field

class MySkillInput(BaseModel):
    """技能输入参数定义（使用 Pydantic 进行类型校验）"""
    query: str = Field(..., description="搜索查询词")
    max_results: int = Field(default=5, description="最大结果数")

@register_skill(
    name="my_custom_skill",          # 技能唯一标识
    description="我的自定义搜索技能，用于...",  # Agent 用来决策是否调用
    category="search",               # 分类：search/code/file/api/custom
    version="1.0.0",
    author="your-name",
)
class MyCustomSkill(BaseSkill):
    """自定义技能示例"""
    
    # 声明输入 Schema
    InputSchema = MySkillInput
    
    async def execute(self, input: MySkillInput) -> SkillResult:
        """
        技能核心执行逻辑。
        - 支持 async/await
        - 支持流式输出（yield）
        - 可访问 self.context 获取当前 Agent 上下文
        """
        try:
            # 你的实现逻辑
            results = await self._do_search(input.query, input.max_results)
            
            return SkillResult(
                success=True,
                data=results,
                metadata={
                    "source": "my_custom_skill",
                    "query": input.query,
                    "count": len(results),
                }
            )
        except Exception as e:
            return SkillResult(success=False, error=str(e))
    
    async def _do_search(self, query: str, max_results: int) -> list:
        # 实现具体逻辑
        return []

# 任务命令
/research <topic>    - 启动深度研究任务
/code <description> - 生成并执行代码
/analyze <file>     - 分析上传的文件

# 记忆管理
/memory list        - 查看我的记忆
/memory clear       - 清空记忆
/memory edit <id>  - 编辑记忆条目

# 任务控制
/status             - 查看当前任务状态
/cancel             - 取消正在执行的任务
/history            - 查看历史任务

# 配置
/config             - 查看/修改个人配置
/model <name>       - 切换使用的 LLM 模型
/help               - 显示帮助信息

# config.yaml — Telegram 集成配置
telegram:
  enabled: true
  
  # 允许使用的用户（user_id 列表，留空则允许所有）
  allowed_users:
    - 123456789
    - 987654321
  
  # 允许使用的群组（group_id 列表）
  allowed_groups:
    - -100123456789
  
  # 消息格式（markdown / html）
  parse_mode: "markdown"
  
  # 流式输出（实时更新消息）
  streaming: true
  
  # 文件大小限制（MB）
  max_file_size: 50
  
  # 每用户并发任务限制
  max_concurrent_tasks_per_user: 2

{
  "user_id": "123456789",
  "preferences": {
    "language": "zh-CN",
    "model": "deepseek-chat",
    "response_style": "concise",
    "enable_memory": true,
    "max_research_depth": 3,
    "code_sandbox": "docker",
    "timezone": "Asia/Shanghai",
    "notification": {
      "task_complete": true,
      "task_failed": true,
      "progress_updates": false
    }
  }
}

pip install deerflow-sdk
# 或使用 uv
uv add deerflow-sdk

import asyncio
from deerflow import DeerFlowClient

async def main():
    # 初始化客户端
    client = DeerFlowClient(
        base_url="http://localhost:8000",
        api_key="your-api-key",        # 可选
        timeout=300,                   # 全局超时（秒）
    )
    
    # 同步调用（等待完整结果）
    result = await client.run(
        task="分析 Python 中 asyncio 的工作原理，并给出最佳实践",
        user_id="user_001",            # 用于记忆隔离
        model="deepseek-chat",         # 覆盖默认模型
    )
    print(result.content)
    
    # 流式调用（实时接收输出）
    async for event in client.stream(
        task="对 DeerFlow 项目进行深度技术分析",
        user_id="user_001",
    ):
        if event.type == "text_delta":
            print(event.data, end="", flush=True)
        elif event.type == "tool_call":
            print(f"\n[🔧 调用工具: {event.data['tool_name']}]")
        elif event.type == "done":
            print(f"\n\n✅ 任务完成，耗时: {event.data['duration']}s")
    
    await client.close()

asyncio.run(main())

from deerflow import DeerFlowClient
from deerflow.models import UserConfig, TaskConfig

client = DeerFlowClient(base_url="http://localhost:8000")

# 更新用户配置
await client.users.update_config(
    user_id="user_001",
    config=UserConfig(
        model="claude-sonnet-4-5",
        language="zh-CN",
        enable_memory=True,
        response_style="detailed",
    )
)

# 查看用户记忆
memories = await client.memory.list(user_id="user_001")
for mem in memories:
    print(f"[{mem.created_at}] {mem.content}")

# 删除特定记忆
await client.memory.delete(user_id="user_001", memory_id="mem_123")

# 获取任务状态
task = await client.tasks.get(task_id="task_abc123")
print(f"状态: {task.status}, 进度: {task.progress}%")

# 取消任务
await client.tasks.cancel(task_id="task_abc123")

from pathlib import Path

# 上传文件作为任务输入
with open("data.csv", "rb") as f:
    result = await client.run(
        task="分析这份 CSV 数据，找出异常值并生成可视化图表",
        files=[("data.csv", f, "text/csv")],
        user_id="user_001",
    )

# 下载任务生成的输出文件
task_id = result.task_id
files = await client.tasks.list_files(task_id=task_id)
for file_info in files:
    content = await client.tasks.download_file(
        task_id=task_id,
        file_path=file_info.path
    )
    Path(file_info.name).write_bytes(content)
    print(f"✅ 已下载: {file_info.name} ({file_info.size} bytes)")

from deerflow import DeerFlowClient
from deerflow.integrations.claude_code import ClaudeCodeBridge

# 初始化 Claude Code 桥接
bridge = ClaudeCodeBridge(
    deerflow_client=DeerFlowClient(base_url="http://localhost:8000"),
    workspace_path="/path/to/your/project",
)

# 将整个代码库作为上下文进行深度分析
async for event in bridge.analyze_codebase(
    task="找出所有潜在的安全漏洞，并提供修复建议",
    include_patterns=["*.py", "*.js"],
    exclude_patterns=["node_modules/**", "*.pyc"],
):
    print(event.content, end="")

# 自动重构
result = await bridge.refactor(
    description="将所有同步数据库调用改为异步版本",
    files=["app/db.py", "app/models.py"],
    dry_run=True,  # 先预览，不实际写入
)