尧图网站建设 
深度理解LangGraph核心-Graph一、 理解什么是Graph图1、State状态2、Node节点3、Edge边4、子图5、图的Stream支持五、总结在了解了LangGraph中如何构建Agent智能体之后接下来就要进入LangGraph的重头戏Graph了。Graph是LangGraph的核心它以有向无环图的方式来整合多个Agent构建更复杂的Agent大模型应用形成更复杂的工作流。并且提供了很多产品级的特性保证这些应用可以更稳定高效的执行。一、 理解什么是Graph图Graph是LangGraph的基本构建模块它是一个有向无环图(DAG)用于描述任务之间的依赖关系。主要包含三个基本的元素:State:在整个应用当中共享的一种数据结构。Node:一个处理数据的节点。LangGraph中通常是一个Python的函数以State为输入经过一些操作后返回更新后的State。Edge:表示Node之前的依赖关系。LangGraph中通常也是一个Python函数根据当前State来决定接下来执行哪个Node。接下来用一个最简单的案例来看一下Graph的基本用法。#安装依赖 pip install -U langgraphfromtypingimportTypedDictfromlanggraph.constantsimportEND,STARTfromlanggraph.graphimportStateGraphclassInputState(TypedDict):user_input:strclassoutputState(TypedDict):graph_output:strclassOverallState(TypedDict):foo:struser_input:strgraph_output:strclassPrivateState(TypedDict):bar:strdefnode_1(state:InputState)-OverallState:# Write to OverallStatereturn{foo:state[user_input]博客}defnode_2(state:OverallState)-PrivateState:# Read from OverallState, write to PrivateStatereturn{bar:state[foo]非常}defnode_3(state:PrivateState)-OutputState:# Read from PrivateState, write to OutputStatereturn{graph_output:state[bar]靠谱}# 构建图builderStateGraph(OverallState,inputInputState,outputOutputState)# 添加Nodebuilder.add_node(node_1,node_1)builder.add_node(node_2,node_2)builder.add_node(node_3,node_3)#添加Edgebuilder.add_edge(START,node_1)builder.add_edge(node_1,node_2)builder.add_edge(node_2,node_3)builder.add_edge(node_3,END)#编译图graphbuilder.compile()#调用图graph.invoke({user_input:wuyuanshun}){‘graph_output’:‘wuyuanshun博客非常靠谱’}这个案例当中请求的参数从固定的START传入依次经过三个节点处理每个节点的处理结果都会被保存到不同的state当中最后进入到END节点结束。如果你觉得这个流程不够明显还可以直接用真正的图来看看请求是如何夕上理的。1、State状态State是所有节点共享的状态它是一个字典包含了所有节点的状态。有几个需要注意的地方:State形式上可以是TypedDict字典也可以是Pydantic中的一个BaseModel。例如:frompydanticimportBaseModel# The overall state of the graph (this is the public state shared across nodes)classOverallState(BaseModel):a:str这两种实现本质上没有太多的区别。State中定义的属性通常不需要指定默认值。如果需要默认值可以通过在START节点后定义一个node来指定默认值。defnode(state:OverallState):return{a:goodbye}State中的属性除了可以修改值之外也可以定义一些操作。来旨定如何更新State中的值。例如fromlanggraph.graph.messageimportadd_messagesclassState(TypedDict):messages:Annotated[list[AnyMessage],add_messages]list_field:Annotated[list[int],add]extra_field:int此时如果有一个node返回了State中更新的值那么messages和lis_field的值就会添加到原有的旧集合中而extra_field的值则会被替换。fromlangchain_core.messagesimportAnyMessage,AIMessagefromlanggraph.graphimportStateGraphfromlanggraph.graph.messageimportadd_messagesfromtypingimportAnnotated,TypedDictfromoperatorimportaddclassState(TypedDict):messages:Annotated[list[AnyMessagel,add_messages]list_field:Annotated[list[int],add]extra_field:intdefnodel(state:State):new_messageAIMessage(Hello!)return{messages:[new_message],list_field:[10],extra_field:10}defnode2(state:State):new_messageAIMessage(LangGraph!)return{messages:[new_messagel,list_field:[20],extra_field:20}graph(StateGraph(State).add_node(node1,node1).add_node(node2,node2).set_entry_point(node1).add_edge(node1,node2).compile())input_message{role:user,content:Hi}resultgraph.invoke({messages:[input_message],list_field:[1,2,3]})print(result)在LangGraph的应用当中State通常都会要保存聊天消息。为此LangGaph中还提供了一个langgraph.graph.MessagesState可以用来快速保存消息。他的声明方式就是这样的:classMessagesState(TypedDict):messages:Annotated[list[AnyMessage],add_messages然后对于Messages也可以用序列化的方式来声明例如下面两种方式都是可以的{messages: [HumanMessage(contentmessage)]} {messages:[{type:human,content:message}]2、Node节点Node是图中的一个处理数据的节点。也有以下几个需要注意的地方:在LangGraph中Node通常是一个Python的函数它接受一个State对象作为输入返回一个State对象作为输出。每个Node都有一个唯一的名称通常是一个字符串。如果没有提供名称LangGraph会自动生成一个和函数名一样的名称。在具体实现时通常包含两个具体的参数第一个是State这个是必选的。第二个是一个可选的配置项config。这里面包含了一些节点运行的配置参数。LangGraph对每个Node提供了缓存机制。只要Node的传入参数相同。LangGraph就会优先从缓存当中获取Node的执行结果。从而提升Node的运行速度。importtimefromtypingimportTypedDictfromlangchain_core.runnablesimportRunnableConfigfromlanggraph.constantsimportSTART,ENDfromlanggraph.graphimportStateGraphfromlanggraph.typesimportCachePolicyfromlanggraph.cache.memoryimportInMemoryCache#是langgraph中的,而不是langchain中的.#配置状态classState(TypedDict):number:intuser_id:str配置信息#classConfigSchema(TypedDict):user_id:strdefnode_1(state:State,config:RunnableConfig):time.sleep(3)user_idconfig[configurable][user_id]return{number:state[number]1,user_id:user_id}builderStateGraph(State,config_schemaConfigSchema)#Node缓存5秒builder.add_node(node1,node_1,cache_policyCachePolicy(ttl5))builder.add_edge(START,node1)]builder.add_edge(START,nodel)builder.add_edge(nodel,IEND)graphbuilder.compile(cacheInMemoryCache())print(graph.invoke((number:5},config(configurable:(user_id:123}},stream modeupdates))# [{nodel:{number:6, user_id:123}}]# node入参相同就会走缓存print(graph.invoke((number:5},config(configurable:(user_id:456},stream_modeupdates))# [nodel: {number:6, user_id: 123}, metadata_: (cached:True}}]对于NodeLangGraph除了提供缓存机制还提供了重试机制。可以针对单个节点指定例如:fromlanggraph.typesimportRetryPolicybuilder.add_node(node1,node_1,retryRetryPolicy(max_attempts4))另外也可以针对某一次任务调用指定例如print(graph.invoke(xxxxx,config{recursion_limit:25}))3、Edge边在Graph图中通过Edge(边)把Node(节点)连接起来从而决定State应该如何在Graph中传递。LangGraph中也提供了非常灵活的构建方式。普通Edge和EntryPointEdge通常是用来把两个Node连接起来形成逻辑处理路线例如 graph add edge(node. 1node-22)。LangGraph中提供了两个默认的NodeSTART和END用来作为Graph的入口和出口。同时也可以自行指定EntryPoint。例如builder StateGraph(State)builder.set_entry_point(“node1”)builder.set_finish_point(“node2”)条件Edge和EntryPoint我们也可以添加带有条件判断的Edge和EntryPoint用来动态构建更复杂的工作流程。具体实现时可以指定一个函数函数的返回值就可以是下一个Node的名称。fromtypingimportTypedDictfromlangchain_core.runnablesimportRunnableConfigfromlanggraph.constantsimportSTART,ENDfromlanggraph.graphimportStateGraph# 配置状态classState(TypedDict):number:intdefnode_1(state:State,config:RunnableConfig):return{number:state[number]1}builderStateGraph(State)#Node缓存5秒builder.add_node(node1,node_1)defrouting_func(state:State)-str:ifstate[number]5:returnnodelelse:returnEND builder.add_edge(node1,END)builder.add_conditional_edges(START,routing_func)graphbuilder.compile()print(graph.invoke({number:7})){‘number’: 8}#补充看一下Graph的结构fromIPython.displayimportImage,display display(Image(graph.get_graph().draw_mermaid_png()))另外如果不想在路由函数中写入过多具体的节点名称也可以在函数中退回一个自定义的结果然后将这个结果解析到某一个具体的Node上。例如defrouting_func(state:State)-bool:ifstate[number]5:returnTrueelse:returnFalsebuilder.add_conditional_edges(START,routing_func,{True:node_a,False:node_b})Send动态路由在条件边中如果希望一个Node后同时路由到多个Node就可以返回Send动态路由的方式实现。Send对象可传入两个参数第一个是下一个Node的名称第二个是Node的输入。fromoperatorimportaddfromtypingimportTypedDict,Annotatedfromlanggraph.constantsimportSTART,ENDfromlanggraph.graphimportStateGraphfromlanggraph.typesimportSend#配置状态classState(TypedDict):messages:Annotated[list[str],add]classPrivateState(TypedDict):msg:strdefnode_1(state:PrivateState)-State:resstate[msg]!return{messages:[res]}builderStateGraph(State)# Node缓存5秒builder.add_node(node1,node_1)defrouting_func(state:State):result[]formessageinstate[messages]:result.append(Send(nodel,{msg:message}))returnresult#通过路由函数将消息中每个字符串分别传入node1处理。builder.add_conditional_edges(START,routing_func,[node1])builder.add_edge(node1,END)graphbuilder.compile()print(graph.invoke((messages:[hello,world,hello,graph]}))#{messages: [hello, world, hello, ‘graph, hello!, ‘world!, hello!, graph!])#补充看一下Graph的结构fromIPython.displayimportImage,display display(Image(graph.get_graph().draw_mermaid_png()))Command命令通常Graph中一个典型的业务步骤是State进入一个Node处理。在Node中先更新State状态然后再通过Edges传递给下一个Node。如果希望将这两个步骤合并为一个命令那么还可以使用Command命。fromoperatorimportaddfromtypingimportTypedDict,Annotatedfromlanggraph.constantsimportSTART,ENDfromlanggraph.graphimportStateGraphfromlanggraph.typesimportCommand#配置状态classState(TypedDict):messages:Annotated[list[str],add]defnode_1(state:State):new_message[]formessageinstate[messages]:new_message.append(message!)returnCommand(gotoEND update{messages:new_message})builderStateGraph(State)builder.add_node(node1,node_1)# node1中通过Command同时集成了更新State和指定下个Nodebuilder.add_edge(START,node1)graphbuilder.compile()print(graph.invoke({messages:[hello,world,hello,graph]}))# (messages: [hello, ‘world, hello, ‘graph, hello!, ‘world!, hello!, ‘graph!])4、子图在LangGraph中一个Graph除了可以单独使用还可以作为一个Node嵌入到另一个Graph中。这种用法就称为子图。通过子图我们可以更好的重用Graph构建更复杂的工作流。尤其在构建多Agent时非常有用。在大型项目中通常都是由一个团队专门开发Agent再通过其他团队来完整Agent整合。使用子图时基本和使用Node没有太多的区别。唯一需要注意的是当触发了SubGraph代表的Node后实际上是相当于重新调用了一次subgraph.invoke(state)方法。# subgraph与graph使用相同StatefromoperatorimportaddfromtypingimportTypedDict,Annotatedfromlanggraph.constantsimportENDfromlanggraph.graphimportStateGraph,MessagesState,STARTclassState(TypedDict):messages:Annotated[list[str],add]# Subgraphdefsub_node_1(state:State)-MessagesState:return{messages:[response from subgraph]}subgraph_builderStateGraph(State)subgraph_builder.add_node(sub_node_1,sub_node_1)subgraph_builder.add_edge(START,sub_node_1)subgraph_builder.add_edge(sub_node_1,END)subgraphsubgraph_builder.compile()# Parent graphbuilderStateGraph(State)builder.add_node(subgraph_node,subgraph)builder.add_edge(START,subgraph_node)builder.add_edge(subgraph_node,END)graphbuilder.compile()print(graph.invoke({messages:[hello subgraph]}))#结果hello subgraph会出现两次。这是因为在subgraph_ node中默认调用了一次subgraph.invoke(state)方法。主图里也调用了一次invoke.这就会往state中添加两次语句#{messages: [hello subgraph, hello subgraph, response from subgraph]}5、图的Stream支持和调用大模型相似Graph除了可以通过invoke方法进行直接调用外也支持通过stream()方法进行流式调用。不过大模型的流式调用是依次返回大模型响应的Token。而Graph的流式输出则是依次返回State的数据处理步骤。graph提供了stream()方法进行同步的流式调用也提供了astream()方法进行异步的流式调用。forchunkingraph.stream({messages:[hello subgraph]},stream_modedebug):print(chunk)# (subgraph_node: {messages: [hello subgraph, response from subgraph]}}LangGraph支持几种不同的stream mode:values:在图的每一步之后流式传输状态的完整值.updates:在图的每一步之后将更新内容流式传输到状态。如果在同:步骤中进行了多次更新(例如运行了多个节点)这些更新将分别进行流式传输。custom:从图节点内部流式传输自定义数据。通常用于调试。messages:从任何调用大语言模型(LLM)的图节点中流式传输二元组(LLM的Token元数据)debug:在图的执行过程中尽可能多地传输信息。用得比较少。values、updates、debug输出模式使用之前案例验证就能很快感受到其中的区别。messages输出模式由于在之前案例中并没有调用大模型所以不会有输出结果。而custom输出模式可以自定义输出内容。在Node节点内或者Tools工具内通过get_stream_writer()方法获取一个StreamWriter对象然后使用write()方法将自定义数据写入流中。fromtypingimportTypedDictfromlanggraph.configimportget_stream_writerfromlanggraph.graphimportStateGraph,STARTclassState(TypedDict):query:stranswer:strdefnode(state:State):writerget_stream_writer()writer({自定义key:“在节点内返回自定义信息})return{answer:some data}graph(StateGraph(state].add_node(node).add_edge(START,node).compile())inputs{query:example}# Usageforchunkingraph.stream(inputs,stream_modecustom):print(chunk)# {自定义key:在节点内返回自定义信息}最后在langChain中构建LLM对象时大都支持desable_streaming属性禁止流式输出。例如:llmChatOpenAI(model,disable_streamingTrue)llmChatOpenAI(model,disable_streamingTrue)五、总结在这一章节我们详细演练了LangGraph中的Graph构建以及工作方式。可以看到Graph图的构建非常灵活我们可以自由地构建各种复杂的图结构。即使是没有与大模型交互的图也可以通过LangGraph来构建。这对于处理传统任务也是非常有用的。当然LangGraph中的图还是要有大模型的加持才能更好的体现他的强大之处。下一章节我们就着重去演练大模型加持下的LangGraph。在这里不妨回顾一下LangChain中的Chain是如何构建的并与Graph做一下对比。可以看到这两个框架都是着眼于将多个独立的功能模块组合进行调度、组合形成复杂的智能体。只不过LangChain使用的是Chain的方式而LangGraph是使用Graph的方式。或许这样能够更好的体会到为什么LangGraph是LangChain的一个子项目而不是一个独立的框架了。 LangGraph原理逻辑-深度理解LangGraph核心-Graph（四）
深度理解LangGraph核心-Graph一、 理解什么是Graph图1、State状态2、Node节点3、Edge边4、子图5、图的Stream支持五、总结在了解了LangGraph中如何构建Agent智能体之后接下来就要进入LangGraph的重头戏Graph了。Graph是LangGraph的核心它以有向无环图的方式来整合多个Agent构建更复杂的Agent大模型应用形成更复杂的工作流。并且提供了很多产品级的特性保证这些应用可以更稳定高效的执行。一、 理解什么是Graph图Graph是LangGraph的基本构建模块它是一个有向无环图(DAG)用于描述任务之间的依赖关系。主要包含三个基本的元素:State:在整个应用当中共享的一种数据结构。Node:一个处理数据的节点。LangGraph中通常是一个Python的函数以State为输入经过一些操作后返回更新后的State。Edge:表示Node之前的依赖关系。LangGraph中通常也是一个Python函数根据当前State来决定接下来执行哪个Node。接下来用一个最简单的案例来看一下Graph的基本用法。#安装依赖 pip install -U langgraphfromtypingimportTypedDictfromlanggraph.constantsimportEND,STARTfromlanggraph.graphimportStateGraphclassInputState(TypedDict):user_input:strclassoutputState(TypedDict):graph_output:strclassOverallState(TypedDict):foo:struser_input:strgraph_output:strclassPrivateState(TypedDict):bar:strdefnode_1(state:InputState)-OverallState:# Write to OverallStatereturn{foo:state[user_input]博客}defnode_2(state:OverallState)-PrivateState:# Read from OverallState, write to PrivateStatereturn{bar:state[foo]非常}defnode_3(state:PrivateState)-OutputState:# Read from PrivateState, write to OutputStatereturn{graph_output:state[bar]靠谱}# 构建图builderStateGraph(OverallState,inputInputState,outputOutputState)# 添加Nodebuilder.add_node(node_1,node_1)builder.add_node(node_2,node_2)builder.add_node(node_3,node_3)#添加Edgebuilder.add_edge(START,node_1)builder.add_edge(node_1,node_2)builder.add_edge(node_2,node_3)builder.add_edge(node_3,END)#编译图graphbuilder.compile()#调用图graph.invoke({user_input:wuyuanshun}){‘graph_output’:‘wuyuanshun博客非常靠谱’}这个案例当中请求的参数从固定的START传入依次经过三个节点处理每个节点的处理结果都会被保存到不同的state当中最后进入到END节点结束。如果你觉得这个流程不够明显还可以直接用真正的图来看看请求是如何夕上理的。1、State状态State是所有节点共享的状态它是一个字典包含了所有节点的状态。有几个需要注意的地方:State形式上可以是TypedDict字典也可以是Pydantic中的一个BaseModel。例如:frompydanticimportBaseModel# The overall state of the graph (this is the public state shared across nodes)classOverallState(BaseModel):a:str这两种实现本质上没有太多的区别。State中定义的属性通常不需要指定默认值。如果需要默认值可以通过在START节点后定义一个node来指定默认值。defnode(state:OverallState):return{a:goodbye}State中的属性除了可以修改值之外也可以定义一些操作。来旨定如何更新State中的值。例如fromlanggraph.graph.messageimportadd_messagesclassState(TypedDict):messages:Annotated[list[AnyMessage],add_messages]list_field:Annotated[list[int],add]extra_field:int此时如果有一个node返回了State中更新的值那么messages和lis_field的值就会添加到原有的旧集合中而extra_field的值则会被替换。fromlangchain_core.messagesimportAnyMessage,AIMessagefromlanggraph.graphimportStateGraphfromlanggraph.graph.messageimportadd_messagesfromtypingimportAnnotated,TypedDictfromoperatorimportaddclassState(TypedDict):messages:Annotated[list[AnyMessagel,add_messages]list_field:Annotated[list[int],add]extra_field:intdefnodel(state:State):new_messageAIMessage(Hello!)return{messages:[new_message],list_field:[10],extra_field:10}defnode2(state:State):new_messageAIMessage(LangGraph!)return{messages:[new_messagel,list_field:[20],extra_field:20}graph(StateGraph(State).add_node(node1,node1).add_node(node2,node2).set_entry_point(node1).add_edge(node1,node2).compile())input_message{role:user,content:Hi}resultgraph.invoke({messages:[input_message],list_field:[1,2,3]})print(result)在LangGraph的应用当中State通常都会要保存聊天消息。为此LangGaph中还提供了一个langgraph.graph.MessagesState可以用来快速保存消息。他的声明方式就是这样的:classMessagesState(TypedDict):messages:Annotated[list[AnyMessage],add_messages然后对于Messages也可以用序列化的方式来声明例如下面两种方式都是可以的{messages: [HumanMessage(contentmessage)]} {messages:[{type:human,content:message}]2、Node节点Node是图中的一个处理数据的节点。也有以下几个需要注意的地方:在LangGraph中Node通常是一个Python的函数它接受一个State对象作为输入返回一个State对象作为输出。每个Node都有一个唯一的名称通常是一个字符串。如果没有提供名称LangGraph会自动生成一个和函数名一样的名称。在具体实现时通常包含两个具体的参数第一个是State这个是必选的。第二个是一个可选的配置项config。这里面包含了一些节点运行的配置参数。LangGraph对每个Node提供了缓存机制。只要Node的传入参数相同。LangGraph就会优先从缓存当中获取Node的执行结果。从而提升Node的运行速度。importtimefromtypingimportTypedDictfromlangchain_core.runnablesimportRunnableConfigfromlanggraph.constantsimportSTART,ENDfromlanggraph.graphimportStateGraphfromlanggraph.typesimportCachePolicyfromlanggraph.cache.memoryimportInMemoryCache#是langgraph中的,而不是langchain中的.#配置状态classState(TypedDict):number:intuser_id:str配置信息#classConfigSchema(TypedDict):user_id:strdefnode_1(state:State,config:RunnableConfig):time.sleep(3)user_idconfig[configurable][user_id]return{number:state[number]1,user_id:user_id}builderStateGraph(State,config_schemaConfigSchema)#Node缓存5秒builder.add_node(node1,node_1,cache_policyCachePolicy(ttl5))builder.add_edge(START,node1)]builder.add_edge(START,nodel)builder.add_edge(nodel,IEND)graphbuilder.compile(cacheInMemoryCache())print(graph.invoke((number:5},config(configurable:(user_id:123}},stream modeupdates))# [{nodel:{number:6, user_id:123}}]# node入参相同就会走缓存print(graph.invoke((number:5},config(configurable:(user_id:456},stream_modeupdates))# [nodel: {number:6, user_id: 123}, metadata_: (cached:True}}]对于NodeLangGraph除了提供缓存机制还提供了重试机制。可以针对单个节点指定例如:fromlanggraph.typesimportRetryPolicybuilder.add_node(node1,node_1,retryRetryPolicy(max_attempts4))另外也可以针对某一次任务调用指定例如print(graph.invoke(xxxxx,config{recursion_limit:25}))3、Edge边在Graph图中通过Edge(边)把Node(节点)连接起来从而决定State应该如何在Graph中传递。LangGraph中也提供了非常灵活的构建方式。普通Edge和EntryPointEdge通常是用来把两个Node连接起来形成逻辑处理路线例如 graph add edge(node. 1node-22)。LangGraph中提供了两个默认的NodeSTART和END用来作为Graph的入口和出口。同时也可以自行指定EntryPoint。例如builder StateGraph(State)builder.set_entry_point(“node1”)builder.set_finish_point(“node2”)条件Edge和EntryPoint我们也可以添加带有条件判断的Edge和EntryPoint用来动态构建更复杂的工作流程。具体实现时可以指定一个函数函数的返回值就可以是下一个Node的名称。fromtypingimportTypedDictfromlangchain_core.runnablesimportRunnableConfigfromlanggraph.constantsimportSTART,ENDfromlanggraph.graphimportStateGraph# 配置状态classState(TypedDict):number:intdefnode_1(state:State,config:RunnableConfig):return{number:state[number]1}builderStateGraph(State)#Node缓存5秒builder.add_node(node1,node_1)defrouting_func(state:State)-str:ifstate[number]5:returnnodelelse:returnEND builder.add_edge(node1,END)builder.add_conditional_edges(START,routing_func)graphbuilder.compile()print(graph.invoke({number:7})){‘number’: 8}#补充看一下Graph的结构fromIPython.displayimportImage,display display(Image(graph.get_graph().draw_mermaid_png()))另外如果不想在路由函数中写入过多具体的节点名称也可以在函数中退回一个自定义的结果然后将这个结果解析到某一个具体的Node上。例如defrouting_func(state:State)-bool:ifstate[number]5:returnTrueelse:returnFalsebuilder.add_conditional_edges(START,routing_func,{True:node_a,False:node_b})Send动态路由在条件边中如果希望一个Node后同时路由到多个Node就可以返回Send动态路由的方式实现。Send对象可传入两个参数第一个是下一个Node的名称第二个是Node的输入。fromoperatorimportaddfromtypingimportTypedDict,Annotatedfromlanggraph.constantsimportSTART,ENDfromlanggraph.graphimportStateGraphfromlanggraph.typesimportSend#配置状态classState(TypedDict):messages:Annotated[list[str],add]classPrivateState(TypedDict):msg:strdefnode_1(state:PrivateState)-State:resstate[msg]!return{messages:[res]}builderStateGraph(State)# Node缓存5秒builder.add_node(node1,node_1)defrouting_func(state:State):result[]formessageinstate[messages]:result.append(Send(nodel,{msg:message}))returnresult#通过路由函数将消息中每个字符串分别传入node1处理。builder.add_conditional_edges(START,routing_func,[node1])builder.add_edge(node1,END)graphbuilder.compile()print(graph.invoke((messages:[hello,world,hello,graph]}))#{messages: [hello, world, hello, ‘graph, hello!, ‘world!, hello!, graph!])#补充看一下Graph的结构fromIPython.displayimportImage,display display(Image(graph.get_graph().draw_mermaid_png()))Command命令通常Graph中一个典型的业务步骤是State进入一个Node处理。在Node中先更新State状态然后再通过Edges传递给下一个Node。如果希望将这两个步骤合并为一个命令那么还可以使用Command命。fromoperatorimportaddfromtypingimportTypedDict,Annotatedfromlanggraph.constantsimportSTART,ENDfromlanggraph.graphimportStateGraphfromlanggraph.typesimportCommand#配置状态classState(TypedDict):messages:Annotated[list[str],add]defnode_1(state:State):new_message[]formessageinstate[messages]:new_message.append(message!)returnCommand(gotoEND update{messages:new_message})builderStateGraph(State)builder.add_node(node1,node_1)# node1中通过Command同时集成了更新State和指定下个Nodebuilder.add_edge(START,node1)graphbuilder.compile()print(graph.invoke({messages:[hello,world,hello,graph]}))# (messages: [hello, ‘world, hello, ‘graph, hello!, ‘world!, hello!, ‘graph!])4、子图在LangGraph中一个Graph除了可以单独使用还可以作为一个Node嵌入到另一个Graph中。这种用法就称为子图。通过子图我们可以更好的重用Graph构建更复杂的工作流。尤其在构建多Agent时非常有用。在大型项目中通常都是由一个团队专门开发Agent再通过其他团队来完整Agent整合。使用子图时基本和使用Node没有太多的区别。唯一需要注意的是当触发了SubGraph代表的Node后实际上是相当于重新调用了一次subgraph.invoke(state)方法。# subgraph与graph使用相同StatefromoperatorimportaddfromtypingimportTypedDict,Annotatedfromlanggraph.constantsimportENDfromlanggraph.graphimportStateGraph,MessagesState,STARTclassState(TypedDict):messages:Annotated[list[str],add]# Subgraphdefsub_node_1(state:State)-MessagesState:return{messages:[response from subgraph]}subgraph_builderStateGraph(State)subgraph_builder.add_node(sub_node_1,sub_node_1)subgraph_builder.add_edge(START,sub_node_1)subgraph_builder.add_edge(sub_node_1,END)subgraphsubgraph_builder.compile()# Parent graphbuilderStateGraph(State)builder.add_node(subgraph_node,subgraph)builder.add_edge(START,subgraph_node)builder.add_edge(subgraph_node,END)graphbuilder.compile()print(graph.invoke({messages:[hello subgraph]}))#结果hello subgraph会出现两次。这是因为在subgraph_ node中默认调用了一次subgraph.invoke(state)方法。主图里也调用了一次invoke.这就会往state中添加两次语句#{messages: [hello subgraph, hello subgraph, response from subgraph]}5、图的Stream支持和调用大模型相似Graph除了可以通过invoke方法进行直接调用外也支持通过stream()方法进行流式调用。不过大模型的流式调用是依次返回大模型响应的Token。而Graph的流式输出则是依次返回State的数据处理步骤。graph提供了stream()方法进行同步的流式调用也提供了astream()方法进行异步的流式调用。forchunkingraph.stream({messages:[hello subgraph]},stream_modedebug):print(chunk)# (subgraph_node: {messages: [hello subgraph, response from subgraph]}}LangGraph支持几种不同的stream mode:values:在图的每一步之后流式传输状态的完整值.updates:在图的每一步之后将更新内容流式传输到状态。如果在同:步骤中进行了多次更新(例如运行了多个节点)这些更新将分别进行流式传输。custom:从图节点内部流式传输自定义数据。通常用于调试。messages:从任何调用大语言模型(LLM)的图节点中流式传输二元组(LLM的Token元数据)debug:在图的执行过程中尽可能多地传输信息。用得比较少。values、updates、debug输出模式使用之前案例验证就能很快感受到其中的区别。messages输出模式由于在之前案例中并没有调用大模型所以不会有输出结果。而custom输出模式可以自定义输出内容。在Node节点内或者Tools工具内通过get_stream_writer()方法获取一个StreamWriter对象然后使用write()方法将自定义数据写入流中。fromtypingimportTypedDictfromlanggraph.configimportget_stream_writerfromlanggraph.graphimportStateGraph,STARTclassState(TypedDict):query:stranswer:strdefnode(state:State):writerget_stream_writer()writer({自定义key:“在节点内返回自定义信息})return{answer:some data}graph(StateGraph(state].add_node(node).add_edge(START,node).compile())inputs{query:example}# Usageforchunkingraph.stream(inputs,stream_modecustom):print(chunk)# {自定义key:在节点内返回自定义信息}最后在langChain中构建LLM对象时大都支持desable_streaming属性禁止流式输出。例如:llmChatOpenAI(model,disable_streamingTrue)llmChatOpenAI(model,disable_streamingTrue)五、总结在这一章节我们详细演练了LangGraph中的Graph构建以及工作方式。可以看到Graph图的构建非常灵活我们可以自由地构建各种复杂的图结构。即使是没有与大模型交互的图也可以通过LangGraph来构建。这对于处理传统任务也是非常有用的。当然LangGraph中的图还是要有大模型的加持才能更好的体现他的强大之处。下一章节我们就着重去演练大模型加持下的LangGraph。在这里不妨回顾一下LangChain中的Chain是如何构建的并与Graph做一下对比。可以看到这两个框架都是着眼于将多个独立的功能模块组合进行调度、组合形成复杂的智能体。只不过LangChain使用的是Chain的方式而LangGraph是使用Graph的方式。或许这样能够更好的体会到为什么LangGraph是LangChain的一个子项目而不是一个独立的框架了。