deer-flow/
├── backend/
│   ├── app/gateway/           ← FastAPI 网关（HTTP API + LangGraph Runtime）
│   ├── deerflow/              ← 核心 Harness 包
│   │   ├── agents/            ← Agent 定义
│   │   │   └── lead_agent/    ← 主 Agent（编排入口）
│   │   ├── subagents/         ← 子 Agent 系统
│   │   ├── tools/             ← 工具系统
│   │   ├── skills/            ← 技能系统
│   │   ├── sandbox/           ← 沙箱执行环境
│   │   ├── memory/            ← 记忆系统
│   │   ├── models/            ← 模型接入
│   │   ├── runtime/           ← LangGraph 运行时
│   │   ├── persistence/       ← 持久化层
│   │   ├── community/         ← 社区工具（搜索、图片等）
│   │   ├── config/            ← 配置解析
│   │   ├── mcp/               ← MCP 协议支持
│   │   └── uploads/           ← 文件上传
│   └── packages/harness/      ← 可安装的 Harness 包
├── frontend/
│   ├── src/app/               ← Next.js 路由
│   │   ├── workspace/         ← 工作区页面
│   │   └── ...
│   ├── src/components/        ← UI 组件
│   └── src/core/              ← 前端核心逻辑
├── skills/                    ← 内置技能
│   └── public/                ← 公开技能包
└── config.yaml                ← 主配置文件

用户输入
    │
    ▼
┌──────────────────────┐
│   Gateway (FastAPI)   │  ← 接收 HTTP/SSE 请求
│   AuthMiddleware      │  ← JWT 认证
│   LangGraph Runtime   │  ← LangGraph 驱动
└──────┬───────────────┘
       │
       ▼
┌──────────────────────┐
│   Lead Agent         │  ← 主 Agent（LangGraph 图）
│   ├── 理解意图        │
│   ├── 选择工具/Skill  │
│   └── 编排 Sub-Agent  │
└──────┬───────────────┘
       │
       ▼
┌──────────────────────┐
│   LLM Model          │  ← 调用大模型
│   Tools              │  ← 执行工具
│   Sandbox            │  ← 沙箱环境
│   Memory             │  ← 读写记忆
└──────────────────────┘

# deerflow/agents/lead_agent/agent.py
 
# State 定义 — 在图节点之间传递的数据对象
class AgentState(TypedDict):
    messages: list[BaseMessage]     # 消息历史
    context: dict                    # 上下文（agent_name, thinking_enabled 等）
    is_plan_mode: bool               # 是否计划模式
    deep_research_exhausted: bool    # 深度研究是否已用完
    training_data: list | None       # Agent 训练数据（memory）
    memory_updated: bool             # 记忆是否已更新

# 简化的图结构
graph = StateGraph(AgentState)
 
# 节点注册
graph.add_node("call_model", call_model_node)        # 调用 LLM
graph.add_node("call_tool", call_tool_node)          # 执行工具
graph.add_node("memory_agent", memory_agent_node)    # 更新记忆
 
# 边定义
graph.add_conditional_edges(
    "call_model",
    should_continue,              # 条件函数：判断是否还有工具调用
    {"continue": "call_tool", "end": "__end__"}
)
graph.add_edge("call_tool", "call_model")  # 工具执行完后回到 LLM

call_model (LLM 思考)
    │
    ├── 如果 LLM 返回了 tool_calls → call_tool
    │       │
    │       └── 执行完工具 → 回到 call_model
    │
    └── 如果 LLM 返回了最终答案 → __end__

# 编译图
agent = graph.compile(
    checkpointer=checkpointer,     # 状态检查点（支持中断/恢复）
    interrupt_before=[],           # 在哪些节点前中断
    interrupt_after=[],            # 在哪些节点后中断
)
 
# 执行
result = await agent.ainvoke(
    input={"messages": [user_message]},
    config={"configurable": {"thread_id": thread_id}},
)

用户消息
    │
    ▼
1. 构建 System Prompt
   ├── 加载 config 中的模型配置
   ├── 注入工具列表（tools）
   ├── 注入技能（skills）
   ├── 注入记忆（memory）
   ├── 注入上下文（agent_name, user_id 等）
   └── 注入 SOUL.md（Agent 人格）
    │
    ▼
