【LangGraph】LangGraph 协调者-工作者模式完全解析：从零构建一个智能报告生成系统

LangGraph 协调者-工作者模式完全解析：从零构建一个智能报告生成系统

一、这个系统是做什么的？

一句话：输入一个主题，自动生成一份完整的报告。

比如你输入“中国近代史”，系统会自动：

1. 分析主题，决定报告分几个章节
2. 为每个章节分配一个写作任务
3. 多个“员工”同时写作不同章节
4. 把所有章节拼成完整报告

二、整体流程图

用户输入: {"topic": "中国近代史"} │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ 节点1: orchestrator（协调者/领导） │ │ │ │ 作用：分析主题，制定章节计划 │ │ 输入：{"topic": "中国近代史"} │ │ 输出：{"sections": [章1, 章2, 章3]} │ │ │ │ 内部逻辑：调用 LLM，强制输出固定格式的章节列表 │ └─────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ 函数: assign_workers（分配任务）(边) │ │ │ │ 作用：为每个章节创建一个 Send 对象 │ │ 输入：{"sections": [章1, 章2, 章3]} │ │ 输出：[Send("llm_call", {"section": 章1}), │ │ Send("llm_call", {"section": 章2}), │ │ Send("llm_call", {"section": 章3})] │ └─────────────────────────────────────────────────────────────┘ │ │ 动态创建 3 条边 │ ┌───────────┼───────────┐ │ │ │ ▼ ▼ ▼ ┌────────┐ ┌────────┐ ┌────────┐ │ 节点2 │ │ 节点2 │ │ 节点2 │ │llm_call│ │llm_call│ │llm_call│ │ (员工) │ │ (员工) │ │ (员工) │ │ │ │ │ │ │ │ 写章1 │ │ 写章2 │ │ 写章3 │ └───┬────┘ └───┬────┘ └───┬────┘ │ │ │ └──────────┼──────────┘ ▼ ┌─────────────────────────────────────────────────────────────┐ │ 节点3: synthesizer（合成器/文员） │ │ │ │ 作用：把所有章节拼成完整报告 │ │ 输入：{"completed_sections": ["章1内容", "章2内容", ...]} │ │ 输出：{"final_report": "章1内容\n---\n章2内容\n---\n..."} │ └─────────────────────────────────────────────────────────────┘ │ ▼ 输出报告

三、核心概念：协调者-工作者模式

3.1 和普通并行化的区别

3.2 生活例子

关键：领导可能决定写3章，也可能写5章。员工数量取决于领导的决定。

四、状态定义（State）

classState(TypedDict):topic:str# 用户输入的主题sections:list# 协调者生成的章节列表completed_sections:Annotated[List,operator.add]# 员工写好的内容final_report:str# 最终报告

4.1 字段说明

4.2 重要：operator.add的作用

completed_sections:Annotated[List,operator.add]

如果不加这个，多个员工同时交稿，只有最后一个会被保留。

五、节点详解

5.1 节点1：orchestrator（协调者）

deforchestrator(state:State):response=planner.invoke(f"为主题'{state['topic']}'制定报告大纲，包含3个章节")return{"sections":response.sections}

5.2 分配函数：assign_workers（关键！）（边）

defassign_workers(state:State):worker_tasks=[]forsectioninstate["sections"]:worker_tasks.append(Send("llm_call",{"section":section}))returnworker_tasks

重点：Send是什么？

Send("llm_call",{"section":section})

Send的作用：动态创建一条边，并携带数据给目标节点。
注意：

• llm_call收到的state只有Send传递的{"section": ...}。

5.3 节点2：llm_call（工作者）

defllm_call(state:dict):section=state["section"]result=model.invoke(f"编写报告章节:{section.name}, 内容要求:{section.description}")return{"completed_sections":[result.content]}

5.4 节点3：synthesizer（合成器）

defsynthesizer(state:State):completed_sections=state["completed_sections"]final_report="\n\n---\n\n".join(completed_sections)return{"final_report":final_report}

"\n\n---\n\n".join(completed_sections)：把列表里的多个字符串，用 \n\n—\n\n 作为分隔符，拼接成一个字符串。

六、图的构建（核心难点）

6.1 完整代码

builder=StateGraph(State)# 添加节点builder.add_node("orchestrator",orchestrator)builder.add_node("llm_call",llm_call)builder.add_node("synthesizer",synthesizer)# 边1：开始 → 协调者builder.add_edge(START,"orchestrator")# 边2：条件边（关键！）builder.add_conditional_edges("orchestrator",assign_workers,["llm_call"]# 用于类型检查，声明可能去的节点。实际去多少次、每次带什么数据，由 `assign_workers` 返回的 `Send` 列表决定。)# 边3：工作者 → 合成器builder.add_edge("llm_call","synthesizer")builder.add_edge("synthesizer",END)

