邯郸网站seo长沙小红书推广公司

邯郸网站seo,长沙小红书推广公司,网站建设哪些,站酷网素材图库排版LangGraph的核心主要是Graph#xff0c;它是⼀个有向⽆环图#xff0c;⽤于描述任务之间的依赖关系。 主要包含三个基本元素#xff1a;State#xff1a;一种数据结构Node#xff1a;处理数据的节点#xff0c;LangGraph中通常是一个python函数#xff0c;以State为…LangGraph的核心主要是Graph它是⼀个有向⽆环图⽤于描述任务之间的依赖关系。主要包含三个基本元素· State一种数据结构· Node处理数据的节点LangGraph中通常是一个python函数以State为输入经过一些操作处理后返回更新后的State· Edge在LangGraph中通常也是一个python函数作用是把两个Node连接起来形成逻辑处理路线。举个简单示例看下Graph的基本用法。先安装对应包pipinstalllanggraph代码fromtypingimportTypedDictfromlanggraph.constantsimportEND,STARTfromlanggraph.graphimportStateGraphclassInputState(TypedDict):user_input:strclassOutputState(TypedDict):graph_output:strclassGraphState(TypedDict):user_input:strgraph_output:strdefnode1(state:InputState)-OutputState:return{graph_output:state[user_input]这个是节点1}# 构件图builderStateGraph(GraphState)# 1. 添加节点到状态图指定节点名称和对应的处理函数builder.add_node(node1,node1)# 2. 添加起始节点到node1的边START - node1builder.add_edge(START,node1)# 3. 添加node1到结束节点的边node1 - ENDbuilder.add_edge(node1,END)# 4. 编译状态图生成可运行的图实例graphbuilder.compile()if__name____main__:initial_state{user_input:测试输入}# 运行图两种方式stream流式输出 / invoke直接获取最终结果# 方式1invoke直接获取最终状态final_stategraph.invoke(initial_state)print(ffinal_state的值{final_state})# 方式2stream流式遍历执行过程forstepingraph.stream(initial_state):fornode_name,node_outputinstep.items():print(f节点名称{node_name} 节点输出{node_output})构建出来的图是这样的自己想测验也可以在代码后边加这一段生成然后查看graph_structuregraph.get_graph()png_datagraph_structure.draw_mermaid_png()# 保存到本地文件withopen(langgraph_flow.png,wb)asf:f.write(png_data)这个是Graph的简单流程下边我们展开详细讲解下。StateState是所有节点共享的状态它是⼀个字典可以是TypedDict字典也可以是Pydantic中的⼀个BaseModel。例如fromtypingimportTypedDictfrompydanticimportBaseModelclassInputState1(TypedDict):user_input:strclassInputState2(BaseModel):user_input:str这两种都可以实现定义State。State中也可以定义一些操作来指定更新比如fromtypingimportTypedDict,AnnotatedfromoperatorimportaddclassInputState(TypedDict):user_input:strlist_input:Annotated[list[int],add]# Annotated为原始类型附加任意元数据不改变类型本身的本质这里如果node中有返回InputState中更新的值用list_input举例那对应的值会发生变化比如fromtypingimportTypedDict,Annotatedfromoperatorimportaddfromlanggraph.graphimportStateGraphfromlanggraph.constantsimportEND,STARTclassInputState(TypedDict):user_input:strlist_input:Annotated[list[int],add]# Annotated为原始类型附加任意元数据不改变类型本身的本质defnode1(state:InputState):return{list_input:[123],user_input:node1}defnode2(state:InputState):return{user_input:node2}graph(StateGraph(InputState).add_node(node1,node1)# 添加节点1.add_node(node2,node2)# 添加节点2.add_edge(START,node1)# 补充起始节点 - node1.add_edge(node1,node2)# node1 - node2.add_edge(node2,END)# node2 - 结束节点.compile()# 编译状态图)if__name____main__:initial_state{user_input:测试输入,list_input:[1,2,]}final_stategraph.invoke(initial_state)print(ffinal_state的值{final_state})# final_state的值{user_input: node2, list_input: [1, 2, 123]}可以看到list_input在node1中有进行操作打印出来的结果有变化。在LangGraph中State应用场景可以用于保存聊天消息比如我们在豆包中一个对话中往上滑看见的上一条问题NodeState中定义的属性通常不指定默认值而是在Node中进行指定并且Node通常是一个Python函数它接受⼀个State对象作为输⼊返回⼀个State对象作为输出。在具体实现时通常包含两个具体的参数第⼀个是State这个是必选的。第⼆个是⼀个可选的配置项config。并且LangGraph对每个Node提供了缓存机制。只要Node的传⼊参数相同LangGraph就会优先从缓存当中获取Node的执⾏结果。从⽽提升Node的运⾏速度。示例fromtypingimportTypedDictfromlangchain_core.runnablesimportRunnableConfigfromlanggraph.graphimportStateGraphfromlanggraph.constantsimportEND,STARTfromlanggraph.typesimportCachePolicyfromlanggraph.cache.memoryimportInMemoryCache#是langgraph中的,⽽不是langchain中的。classInputState(TypedDict):user_id:strnumber:intdefnode1(state:InputState,config:RunnableConfig)-InputState:user_idconfig[configurable][user_id]return{number:state[number]1,user_id:user_id}graph(StateGraph(InputState).add_node(node1,node1,cache_policyCachePolicy(ttl5))# 添加节点1Node缓存5秒.add_edge(START,node1)# 补充起始节点 - node1.add_edge(node1,END)# node1 - 结束节点.compile(cacheInMemoryCache())# 编译状态图)if__name____main__:initial_state{user_id:user_01,number:5}final_stategraph.invoke(inputinitial_state,config{configurable:{user_id:user_02}})print(ffinal_state的值{final_state})# final_state的值{user_id: user_02, number: 6}Edge在Graph图中通过Edge(边)把Node(节点)连接起来从⽽决定State应该如何在Graph中传递。类似于工作流中的箭头指向线。LangGraph中提供了两个默认的NodeSTART和END⽤来作为Graph的⼊⼝和出⼝。上边有普通Edge代码示例这里讲一下另一个操作set_entry_pointfromtypingimportTypedDict,Annotatedfromoperatorimportaddfromlanggraph.graphimportStateGraphclassInputState(TypedDict):user_input:strlist_input:Annotated[list[int],add]# Annotated为原始类型附加任意元数据不改变类型本身的本质defnode1(state:InputState):return{list_input:[123],user_input:node1}defnode2(state:InputState):return{user_input:node2}graph(StateGraph(InputState).add_node(node1,node1)# 添加节点1.add_node(node2,node2)# 添加节点2.set_entry_point(node1).add_edge(node1,node2).compile()# 编译状态图)if__name____main__:initial_state{user_input:测试输入,list_input:[1,2,]}final_stategraph.invoke(initial_state)print(ffinal_state的值{final_state})graph_structuregraph.get_graph()png_datagraph_structure.draw_mermaid_png()# 保存到本地文件withopen(langgraph_flow3.png,wb)asf:f.write(png_data)图片set_entry_point是 LangGraph中用于修改/指定图执行入口节点的方法但是比add_edge优先级更高若同时使用 add_edge(START, “node1”)和set_entry_point(“node2”)最终入口为 node2。如果遇到条件走不同节点怎么办可以写动态路由fromtypingimportTypedDict,Annotatedfromoperatorimportaddfromlanggraph.graphimportStateGraphfromlanggraph.constantsimportEND,STARTclassInputState(TypedDict):user_input:strlist_input:Annotated[list[int],add]# Annotated为原始类型附加任意元数据不改变类型本身的本质defnode1(state:InputState):return{list_input:[123],user_input:This is node1}defnode2(state:InputState):return{user_input:This is node2}defnode3(state:InputState):return{user_input:This is node3}defrouting_func(state:InputState):ifstate[user_input]111:returnnode1elifstate[user_input]222:returnnode2elifstate[user_input]333:returnnode3else:returnEND graph(StateGraph(InputState).add_node(node1,node1)# 添加节点1.add_node(node2,node2)# 添加节点2.add_node(node3,node3)# 添加节点3.add_conditional_edges(START,routing_func).compile()# 编译状态图)if__name____main__:initial_state{user_input:111,list_input:[1,2,]}final_stategraph.invoke(initial_state)print(ffinal_state的值{final_state})graph_structuregraph.get_graph()png_datagraph_structure.draw_mermaid_png()# 保存到本地文件withopen(langgraph_flow4.png,wb)asf:f.write(png_data)路线