1. 从单体Agent到图式编排为什么我们需要LangGraph如果你在过去一年里尝试过构建基于大语言模型的智能应用大概率会经历过这样的场景你写了一个功能强大的Agent它能调用工具、能联网搜索、能写代码看起来无所不能。但当你试图让它处理一个稍微复杂点的业务流程时比如一个需要多轮审批的报销申请或者一个涉及客户咨询、方案匹配、报价生成的销售流程事情就开始变得棘手了。你的代码里很快会塞满各种if-else、while循环和状态标志位整个逻辑变得像一团乱麻难以维护和调试。更糟糕的是一旦流程需要长时间运行比如等待人工审批如何保存和恢复Agent的“记忆”和“状态”就成了一个令人头疼的问题。这正是LangGraph要解决的核心痛点。它不是一个用来替代LangChain的新框架而是LangChain生态中一个专门用于构建有状态、多步骤、可长时间运行的智能体工作流的库。你可以把它想象成给AI智能体装上了“流程图”和“记忆中枢”。传统的单体Agent像是一个聪明的、但只能处理单次问答的专家而基于LangGraph构建的Agentic AI系统则更像一个配备了明确SOP标准作业程序和持久化工作记忆的自动化团队能够协同处理复杂的、有状态的业务流程。简单来说LangGraph将你的业务逻辑从线性的、脆弱的代码控制流升级为可视化的、健壮的图Graph结构。图中的节点Node代表一个具体的处理步骤比如“解析用户意图”、“调用数据库查询”、“生成报告草稿”而边Edge则定义了步骤之间的流转条件比如“如果查询成功则进入下一步如果失败则跳转到错误处理节点”。这种范式转变带来了几个关键优势流程可视化与可维护性、复杂条件分支的优雅处理、状态的持久化与恢复以及对长时间运行任务的天然支持。接下来我们将深入拆解如何利用LangGraph的这些特性来设计和实现那些过去难以驾驭的长周期、有状态业务流程。2. 核心概念拆解图、状态与工作流到底是什么在深入实战之前我们必须先统一语言理解LangGraph中几个最核心的抽象。这些概念是构建一切复杂工作流的基础。2.1 图Graph业务流程的可视化骨架在LangGraph中“图”不是一个存储数据的结构如知识图谱而是一个计算图或工作流图。它定义了整个智能体系统的执行蓝图。节点Node这是图的基本执行单元。一个节点通常是一个函数在Python中就是一个callable它接收当前的系统状态执行一些操作比如调用LLM、运行工具、处理数据然后返回更新后的状态。例如你可以有一个名为validate_input的节点来校验用户输入一个名为search_knowledge_base的节点来查询内部知识库。边Edge边决定了执行流程的方向。在LangGraph中边通常与条件判断相关联。这通过“条件边”来实现。例如从validate_input节点可能引出两条边一条指向process_request节点条件为“输入有效”另一条指向ask_for_clarification节点条件为“输入无效”。这种设计使得基于动态结果的流程路由变得非常直观。这种图结构的最大好处是显式化。传统的代码中业务流程隐藏在层层嵌套的条件和循环语句里而在LangGraph中整个流程被摊开成一张图任何人都能一目了然地看到业务的全貌和所有可能的分支路径。这对于团队协作、代码审查和后期维护至关重要。2.2 状态State工作流的长期记忆与上下文这是LangGraph处理“有状态”流程的核心。状态是一个字典-like的对象它在整个图执行过程中被传递和修改。你可以把它理解为工作流的“记忆体”或“共享白板”。状态模式State Schema在定义图之前你需要先定义一个状态模式。这类似于为你的工作流声明一个强类型的数据结构。使用TypedDict或Pydantic模型来定义状态中包含哪些字段以及它们的类型。例如一个客户服务流程的状态可能包含user_query: str用户问题、conversation_history: List[dict]对话历史、retrieved_docs: List[str]检索到的文档、current_step: str当前步骤标识、final_answer: Optional[str]最终答案。状态的流转每个节点函数都接收一个状态对象作为参数并返回一个包含更新字段的字典。LangGraph会自动将这个返回的字典与原有状态进行合并merge。例如search_knowledge_base节点可能接收包含user_query的状态然后返回{retrieved_docs: [doc1内容, doc2内容]}。合并后状态中就新增了retrieved_docs字段。持久化Persistence这是实现“长时间运行”的关键。LangGraph提供了开箱即用的持久化接口可以将图的状态完整地保存到数据库如SQLite、PostgreSQL或内存中。这意味着一个处理到一半的流程比如等待人工审核其全部上下文状态都可以被序列化存储。当需要恢复时如审核通过只需从存储中加载该状态并重新执行图它就能从上次中断的地方继续运行仿佛从未停止过。这解决了传统Agent在服务重启或长时间等待后上下文丢失的根本问题。2.3 工作流Workflow与智能体Agent的融合在LangGraph的语境下我们构建的“图”本身就是一个高级的、可编排的智能体工作流。它与传统单体Agent的关系是互补而非替代传统LangChain Agent通常是一个AgentExecutor它内部有一个循环LLM思考 - 决定调用工具 - 执行工具 - 将结果返回给LLM继续思考。这个循环是隐式的、内聚的适合解决目标明确、步骤线性的单任务。LangGraph Workflow它将这个“思考-行动”循环以及更多其他步骤如输入预处理、结果后处理、分支判断显式地分解为图中的多个节点。一个节点内部可以封装一个完整的LangChain Agent另一个节点可能只是一个纯数据校验函数。这样你就拥有了更高的控制粒度和流程透明度。你可以基于业务需求自由设计图的复杂度。一个简单的线性审批流可能只有3-4个节点而一个复杂的对公信贷尽职调查系统可能包含数十个节点涉及多个专业Agent如财务分析Agent、法律合规Agent、风险评估Agent的协同以及大量的人工审核和条件跳转节点。3. 实战构建一个订单处理与客户跟进工作流理论说得再多不如动手构建一个。假设我们有一个电商场景的订单处理流程它不是一个简单的下单即结束的动作而是一个包含状态跟踪、异常处理和主动跟进的长周期业务流程。我们将用LangGraph来实现它。3.1 第一步定义状态模式与工具首先明确我们的工作流需要记住什么。我们使用Pydantic来定义状态这样能获得良好的类型提示和校验。from typing import TypedDict, List, Optional, Annotated from langgraph.graph import StateGraph, END from pydantic import BaseModel, Field import operator # 1. 定义状态模式 class OrderState(TypedDict): # 输入与核心数据 order_id: str customer_id: str items: List[dict] # 商品列表 total_amount: float # 流程状态与决策 current_status: str # e.g., received, payment_verified, inventory_checked, shipped, delivered, cancelled payment_verified: bool inventory_available: bool shipping_address_valid: bool # 沟通与日志 customer_messages: List[str] # 发送给客户的消息记录 internal_notes: List[str] # 内部处理日志 # 异常与等待 pending_action: Optional[str] # e.g., wait_for_payment, wait_for_inventory, need_address_confirmation error_reason: Optional[str]接下来定义一些工具函数它们将被封装在节点中。在真实场景中这些工具可能会调用外部API支付网关、库存系统、物流接口。# 2. 定义工具函数模拟 def verify_payment(order_id: str) - bool: 模拟支付验证。真实情况应调用支付网关API。 print(f[工具] 正在验证订单 {order_id} 的支付状态...) # 模拟逻辑假设90%的支付是成功的 import random is_success random.random() 0.1 return is_success def check_inventory(items: List[dict]) - bool: 模拟库存检查。 print(f[工具] 正在检查商品库存...) # 模拟逻辑检查每个商品的库存是否大于0 # 这里简化处理假设库存充足 return True def validate_shipping_address(customer_id: str) - bool: 模拟地址验证。 print(f[工具] 正在验证客户 {customer_id} 的收货地址...) # 模拟逻辑假设地址有效 return True def notify_customer(customer_id: str, message: str) - str: 模拟发送通知给客户邮件、短信等。 print(f[通知客户 {customer_id}]: {message}) return f通知已发送: {message} def update_order_status_in_db(order_id: str, status: str, note: str): 模拟更新订单主状态到数据库。 print(f[数据库] 更新订单 {order_id} 状态为: {status}, 备注: {note})3.2 第二步构建节点与图现在我们将业务流程的每个步骤定义为一个节点函数。每个函数接收并返回状态的一部分。# 3. 定义节点函数 def node_receive_order(state: OrderState) - dict: 节点接收订单初始化状态。 print(f\n--- 开始处理订单 {state[order_id]} ---) state[current_status] received state[internal_notes].append(f订单已接收金额 {state[total_amount]}) return {current_status: state[current_status], internal_notes: state[internal_notes]} def node_verify_payment(state: OrderState) - dict: 节点验证支付。 is_verified verify_payment(state[order_id]) state[payment_verified] is_verified if is_verified: state[current_status] payment_verified state[internal_notes].append(支付验证成功。) # 决定下一个节点去检查库存 state[pending_action] None else: state[current_status] payment_failed state[error_reason] 支付授权失败 state[internal_notes].append(支付验证失败。) state[pending_action] wait_for_payment # 发送通知给客户 msg f您的订单 {state[order_id]} 支付未成功请检查支付方式。 notify_customer(state[customer_id], msg) state[customer_messages].append(msg) return state def node_check_inventory(state: OrderState) - dict: 节点检查库存。 is_available check_inventory(state[items]) state[inventory_available] is_available if is_available: state[current_status] inventory_checked state[internal_notes].append(库存检查通过。) state[pending_action] None else: state[current_status] inventory_out_of_stock state[error_reason] 部分商品缺货 state[internal_notes].append(库存不足。) state[pending_action] wait_for_inventory msg f您的订单 {state[order_id]} 中部分商品暂时缺货我们将为您补货后尽快发出。 notify_customer(state[customer_id], msg) state[customer_messages].append(msg) return state def node_validate_address(state: OrderState) - dict: 节点验证收货地址。 is_valid validate_shipping_address(state[customer_id]) state[shipping_address_valid] is_valid if is_valid: state[current_status] address_validated state[internal_notes].append(收货地址验证通过。) state[pending_action] None else: state[current_status] address_invalid state[error_reason] 收货地址不完整或无法配送 state[internal_notes].append(收货地址无效。) state[pending_action] need_address_confirmation msg f您的订单 {state[order_id]} 收货地址信息有误请登录账户确认或联系客服。 notify_customer(state[customer_id], msg) state[customer_messages].append(msg) return state def node_prepare_for_shipping(state: OrderState) - dict: 节点准备发货最终处理节点。 if all([state[payment_verified], state[inventory_available], state[shipping_address_valid]]): state[current_status] shipped state[internal_notes].append(订单已打包等待物流取件。) msg f您的订单 {state[order_id]} 已发货 notify_customer(state[customer_id], msg) state[customer_messages].append(msg) update_order_status_in_db(state[order_id], shipped, 已发货) else: # 理论上不应该走到这里因为前面有条件边控制 state[internal_notes].append(错误尝试发货但前置条件未满足。) return state def node_handle_pending(state: OrderState) - dict: 节点处理挂起状态模拟等待后重新检查。 action state.get(pending_action) note f检测到挂起动作: {action}。执行恢复逻辑... state[internal_notes].append(note) print(f[处理挂起] {note}) # 简化处理清除挂起状态并视情况决定下一步 # 真实场景可能查询支付状态是否更新、库存是否到货、客户是否回复 state[pending_action] None state[error_reason] None # 这里我们简单地将状态重置为‘received’让流程重新开始验证支付。 # 更复杂的图可以设计循环回到特定节点。 state[current_status] received return state有了节点我们开始构建图并定义节点之间的流转逻辑边。# 4. 构建图 builder StateGraph(OrderState) # 添加节点 builder.add_node(receive_order, node_receive_order) builder.add_node(verify_payment, node_verify_payment) builder.add_node(check_inventory, node_check_inventory) builder.add_node(validate_address, node_validate_address) builder.add_node(prepare_for_shipping, node_prepare_for_shipping) builder.add_node(handle_pending, node_handle_pending) # 设置入口点 builder.set_entry_point(receive_order) # 添加边定义流程 builder.add_edge(receive_order, verify_payment) # 从 verify_payment 出发的条件边 def route_after_payment(state: OrderState) - str: if state.get(payment_verified): return proceed_to_inventory else: return handle_pending # 支付失败进入挂起处理 builder.add_conditional_edges( verify_payment, route_after_payment, { proceed_to_inventory: check_inventory, handle_pending: handle_pending, } ) # 从 check_inventory 出发的条件边 def route_after_inventory(state: OrderState) - str: if state.get(inventory_available): return proceed_to_address else: return handle_pending # 库存不足进入挂起处理 builder.add_conditional_edges( check_inventory, route_after_inventory, { proceed_to_address: validate_address, handle_pending: handle_pending, } ) # 从 validate_address 出发的条件边 def route_after_address(state: OrderState) - str: if state.get(shipping_address_valid): return proceed_to_shipping else: return handle_pending # 地址无效进入挂起处理 builder.add_conditional_edges( validate_address, route_after_address, { proceed_to_shipping: prepare_for_shipping, handle_pending: handle_pending, } ) # 从 handle_pending 出发我们让它回到 verify_payment 重新尝试简化逻辑 builder.add_edge(handle_pending, verify_payment) # 从 prepare_for_shipping 到结束 builder.add_edge(prepare_for_shipping, END) # 编译图 graph builder.compile()3.3 第三步执行与持久化演示现在我们可以运行这个工作流并演示持久化如何让一个“长时间运行”的流程得以暂停和恢复。# 5. 执行工作流第一次运行 print( 第一次执行正常流程 ) initial_state { order_id: ORD-001, customer_id: CUST-1001, items: [{name: 商品A, qty: 2}], total_amount: 199.99, current_status: , payment_verified: False, inventory_available: False, shipping_address_valid: False, customer_messages: [], internal_notes: [], pending_action: None, error_reason: None, } # 模拟一个成功的执行 final_state graph.invoke(initial_state) print(f\n最终状态: {final_state[current_status]}) print(f内部笔记: {final_state[internal_notes]}) print(f客户消息: {final_state[customer_messages]}) # 6. 演示持久化模拟支付失败场景 print(\n\n 第二次执行模拟支付失败与恢复 ) # 为了演示我们手动设置一个会触发支付失败的状态 state_with_pending { order_id: ORD-002, customer_id: CUST-1002, items: [{name: 商品B, qty: 1}], total_amount: 89.99, current_status: received, payment_verified: False, # 初始化为False但我们需要模拟工具返回False inventory_available: False, shipping_address_valid: False, customer_messages: [], internal_notes: [], pending_action: None, error_reason: None, } # 为了控制工具行为我们这里临时“作弊”让 verify_payment 工具固定返回False import unittest.mock with unittest.mock.patch(__main__.verify_payment, return_valueFalse): # 第一次执行会停在 handle_pending intermediate_state graph.invoke(state_with_pending) print(f\n第一次执行后状态: {intermediate_state[current_status]}) print(f挂起动作: {intermediate_state[pending_action]}) print(f内部笔记: {intermediate_state[internal_notes]}) # 假设此时我们将状态持久化到数据库 # persisted_state_id persistence_layer.save(intermediate_state) print([模拟] 状态已持久化到数据库。) # 模拟一段时间后如客户重新支付我们从数据库加载状态 # loaded_state persistence_layer.load(persisted_state_id) loaded_state intermediate_state # 这里用中间状态代替加载 # 并且我们假设支付问题已解决所以让工具返回True with unittest.mock.patch(__main__.verify_payment, return_valueTrue): print(\n[模拟] 客户已完成支付从持久化状态恢复工作流...) # 从加载的状态继续执行 final_state_restored graph.invoke(loaded_state, config{recursion_limit: 50}) print(f恢复执行后最终状态: {final_state_restored[current_status]}) print(f内部笔记: {final_state_restored[internal_notes]})通过这个例子你可以清晰地看到流程可视化整个订单处理流程接收-支付验证-库存检查-地址验证-发货被清晰地定义在图中。条件路由在每个关键节点支付、库存、地址都根据业务结果成功/失败决定下一步走向。状态管理OrderState对象承载了流程的全部记忆包括订单数据、流程状态、沟通记录和错误信息。持久化与恢复当流程因支付失败而进入handle_pending挂起状态时我们可以将整个state保存。当外部条件满足支付成功后加载该状态并重新invoke图工作流便能从上次中断处handle_pending节点继续执行后续检查最终完成发货。这完美诠释了“长时间运行的有状态业务流程”。4. 高级模式与架构设计构建企业级复杂系统简单的线性审批流只是开始。LangGraph的真正威力在于构建涉及多智能体协作、动态子图调用和复杂循环的企业级系统。以“对公信贷尽职调查报告生成系统”为例我们来探讨如何设计这样的架构。4.1 多智能体协作与角色分配一个尽职调查涉及财务、法律、风险等多个领域。我们可以为每个领域设计一个专门的Agent作为图中的一个节点。# 定义更复杂的全局状态 class DueDiligenceState(TypedDict): company_name: str raw_data: dict # 收集的原始数据 financial_analysis: Optional[str] legal_assessment: Optional[str] risk_report: Optional[str] executive_summary: Optional[str] current_phase: str # e.g., data_collection, analysis, review, finalization assigned_analyst: Optional[str] # ... 其他字段 # 定义专业Agent节点简化示意 def node_financial_agent(state: DueDiligenceState) - dict: 财务分析Agent节点。 # 1. 从state中提取财务相关原始数据 financial_data state[raw_data].get(financial_statements, {}) # 2. 构造一个专业的财务分析Prompt调用LLM # 3. 可能还会调用工具如计算财务比率、与行业基准对比等 # 4. 将分析结果写入state analysis_result f基于{state[company_name]}的财报其偿债能力...盈利能力... return {financial_analysis: analysis_result, current_phase: analysis} def node_legal_agent(state: DueDiligenceState) - dict: 法律合规Agent节点。 legal_docs state[raw_data].get(legal_documents, []) # 调用法律知识库检索、合同解析工具等 assessment f法律审查发现...合规风险点在于... return {legal_assessment: assessment} def node_risk_agent(state: DueDiligenceState) - dict: 风险评估Agent节点。 # 综合财务分析和法律评估进行风险量化 risk_score 0.65 # 模拟计算 report f综合风险评级为B。主要风险集中在... return {risk_report: report}在图中这些Agent节点可以是并行执行的如果彼此独立也可以是串行执行的如果后者依赖前者的输出。LangGraph允许你灵活编排。4.2 动态工作流与子图Subgraph对于超复杂的流程你可以使用子图来模块化。例如“数据分析”阶段本身可能就是一个包含数据清洗、特征提取、模型预测等多个步骤的子图。主图中的一个节点trigger_analysis实际上就是调用这个子图。# 假设我们已定义了一个 analysis_subgraph # 在主图构建中 builder.add_node(conduct_analysis, analysis_subgraph)更高级的模式是动态决定调用哪个子图。例如根据公司所属行业制造业 vs 互联网调用不同的分析流水线。这可以通过条件边让一个路由节点Router Node来决定下一个节点是manufacturing_analysis_subgraph还是internet_analysis_subgraph。4.3 状态设计的核心经验与陷阱设计一个好的State是LangGraph项目成功的一半。这里有几个关键经验扁平化与结构化并存状态应该尽量扁平方便每个节点读写。但对于复杂数据使用Pydantic模型嵌套是更好的选择它能提供自动校验和清晰的文档。例如raw_data字段可以是一个包含financial_statements、legal_documents等子字段的Pydantic模型。区分“流程状态”与“业务数据”像current_phase、pending_action这类控制流程的字段与financial_analysis这类业务内容字段分开管理。这使流程逻辑更清晰。为并发设计如果多个节点可能并行修改状态的不同部分需要仔细考虑状态合并策略。LangGraph默认使用dict.update式的合并对于列表如internal_notes这会导致后一个节点的列表完全覆盖前一个。通常建议对于列表字段在节点内采用state[“notes”].append(new_note)的方式修改原列表而不是返回一个新的列表去覆盖。序列化兼容性由于状态需要被持久化pickle或JSON序列化确保状态中所有字段的数据类型都是可序列化的。避免使用数据库连接对象、文件句柄等不可序列化的对象。对于复杂对象考虑将其转换为字典或字符串存储。注意一个常见的陷阱是直接在节点函数中修改传入的state字典并返回None或修改后的部分字段。LangGraph的合并机制依赖于节点函数的返回值。最安全的做法是将需要更新的字段以字典形式返回。例如return {“current_status”: “updated”, “value”: new_value}。LangGraph会自动将其合并到完整状态中。5. 生产环境部署、监控与调试心法将LangGraph工作流从原型推向生产会面临一系列新的挑战。以下是来自实战的几点核心心法。5.1 持久化后端选型与配置LangGraph支持多种持久化存储。对于生产环境开发/轻量级环境SqliteSaver足够简单快速。生产环境优先选择PostgresSaver或RedisSaver。PostgreSQL可靠且支持复杂的查询方便你后期根据state中的业务字段如order_id,current_status来检索和监控工作流实例。Redis则速度极快适合状态较小且吞吐量高的场景。配置关键务必为每个工作流实例设置唯一的thread_id。这个thread_id是加载和恢复状态的钥匙。它通常与你的业务ID强关联例如f”order_{order_id}”或f”dd_report_{company_name}_{date}”。from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.graph import StateGraph, START, END # 配置持久化 checkpointer SqliteSaver.from_conn_string(“:memory:”) # 生产环境换为Postgres连接 builder StateGraph(MyState, config_schemaMyConfig) # ... 添加节点和边 ... graph builder.compile(checkpointercheckpointer) # 在编译时注入检查点管理器 # 执行时传入 config其中包含 thread_id config {“configurable”: {“thread_id”: “order_12345”}} initial_state {…} # 第一次调用会创建或加载这个thread_id对应的状态 result graph.invoke(initial_state, configconfig) # 后续再以相同的 config 调用 invoke就会基于上次的状态继续执行5.2 可视化、日志与监控LangGraph Studio这是官方提供的本地开发调试神器。它能将你的图可视化并逐步播放每个节点的状态变化是理解流程和调试逻辑不可或缺的工具。强烈建议在开发阶段使用。结构化日志在每个节点的入口和出口打印结构化日志记录thread_id、node_name、input_state_snapshot、output_state_snapshot。这能帮你追踪每个工作流实例的完整生命周期。可以将日志集成到ELK或Datadog等系统中。监控指标在关键节点埋点监控工作流的吞吐量、各节点耗时、错误率以及停滞在pending状态实例的数量。例如监控“平均订单处理时长”、“支付验证失败率”等业务指标。5.3 错误处理、重试与超时机制图中的任何一个节点都可能失败网络超时、API限流、逻辑错误。LangGraph本身不提供自动重试需要你在节点内部或外部架构中实现。节点级容错在节点函数内部使用try...catch将可预见的错误转化为状态字段如error_reason并让条件边路由到专门的error_handling节点。工作流级容错对于瞬态错误如网络抖动可以在调用graph.invoke()的外层包装重试逻辑如使用tenacity库。但要注意幂等性确保重试不会导致重复业务操作如重复扣款。超时控制为整个工作流或单个节点设置超时。可以使用asyncio.wait_for包装节点函数或者在任务队列如Celery层面控制整个工作流的执行时间。死循环预防图中如果存在循环比如我们的示例中handle_pending会跳回verify_payment必须设置recursion_limit。在graph.invoke(config{“recursion_limit”: 100})中传入防止因逻辑错误导致无限循环。5.4 与现有系统集成LangGraph工作流不应该是一个孤岛。它需要与你的微服务、消息队列、数据库交互。作为后台服务将编译好的graph对象封装为一个FastAPI或Flask服务。提供/start、/status/{thread_id}、/resume/{thread_id}等端点。前端或其它服务可以通过API触发和查询工作流。响应事件使用消息队列如RabbitMQ、Kafka。让一个消费者监听特定事件如payment_succeeded当事件到来时根据事件中的业务ID对应thread_id加载对应的工作流状态并调用graph.invoke使其从等待中恢复。数据同步工作流状态是它的“私有内存”。对于需要持久化到业务数据库的结果如生成的尽职调查报告应在最终的节点中显式调用数据库写入API而不是依赖状态持久化作为唯一存储。从我的实践经验来看成功的关键在于前期花足够的时间进行图的设计和状态建模。在白板上画出完整的流程图明确每个节点的输入、输出和异常分支讨论清楚状态的每个字段如何被读写。这能避免在开发中期陷入混乱的状态管理和复杂的条件边调试。LangGraph是一把强大的瑞士军刀但清晰的架构设计才是用好它的前提。