LangGraph状态机实战构建有思考链的多步骤Agent工作流 ⚙️导读前面我们学了推理范式的理论今天落地到代码用LangGraph构建一个有思考链的多步骤Agent工作流。从状态定义到节点编排从条件分支到循环控制一篇文章全部搞定一、LangGraph是什么LangGraph 用图Graph来编排Agent的工作流。概念类比说明State状态 共享白板所有节点都能读写的数据Node节点 工作站每个节点做一个具体的事Edge边️ 传送带连接节点决定下一步去哪Conditional Edge 岔路口根据条件走不同的路1.1 一个简单的LangGraph工作流是否START理解意图需要工具?调用工具整合结果END二、LangGraph核心概念详解 2.1 State状态—— 共享的白板fromtypingimportAnnotatedfromlanggraph.graphimportMessagesStatefromlanggraph.graph.messageimportadd_messagesfromlangchain_core.messagesimportAnyMessage# 定义Agent的状态classAgentState(MessagesState):Agent的状态包含所有需要共享的数据messages:Annotated[list[AnyMessage],add_messages]# 对话消息# 可以添加更多状态字段# current_task: str # plan: list []# results: dict {}2.2 Node节点—— 工作站点fromlangchain_openaiimportChatOpenAI# 每个节点就是一个函数defunderstand_intent(state:AgentState)-dict:节点1理解用户意图llmChatOpenAI(modelgpt-4o)responsellm.invoke(state[messages])return{messages:[response]}defuse_tool(state:AgentState)-dict:节点2调用工具# 执行工具调用逻辑tool_result工具执行结果...return{messages:[(tool,tool_result)]}defgenerate_response(state:AgentState)-dict:节点3生成最终回答llmChatOpenAI(modelgpt-4o)responsellm.invoke(state[messages])return{messages:[response]}2.3 Graph图—— 编排工作流fromlanggraph.graphimportStateGraph,START,END# 创建图workflowStateGraph(AgentState)# 添加节点workflow.add_node(understand,understand_intent)workflow.add_node(tool_call,use_tool)workflow.add_node(respond,generate_response)# 添加边连接节点workflow.add_edge(START,understand)# 入口 → 理解意图workflow.add_conditional_edges(# 理解意图 → 条件分支understand,should_use_tool,# 条件判断函数{use_tool:tool_call,direct:respond})workflow.add_edge(tool_call,respond)# 工具调用 → 生成回答workflow.add_edge(respond,END)# 生成回答 → 结束# 编译agentworkflow.compile()三、实战构建多步骤分析Agent 3.1 目标构建一个数据分析Agent工作流如下否需要更多数据是否是START 理解需求 制定计划 收集数据 分析数据分析完成? 生成报告✅ 质量检查质量合格?END3.2 完整代码importjsonfromtypingimportAnnotated,Literalfromlangchain_openaiimportChatOpenAIfromlangchain.toolsimporttoolfromlanggraph.graphimportStateGraph,START,END,MessagesStatefromlanggraph.graph.messageimportadd_messagesfromlangchain_core.messagesimportAnyMessage,SystemMessage,HumanMessage# 定义状态 classAnalysisState(MessagesState):messages:Annotated[list[AnyMessage],add_messages]plan:str# 分析计划data:str# 收集到的数据analysis_result:str# 分析结果report:str# 最终报告quality_passed:bool# 质量检查是否通过# 定义工具 tooldefweb_search(query:str)-str:搜索互联网获取数据。 Args: query: 搜索关键词 returnf搜索结果关于{query}的数据...tooldefexecute_analysis(code:str)-str:执行数据分析代码。 Args: code: Python分析代码 returnf分析结果数据处理完成...# 定义节点 llmChatOpenAI(modelgpt-4o,temperature0)defunderstand_node(state:AnalysisState)-dict:理解用户需求responsellm.invoke([SystemMessage(content你是一个数据分析专家。请理解用户的分析需求用一句话总结核心目标。),*state[messages]])return{messages:[response]}defplan_node(state:AnalysisState)-dict:制定分析计划responsellm.invoke([SystemMessage(content根据用户需求制定详细的分析计划。列出需要执行的步骤。),*state[messages]])return{messages:[response],plan:response.content}defcollect_data_node(state:AnalysisState)-dict:收集数据responsellm.invoke([SystemMessage(content根据分析计划使用搜索工具收集所需数据。),*state[messages]])return{messages:[response],data:response.content}defanalyze_node(state:AnalysisState)-dict:分析数据responsellm.invoke([SystemMessage(content对收集到的数据进行深入分析提取关键洞察。),*state[messages]])return{messages:[response],analysis_result:response.content}defreport_node(state:AnalysisState)-dict:生成报告responsellm.invoke([SystemMessage(content基于分析结果生成一份结构化的分析报告。使用Markdown格式。),*state[messages]])return{messages:[response],report:response.content}defquality_check_node(state:AnalysisState)-dict:质量检查responsellm.invoke([SystemMessage(content检查报告质量是否完整数据是否准确结论是否有支撑回答PASS或FAIL。),*state[messages]])passedPASSinresponse.content.upper()return{messages:[response],quality_passed:passed}# 定义条件函数 defshould_continue(state:AnalysisState)-Literal[analyze,collect_data]:是否需要收集更多数据# 简单判断第一次直接去分析returnanalyzedefquality_gate(state:AnalysisState)-Literal[end,revise]:质量关卡ifstate.get(quality_passed,False):returnendreturnrevise# 构建工作流 workflowStateGraph(AnalysisState)# 添加节点workflow.add_node(understand,understand_node)workflow.add_node(plan,plan_node)workflow.add_node(collect_data,collect_data_node)workflow.add_node(analyze,analyze_node)workflow.add_node(report,report_node)workflow.add_node(quality_check,quality_check_node)# 添加边workflow.add_edge(START,understand)workflow.add_edge(understand,plan)workflow.add_edge(plan,collect_data)workflow.add_conditional_edges(collect_data,should_continue)workflow.add_edge(analyze,report)workflow.add_edge(report,quality_check)workflow.add_conditional_edges(quality_check,quality_gate,{end:END,revise:report# 质量不合格重新生成报告})# 编译analysis_agentworkflow.compile()# 运行 resultanalysis_agent.invoke({messages:[HumanMessage(content分析2025年AI Agent市场的发展趋势)],})print(result[report])四、LangGraph高级模式 4.1 子图Subgraph—— 模块化工作流# 创建一个子图数据处理子流程data_workflowStateGraph(AnalysisState)data_workflow.add_node(clean,clean_data)data_workflow.add_node(transform,transform_data)data_workflow.add_node(validate,validate_data)data_workflow.add_edge(START,clean)data_workflow.add_edge(clean,transform)data_workflow.add_edge(transform,validate)data_workflow.add_edge(validate,END)# 将子图作为主图的一个节点main_workflow.add_node(data_processing,data_workflow.compile())4.2 人机协作Human-in-the-Loopfromlanggraph.checkpoint.memoryimportMemorySaver# 在关键节点暂停等待人工确认workflow.add_node(human_review,human_review_node)# 设置中断点appworkflow.compile(checkpointerMemorySaver(),interrupt_before[human_review]# 在人工审核前暂停)# 运行到中断点config{configurable:{thread_id:session-1}}resultapp.invoke({messages:[...]},config)# 人工审核后继续app.invoke(None,config)# 传入None表示继续执行4.3 持久化状态# 使用SQLite持久化状态fromlanggraph.checkpoint.sqliteimportSqliteSaver checkpointerSqliteSaver.from_conn_string(agent_state.db)appworkflow.compile(checkpointercheckpointer)# 状态会自动持久化重启后恢复五、LangGraph vs 其他框架 特性LangGraphAutoGenCrewAI编排方式状态机图对话流角色扮演控制精度⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐学习曲线⭐⭐⭐较陡⭐⭐⭐⭐⭐⭐⭐⭐⭐灵活性⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐可视化⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐生产就绪⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐LangGraph的优势精确控制每一步的执行流程适合需要复杂逻辑的生产级应用。六、本期小结 知识点核心内容LangGraph是什么用图状态机编排Agent工作流的框架核心概念State状态、Node节点、Edge边条件分支add_conditional_edges根据条件走不同路径循环控制边可以指回之前的节点形成循环高级模式子图、人机协作、状态持久化LangGraph是构建生产级Agent的首选框架。它让你像画流程图一样设计Agent的工作逻辑——清晰、可控、可调试。 下期预告记忆与规划篇完结下一篇进入多Agent协作篇——《Multi-Agent架构全景图主从式、辩论式、流水线式哪种最适合你》三连走起LangGraphAgent工作流的终极方案专栏第16/24期记忆与规划篇完结作者高炉炼铁智能化技术研究者专注钢铁冶金与人工智能 交叉领域。 如果觉得有帮助请点赞、收藏、转发版权归作者所有未经许可请勿抄袭套用商用(或其它具有利益性行为)。 关注专栏不错过后续精彩内容