6.2 关键：add_conditional_edges的两种用法

# 普通用法defroute(state):ifstate["type"]=="A":return"node_A"else:return"node_B"builder.add_conditional_edges("node",route,["node_A","node_B"])# 本代码的用法（动态创建多个）defassign_workers(state):return[Send("llm_call",{"section":s})forsinstate["sections"]]builder.add_conditional_edges("orchestrator",assign_workers,["llm_call"])

七、执行过程详解

第1步：用户输入

result=workflow.invoke({"topic":"中国近代史"})

第2步：orchestrator执行

# 输入state={"topic":"中国近代史"}# 调用 LLM，生成3个章节response=planner.invoke("为主题'中国近代史'制定报告大纲，包含3个章节")# response 是 Sections 对象# response.sections = [章1, 章2, 章3]# 返回return{"sections":response.sections}

第3步：assign_workers执行

# 输入state={"topic":"中国近代史","sections":[章1,章2,章3]}# 为每个章节创建 Sendreturn[Send("llm_call",{"section":章1}),Send("llm_call",{"section":章2}),Send("llm_call",{"section":章3})]

第4步：llm_call执行（3次并行）

# 第一次调用state={"section":章1}# 写第一章内容return{"completed_sections":["第一章内容..."]}# 第二次调用state={"section":章2}return{"completed_sections":["第二章内容..."]}# 第三次调用state={"section":章3}return{"completed_sections":["第三章内容..."]}

由于completed_sections使用了operator.add，三个结果自动合并成：

{"completed_sections":["第一章内容...","第二章内容...","第三章内容..."]}

第5步：synthesizer执行

# 输入state={"completed_sections":["第一章内容...","第二章内容...","第三章内容..."]}# 拼接final_report="第一章内容...\n\n---\n\n第二章内容...\n\n---\n\n第三章内容..."# 返回return{"final_report":final_report}

第6步：输出

print(result["final_report"])# 输出完整的报告

八、完整可运行代码

fromtypingimportAnnotated,List,TypedDictimportoperatorfromlangchain.chat_modelsimportinit_chat_modelfromlanggraph.constantsimportSTART,ENDfromlanggraph.graphimportStateGraphfromlanggraph.typesimportSendfrompydanticimportBaseModel# ============================================================# 1. 定义状态# ============================================================classState(TypedDict):topic:strsections:listcompleted_sections:Annotated[List,operator.add]final_report:str# ============================================================# 2. 定义结构化输出# ============================================================classSection(BaseModel):name:strdescription:strclassSections(BaseModel):sections:List[Section]# ============================================================# 3. 创建模型# ============================================================model=init_chat_model("gpt-4o-mini")# 普通模型，用于生成内容planner=model.with_structured_output(Sections)# 结构化模型，用于生成大纲# ============================================================# 4. 协调者节点# ============================================================deforchestrator(state:State):response=planner.invoke(f"为主题'{state['topic']}'制定报告大纲，包含3个章节")return{"sections":response.sections}# ============================================================# 5. 工作者节点# ============================================================defllm_call(state:dict):section=state["section"]result=model.invoke(f"编写报告章节:{section.name}, 内容要求:{section.description}")return{"completed_sections":[result.content]}# ============================================================# 6. 合成器节点# ============================================================defsynthesizer(state:State):completed_sections=state["completed_sections"]final_report="\n\n---\n\n".join(completed_sections)return{"final_report":final_report}# ============================================================# 7. 任务分配函数# ============================================================defassign_workers(state:State):worker_tasks=[]forsectioninstate["sections"]:worker_tasks.append(Send("llm_call",{"section":section}))returnworker_tasks# ============================================================# 8. 构建图# ============================================================builder=StateGraph(State)builder.add_node("orchestrator",orchestrator)builder.add_node("llm_call",llm_call)builder.add_node("synthesizer",synthesizer)builder.add_edge(START,"orchestrator")builder.add_conditional_edges("orchestrator",assign_workers,["llm_call"])builder.add_edge("llm_call","synthesizer")builder.add_edge("synthesizer",END)workflow=builder.compile()# ============================================================# 9. 运行# ============================================================if__name__=="__main__":result=workflow.invoke({"topic":"中国近代史"})print("\n"+"="*50)print("最终报告：")print("="*50)print(result["final_report"])

九、总结

一句话总结：协调者动态决定任务数量，Send动态创建边并携带数据，工作者并行执行，结果自动合并，最终合成完整报告。