多Agent编排之手写调度器

1
2
3
4
5
6
7
8
9

用户问题
  -> 规划研究大纲
  -> 搜索资料
  -> 数据分析
  -> 生成图表
  -> 撰写报告
  -> 审核报告
  -> 必要时补充搜索/修订
  -> 最终报告

1
2
3
4
5
6

ChiefArchitect  规划
DeepScout       搜索
DataAnalyst     数据分析
CodeWizard      图表/代码
LeadWriter      写报告
CriticMaster    审核

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

query
outline
facts
data_points
charts
final_report
references
critic_feedback
quality_score
iteration
phase

1
2
3
4
5

资料不够
引用缺失
逻辑有问题
数据过旧
报告不完整

1

review -> revise -> review

1

review -> re_researching -> writing -> review

1
2
3
4
5
6
7
8
9

while iteration < max_iterations:
    critic.process(state)
    if state["phase"] == "completed":
        break
    elif state["phase"] == "re_researching":
        scout.process(state)
        writer.process(state)
    elif state["phase"] == "revising":
        writer.process(state)

1
2
3
4
5
6
7
8
9

本地知识库检索
联网搜索
招投标数据分析
政策文档分析
人类审核
后台批处理
中断恢复
不同模型分配
不同策略切换

1
2
3
4
5
6

节点是什么
输入输出是什么
成功后去哪
失败后去哪
审核不通过去哪
达到最大轮次去哪

1
2
3
4
5
6

planning
researching
analyzing
writing
reviewing
revising

1
2
3
4

当前执行到哪个节点？
为什么从 review 回到了 revise？
为什么进入 re_researching？
哪一轮审核通过了？

1
2
3
4
5
6

当前 phase 是什么
上一个完成节点是什么
下一个应该执行哪个节点
当前 iteration 是第几轮
哪些状态已经生成
哪些节点不能重复跑

1
2
3
4
5
6

暂停
恢复
重试
跳过
回滚
人工介入

1
2
3
4
5
6
7
8
9

StateGraph(ResearchState)
  add_node("plan", _plan_node)
  add_node("research", _research_node)
  add_node("analyze", _analyze_node)
  add_node("write", _write_node)
  add_node("review", _review_node)
  add_node("revise", _revise_node)
  add_conditional_edges("review", _should_revise, {"revise": "revise", "complete": END})
  add_edge("revise", "review")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

while state["iteration"] < state["max_iterations"]:
    ...
    async for msg in run_agent_with_streaming(self.critic):
        yield msg

    if state["phase"] == ResearchPhase.RE_RESEARCHING.value:
        async for msg in run_agent_with_streaming(self.scout):
            yield msg
        state["phase"] = ResearchPhase.WRITING.value
        async for msg in run_agent_with_streaming(self.writer):
            yield msg

    elif state["phase"] == ResearchPhase.REVISING.value:
        async for msg in run_agent_with_streaming(self.writer):
            yield msg

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

ResearchRouter.stream_research
  -> DeepResearchV2Service.research
    -> DeepResearchGraph.run
      -> _run_simplified
        -> run_agent_with_streaming
          -> asyncio.create_task(agent.process(state))
          -> BaseAgent.add_message
          -> asyncio.Queue
    -> DeepResearchV2Service._format_sse
  -> StreamingResponse

1
2
3

# 主函数
async def _run_simplified(self, state: ResearchState):
    ...

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165

class ResearchState(TypedDict):
    """
    LangGraph 状态定义

    这是整个研究过程的全局状态，所有Agent都在读写这个状态。
    """
    # 基础信息
    query: str                              # 用户原始问题
    session_id: str                         # 会话ID
    phase: str                              # 当前阶段
    iteration: int                          # 当前迭代轮次
    max_iterations: int                     # 最大迭代次数

    # 搜索模式配置
    search_web: bool                        # 是否启用网络搜索
    search_local: bool                      # 是否启用本地知识库搜索

    # 规划输出
    outline: List[Dict[str, Any]]           # 动态大纲 (Section序列化)
    mind_map: Dict[str, Any]                # 知识图谱/思维导图
    key_entities: List[str]                 # 关键实体
    research_questions: List[str]           # 待研究的子问题
    hypotheses: List[Dict[str, Any]]        # 研究假设（假设驱动研究）
    knowledge_graph: Dict[str, Any]         # 知识图谱 {nodes: [], edges: []}

    # 知识库
    facts: List[Dict[str, Any]]             # 结构化事实库
    data_points: List[Dict[str, Any]]       # 数据点
    raw_sources: List[Dict[str, Any]]       # 原始来源（网页内容）

    # 分析输出
    charts: List[Dict[str, Any]]            # 生成的图表
    code_executions: List[Dict[str, Any]]   # 代码执行记录
    insights: List[str]                     # 数据洞察

    # 写作输出
    draft_sections: Dict[str, str]          # 章节草稿 {section_id: content}
    final_report: str                       # 最终报告
    references: List[Dict[str, Any]]        # 参考文献

    # 审核反馈
    critic_feedback: List[Dict[str, Any]]   # 评论家反馈
    unresolved_issues: int                  # 未解决问题数
    quality_score: float                    # 质量评分
    pending_search_queries: List[str]       # 待执行的补充搜索查询（审核后需要补充的）

    # 元数据
    logs: List[Dict[str, Any]]              # 执行日志
    errors: List[str]                       # 错误记录
    messages: List[Dict[str, Any]]          # Agent间消息（用于流式输出）


