AI Agent 从入门到封神：24 讲打造你的超级智能体~系列文章16：LangGraph状态机实战：构建有“思考链“的多步骤Agent工作流-北京尧图网络

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期记忆与规划篇完结作者高炉炼铁智能化技术研究者专注钢铁冶金与人工智能交叉领域。如果觉得有帮助请点赞、收藏、转发版权归作者所有未经许可请勿抄袭套用商用(或其它具有利益性行为)。关注专栏不错过后续精彩内容

AI Agent 从入门到封神：24 讲打造你的超级智能体~系列文章16：LangGraph状态机实战：构建有“思考链“的多步骤Agent工作流

相关推荐

实时通信架构选型：WebSocket、Server-Sent Events 与长轮询的工程权衡

C++队列数据结构实现：从数组循环队列到链表动态队列的底层原理与实战

把 Ace Data Cloud 平台能力接入你的 AI 助手：MCP 文档、价格、API 与账户管理实战

更多资讯

Linux下USB转串口设备找不到？从驱动到权限的完整排查指南

基于COMSOL的油浸式变压器多物理场流固耦合温升仿真建模

基于线性MPC的弹簧阻尼系统轨迹跟踪控制：从建模到代码实现

营销能力的本质是用户感知力：从信任构建到行为洞察

AI助手选型指南：场景化分工与评估框架实战

Cursor C++调试实战：用cppvsdbg玩转Windows进程附加

今日更新

在S3C2410上移植Nucleus RTOS：启动、驱动与实时性调优实战

2026年Turnitin降重降AI二合一红黑榜：能过和别碰

2026期刊投稿AI写作工具实测：6款打分谁更值得用

本周精选

超人会飞不算本事：系统稳定依赖清晰规则与边界设计

超人VS蜘蛛侠：拆解超级IP的影响力与传播方法论

基于CNN的调制信号识别：MATLAB实现时频图分类实战

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？