LangGraph笔记：三、节点与边

LangGraph中节点和边构成了整个工作流的框架。节点是负责处理业务的边负责将节点处理完毕的数据传动给下一个节点。LangGraph是从START位置开始然后到节点最后END终止位置。START节点代表用户输入进入图的位置所有从START出发的边决定了图的入口点。END节点表示图执行结束的位置指向END的边告诉 LangGraph 在这里停止执行。LangGraph支持多种连接方式并行连接盒串行连接。串行连接START----节点1------边-------节点2------边-------节点3------边-------END并行连接START----节点1--| |------边-------节点3------边-------END START----节点2--|3.1、节点节点实际就是一个函数接受当前的状态作为输入执行相应的业务之后输出返回或更新状态。这个节点可以是一个简单的函数可以是一个完整的LangGraph子图。节点函数返回是字典只包含需要更新的状态字段即可不需要整个状态对象。LangGraph会自动将这个更新合并到整体状态中。简单的函数:defsimple_node(state:State)-dict:return{message:state[text]}完整的LangGraph子图fromtyping_extensionsimportTypedDictclassState(TypedDict):messages:Annotated[list[AnyMessage],add_messages]defnode_a(state:State):return{messages:value_a}defnode_b(state:State):return{messages:value_b}defgraph_demo(state:State):builderStateGraph(State)builder.add_node(node_a)builder.add_node(node_b)builder.add_edge(START,node_a)builder.add_edge(START,node_b)demo1StateGraph(State)demo1.add_node(graph_demo,graph_demo)demo1.add_edge(START,graph_demo)demo1.add_edge(graph_demo,END)节点函数可以接受更多参数来获取运行时信息多参数接收fromlangchain_core.runnablesimportRunnableConfigfromlanggraph.runtimeimportRuntimedefadv_node(state:State,config:RunnableConfig,runtime:Runtime):# config 包含线程ID、标签等配置信息thread_idconfig[configurable][thread_id]# runtime 提供上下文、存储等运行时能力contextruntime.context# 如果有定义上下文的话return{text:fProcessed in thread{thread_id}}builder.add_node(advanced,advanced_node)如果想添加多个节点可以如下操作多节点# 方式一显式指定节点名称builder.add_node(my_node,simple_node)# 方式二使用函数名作为节点名称builder.add_node(simple_node)# 节点名称为 simple_node多边fromlanggraph.graphimportSTART,END# 定义从起点到第一个节点的边builder.add_edge(START,processor)# 定义从最后一个节点到终点的边builder.add_edge(processor,END)3.2、节点缓存机制LangGraph支持节点缓存功能主要用于解决一些耗时的功能调用缓存可以显著减少重复计算。importtimefromlanggraph.cache.memoryimportInMemoryCachefromlanggraph.typesimportCachePolicy memberInMemoryCache()defdemo_node(state:State)-dict:time.sleep(15)return{message:state[message]}# 添加缓存策略TTL 为 120 秒builder.add_node(demo_node,demo_node,cache_policyCachePolicy(ttl120))# 编译图时启用缓存graphbuilder.compile(cachemember)# 第一次调用会执行耗时操作result1graph.invoke({message:5})# 第二次调用会命中缓存立即返回result2graph.invoke({message:5})3.3、边类型边类型分为普通边和条件边。3.3.1、普通边普通边是确定的方向、没有条件的流转路径一旦定义流传路径是固定的不依赖任何运行时状态适合确定性分支流程。START----节点1------边-------节点2------边-------节点3------边-------ENDbuilder.add_edge(node_a,node_b)3.3.2、条件边条件边动态路由逻辑允许根据运行时状态动态的决定下一个节点如根据LLM的输出决定调用哪个工具。defrouting_function(state:State)-str:# 根据状态决定下一个节点ifstate[condition]A:returnnode_belse:returnnode_c# 添加条件边builder.add_conditional_edges(node_a,routing_function)条件边多节点并行条件边的路由函数返回目标节点的名称如果返回字符串列表所有对应的节点会并行执行。defmulti_route(state:State)-list[str]:return[node_b,node_c,node_d]# 入口点builder.add_edges(START,node_a)# 条件边返回节点列表并且并行执行builder.add_conditional_edges(node_a,multi_route)**条件入口点动态选择起点**允许根据输入状态动态选择从哪个节点开始执行。defadmin_node(state:State):return{message:one}defnormal_node(state:State):return{message:two}defentry_routing(state:State)-str:# 根据输入决定从哪个节点开始ifstate[user_type]admin:returnadmin_nodeelse:returnnormal_node# 从 START 添加条件边builder.add_conditional_edges(START,entry_routing)**路径映射显示定义路由**当路由函数的返回值不是直接的节点名称而是根据返回的一个属性值判断使用路径映射将标识转换为实际的节点名称。路径映射让路由逻辑与节点名称解耦提高了代码的可维护性。当节点名称变化时只需修改映射表不需要改动路由函数。defstatus_router(state:State)-str:returnsuccess# return retry# return faildefcomplete_node(state:State):return{message:complete_node}defretry_node(state:State):return{message:retry_node}deferror_node(state:State):return{message:error_node}# 使用路径映射将状态码映射到节点builder.add_conditional_edges(checker,status_router,{success:complete_node,retry:retry_node,fail:error_node})终止条件路由到END路由函数可以返回END来终止图执行。这在满足特定条件时提前结束流程很有用。defnext_node(state:State):return{message:next_node}defrouter(state:State)-str:ifstate[should_continue]isFalse:returnEND# 终止执行returnnext_nodebuilder.add_conditional_edges(checker,router)允许边可以通过END终止图执行。defnext_node(state:State):return{message:next_node}builder.add_edges(next_node,END)**多入口多出口**对于复杂的工作流可以多个入口或出口。builder.add_edge(START,start_node)builder.add_edge(success_node,END)builder.add_edge(error_node,END)builder.add_edge(partial_node,END)