Agent-10-09.多Agent编排与LangGraph

发表于 | 分类于
多 Agent 协作、显式状态图、LangGraph checkpoint 与本仓库 orchestrator 实现。

系列：00 索引 · 上一篇：08 HITL · 下一篇：10 长任务

1. 行业常见问题

现象 痛点
单 Agent 提示词过长 角色混乱、工具误用
if/else 散落业务代码 不可测试、不可视化
进程崩溃丢进度 长流程需恢复
无法「回到文献阶段改方向」 缺显式回退

多角色产品（调研+执行+审核）需要 分工 + 显式控制流。

2. 该技术如何解决

多 Agent：每角色独立 agent.yaml + MCP（sequence / literature / method_kb / planner / analyst / reporter）。
编排器：agent/orchestrator/graph.py 用 LangGraph 串联 Specialist 节点，不内嵌 RAG/Argo 细节。
Checkpoint：workspace/checkpoints.db 持久化 PipelineState，配合 HITL resume（见 08）。

3. 核心原理

3.1 PipelineState 存什么

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# agent/orchestrator/state.py
class PipelineState(TypedDict, total=False):
    run_id: str                      # LangGraph configurable.thread_id；RunStore 主键
    gene_symbol: str                 # 传入各 Specialist 的基因 symbol
    stage: str                       # 当前阶段名；CLI/SSE 展示进度
    sequence_summary: str            # node_sequence 写入；report / llm_report 输入
    sequence_artifact_path: str      # transcripts.json 等路径；审计链接
    literature_summary: str          # node_literature 摘要
    literature_artifact_path: str    # literature_manifest.json 路径
    method_kb_summary: str           # node_method_kb 摘要
    method_kb_artifact_path: str     # method_manifest.json 路径
    planner_summary: str             # node_planner 摘要；HITL 待办展示
    plan_artifact_path: str          # research_plan.json；HITL payload_path
    analyst_summary: str             # node_analyst 执行摘要
    execution_artifact_path: str     # execution_manifest.json 路径
    report: str                      # 最终 Markdown 正文（内存）
    report_artifact_path: str        # final_report.md 落盘路径
    approved: bool                   # HITL：True 才允许 analyst submit（见 08）
    iteration: int                  # 循环轮次（build_pipeline_loop）
    max_iterations: int             # 最大循环次数，超限 → END
    revise_feedback: str            # 人工调整方向，literature 重检索时消费
    review_approved: bool           # analyst 后审核通过 → reporter

只存 摘要 + artifact 路径，不存 FASTA/PDF 全文。

3.2 GraphConfig 编译开关

1
2
3
4
5
6
7
8
9
# agent/orchestrator/graph.py
@dataclass
class GraphConfig:
    workspace_root: Path = Path("workspace")  # RunStore / artifact / checkpoint 根目录
    use_llm_agents: bool = False   # True：Specialist 走 AgentRuntime.chat + tool loop
    use_llm_planner: bool = False  # True：planner 节点用 LangChain structured output
    force_mock_llm: bool = False   # True：无 API Key 时用 mock 回复
    collection: str = "brca1_gene"   # literature 节点 query_hybrid 的 collection
    method_domain: str = "sirna_methods"  # method_kb 节点 query_graph 的 domain

3.3 节点粒度

一个 LangGraph 节点 = 一次 Specialist 入口（或 HITL 闸门），不是一个 MCP tool：

1
2
3
node_literature → run_literature_agent() / run_literature_direct()
node_planner    → run_planner_agent() / run_planner_direct()
node_analyst    → run_analyst_direct(hitl_approved=state["approved"])

--llm-agents 时走 AgentRuntime.chat + tool loop；默认 direct 模式调 MCP/函数，便于 CI 与无密钥验收。

3.4 两种流水线形态

模式 构建函数 边 用途
短流水线 build_pipeline(full=False) method_kb → report 三阶段调研 + 合成报告
线性 full build_pipeline(full=True) … → planner → hitl_plan → analyst → reporter 含方案与 Argo，无回环
审核循环 full build_pipeline_loop() analyst → hitl_review → literature（revise）或 reporter（approve） 执行结果不满意时人工给方向，从 literature 重跑

4. 典型实现与代码示例

基于 agent/orchestrator/graph.py，理解框架如何 构图 + 编译 + 调用。