2. Call Model（调用 LLM）
   ├── 创建 ChatModel 实例
   ├── 绑定工具（bind_tools）
   ├── 设置参数（temperature, max_tokens 等）
   └── 调用 LLM（流式或非流式）
    │
    ▼
3. Should Continue（判断下一步）
   ├── LLM 返回了 tool_calls → 去 call_tool
   ├── LLM 返回了最终答案 → 结束
   └── LLM 返回了空内容 → 去 memory_agent
    │
    ▼
4. Call Tool（执行工具）
   ├── 解析 tool_call
   ├── 检查权限/护栏
   ├── 执行工具函数
   └── 将结果放入 messages
    │
    ▼
5. Memory Agent（更新记忆）
   ├── 从对话中提取关键信息
   └── 更新 memory.json

# prompt.py 中的核心函数
def get_system_prompt(config, user_context, tools, skills) -> str:
    parts = []
 
    # 1. 角色定义
    parts.append(get_role_prompt())
 
    # 2. 工具列表
    parts.append(get_tools_prompt(tools))
 
    # 3. 技能描述
    parts.append(get_skills_prompt(skills))
 
    # 4. 记忆上下文
    parts.append(get_memory_prompt(user_context))
 
    # 5. SOUL.md（Agent 人格）
    parts.append(get_soul_prompt(user_context))
 
    # 6. 规则约束
    parts.append(get_rules_prompt())
 
    return "\n\n".join(parts)

# agents/lead_agent/agent.py — 核心执行函数
 
async def call_model_node(state: AgentState, config: RunnableConfig):
    """LLM 调用节点 — Agent 的"大脑" """
 
    # 1. 构建消息列表
    messages = state["messages"]
 
    # 2. 获取绑定了工具的 LLM
    runtime = config["configurable"]["runtime"]
    llm = runtime.get_chat_model()
 
    # 3. 注入 Agent 上下文
    llm = llm.bind_tools(tools)
 
    # 4. 调用 LLM
    response = await llm.ainvoke(messages)
 
    # 5. 返回新的消息
    return {"messages": [response]}

# subagents/executor.py
 
class SubAgentExecutor:
    """子 Agent 执行器"""
 
    async def execute(self, task: SubTask) -> SubTaskResult:
        # 1. 为子 Agent 创建独立的 LangGraph 图
        # 2. 配置独立的 LLM 和工具集
        # 3. 执行任务并收集结果
        # 4. 将结果返回给 Lead Agent

Lead Agent
    │
    ├── 分析复杂任务
    ├── 拆分为子任务
    ├── 创建 Sub-Agent（每个一个独立的 LangGraph 实例）
    │
    ├── 模式 1：串行执行
    │   SubAgent1 → SubAgent2 → SubAgent3
    │
    ├── 模式 2：并行执行
    │   SubAgent1 ──┐
    │   SubAgent2 ──┤── 合并结果
    │   SubAgent3 ──┘
    │
    └── 收集子任务结果 → 汇总输出

# config.yaml 中的 subagent 配置
subagents:
  enabled: true
  max_concurrency: 3        # 最大并行数
  default_timeout: 300      # 默认超时（秒）

# tools/tools.py
 
def get_available_tools(
    groups: list[str] | None = None,
    model_name: str | None = None,
    app_config: AppConfig | None = None,
) -> list[BaseTool]:
    """从 config 中读取 tool 配置，动态加载"""
 
    # 1. 从 config.yaml 读取 tools 列表
    config = app_config or load_config()
    tool_configs = config.tools
 
    # 2. 过滤 tool group
    if groups:
        tool_configs = [t for t in tool_configs if t.group in groups]
 
    # 3. 动态加载每个 tool
    tools = []
    for tc in tool_configs:
        # resolve_variable 通过字符串路径加载 Python 对象
        tool_class = resolve_variable(tc.use, BaseTool)
        tool_instance = tool_class()
        tools.append(tool_instance)
 
    return tools

