文章目录场景代码这份代码看上去平平无奇就是一个节点分出两条线再由两个节点汇合到一个节点。实际上langgraph在里面起到了作用。1、参数的兼容问题只覆盖返回的。2、自带并行机制即如果一个节点没有返回汇合节点会等另外一个节点。所以我们用起来觉得很简单。场景模拟langgraph的并行功能文件名叫做develop(意思是开发)。有两个任务backend(后端)front(前段)都不需要实际的代码打印一行日志正在开发后端正在开发前段并延时1-5秒然后汇合到一个节点开发完成。代码importtimeimportrandomimportloggingfromtypingimportAnnotated,Listfromtyping_extensionsimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDimportoperator# 配置日志logging.basicConfig(levellogging.INFO,format%(asctime)s - %(levelname)s - %(message)s)loggerlogging.getLogger(__name__)# 1. 定义共享状态# 使用 Annotated 和 operator.add 确保并行节点的返回结果能安全地累加到列表中classDevelopState(TypedDict):logs:Annotated[List[str],operator.add]backend_duration:floatfrontend_duration:float# 2. 定义后端开发节点defbackend_dev(state:DevelopState):logger.info( [后端] 正在开发后端...)# 随机延时 1-5 秒模拟开发耗时durationrandom.uniform(1,5)time.sleep(duration)logger.info(f✅ [后端] 后端开发完成 (耗时{duration:.2f}秒))return{logs:[[后端] 开发完成],backend_duration:duration}# 3. 定义前端开发节点deffront_dev(state:DevelopState):logger.info( [前端] 正在开发前端...)# 随机延时 1-5 秒模拟开发耗时durationrandom.uniform(1,5)time.sleep(duration)logger.info(f✅ [前端] 前端开发完成 (耗时{duration:.2f}秒))return{logs:[[前端] 开发完成],frontend_duration:duration}# 4. 定义汇合节点项目整体完成defproject_done(state:DevelopState):total_timemax(state.get(backend_duration,0),state.get(frontend_duration,0))logger.info(f 前后端均已就绪项目开发完成(并行阶段总耗时约{total_time:.2f}秒))return{logs:[项目整体完成]}# 5. 构建 LangGraph 流程图defcreate_develop_graph():builderStateGraph(DevelopState)# 添加节点builder.add_node(backend,backend_dev)builder.add_node(front,front_dev)builder.add_node(done,project_done)# 添加边实现 Fan-out (并行) 与 Fan-in (汇合)# 从起点同时引出两条边 - 触发并行执行builder.add_edge(START,backend)builder.add_edge(START,front)# 两个任务都完成后同时指向 done 节点 - 触发汇合等待builder.add_edge(backend,done)builder.add_edge(front,done)builder.add_edge(done,END)returnbuilder.compile()# 创建工作流实例dev_graphcreate_develop_graph()if__name____main__:print(\n 开始启动项目开发流程 (LangGraph 并行演示) \n)start_timetime.time()# 执行工作流final_statedev_graph.invoke({logs:[]})end_timetime.time()total_durationend_time-start_timeprint(\n--------------------------------------------------)print(f 最终状态日志:{final_state[logs]})print(f⏱️ 整个开发流程总耗时:{total_duration:.2f}秒)print( 提示如果总耗时接近后端和前端中较慢的那个时间说明并行成功啦)print(--------------------------------------------------\n)