4.1 线性构建图（build_pipeline）

默认 无环 DAG，适合一次性跑通、CI 验收。与循环版 并存，不互相替代。

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
# agent/orchestrator/graph.py（节选）
def build_pipeline(cfg, *, full: bool = False, with_hitl: bool = False, use_interrupt: bool = False):
    nodes = make_nodes(cfg)
    graph = StateGraph(PipelineState)

    graph.add_node("sequence", nodes["sequence"])
    graph.add_node("literature", nodes["literature"])
    graph.add_node("method_kb", nodes["method_kb"])
    graph.add_edge(START, "sequence")
    graph.add_edge("sequence", "literature")
    graph.add_edge("literature", "method_kb")

    if full:
        graph.add_node("planner", nodes["planner"])
        graph.add_node("analyst", nodes["analyst"])
        graph.add_node("reporter", nodes["reporter"])
        graph.add_edge("method_kb", "planner")
        if with_hitl:
            graph.add_node("hitl_plan", nodes["hitl_plan"])
            graph.add_edge("planner", "hitl_plan")
            graph.add_edge("hitl_plan", "analyst")
        else:
            graph.add_edge("planner", "analyst")
        graph.add_edge("analyst", "reporter")   # 线性：analyst 后只能进 reporter
        graph.add_edge("reporter", END)
        return graph.compile(
            checkpointer=get_checkpointer(),
            interrupt_before=["analyst"] if use_interrupt and with_hitl else None,
        )
    # ... 短流水线 report 分支

4.2 Specialist 节点：编排 vs 业务分离

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# agent/orchestrator/specialists.py
@dataclass
class SpecialistResult:
    summary: str        # 写入 PipelineState 的 *_summary 字段
    artifact_path: Path # 写入 PipelineState 的 *_artifact_path 字段
    stage: str          # 写入 PipelineState.stage（如 literature_done）

def run_literature_agent(symbol, *, run_id, workspace_root, collection, force_mock_llm):
    """加载 agent/literature/agent.yaml → AgentRuntime.chat → 写 literature_manifest artifact。"""
    ...

def run_literature_direct(symbol, *, run_id, workspace_root, collection):
    """无 LLM：ingest + query_hybrid + 写 manifest，供编排 CI 使用。"""
    ...

编排层 只选 agent 还是 direct；RAG 细节留在 literature / mcp_rag_gene。

4.3 调用与 checkpoint

1
2
3
4
5
6
7
8
9
10
11
# agent/orchestrator/cli.py run 命令（节选）
app = build_pipeline(cfg, full=full, with_hitl=hitl, use_interrupt=False)
config = {"configurable": {"thread_id": run_id}}
initial: PipelineState = {
    "run_id": run_id,
    "gene_symbol": symbol,
    "stage": "started",
    "approved": approve if hitl else True,
}
final = app.invoke(initial, config=config)
snapshot = app.get_state(config)  # 从 checkpoints.db 读回

流式场景用 app.stream(..., stream_mode="updates")，见 agent/orchestrator/streaming.py。

4.4 循环构建图（build_pipeline_loop）

在 保留线性 build_pipeline 不变 的前提下，单独提供带 回环 的 full 流水线：analyst 执行完后进入 hitl_review，人工审核 execution 结果；不满意则填写 revise_feedback，图从 literature 重新执行（跳过 sequence，保留已选转录本 artifact）。

1
2
3
4
5
6
7
8
9
10
sequence → literature → method_kb → planner → hitl_plan → analyst
                                                              ↓
                                                         hitl_review
                           ┌──────── revise_feedback ────────┘
                           ↓
                      literature → method_kb → planner → hitl_plan → analyst → …
                           │
              review_approved
                           ↓
                       reporter → END

4.4.1 循环专用 state 字段

1
2
3
4
5
6
7
# agent/orchestrator/state.py（循环字段）
class PipelineState(TypedDict, total=False):
    iteration: int           # 当前循环轮次，每次 revise +1
    max_iterations: int      # 硬上限（初始 state 建议 3），超限 route → END
    revise_feedback: str     # 人工调整方向；literature 节点拼入 query 后清空
    review_approved: bool    # True → hitl_review 后进入 reporter
    approved: bool           # 计划 HITL；revise 回 literature 时重置为 False，需重新批 plan

