AI创想

标题: langgraph子图 [打印本页]

作者: AI小编 时间: 10 小时前
标题: langgraph子图
作者：CSDN博客
      一个图可以实现特定的工作流，同时也可以复用到其他更大的图（可称之为父图）中作为子图，借助子图除了可以实现逻辑重用，同时也是多智能体的物质基础。
      本文首先根据是否共享状态数据说明父图和子图如何集成，然后对父子图集成情况下的持久化进行讲解，最后说明如何查看子图的状态。
   1.集成子图

      父图集成子图需要考虑两种情况，一种是父图和子图的中状态有相同的频道（健），另一种是父子图状态中无相同的频道。
   1.1父子图状态相同

      父子图状态相同时，可以直接把子图作为父图中的一个节点，进入子图时携带了共享的频道，在子图中可是设置和使用自己只有的频道，子图完成专有频道的更新后再回到父图，父图并不能感知到；当然父图也可以有自己专有的频道，传入到子图时，子图对父图的专有频道页无感。
      以下代码定义一个子图，子图与父图共享频道v，在子图中增加了自己的专有频道 delta，子图中把delta与父图传入的v相加，并更新v的值。

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
#子图状态。与父图共享v频道，delta为专有频道
class SubState(TypedDict):
v: int
delta: int
def node1(state: SubState):#子图节点1中给delta赋值为10
return {'delta': 10}
def node2(state: SubState):#子图节点2中执行v+delta
return {'v': state['v'] + state['delta']}
#构建子图sub_graph_builder = StateGraph(SubState)
sub_graph_builder.add_node("node1", node1)
sub_graph_builder.add_node("node2", node2)
sub_graph_builder.add_edge(START, "node1")
sub_graph_builder.add_edge("node1", "node2")
sub_graph_builder.add_edge("node2", END)
sub_graph = sub_graph_builder.compile()
#父图状态。与子图共享 v频道
class State(TypedDict):
v: int
def node(state: State):#在节点中完成v的平方计算
return {'v': state['v'] * state['v']}
#构建父图，把子图作为一个节点
graph_builder = StateGraph(State)
graph_builder.add_node(node)
graph_builder.add_node("sub_graph", sub_graph)
graph_builder.add_edge(START, "node")
graph_builder.add_edge("node", "sub_graph")
graph_builder.add_edge("sub_graph", END)
graph = graph_builder.compile()
graph.invoke({'v':10})

运行输出为v*v + delta，具体结果如下：

{'v': 110}

   1.2父子图状态相异

      如果父子图无共享频道，比如在多智能体情况下，每个智能体保存自己专有的对话数据，此时不能把子图作为父图的节点，只能从父图一个节点内部调用子图，在调用前需要适配父图的状态为子图状态，并且把子图的状态适配为父图的状态后再更新状态。
      以下代码在子图中接收父图传入的参数，做3次方运算，把结果更新到子图频道v中。在父图的node节点中进行输入和输出的适配，具体代码如下：

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
#子图装填，base,exp和v均为专用频道
class SubState(TypedDict):
base: int
exp: int
v: int
def node1(state: SubState): #给exp赋值
return {'exp': 3}
def node2(state: SubState): #执行3次方运算并写入v频道
return {'v': pow(state['base'], state['exp'])}
#创建子图
sub_graph_builder = StateGraph(SubState)
sub_graph_builder.add_node("node1", node1)
sub_graph_builder.add_node("node2", node2)
sub_graph_builder.add_edge(START, "node1")
sub_graph_builder.add_edge("node1", "node2")
sub_graph_builder.add_edge("node2", END)
sub_graph = sub_graph_builder.compile()
#父图状态
class State(TypedDict):
x: int
def node(state: State):#在节点内部调用子图
ret = sub_graph.invoke({'base': state['x']}) #输入参数适配
return {'x': ret['v']} #返回参数适配
#构建父图，此时子图不会作为父图节点存在
graph_builder = StateGraph(State)
graph_builder.add_node(node)
graph_builder.add_edge(START, "node")
graph_builder.add_edge("node", END)
graph = graph_builder.compile()
graph.invoke({'x':10}, subgraphs=True)

   2.查看子图状态

      在使用短期记忆时，可以查看子图的状态。在langgraph中仅可以在子图被中断时才可以查看状态，从中断恢复后不能再访问子图状态。
      以下代码中，在子图的node1节点中断等待用户确认，此时可以通过调用图的get_state方法区查看子图的状态，此时除需要传入config外，还必须传入subgraphs=True。

from langgraph.graph import START, StateGraph
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from typing_extensions import TypedDict
#父图和子图共享频道name和answer
class State(TypedDict):
name: str
answer: str
# 子图节点中产生中断，等待用户确认
def node1(state: State):
value = interrupt(f"Is your name right: {state['name']}")
if value=='y':
      return {"answer": f"Your name is {state['name']}"}
else:
      return {"answer": f"Your name is incorrect, please confirm"}
#构建子图
subgraph_builder = StateGraph(State)
subgraph_builder.add_node(node1)
subgraph_builder.add_edge(START, "node1")
subgraph = subgraph_builder.compile()
# 构建父图
builder = StateGraph(State)
builder.add_node("subgraph", subgraph)
builder.add_edge(START, "subgraph")
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "1"}}
graph.invoke({"name": "davi"}, config)
subgraph_state = graph.get_state(config, subgraphs=True).tasks[0].state
print(subgraph_state)
graph.invoke(Command(resume="y"), config)

输出如下，可以看到子图当前的状态：

StateSnapshot(values={'name': 'davi'}, next=('node1',), config={'configurable': {'thread_id': '1', 'checkpoint_ns': 'subgraph:2cf7ed04-daf3-458c-3807-5afd72314233', 'checkpoint_id': '1f0a8d84-4321-61ce-8000-687a815a3c6b', 'checkpoint_map': {'': '1f0a8d84-4302-6e00-8000-7dd4b6147e42', 'subgraph:2cf7ed04-daf3-458c-3807-5afd72314233': '1f0a8d84-4321-61ce-8000-687a815a3c6b'}}}, metadata={'source': 'loop', 'step': 0, 'parents': {'': '1f0a8d84-4302-6e00-8000-7dd4b6147e42'}}, created_at='2025-10-14T08:32:05.901944+00:00', parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': 'subgraph:2cf7ed04-daf3-458c-3807-5afd72314233', 'checkpoint_id': '1f0a8d84-431a-69f4-bfff-ba93d1a32e32', 'checkpoint_map': {'': '1f0a8d84-4302-6e00-8000-7dd4b6147e42', 'subgraph:2cf7ed04-daf3-458c-3807-5afd72314233': '1f0a8d84-431a-69f4-bfff-ba93d1a32e32'}}}, tasks=(PregelTask(id='6b1b1803-9ca3-736a-d74c-a25353e7795d', name='node1', path=('__pregel_pull', 'node1'), error=None, interrupts=(Interrupt(value='Is your name right: davi', id='0a7f4eeaa60b8b3a5254c36559bf39f5'),), state=None, result=None),), interrupts=(Interrupt(value='Is your name right: davi', id='0a7f4eeaa60b8b3a5254c36559bf39f5'),))

      从中断恢复后，再次查看子图状态将报错。
   3.子图流式输出

      如果调用图时，希望子图也以流式输出，则仅需要在图的graph方法中传入 subgraphs=True参数即可。

for chunk in graph.stream(
{"foo": "foo"},
stream_mode="updates",
subgraphs=True,
):
print(chunk)

原文地址：https://blog.csdn.net/zhangbaolin/article/details/153262563

欢迎光临 AI创想 (https://www.llms-ai.com/)