Multi-Agent-Workflow && Data Flow

Multi-Agent-Workflow

Tip

多代理工作流架构，不同的专业代理协作处理用户查询，并生成一份全面的研究报告

Agent 职责和能力

DeerFLow 的多智能体工作流程由五个专业智能体组成，以下是它们特定的角色以及责任

Coordinator Agent

将用户的请求分为三类

• • 直接处理：问候和闲聊
• • 礼貌拒绝：不当或有害的请求，系统会拒绝处理
• • 交给 Planner：研究问题、事实查询，会被进一步处理

Planner Agent

将复杂查询分解为可管理的步骤，以便创建详细的研究计划

1. 1. 评估是否有足够的上下文可以直接回答查询
2. 2. 若上下文不足，则创建一个结构化的研究计划，包含具体步骤
3. 3. 将步骤分类为“研究”（网络搜索）或处理（计算任务）
4. 4. 限制计划的步骤数量，最大思考步数（默认为 3 次）

Researcher Agent

执行需要收集信息的研究步骤

1. 1. 使用内置工具，如 web_search_tool 和 crawl_tool
2. 2. 在可用时动态加载工具
3. 3. 记录文档来源和属性信息，以确保引用
4. 4. 结构化综合收集的信息

Coder Agent

执行涉及到 Python 代码的数据处理

1. 1. 分析需求并规划解决方案
2. 2. 使用 Python 脚本实现解决方案
3. 3. 利用可用的库，如 pandas、numpy
4. 4. 记录方法并可视化结果

Reporter Agent

负责所有收集的信息综合成完整的最终报告

• • 以清晰的结构，逻辑地组织信息
• • 突出关键发现和重要见解
• • 包含研究过程中相关的图片
• • 正确引用所有来源
• • 保持用户偏好的语言

工作流执行过程

工作流配置

提示到 Agent 的映射

将代理的提示映射到定义其行为的实际代码文件

计划结构化和步骤执行

步骤类型和执行

步骤分为研究、处理两种类型

StepType.RESEARCH

• • 需要进行网络搜索 need_web_search: true
• • 重点在于收集信息，统计数据和数据点
• • 由 Researcher Agenet 执行，使用网络搜索和爬取工具

StepType.PROCESSING

• • 不需要进行网络搜索 need_web_search: false
• • 涉及计算、数据分析和计算任务
• • 由 Coder Agent 执行，使用 Python 进行数据处理

工具整合

DeerFlow 集成了代理可用于收集信息和处理数据的各种工具

内置工具

动态工具加载随着 MCP

支持通过 MCP 动态加载其他工具

这里给出 GitHub 趋势存储库的 MCP 示例

"mcp_settings": {
    "servers": {
        "mcp-github-trending": {
            "transport": "stdio",
            "command": "uvx",
            "args": ["mcp-github-trending"],
            "enabled_tools": ["get_github_trending_repositories"],
            "add_to_agents": ["researcher"]
        }
    }
}

Data Flow

Tip

本文的记录数据的流转，用户输入到最终研究报告的产出，前后端交互时的消息、事件、状态，以及每个组件内的内部数据转换

Data Flow 的预览

前端 Data Flow

用户输入作为 API Request

sendMessage 函数（web/src/core/store/store.ts 76-154）的处理流程如下：

1. 1. 创建新的用户消息，生成唯一 ID，并将其添加到 Zustand 状态存储
2. 2. 设置存储中检索聊天设置，使用 getChatStreamSettings()
3. 3. 使用用户消息和配置参数发起 API 调用，调用 chatStream()
4. 4. 在对话期间将前端状态设置为“responding”

chat 的 Stream 参数包括如下

事件处理与状态管理

前端使用 Zustand 来管理应用状态，关键状态组件包括

当后端接收到事件时，它们会被处理并更新上述状态

• • message_chunk：更新消息内容
• • tool_calls：处理工具调用
• • tool_call_result：更新工具调用结果
• • interrupt：处理用户反馈

以下这些状态更新函数用于管理聊天、研究过程中的状态变化，确保前端能够正确显示对话和研究进度

• • appendMessage：向存储中添加新消息
• • updateMessage：使用新内容更新现有消息
• • updateMessages：批量更新多个消息
• • setOngoingResearch：跟踪当前研究会话
• • openResearch：控制 UI 中显示的研究内容

API 通信

前后端通过 SSE 进行流式响应，支持两种模式

• • 常规模式（Regular mode）：发送实际请求到后端，获取实时响应
• • 回放模式（Replay mode）：使用预录的响应进行演示或测试

后端 Data Flow

请求处理

当后端接收到聊天请求时，通过 /api/chat/stream 断点进行处理

• • 解析用户输入并生成 UUID
• • 初始化流式事件
• • 启动 StreamResponse，使用 _astream_workflow_generator() 处理请求并生成流式响应

工作流事件生成

_astream_workflow_generator() 职责如下

1. 1. 设置 LangGraph 工作流的初始状态
2. 2. 将代理交互和结果流式传输回客户端
3. 3. 将 LangGraph 事件转换为 SSE（服务器发送事件）

LangGraph 状态流

管理 Agent 节点之间的状态转换，每个节点处理当前状态，并返回一个对象，该对象用于更新状态并引导工作流进入下一个节点

LangGraph 的状态对象包含

数据结构

Message 结构

Message 是 DeerFlow 系统的主要数据结构，在前后端之间流动

ChatEvent 结构

从后端流式传递到前端的事件具有如下结构

{
  "event": "<event_type>",
  "data": "<JSON 数据>"
}

其中 event_type 可以是：

• • message_chunk：来自代理的文本内容
• • tool_calls：代理请求使用工具
• • tool_call_chunks：工具调用的部分内容
• • tool_call_result：工具执行的结果
• • interrupt：请求用户反馈

ChatRequest 结构

从前端传递到后端的请求结构如下

{
  "messages": [{ "role": "user", "content": "string" }],
  "thread_id": "string",
  "auto_accepted_plan": true,
  "max_plan_iterations": 3,
  "max_step_num": 5,
  "interrupt_feedback": "string",
  "enable_background_investigation": true,
  "mcp_settings": {
    "servers": {
      "server_name": {
        "transport": "string",
        "command": "string",
        "args": ["string"],
        "url": "string",
        "env": { "key": "value" },
        "enabled_tools": ["string"],
        "add_to_agents": ["string"]
      }
    }
  }
}