4.4.2 构图与条件路由

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
# agent/orchestrator/graph.py
def build_pipeline_loop(cfg, *, with_hitl=True, use_interrupt=False):
    nodes = make_nodes(cfg)
    graph = StateGraph(PipelineState)
    # … 注册 sequence / literature / method_kb / planner / analyst / reporter / hitl_review
    graph.add_edge("analyst", "hitl_review")
    graph.add_conditional_edges(
        "hitl_review",
        route_after_review,
        {"reporter": "reporter", "literature": "literature", "end": END},
    )
    # literature → method_kb → planner → [hitl_plan] → analyst 形成环
    return graph.compile(
        checkpointer=get_checkpointer(),
        interrupt_before=["analyst", "hitl_review"] if use_interrupt else None,
    )

def route_after_review(state) -> Literal["reporter", "literature", "end"]:
    if state.get("review_approved"):
        return "reporter"
    if int(state.get("iteration") or 0) >= int(state.get("max_iterations") or 3):
        return "end"
    if (state.get("revise_feedback") or "").strip():
        return "literature"
    return "end"

node_literature 读取 revise_feedback 拼入检索问题；消费后清空并设 approved=False，下一轮 planner 后需再次走 hitl_plan（见 Agent-10-08）。

node_hitl_review 使用 HITLGate.S5_EXECUTE，待办摘要为 analyst_summary，payload 为 execution_artifact_path。

4.4.3 interrupt + resume（两档人工闸门）

闸门 interrupt 点 人工操作 state 写入
计划审批 analyst 前 approve plan approved=True
执行审核 hitl_review 前 结果 OK / 需 revise review_approved=True 或 revise_feedback="…" 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 启动循环流水线（伪代码，同 Agent-10-08 streaming 模式）
app = build_pipeline_loop(cfg, with_hitl=True, use_interrupt=True)
initial: PipelineState = {
    "run_id": run_id,
    "gene_symbol": "BRCA1",
    "approved": False,
    "iteration": 0,
    "max_iterations": 3,
    "revise_feedback": "",
    "review_approved": False,
}
app.stream(initial, config)  # 可在 analyst 前、hitl_review 前各 interrupt 一次

# 计划批准后 resume → analyst 跑完 → 再次 interrupt 在 hitl_review 前
app.update_state(config, {"review_approved": False, "revise_feedback": "扩大检索 BRCA1 其它 isoform 文献"})
app.stream(Command(resume=True), config)  # → literature 重跑 → … → analyst

4.4.4 与线性版的选型

build_pipeline(full=True) build_pipeline_loop()
analyst 后 固定 → reporter 条件 → reporter / literature
适用 一次性交付、自动化 CI 科研试错、人工质控后改检索方向
sequence 每 run 一次 仅首轮；revise 从 literature 重入

4.5 工程验收

1
2
3
4
5
6
7
8
9
10
11
12
# 线性 mermaid（原有）
uv run python -m agent.orchestrator.cli graph --full --hitl

# 循环 mermaid（新增）
uv run python -m agent.orchestrator.cli graph --loop

# 线性 full 跑通
uv run python -m agent.orchestrator.cli run --symbol BRCA1 --full

# HITL interrupt + resume（线性 / 循环均适用，循环需 build_pipeline_loop + 上述 state）
uv run python -m agent.orchestrator.cli start-hitl --symbol BRCA1
uv run python -m agent.orchestrator.cli resume <run_id> --approve

产物路径：workspace/runs/<run_id>/ 下各 Specialist 的 manifest / plan / report。

5. 替代方案与优缺点

方案 优点 缺点
LangGraph + 本仓库 orchestrator checkpoint、HITL、Specialist 解耦 与 LangChain 生态绑定
OpenAI Agents handoff 轻量多 Agent 复杂回退、长流程弱
CrewAI 角色声明快 自定义 MCP 集成成本高
单 Agent 强 prompt 原型最快 规模上限低

6. 自检题

  1. 为什么 node_literature 不应直接写 BM25 逻辑？
  2. build_pipeline 与 build_pipeline_loop 各适合什么场景？
  3. revise 回 literature 时为什么要重置 approved=False？

7. 延伸阅读

-------------本文结束感谢您的阅读-------------