# config.yaml
tools:
  - name: web_search
    group: web
    use: deerflow.community.ddg_search.tools:web_search_tool
    max_results: 5
 
  - name: create_plan
    group: plan
    use: deerflow.plans.tools:create_plan

def resolve_variable(use_path: str, expected_type: type) -> Any:
    """解析 'module.submodule:object' 格式的路径"""
    module_path, obj_name = use_path.split(":", 1)
    module = importlib.import_module(module_path)
    return getattr(module, obj_name)

my-skill.skill/
├── SKILL.md          ← 技能定义（必选）
├── scripts/          ← 脚本（可选）
├── templates/        ← 模板（可选）
├── references/       ← 参考文件（可选）
└── evals/            ← 评估数据（可选）

---
name: my-skill
description: 当用户请求 X 时使用此技能。触发词：Y、Z。
---
 
# My Skill
 
## Overview
此技能的作用是...
 
## Workflow
### Step 1: 理解需求
- 识别用户输入
- 确定需要的数据
 
### Step 2: 执行
- 写代码/查资料/生成内容
 
### Step 3: 输出
- 格式化结果
- 保存到 /mnt/user-data/outputs/

# skills/parser.py — 解析 SKILL.md
 
def parse_skill(skill_dir: Path) -> Skill:
    """解析 skill 目录，提取元数据和指令"""
 
    skill_file = skill_dir / "SKILL.md"
    with open(skill_file) as f:
        content = f.read()
 
    # 解析 YAML front-matter
    _, front_matter, body = content.split("---", 2)
    meta = yaml.safe_load(front_matter)
 
    return Skill(
        name=meta["name"],
        description=meta["description"],
        instructions=body.strip(),
        scripts_dir=skill_dir / "scripts",
        templates_dir=skill_dir / "templates",
    )

# 在 system prompt 中注入技能描述
skills_section = "## Available Skills\n\n"
for skill in available_skills:
    skills_section += f"- {skill.name}: {skill.description}\n"

# sandbox/local/provider.py
 
