LangGraph 基础入门2(小白必看)

LangGraph 控制流与节点执行顺序、分支、循环、容错一篇梳理清楚上一篇整理了 LangGraph 的基础State、Node、Edge、Reducer和MessagesState。这一篇继续往下走重点看 LangGraph 里最关键的能力控制流。所谓控制流就是图运行时到底怎么走是按顺序执行 是根据条件选择分支 是多个节点并行 是多个分支汇总 是工具调用失败后循环重试 节点报错了要不要自动重试这篇文章按从简单到复杂的顺序整理顺序结构、分支结构、Fan-in 汇聚、MapReduce、循环结构、递归限制、节点容错、重试、超时、错误处理和缓存。一、顺序结构最普通的执行流程顺序结构就是最基础的START - node_a - node_b - node_c - END在 LangGraph 中可以用add_edge()一条一条连接。builder.add_edge(START,node_a)builder.add_edge(node_a,node_b)builder.add_edge(node_b,node_c)builder.add_edge(node_c,END)这表示从 START 开始 先执行 node_a 再执行 node_b 再执行 node_c 最后到 ENDadd_edge 的核心作用add_edge()就是添加一条边告诉 LangGraph 一个节点执行完后下一步去哪。builder.add_edge(node_a,node_b)意思是node_a 执行完后触发 node_b。二、add_sequence快速写顺序流程如果节点很多一条一条写add_edge()会比较啰嗦。比如builder.add_edge(START,node_a)builder.add_edge(node_a,node_b)builder.add_edge(node_b,node_c)builder.add_edge(node_c,END)可以用add_sequence()简化。builder.add_sequence([node_a,node_b,node_c])builder.add_edge(START,node_a)builder.add_edge(node_c,END)它表示把这些节点按顺序串起来node_a - node_b - node_c注意add_sequence()只是连接节点之间的顺序不一定自动帮你处理START和END实际写代码时仍要看版本和具体用法最好明确写出入口和出口。三、可以省略指向 END 的边吗有些情况下LangGraph 可以推断终止节点。但初学时不建议依赖隐式行为。推荐明确写builder.add_edge(last_node,END)这样图结构最清楚也方便自己调试。四、静态分支图编译前就确定候选路径分支结构分两类静态分支候选下游节点在图编译前已经确定动态分支运行时才决定触发哪些节点、触发多少个任务先看静态分支。1. 静态并行节点如果一个节点后面接多个节点builder.add_edge(node_a,node_b)builder.add_edge(node_a,node_c)意思是node_a 执行完后同时触发 node_b 和 node_c。流程图大概是- node_b node_a - node_c如果node_b和node_c都写同一个状态字段就要注意 Reducer否则可能发生冲突。五、条件分支add_conditional_edges条件分支用builder.add_conditional_edges(...)它的作用是从某个节点出发根据当前状态决定下一步去哪。常见写法defrouter(state):ifstate[content_type]poem:returnnode_areturnnode_bbuilder.add_conditional_edges(START,router)这里的运行逻辑是START - router 判断 如果返回 node_a就去 node_a 如果返回 node_b就去 node_bpath_map 是干什么的path_map用来把路由函数的返回值映射到真实节点名。比如defrouter(state):ifstate[content_type]poem:returnareturnbbuilder.add_conditional_edges(START,router,path_map{a:node_a,b:node_b})意思是router 返回 a - 执行 node_a router 返回 b - 执行 node_b如果不写path_map那router返回的字符串通常就必须是真实节点名。path_map 列表写法也可以写path_map[node_a,node_b]这种可以理解成声明候选节点router 可能返回 node_a 或 node_b。它更像是{node_a:node_a,node_b:node_b}六、deferTrue延迟节点执行注册节点时可以写builder.add_node(audit_node,audit_node,deferTrue)deferTrue的意思是这个节点即使被连接到了流程里也会尽量延迟到普通任务结束后执行。常见场景是审计、日志、收尾检查。比如builder.add_node(audit_node,audit_node,deferTrue)builder.add_edge(START,audit_node)虽然它从START就可达但由于deferTrue它更像一个“最后检查节点”。可以理解成主任务先跑audit_node 尽量最后跑。七、动态分支Send 动态分发任务动态分支的重点是运行时才决定要创建多少个任务。典型用法是Send。fromlanggraph.typesimportSendSend的结构可以简单理解为Send(目标节点名,传给目标节点的私有状态)比如Send(worker_node,{content_type:poem,prompt:请生成一首关于布偶猫的诗})意思是给 worker_node 发一个任务并传入这份私有状态。动态生成多个任务比如CONTENT_TYPES[poem,ci_poem,joke]defrouter(state):topicstate[topic]return[Send(worker_node,{content_type:content_type,prompt:f请生成关于{topic}的{content_type}})forcontent_typeinCONTENT_TYPES]如果输入{topic:布偶狗}那么会动态创建 3 个任务worker_node 生成 poem worker_node 生成 ci_poem worker_node 生成 joke注意动态不是运行时创建新的节点定义节点仍然要提前注册builder.add_node(worker_node,worker_node)动态的是运行时触发多少次节点、每次传什么数据。八、Command(goto…)节点内部决定下一步除了add_conditional_edges()还有一种动态跳转方式fromlanggraph.typesimportCommand例子defrouter(state):content_typestate[content_type]ifcontent_typepoem:returnCommand(gotopoem_node)elifcontent_typejoke:returnCommand(gotojoke_node)returnCommand(gotoEND)这表示如果 content_type 是 poem就跳到 poem_node 如果 content_type 是 joke就跳到 joke_node 否则直接 END和add_conditional_edges()的区别可以这样记add_conditional_edges路由规则写在边上 Command(goto...)路由规则写在节点返回值里九、多分支汇聚Fan-inFan-in 就是多个分支最后汇聚到一个节点。1. “与”触发等所有上游都完成builder.add_edge([node_c,node_d],node_e)意思是node_c 和 node_d 都完成后node_e 才执行一次。这叫“等待所有上游”。2. “或”触发任意上游到达就触发builder.add_edge(node_c,node_e)builder.add_edge(node_d,node_e)意思是node_c 完成后可以触发一次 node_e node_d 完成后也可以触发一次 node_e所以node_e可能执行两次。一句话区分列表上游等所有上游触发一次 分开写边谁到谁触发可能多次十、MapReduce动态扇出 汇总MapReduce 是动态分支里很典型的模式。比如我们要统计多个句子的词频{input_values:[hello world,hello Atguigu,hello LLM]}可以分成三步Map每句话拆成 (word, 1) Shuffle相同单词放一起 Reduce统计每个单词出现次数状态可以这样定义fromtypingimportTypedDict,AnnotatedfromoperatorimportaddclassOverAllState(TypedDict):input_values:list[str]entries:Annotated[list[tuple[str,int]],add]word_counts:dict[str,int]这里的关键是entries:Annotated[list[tuple[str,int]],add]多个 mapper 节点返回的entries会自动拼接。动态分发defrouter_map(state:OverAllState):tasks[]forinput_valueinstate[input_values]:tasks.append(Send(mapper_node,{input_value:input_value}))returntasks每一句话都会生成一个mapper_node任务。最后reducer_node统一拿到所有entries统计成word_counts。这就是router_map - 多个 mapper_node - reducer_node十一、循环结构ReAct 的核心Agent 里常见的工具调用流程本质就是循环。典型 ReAct 流程LLM 思考 如果需要工具 - 调用工具 工具结果返回给 LLM LLM 再思考 直到不再需要工具 - 输出最终答案在 LangGraph 里可以这样设计input_node - llm_node - router ↓ 有工具调用 - tool_node - llm_node 无工具调用 - output_node - END关键边是builder.add_conditional_edges(llm_node,router)builder.add_edge(tool_node,llm_node)builder.add_edge(output_node,END)其中llm_node调用模型router判断模型是否产生tool_callstool_node执行工具并把结果作为ToolMessage放回消息列表tool_node - llm_node让模型基于工具结果继续推理output_node取最终答案这就是一个简化版 ReAct 循环。十二、静态循环和动态循环循环结构有两种常见实现方式静态实现静态实现通常通过add_conditional_edges()提前定义好候选路径。比如llm_node 后面只可能去 tool_node 或 output_node然后运行时只是在这两个候选节点中选一个。动态实现动态实现通常通过Command(goto...)让节点自己在返回值里决定下一步去哪。比如returnCommand(gototool_node)或者returnCommand(gotoEND)它更适合“节点执行时同时更新状态并决定下一步”的场景。十三、递归限制防止死循环循环结构很强但也有风险如果一直循环不结束就会无限跑下去。LangGraph 支持递归限制。调用时可以配置graph.invoke(input_data,config{recursion_limit:10})意思是最多允许图运行 10 步左右超过就中断。主动退出主动退出就是你自己在逻辑里判断ifstep_count5:returnCommand(gotoEND)优点是可控、优雅。被动中断被动中断就是依赖recursion_limit。如果循环超过限制LangGraph 抛异常。这种更像兜底保护不适合作为主要业务逻辑。推荐业务上用主动退出 系统安全上加 recursion_limit 兜底十四、LangGraph 里的边总结LangGraph 常见边可以分为类型用法含义普通边add_edge(a, b)a 执行完触发 b条件边add_conditional_edges(a, router)根据 router 返回值决定去哪入口点add_edge(START, node)图从哪个节点开始条件入口点add_conditional_edges(START, router)从入口处就根据条件分发终止边add_edge(node, END)node 执行完结束条件终止router返回END满足条件时结束END不是普通节点它表示图运行终止。十五、节点容错机制现实中节点可能失败比如API 网络波动模型调用超时工具返回异常数据库连接失败LangGraph 提供了节点级别的容错能力比如重试策略超时控制错误处理节点缓存这些能力可以让图更接近真实工程项目而不是只适合写 demo。十六、重试机制 RetryPolicy如果某个节点容易失败可以给节点配置重试策略。核心配置一般包括max_attempts initial_interval backoff_factor max_interval jitter retry_onmax_attempts最大尝试次数。比如max_attempts3表示最多执行 3 次第一次失败后还可以重试。initial_interval第一次重试前等待多久。backoff_factor退避系数。比如第一次等 1 秒第二次可能等 2 秒第三次等 4 秒。max_interval最大等待时间防止等待时间越来越夸张。jitter随机抖动避免多个请求同时失败后又同时重试造成新的流量高峰。retry_on指定哪些异常需要重试。比如只对网络异常重试对代码逻辑错误不重试。十七、超时控制 TimeoutPolicy超时控制用于限制节点最长执行时间。比如一个节点调用外部 API如果一直没有响应就不能无限等下去。核心思想是超过指定时间还没完成就认为这个节点失败。这通常会和重试机制配合节点超时 - 触发失败 - 根据 RetryPolicy 决定是否重试十八、错误处理错误处理主要解决节点失败后要不要返回一个兜底结果 要不要进入另一个节点 要不要记录日志 要不要中断整个图简单项目可以直接让异常抛出。工程项目里通常会设计专门的错误节点比如normal_node - 成功 - next_node normal_node - 失败 - error_node这样用户体验会更稳定。十九、节点缓存节点缓存用于避免重复执行相同计算。适合场景节点计算成本高输入相同输出也稳定调用外部 API 比较慢或比较贵结果短时间内可以复用例如同一个城市的天气查询短时间内不需要反复请求。节点缓存通常需要两步1. 给节点配置 cache_policy 2. 编译图时启用缓存后端缓存策略里常见配置key_func决定缓存 key 怎么生成ttl缓存有效期key_funckey_func用来决定“什么输入算同一次请求”。比如city杭州可以生成一个缓存 keyweather:杭州ttlttl表示缓存保留多久。比如天气数据可以缓存 5 分钟文档解析结果可以缓存更久。二十、全图默认配置如果很多节点都需要同样的重试、超时、缓存策略可以考虑全图默认配置。这样不用每个节点都重复写一遍。但是要注意不同节点的风险不同。比如模型调用节点适合配置重试本地纯计算节点可能不需要重试写数据库节点不能随便重试否则可能重复写入所以默认配置要谨慎最好结合节点类型来设计。二十一、学习路线总结LangGraph 控制流可以按这个顺序理解顺序结构add_edge / add_sequence 静态分支add_conditional_edges 动态分支Send / Command 多分支汇聚Fan-in 循环结构LLM - Tool - LLM 安全控制recursion_limit 工程增强retry / timeout / error handling / cache如果你刚开始学最重要的是先抓住这几组区别概念重点add_edge固定下一步add_conditional_edges根据状态选择下一步Send动态创建多个任务Command(goto...)节点自己决定跳转add_edge([a, b], c)等多个上游都完成再执行一次分开写多条边任意上游完成都可能触发tool_node - llm_node构成 ReAct 循环recursion_limit防止无限循环总结LangGraph 的控制流本质上是在回答一个问题当前节点执行完以后下一步到底该去哪简单流程用add_edge()。条件选择用add_conditional_edges()。动态多任务用Send。节点内部跳转用Command(goto...)。多个分支汇总用 Fan-in。Agent 工具调用靠循环结构。真正做项目时再加上重试、超时、错误处理和缓存图就从“能跑”变成了“更稳定、更工程化”。