1. 为什么复杂Agent必须显式建模“分支”而不是靠if/else堆叠写AI Agent的时候我一开始也是按最朴素的思路来每个意图配一个函数外层套个 if/else 或者 match-case。用户问价格走价格函数用户问售后走售后函数看起来很清楚。但代码一旦超过两三百行情况就变了——分支逻辑开始散落到各个回调里改一条流程要在好几个文件里来回找。更头疼的是当一次对话要经过“分类 - 检索 - 组装回答”这三层而每一层都可能根据中间结果改变下一步走向时纯代码堆叠的方式基本就失控了。所以我转向了LangGraph。它的核心思路其实很简单把Agent的执行过程建模成一个图。节点node负责干活边edge负责把节点连起来而“接下来到底去哪个节点”这件事则交给条件边conditional edge来动态决定。这听起来只是工程上的解耦但实际用下来它带来的最大收益是分支逻辑变成了图的可视化结构而不是埋在代码逻辑里的隐式跳转。我可以直接看到什么条件下走“VIP客服节点”什么条件下走“普通回复节点”什么条件下直接终止。这个可视化能力对排错和协作都有巨大帮助。1.1 真实场景里的分支痛点决策矩阵远比想象中复杂举一个我在工单系统里做过的例子。用户提交一个工单Agent要先做意图分类是“退款”“维修”“投诉”还是“闲聊”。光这四类就已经需要四路分支。但如果按标签再细分退款又分“已收到货”和“未收到货”维修又分“保内”和“保外”投诉还得根据用户情绪等级走不同处理流程。把这张决策矩阵画出来它实际上是一棵树每个节点都要做一次判断然后选择不同的下游节点。如果这个逻辑全部写在Python函数里你会得到大量嵌套的 if/else而且每一步判断之间还穿插着API调用、数据库查询、LLM调用。你很难回答一个基本问题“某一个具体的请求到底经过了哪些节点”因为执行路径完全隐式只有等出问题时才打日志去猜。LangGraph把“判断”本身也从业务节点里抽了出来。节点只做一件事返回一个决定。条件边拿着这个决定去查路由表。这样每个请求的执行路径都是由图中真实存在的节点构成的而不是一堆散落的函数调用。1.2 图模型中的三类分支要素节点、边、路由函数在LangGraph里构建一个分支你只需要理解三个东西节点node业务函数。输入一个state对象做处理返回一个更新dict。条件边conditional edge连接源节点和多个目标节点的“路由器”。它包含一个路由函数以及一张路径映射表。路径映射表path map路由函数返回的键值与目标节点名的对应关系。例如返回键refund映射到refund_node。举个例子用LangGraph定义一个最简分支from typing import TypedDict from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): intent: str message: str def classify_intent(state: AgentState) - AgentState: # 实际场景里这里一般会调用LLM做意图识别 if 退 in state[message]: return {intent: refund} if 修 in state[message]: return {intent: repair} return {intent: chat} def handle_refund(state: AgentState) - AgentState: return {message: 退款流程请提供订单号} def handle_repair(state: AgentState) - AgentState: return {message: 维修流程请提供设备型号} def handle_chat(state: AgentState) - AgentState: return {message: 闲聊模式开启} # 路由函数从state里取intent返回一个键 def route_by_intent(state: AgentState) - str: return state[intent] # 路径映射表键 - 目标节点 path_map { refund: refund_node, repair: repair_node, chat: chat_node, } builder StateGraph(AgentState) builder.add_node(classify, classify_intent) builder.add_node(refund_node, handle_refund) builder.add_node(repair_node, handle_repair) builder.add_node(chat_node, handle_chat) builder.add_edge(START, classify) builder.add_conditional_edges(classify, route_by_intent, path_map) builder.add_edge(refund_node, END) builder.add_edge(repair_node, END) builder.add_edge(chat_node, END) graph builder.compile()执行时LangGraph会先跑classify_intent拿到返回的{intent: refund}更新state然后调用route_by_intent(state)得到键refund再查path_map找到目标节点refund_node。整个过程透明、可观测分支选择完全由state驱动。2. 条件边路由add_conditional_edges 的分派原理与最小实例add_conditional_edges是理解LangGraph分支绕不开的API。它做的事情是从某个源节点开始根据路由函数的结果动态决定下一步跳到哪个节点。但真正上手时很多人会卡在“路由函数返回什么”和“路径映射表怎么填”这两个问题上。2.1 路由函数返回键、路径映射表与END的配合方式路由函数本质上是一个state - str的函数。它读state中已有的字段然后返回一个字符串键。LangGraph拿到这个键之后会去路径映射表里查目标节点。路径映射表可以是dict也可以直接用list。dict形态键是路由结果值是目标节点名。最灵活因为允许“多个键指向同一个节点”。list形态路由函数返回的不再是键而是直接返回目标节点名。LangGraph会直接拿着这个返回值去找同名节点。实测下来我建议优先使用dict形态。原因很实际业务里经常出现两种意图走同一个节点的场景。比如“退款”和“取消订单”最终都要走“售后确认节点”用dict可以直接把两个键映射到同一个目标用list的话就得在路由函数内部多做一层归一化。路径映射表里有一个特殊目标END。当路由函数发现某个分支已经完成、不需要后续节点时可以直接让键指向END。这让“提前终止”这种分支模式变得很干净。例如builder.add_conditional_edges( audit_node, route_by_audit_result, { pass: finalize_node, reject: notify_node, spam: END, # 直接结束不做处理 }, )2.2 最小可运行实例工单自动分类器的条件边实现我写一个真实可跑的最小实例场景是“工单自动分类”。这个Agent接收用户输入判断工单类型然后走三种不同流程。from typing import TypedDict from langgraph.graph import StateGraph, START, END class TicketState(TypedDict): user_input: str ticket_type: str response: str def detect_type(state: TicketState) - dict: text state[user_input] if 退 in text or 退款 in text: return {ticket_type: refund} if 慢 in text or 等待 in text: return {ticket_type: delay} return {ticket_type: other} def refund_handler(state: TicketState) - dict: return {response: 退款类工单已创建优先处理队列} def delay_handler(state: TicketState) - dict: return {response: 延迟类工单已通知物流团队跟进} def other_handler(state: TicketState) - dict: return {response: 兜底处理转人工客服审核} def choose_handler(state: TicketState) - str: return state[ticket_type] builder StateGraph(TicketState) builder.add_node(detect, detect_type) builder.add_node(refund, refund_handler) builder.add_node(delay, delay_handler) builder.add_node(other, other_handler) builder.add_edge(START, detect) builder.add_conditional_edges( detect, choose_handler, { refund: refund, delay: delay, other: other, }, ) builder.add_edge(refund, END) builder.add_edge(delay, END) builder.add_edge(other, END) ticket_graph builder.compile()执行时调用result ticket_graph.invoke({user_input: 我的退款怎么还没到账}) print(result[ticket_type]) # refund print(result[response]) # 退款类工单已创建优先处理队列这个实例里detect节点只是负责“改state”choose_handler只是“读state”。节点本身不知道下一站是谁边才负责路由。这种职责分离的好处是你可以随时改路径映射表而不用动业务节点的代码。2.3 路由函数里的边界情况键不在映射表中会怎样这是新手最容易踩的坑。如果路由函数返回的键没有出现在路径映射表里LangGraph会直接抛异常执行中断。比如上面例子中如果LLM识别出一个新类型invoice而映射表里没有这个键整个图就会崩溃。我的建议是两种处理方式二选一路由函数内部做兜底保证永远返回映射表中存在的键。比如在detect_type末尾加一个else: return {ticket_type: other}。在路径映射表中增加一个unknown键专门承接所有意料之外的结果。我实际开发时通常两个都做函数里保留兜底映射表里也放unknown。多一层防护生产环境少一次事故。3. 动态分支的两种高频场景工具调用的自寻路与Send并行扇出条件边解决的是“静态的四选一”问题。但Agent真实场景里还有两种更复杂的动态分支一种是节点在执行途中决定下一步去哪另一种是一个节点派生出多个并行子任务。这两种分别对应LangGraph里的Command机制和SendAPI。3.1 节点内部用Command实现“自己决定下一步去哪”条件边是图级别路由路由函数在节点执行完后再被调用。但有些场景下节点执行到一半就需要改变方向了。最典型的是工具调用循环LLM生成一个工具调用请求Agent执行工具拿回结果后LLM又要生成下一个工具调用。直到LLM不再请求工具Agent才把最终回复交给用户。这个循环如果只靠条件边表达你得拆成好多节点互相指来指去。LangGraph提供了一个更直接的办法节点返回Command对象直接在返回值里声明“下一步去哪”。看一个简化版from langgraph.types import Command def tool_agent_node(state: AgentState): result llm_with_tools.invoke(state[messages]) if result.tool_calls: return Command( gototool_executor, # 动态指定下一跳 update{messages: [result]}, # 同时更新state ) return Command( gotoEND, # 工具调用结束回到终点 update{messages: [result]}, ) builder.add_node(tool_agent, tool_agent_node) builder.add_node(tool_executor, execute_tools) builder.add_edge(START, tool_agent) builder.add_edge(tool_executor, tool_agent) # 工具执行完回到Agent重新决策这段代码里tool_agent_node不再是一个“只管返回dict的普通节点”它在返回值里直接指定了路由目标。goto可以是任意已注册的节点名也可以是END。这就是分支执行逻辑里最灵活的一种动态自寻路。它相当于把条件边缩小到了节点内部但你仍然拥有完整的状态更新能力和图级观测能力。用Command的时候有两点要留意goto的优先级高于边。如果同时给节点挂了一条出边而节点返回Command(goto...)那么跟这个goto指定的目标为准。实测中我尽量不对这类节点再画普通边避免混淆。update字段照常参与state合并。哪怕节点动态跳转它对state的修改也不会丢。3.2 用Send API实现一个节点派生出多个并行子分支另一种高频场景是并行执行。比如一个内容审核Agent收到一篇长文要同时跑“敏感信息识别”“版权检测”“格式校验”三个子任务。三个子任务之间没有先后依赖如果串行跑响应时间就是三者之和如果并行跑响应时间只取决于最慢的一个。LangGraph支持通过条件边返回一批Send对象来实现并行扇出。Send的两个参数是目标节点名、传给那个子实例的state。from langgraph.types import Send def dispatch_checks(state: AgentState): return [ Send(sensitivity_check, {article: state[article], check_type: sensitive}), Send(copyright_check, {article: state[article], check_type: copyright}), Send(format_check, {article: state[article], check_type: format}), ] builder StateGraph(AgentState) builder.add_node(dispatch, dispatch_checks) builder.add_node(sensitivity_check, run_check) builder.add_node(copyright_check, run_check) builder.add_node(format_check, run_check) builder.add_edge(START, dispatch) builder.add_conditional_edges(dispatch, dispatch_checks, [sensitivity_check, copyright_check, format_check])这里的关键点dispatch_checks虽然是普通函数但它返回的不是dict而是Send列表。LangGraph看到这个列表后会为目标列表里的每个节点各创建一个子图实例并传给对应的state。这些子实例彼此独立可以并行执行。需要注意add_conditional_edges的第三参在这里从映射表变成了list。因为Send已经明确指定了目标节点名不再需要键值映射。4. 分支汇合时的状态更新语义别让并行分支互相踩踏有了并行分支就必然有汇合问题。多个子分支跑完之后它们的返回值怎么合并回总state这个问题如果不搞透彻并行分支跑得越欢数据覆盖得越惨。4.1 LangGraph默认的last-value语义最后的写入覆盖先前的LangGraph的state每个字段默认是“单值覆盖”语义。也就是说多个分支如果往同一个字段写值最终谁能留下取决于运行顺序而不是合并规则。并行分支在这种情况下会产生典型的竞态分支A计算出的统计结果被分支B的结果覆盖。这个行为在一个分支里不会出问题因为顺序是确定的。但在Send并行场景下两个子实例同时写state[result]最终结果就不确定了。我第一次用并行分支时就踩了这坑三个审核子任务各自把结果写进state[result]等全部跑完result里只剩最后一个完成的任务的数据前两个全丢了。4.2 用Reducer合并并行子分支的结果Annotated operator.add解决方案是用带Reducer的字段。在定义state的TypedDict时通过Annotated指定合并函数LangGraph每次收到该字段的更新时都会先跑一次合并逻辑而不是直接覆盖。import operator from typing import Annotated, TypedDict class AuditState(TypedDict): article: str issues: Annotated[list, operator.add] # 每有新问题就追加而不是覆盖这里operator.add对list的语义是拼接。每个子分支返回{issues: [发现敏感词:xxx]}时LangGraph会把新list追加到旧list后面而不是替换。跑完三个分支issues里自然就是三年份的结果。实际工程项目里我还会自定义Reducer函数来处理更复杂的合并逻辑。比如需要去重的场景operator.add直接拼接会产生重复项可以写一个dedupe函数def dedupe(left: list, right: list) - list: combined left right seen set() result [] for item in combined: key item[code] if isinstance(item, dict) else item if key not in seen: seen.add(key) result.append(item) return result class AuditState(TypedDict): issues: Annotated[list, dedupe]这个设计给分支汇合提供了确定性。任何并行子分支的写入都有明确合并规则可循而不是碰运气。4.3 控制递归深度并行分支与循环都会消耗recursion_limit还有一个看不见的坑LangGraph对一次执行的总步数有上限控制默认值是25。这个上限意图是防止Agent陷入死循环。但在并行分支场景下每个子分支的节点执行都会计入总步数。假设你的图有6个节点并行扇出3个子分支每个子分支都要跑两个节点再加上汇合节点一次invoke可能就消耗掉将近一半的配额。如果Agent还需要多轮工具调用25的上限非常容易触顶。触顶后LangGraph会抛出GraphRecursionError执行被强制中断。实测中我会根据图的复杂度调整这个值config {recursion_limit: 50} graph.invoke(initial_state, configconfig)也可以给工具调用循环设置更精细的轮数预算在节点内部用计数器控制而不是完全依赖全局限额。5. 我实际踩过的分支执行坑从死循环到路由键不匹配到这里LangGraph分支的核心机制基本过完了。但纸上得来终觉浅真正在项目里跑起来还有一堆细节不试不知道。我把踩过的坑整理一下每个都有明确的复现路径和解决思路希望对你有用。5.1 路由键不在路径映射表中报错InvalidUpdate或KeyError现象执行到某个条件边时Agent突然中断日志里出现类似InvalidUpdate或KeyError的异常。根因路由函数返回了一个映射表中不存在的键。这种情况在路由结果来自LLM输出时尤其常见。LLM的返回永远有不确定性你预设了四个意图它有概率吐出一个第五个。定位方式把路由函数的输出打印出来或者用graph.stream观察每一步的返回值。确认那个“肇事键”到底是什么。解决我在生产代码里做了一个短小而关键的习惯——每个路由函数最后写一个默认返回值并且在路径映射表中设置一个unknown目标节点或直接指向END。def safe_route(state) - str: intent state.get(intent, ) if intent in path_map: return intent return unknown5.2 并行分支写入同一个状态字段结果互相覆盖这个前面已经提过。现象是从Send并行分支跑完后最终state里只保留了最后一次写入其他分支的结果神秘消失。根因是默认last-value覆盖语义。解决就是给共享字段加AnnotatedReducer。这块一定要在设计state结构时就考虑进去等并行分支都写完了再去追数据丢失排查成本会高得多。5.3 工具调用循环找不到出口现象Agent在工具调用节点和工具执行节点之间反复横跳一直不结束直到触发GraphRecursionError。根因LLM决策循环里模型反复请求调用同一个工具。常见原因是工具返回结果不够明确模型判定需要再次调用或者节点代码里Command(goto...)逻辑写成了“只要有过一次工具调用就继续循环”没有检查当前消息里是否还有新的tool_calls。解决三层防护。第一设置合理的recursion_limit防止无限资源消耗。第二在负责决策的节点里加入判断如果工具调用的次数已经达到上限比如5轮强制Command(gotoEND)。第三仔细检查循环出口条件确保只有在tool_calls列表不为空时才跳回工具执行节点。我见过一种很隐蔽的写法错误决策节点已经返回了Command(gotoEND)但因为Command的update里没有清空state里的残留标志位图里的另一条边又把它拉了回去。排查这类问题最有效的方式就是stream流式输出直接看每个节点实际执行顺序。5.4 用stream观察分支走向排错的唯一可靠手段LangGraph提供了graph.stream方法可以把每一步节点的执行结果流式吐出来。格式大致是{节点名: {字段名: 更新值}}for event in graph.stream(initial_state): print(event)输出示例{classify: {intent: refund}} {refund: {response: 退款类工单已创建优先处理队列}}当出现“为什么走错了分支”这类疑问时这个输出能直接告诉你哪个节点被执行了它返回了什么最终进了哪条路。比invoke拿最终结果再倒推要直观得多。5.5 别迷信静态图可视化视图不等于全部执行路径LangGraph自带的Studio工具会把图渲染出来静态图上会画出条件边的多条可能走向。但注意这只是“可能性”的集合不代表一次真实执行会走遍所有路径。有些边看起来连上了实际执行时因为路由函数的返回值影响根本不会走到。所以排错时不要对着可视化图猜。要以stream输出为准。图是设计期的蓝图stream才是运行期的黑匣子记录。6. 从分支设计到工程落地一条我从项目里总结的实战建议这部分不讲API了聊一点更偏工程取舍的经验。刚开始从if/else迁移到LangGraph时我犯过一个错误试图把所有细节都表达成节点和边结果是图越画越复杂光节点就有二十几个连自己都看不清流程图。后来我调整了设计原则只有那些“需要被观测、需要被复用、需要跨模块跳转”的决策点才显式建模为分支节点纯粹的内部计算细节继续留在普通函数里。什么分支值得建模我的判断标准是三个下一步行动会因当前结果而改变改变后的路径对应完全不同的处理逻辑该决策需要被记录、被审计、被可视化满足这三条才把决策点拉出来做成条件边或Command分支。不满足的比如只是某个函数内不同参数的调用就不必上到图级别。另一个重要经验是state结构设计。分支越多state字段越要收敛。我见过一个Agent的state里堆了二十多个字段路由函数读哪个、节点更新哪个全靠约定而非约束。后来改成按模块分组customer_info、order_info、process_flags每个分组内部字段才允许互相读分组之间通过明确的接口字段通信。这虽然麻烦但分支逻辑的可维护性大幅提升。最后聊一下和FastAPI、LangChain搭在一起的真实项目形态。网上热词里提到的“基于FastAPI LangChain LangGraph的AI Agent”通常长这样FastAPI负责接收外部请求把会话历史打包成state交给LangGraph的图执行执行结束后把最终回复返回给调用方。LangChain提供底层的模型封装和工具调用规范LangGraph负责流程编排和分支控制。三者配合起来LangGraph恰恰就承担了最关键的“流程大脑”角色。我在写这类结构化Agent时有一个体会真正决定Agent上限的往往不是模型能力而是流程能否在正确的位置做出正确的分支决策。同样的LLM放在一张设计良好的图里和埋在一堆if/else里产出的稳定性和可维护性是天壤之别。所以如果你正在搭自己的Agent别急着堆功能。先把分支画清楚哪些步骤固定串行哪些步骤需要动态选择哪些步骤必须并行。然后照着LangGraph的条件边、Command和Send去落图。等跑起来之后再回头看你会明显感觉到整个流程就像一张可以随时调整走向的交通图而不是一团长在编辑器里的乱麻。