class LocalSandboxProvider:
    """本地沙箱 — 直接在当前进程执行"""
 
    async def execute(self, command: str) -> SandboxResult:
        """在本地执行命令"""
        proc = await asyncio.create_subprocess_shell(
            command,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await proc.communicate()
        return SandboxResult(
            stdout=stdout.decode(),
            stderr=stderr.decode(),
            exit_code=proc.returncode,
        )

# community/aio_sandbox/
 
class AioSandboxProvider:
    """容器沙箱 — 在 Docker 容器中隔离执行"""
 
    async def execute(self, command: str) -> SandboxResult:
        """在隔离容器中执行命令"""
        container = await self._get_or_create_container()
        result = await container.exec_run(command)
        return SandboxResult(
            stdout=result.output.decode(),
            stderr="",
            exit_code=result.exit_code,
        )

.deer-flow/
├── memory.json              ← 全局记忆
└── users/
    └── {user_id}/
        ├── memory.json      ← 用户级记忆
        └── agents/
            └── {agent_name}/
                └── memory.json  ← Agent 级记忆

# memory/manager.py
 
class MemoryManager:
    """记忆管理器 — 在对话结束后自动更新"""
 
    async def update_memory(self, messages: list[BaseMessage]):
        """从对话中提取关键信息并更新记忆"""
        # 1. 提取关键事实
        facts = self._extract_facts(messages)
 
        # 2. 更新用户偏好
        preferences = self._extract_preferences(messages)
 
        # 3. 写入 memory.json
        memory = self._load_memory()
        memory["facts"].extend(facts)
        memory["preferences"].update(preferences)
        self._save_memory(memory)
 
    def _load_memory(self) -> dict:
        with open(self.memory_path) as f:
            return json.load(f)

# config.yaml
models:
  - name: deepseek-v4-flash
    display_name: DeepSeek V4 Flash
    use: deerflow.models.patched_deepseek:PatchedChatDeepSeek
    model: deepseek-v4-flash
    api_key: $DEEPSEEK_API_KEY
    supports_thinking: true
    when_thinking_enabled:
      extra_body:
        thinking:
          type: enabled

# models/factory.py
 
def create_chat_model(model_config: ModelConfig) -> BaseChatModel:
    """根据配置创建 ChatModel 实例"""
 
    # 1. 解析 use 路径
    model_class = resolve_variable(model_config.use)
 
    # 2. 构建参数字典
    kwargs = {
        "model": model_config.model,
        "api_key": model_config.api_key,
        "timeout": model_config.timeout,
    }
 
    # 3. 添加可选参数
    if model_config.base_url:
        kwargs["base_url"] = model_config.base_url
    if model_config.max_tokens:
        kwargs["max_tokens"] = model_config.max_tokens
    if model_config.temperature:
        kwargs["temperature"] = model_config.temperature
 
    # 4. 实例化
    return model_class(**kwargs)

# models/patched_deepseek.py
 
class PatchedChatDeepSeek(ChatDeepSeek):
    """修复多轮对话中 reasoning_content 丢失的问题"""
 
    def _convert_message_to_dict(self, message):
        """确保 reasoning_content 在消息中正确传递"""
        d = super()._convert_message_to_dict(message)
        if hasattr(message, "reasoning_content"):
            d["reasoning_content"] = message.reasoning_content
        return d

# runtime/ 目录结构
runtime/
├── __init__.py
├── checkpointer/    ← 状态检查点
├── events/          ← 事件系统
│   └── store/       ← 事件存储
├── runs/            ← 运行管理
│   └── store/       ← 运行记录存储
├── store/           ← 通用存储
└── stream_bridge/   ← SSE 流式传输

# runtime/stream_bridge/stream_bridge.py
 
class StreamBridge:
    """将 LangGraph 的事件流转换为 SSE"""
 
    async def stream_events(self, thread_id: str, input: dict):
        """流式传输 Agent 执行事件"""
 
        async for event in self.agent.astream_events(input):
            # 事件类型：
            # on_chat_model_start  — LLM 开始思考
            # on_chat_model_end    — LLM 思考完成
            # on_tool_start        — 工具开始执行
            # on_tool_end          — 工具执行完成
 
            sse_event = self._convert_to_sse(event)
            yield sse_event

/api/
├── v1/auth/          ← 认证
├── threads/          ← 对话线程 CRUD
├── threads/{id}/runs ← 运行管理
├── threads/{id}/artifacts ← 文件制品
├── threads/{id}/uploads   ← 文件上传
├── models           ← 模型列表
├── memory           ← 记忆管理
├── skills           ← 技能管理
├── agents           ← 智能体管理
└── plans            ← 计划任务

# auth_middleware.py
 
class AuthMiddleware(BaseHTTPMiddleware):
    """JWT 认证中间件 — 保护所有 /api/* 路由"""
 
    async def dispatch(self, request, call_next):
        # 公开路径免认证
        if _is_public(request.url.path):
            return await call_next(request)
 
        # 检查 JWT Cookie
        token = request.cookies.get("access_token")
        if not token:
            return JSONResponse(status_code=401, ...)
 
        # 验证 JWT
        user = await decode_token(token)
        request.state.user = user
 
        return await call_next(request)

skills/custom/my-skill/
├── SKILL.md
└── scripts/
    └── run.py

---
name: my-skill
description: >
  当用户请求执行 X 操作时使用此技能。
  触发词：X、Y、Z。
  不适用于 A、B 情况。
---
 
# My Skill
 
## Overview
此技能的作用是...
 
## Workflow
### Step 1: 理解需求
- 识别用户输入中的关键参数
 
### Step 2: 执行
- 使用 bash_tool 运行脚本
 
### Step 3: 输出
- 结果保存到 /mnt/user-data/outputs/
- 使用 present_file 展示结果

skills:
  directories:
    - ./skills/public
    - ./skills/custom

用户上传文件 → data-analysis 读取 → pandas 处理 → 输出结果

用户提问 → deep-research 搜索 → 多角度分析 → 整理报告

用户需求 → ppt-generation 规划 → image-generation 配图 → 合成输出

config.yaml → 动态解析 → 组件实例化 → 注入 Agent → 执行