def create_initial_state(
    query: str,
    session_id: str,
    search_web: bool = True,
    search_local: bool = False
) -> ResearchState:
    """创建初始状态

    Args:
        query: 用户查询
        session_id: 会话ID
        search_web: 是否启用网络搜索（默认True）
        search_local: 是否启用本地知识库搜索（默认False）
    """
    ...

def section_to_dict(section: Section) -> Dict[str, Any]:
    """Section 序列化"""
    ...

def fact_to_dict(fact: Fact) -> Dict[str, Any]:
    """Fact 序列化"""
    ...

class ResearchPhase(str, Enum):
    """研究阶段状态机"""
    INIT = "init"                    # 初始化
    PLANNING = "planning"            # 规划阶段
    RESEARCHING = "researching"      # 深度探索阶段
    ANALYZING = "analyzing"          # 数据分析阶段
    WRITING = "writing"              # 撰写阶段
    REVIEWING = "reviewing"          # 对抗审核阶段
    RE_RESEARCHING = "re_researching"  # 补充搜索阶段（审核发现缺失信息后）
    REVISING = "revising"            # 修订阶段（仅文字修改）
    COMPLETED = "completed"          # 完成


@dataclass
class Section:
    """报告章节"""
    id: str
    title: str
    description: str
    section_type: Literal["qualitative", "quantitative", "mixed"]  # 定性/定量/混合
    status: Literal["pending", "researching", "drafted", "reviewed", "final"]
    content: str = ""
    sources: List[str] = field(default_factory=list)
    subsections: List['Section'] = field(default_factory=list)
    requires_data: bool = False
    requires_chart: bool = False


@dataclass
class Fact:
    """结构化事实"""
    id: str
    content: str
    source_url: str
    source_name: str
    source_type: Literal["official", "academic", "news", "report", "self_media"]  # 来源类型
    credibility_score: float  # 可信度评分 0-1
    extracted_at: datetime
    related_sections: List[str] = field(default_factory=list)  # 关联章节ID
    verified: bool = False
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class DataPoint:
    """数据点"""
    id: str
    name: str
    value: Any
    unit: str
    year: Optional[int]
    source: str
    confidence: float


@dataclass
class Chart:
    """图表配置"""
    id: str
    title: str
    chart_type: Literal["line", "bar", "pie", "scatter", "table", "heatmap"]
    data: Dict[str, Any]
    code: str  # 生成图表的Python代码
    image_path: Optional[str] = None
    section_id: Optional[str] = None


@dataclass
class CriticFeedback:
    """评论家反馈"""
    id: str
    target_section: str
    issue_type: Literal["missing_source", "logic_error", "bias", "hallucination", "outdated", "incomplete"]
    severity: Literal["critical", "major", "minor"]
    description: str
    suggestion: str
    resolved: bool = False


@dataclass
class AgentLog:
    """Agent执行日志"""
    timestamp: datetime
    agent: str
    action: str
    input_summary: str
    output_summary: str
    duration_ms: int
    tokens_used: int = 0

1

state = create_initial_state(query, session_id)

1
2
3
4
5
6
7
8

await self.planner.process(state)
await self.deep_scout.process(state)
await self.architect.process(state)
await self.scout.process(state)
await self.data_analyst.process(state)
await self.wizard.process(state)
await self.writer.process(state)
await self.critic.process(state)

1
2
3
4
5

ChiefArchitect 写 outline
DeepScout 读 outline，写 facts/references
DataAnalyst 读 facts，写 data_points/charts/knowledge_graph
LeadWriter 读 facts/charts/references，写 final_report
CriticMaster 读 final_report，写 critic_feedback/quality_score/phase

1
2

state["_message_queue"] = asyncio.Queue()
message_queue = asyncio.Queue()

1
2
3
4
5
6
7
8

def add_message(state, event_type, content):
    ...
    """添加消息到消息队列"""
    message = {"type": event_type, "content": content}
    state["messages"].append(message)
    if "_message_queue" in state:
        state["_message_queue"].put_nowait(message)
    ...

1
2
3
4
5
6
7

Agent
  -> add_message()
  -> 写入 state["messages"]
  -> 推入 asyncio.Queue
  -> 调度器从 Queue 读取
  -> yield 给 SSE
  -> 前端实时展示

1

task = asyncio.create_task(agent.process(state))

1
2
3
4

while not task.done():
    msg = await asyncio.wait_for(message_queue.get(), timeout=0.5)
    # 让调度器最多等待 0.5 秒，然后回到循环检查取消状态。
    yield msg

