目录1 串行控制2 分支控制3 条件控制实现通过不同的边和连接关系组织节点实现节点之间的执行顺序。1 串行控制添加普通边add_edge节点对应关系1:n节点之间按照串行方式连接。注意最终的输出会将所有节点的输出数据合并后输出也就是包含所有上下文import operator from langgraph.graph import END, START, StateGraph from typing import Annotated, TypedDict # 状态模型字典类型 class State(TypedDict): value: Annotated[str,operator.add] def node1(state: State) - State: print(f{state[value]} node1 ) return {value: f{state[value]} node1 } def node2(state: State) - State: print(f{state[value]} node2 ) return {value: f{state[value]} node2 } def node3(state: State) - State: print(f{state[value]} node3 ) return {value: f{state[value]} node3 } def node4(state: State) - State: print(f{state[value]} node4 ) return {value: f{state[value]} node4 } def node5(state: State) - State: print(f{state[value]} node5 ) return {value: f{state[value]} node5 } graph StateGraph(State) graph.add_node(node1, node1) graph.add_node(node2, node2) graph.add_node(node3, node3) graph.add_node(node4, node4) graph.add_node(node5, node5) graph.add_edge(START, node1) graph.add_edge(node1, node2) graph.add_edge(node2, node3) graph.add_edge(node3, node4) graph.add_edge(node4, node5) graph.add_edge(node5, END) compiled_graph graph.compile() print(compiled_graph.get_graph().draw_ascii()) result compiled_graph.invoke(State(value)) print(result[value])2 分支控制添加普通边add_edge,节点对应关系1:n注意汇聚点node5会包含之前所有分支节点的所有数据import operator from langgraph.graph import END, START, StateGraph from typing import Annotated, TypedDict # 状态模型字典类型 class State(TypedDict): value: Annotated[str,operator.add] def node1(state: State) - State: print(f{state[value]} node1 ) return {value: f{state[value]} node1 } def node2(state: State) - State: print(f{state[value]} node2 ) return {value: f{state[value]} node2 } def node3(state: State) - State: print(f{state[value]} node3 ) return {value: f{state[value]} node3 } def node4(state: State) - State: print(f{state[value]} node4 ) return {value: f{state[value]} node4 } def node5(state: State) - State: print(f{state[value]} node5 ) return {value: f{state[value]} node5 } graph StateGraph(State) graph.add_node(node1, node1) graph.add_node(node2, node2) graph.add_node(node3, node3) graph.add_node(node4, node4) graph.add_node(node5, node5) graph.add_edge(START, node1) graph.add_edge(node1, node2) graph.add_edge(node1, node3) graph.add_edge(node1, node4) graph.add_edge([node2,node3, node4], node5) graph.add_edge(node5, END) compiled_graph graph.compile() print(compiled_graph.get_graph().draw_ascii()) result compiled_graph.invoke(State(value)) print(result[value])3 条件控制添加条件边add_conditional_edges节点对应关系1:n添加分叉节点# 第一个参数表示要进行分叉的起始节点 # 第二个参数表示分叉的规则 graph.add_conditional_edges(node2, route) def route(state: State) - Literal[node1, node3]: if state[count] 0: print(f{state[value]} route node1 ) return node1 else: print(f{state[value]} route node3 ) return node3import operator from langgraph.graph import END, START, StateGraph from typing import Annotated, Literal, TypedDict # 状态模型字典类型 class State(TypedDict): value: Annotated[str,operator.add] count: int def node1(state: State) - State: print(f{state[value]} node1 ) return {value: f{state[value]} node1 } def node2(state: State) - State: print(f{state[value]} node2 ) return {value: f{state[value]} node2 , count: state[count] 1} def node3(state: State) - State: print(f{state[value]} node3 ) return {value: f{state[value]} node3 } def node4(state: State) - State: print(f{state[value]} node4 ) return {value: f{state[value]} node4 } def node5(state: State) - State: print(f{state[value]} node5 ) return {value: f{state[value]} node5 } def route(state: State) - Literal[node1, node3]: if state[count] 0: print(f{state[value]} route node1 ) return node1 else: print(f{state[value]} route node3 ) return node3 graph StateGraph(State) graph.add_node(node1, node1) graph.add_node(node2, node2) graph.add_node(node3, node3) graph.add_node(node4, node4) graph.add_node(node5, node5) graph.add_edge(START, node1) graph.add_edge(node1, node2) graph.add_conditional_edges(node2, route) graph.add_edge(node3, node4) graph.add_edge(node4, node5) graph.add_edge(node5, END) compiled_graph graph.compile() compiled_graph.get_graph().draw_png(graph.png) result compiled_graph.invoke(State(value, count0)) # print(result[value])条件控制用ACII的方式显示存在问题因此这里把graph画成png图。关于画图可参考之前我写的langgraph学习1的文章compiled_graph.get_graph().draw_png(graph.png)