from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END # 1. 定义全局状态 class AgentState(TypedDict): question: str need_tool: bool # 标记是否需要调用外部工具 tool_output: str # 工具返回结果 chat_msg: Annotated[list[str], operator.add] answer: str # 2. 定义节点 def judge_intent(state: AgentState) - dict: 节点1意图判断决定是否要用工具 print(【用户意图判断节点检测是否需要调用工具】begin:) q state[question] # 模拟规则包含查询则判定需要工具 print(【用户意图判断节点检测是否需要调用工具】ongoing:, q) need 查询 in q print(【用户意图判断节点检测是否需要调用工具】done:, need) return { need_tool: need, chat_msg: [f意图判断完成是否调用工具{need}] } def call_external_tool(state: AgentState) - dict: 节点2工具调用节点 print(【工具调用节点】begin:) print(【工具调用节点】ongoing:.....) result 从外部接口拿到业务数据 print(【工具调用节点】done:, result) return { tool_output: result, chat_msg: [工具调用成功] } def direct_answer(state: AgentState) - dict: 节点3直接回答节点无需工具 print(【直接回答节点-大模型调用】begin:) print(【直接回答节点-大模型调用】ongoing:) answer 大模型基于常识直接给出的答案; print(【直接回答节点-大模型调用】done:, answer) return { answer: answer, chat_msg: [无需工具直接作答] } def combine_answer(state: AgentState) - dict: 节点4结合工具结果生成答案 print(【整合答案节点】begin:) print(【整合答案节点】onging:) print(【整合答案节点】done:, [整合工具结果完成作答]) return { answer: f结合工具数据{state[tool_output]} 生成答案, chat_msg: [整合工具结果完成作答] } # 3. 定义路由函数分支核心 def route_tool_or_not(state: AgentState) - str: 路由判断函数 入参当前全局状态 返回目标节点名称字符串必须和 path_map 对应 print(【路由函数】begin:) print(【路由函数】goging:) if state[need_tool]: result call_tool else: result direct_ans print(【路由函数】done返回下个节点名称:, result) return result; # 4. 构建图 配置分支 builder StateGraph(AgentState) # 注册所有节点 builder.add_node(judge, judge_intent) builder.add_node(call_tool, call_external_tool) builder.add_node(direct_ans, direct_answer) builder.add_node(combine_ans, combine_answer) # 入口流向judge节点 builder.add_edge(START, judge) # 配置条件分支从source节点开始到其他节点path_map判决的依据path。 builder.add_conditional_edges( sourcejudge, # 分支起始节点 pathroute_tool_or_not, # 路由判断函数这是条件分支最重要的逻辑是在节点逻辑执行完才执行用于选择下一个节点返回下一个节点名称 path_map{ # 映射函数返回值 → 真实节点名 call_tool: call_tool, direct_ans: direct_ans } ) # 配置后续固定流向 builder.add_edge(call_tool, combine_ans) # 工具执行后 → 整合答案 builder.add_edge(direct_ans, combine_ans) # 直接回答 → 结束 builder.add_edge(combine_ans, END) # 整合答案 → 结束 # 编译 执行combine_ans app builder.compile() if __name__ __main__: # 测试1需要工具 print( 测试1包含查询走工具分支 ) # invoke负责触发执行图中的所有节点包括分支和循环 res1 app.invoke({question: 查询今日订单, chat_msg: [], need_tool: False, tool_output: , answer: }) print(\n 测试2普通问题不走工具 ) # 测试2不需要工具 res2 app.invoke({question: 你好, chat_msg: [], need_tool: False, tool_output: , answer: }) 测试1包含查询走工具分支 【用户意图判断节点检测是否需要调用工具】begin:【用户意图判断节点检测是否需要调用工具】ongoing: 查询今日订单【用户意图判断节点检测是否需要调用工具】done: True【路由函数】begin:【路由函数】goging:【路由函数】done返回下个节点名称: call_tool【工具调用节点】begin:【工具调用节点】ongoing:.....【工具调用节点】done: 从外部接口拿到业务数据【整合答案节点】begin:【整合答案节点】onging:【整合答案节点】done: [整合工具结果完成作答] 测试2普通问题不走工具 【用户意图判断节点检测是否需要调用工具】begin:【用户意图判断节点检测是否需要调用工具】ongoing: 你好【用户意图判断节点检测是否需要调用工具】done: False【路由函数】begin:【路由函数】goging:【路由函数】done返回下个节点名称: direct_ans【直接回答节点-大模型调用】begin:【直接回答节点-大模型调用】ongoing:【直接回答节点-大模型调用】done: 大模型基于常识直接给出的答案【整合答案节点】begin:【整合答案节点】onging:【整合答案节点】done: [整合工具结果完成作答]一、基本定义条件分支依托条件边Conditional Edge实现是 LangGraph 区别于线性普通边的核心能力。它指当前节点执行完毕后框架不再固定跳转下一个节点而是根据全局State中的数据动态判断、选择后续执行路径以此实现多分支、分流逻辑。二、核心组成三要素source源节点分支的触发源头代表该节点执行完成后立即进入条件判断流程。一个源节点可以对应一套分支逻辑。path路由函数纯判断逻辑函数也是条件分支的核心。入参全局状态State职责读取状态里的标记、数据、结果等执行业务规则判断出参固定返回自定义分支标识字符串用于匹配分支走向。特点仅做判断不执行业务逻辑、不修改状态。path_map路由映射表字典结构作用是解耦路由函数与真实节点将路由函数返回的标识字符串一一映射为图中已注册的节点名 /END终点。便于统一管理分支、后期维护。三、执行时序遵循框架通用规则严格遵守节点先执行边后执行的底层逻辑完整流程调度执行源节点业务函数完成逻辑处理并更新全局State源节点执行结束框架自动触发该节点绑定的条件边调用路由函数读取最新State并得到分支标识通过path_map匹配标识确定下一跳目标节点 /END跳转到目标分支节点继续执行若匹配到END则流程终止。四、与普通边的核心区别普通边add_edge跳转路径硬编码固定节点执行完必然走向指定节点只能实现串行线性流程无判断能力。条件边add_conditional_edges跳转路径动态可变依赖运行时的状态数据做判断灵活实现分流、分支、前置校验等复杂逻辑。五、典型应用场景意图分流根据用户提问内容区分问答、查询、闲聊等不同流程能力路由判断是否需要调用外部工具、联网检索、数据库查询拆分 “工具分支” 和 “直接回答分支”结果校验节点执行完成后根据输出结果判断是否继续执行、重试或终止权限 / 流程管控结合状态中的权限、审批标记分流不同权限的执行链路。六、关键特性与补充说明多分支支持path_map可配置多个映射关系不止二选一能实现三路及以上分支无状态耦合路由逻辑、节点逻辑、流向配置相互解耦路由函数只依赖全局State修改规则无需改动业务节点可组合使用条件分支可以和普通边、循环、人在回路搭配构建多层级复杂流程终止兼容分支目标可以是普通节点也可以直接指向END终点实现 “部分分支直接结束流程”。七、总结条件分支是 LangGraph 实现非线性流程的基础能力。本质是在节点执行完成后通过一套「状态读取→规则判断→路径映射」的路由逻辑动态分配后续执行链路解决了线性串行流程无法适配复杂业务分流的问题也是智能体、RAG、对话机器人等场景中最常用的基础能力之一。