1
2
3
4
5
6
7
8

前端点击停止
  -> POST /research/cancel/{session_id}
  -> Redis 写 cancel flag
  -> 调度器每 0.5 秒检查
  -> 发现 cancelled
  -> task.cancel()
  -> 当前 Agent 被取消
  -> 调度器结束流程

1
2
3
4
5

save_checkpoint_async({
    "type": "planning",
    "status": "completed",
    "stats": {...}
})

1
2
3
4
5
6
7
8
9

phase
iteration
outline
facts
charts
final_report
references
critic_feedback
...

1
2
3
4
5
6

research_steps
search_results
charts
knowledge_graph
streaming_report
references

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

def save_checkpoint(
        self,
        session_id: str,
        state: Dict[str, Any],
        user_id: Optional[str] = None,
        ui_state: Optional[Dict[str, Any]] = None,
        final_report: Optional[str] = None,
        db: Optional[Session] = None
    ) -> Optional[str]:
        """
        保存检查点

        Args:
            session_id: 研究会话 ID
            state: ResearchState 字典（后端状态）
            user_id: 用户 ID（可选）
            ui_state: 前端 UI 状态（研究步骤、搜索结果、图表等）
            final_report: 最终报告内容

        Returns:
            检查点 ID，失败返回 None
        """
        try:
            # 提取关键信息
            query = state.get("query", "")
            phase = state.get("phase", "planning")
            iteration = state.get("iteration", 0)

            # 清理 state 中不可序列化的内容
            ...
            # 查找现有检查点
            ...

            if existing:
                # 更新现有检查点
                ...
                existing.status = "running"
                ...
            else:
                # 创建新检查点
                checkpoint = ResearchCheckpoint(
                    session_id=session_id,
                    user_id=UUID(user_id) if user_id else None,
                    query=query,
                    phase=phase,
                    iteration=iteration,
                    state_json=clean_state,
                    ui_state_json=clean_ui_state,
                    final_report=final_report,
                    status="running",
                )
                db.add(checkpoint)
                db.flush()
                checkpoint_id = str(checkpoint.id)

            db.commit()
            # 详细日志
            ...
            return checkpoint_id

        except Exception as e:
            logger.error(f"Failed to save checkpoint: {e}")
            db.rollback()
            return None

1
2
3
4
5

1. GET /research/checkpoint/{session_id}/full
   恢复前端 UI 展示

2. POST /research/resume/{session_id}
   恢复后端研究执行

1

GET /research/checkpoint/{session_id}/full

1
2
3
4
5
6

研究步骤
搜索结果
图表
知识图谱
已生成的报告片段
引用

1
2
3

research_checkpoints.state_json
research_checkpoints.ui_state_json
research_checkpoints.final_report

1
2

用户刷新页面 / 重新进入会话
还能看到之前的研究过程

1

service_v2.research(..., resume=True)

1

state = self._load_checkpoint(session_id)

1

后端任务中断后，希望继续跑

1
2
3
4
5

用户只是想看看历史结果
但后端又开始继续跑 Agent
又开始消耗模型 token
又可能重新搜索
又会改变 checkpoint

1

GET /checkpoint/full

1
2

前端只能恢复显示
后端不会继续执行

1

POST /resume/{session_id}

1
2
3
4
5

前端进入 /chat/{session_id}
  -> GET /research/checkpoint/{session_id}/full
  -> 后端返回 checkpoint
  -> 前端恢复研究步骤、搜索结果、图表、报告
  -> 不继续执行任务

1
2
3
4
5
6

用户点击继续研究
  -> POST /research/resume/{session_id}
  -> 后端加载 state_json
  -> 后端继续执行 Agent 流程
  -> 后端通过 SSE 持续返回新事件
  -> 前端继续更新研究过程 UI

1
2
3
4

取消任务
保存阶段级检查点
恢复前端 UI
加载后端 state

1

从中断的具体 Agent/具体 phase 精确继续

1

Plan -> Research -> Analyze -> Write -> Review...

1

加载旧状态，重新进入流程

1

从上次中断点继续执行下一个节点

1
2
3
4

completed_steps
next_step
paused status
phase-aware scheduler

1
2
3

checkpoint.phase == "analyzing"
completed_steps = ["planning", "researching"]
next_step = "analyzing"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

共享状态：
  ResearchState dict 在所有 Agent 之间传递

实时事件：
  state["_message_queue"] + BaseAgent.add_message()

任务控制：
  asyncio.create_task(agent.process(state))

取消：
  Redis cancel flag + task.cancel()

检查点：
  CheckpointService 保存 state_json + ui_state_json

恢复展示：
  前端加载 full checkpoint 恢复 UI

继续执行：
  后端已有 resume 雏形，但还不是严格断点的精准续跑

1
2
3
4
5

ChiefArchitect
DeepScout
DataAnalyst
CodeWizard
LeadWriter

1
2
3
4

1. 清空 state["messages"]
2. update_ui_state()
3. save_checkpoint_async(step_info)
4. yield checkpoint_saved 事件