LangGraph--基础学习(Subgraphs 子图)

        上一篇详细介绍了人在回路中，下面应该是Breakpoints   断点、Time travel   时间旅行、Tools   工具这几个章节，但是我觉得这些都是调试使用的，使用技巧也是中断的概念，所以大家去官网自己学习，我这边直接进入子图学习，为后续的多智能体学习做准备。

        子图是一个图 ，它被用作另一个图中的节点 -这是应用于 LangGraph 的封装概念。子图允许您构建具有多个组件的复杂系统，这些组件本身就是图。

        

使用子图的一些原因是：

• 构建多智能体系统
• 当您希望在多个图表中重用一组节点时
• 当您希望不同的团队独立地处理图的不同部分时，您可以将每个部分定义为子图，并且只要遵守子图接口（输入和输出模式），就可以在不知道子图的任何细节的情况下构建父图

添加子图时，需要定义父图和子图的通信方式：

        共享状态模式 -父图和子图在其状态模式中具有共享状态键

        不同的状态模式 - 父模式和子模式中没有共享状态键

下面就详细介绍这两种状态模式

Shared state schemas  共享状态模式

        一种常见的情况是父图和子图通过模式中的共享状态键（通道）进行通信。例如，在多代理系统中，代理通常通过共享消息密钥进行通信。

       如果您的子图与父图共享状态键，您可以按照以下步骤将其添加到您的图中：

1. 定义子图工作流（下面示例中的 subgraph_builder）并对其进行编译
2. 定义父图形工作流时，将编译的子图形传递给 .add_node 方法

例子如下：

from typing_extensions import TypedDict
from langgraph.graph import StateGraph,START,END# 定义子图状态
class SubGraphState(TypedDict):foo:strbar:str# 定义子图节点
def subgraph_node1(state:SubGraphState):return{"bar":"bar"}def subgraph_node2(state:SubGraphState):return{"foo":state["foo"]+state["bar"]} # 这里需要注意，使用了子图独立的bar键，父图是没有这个键的# 构建子图
subgraph_builder = StateGraph(SubGraphState)
subgraph_builder.add_node("subgraph_node1", subgraph_node1)
subgraph_builder.add_node("subgraph_node2", subgraph_node2)
subgraph_builder.add_edge(START, "subgraph_node1")
subgraph_builder.add_edge("subgraph_node1", "subgraph_node2")
subgraph_builder.add_edge("subgraph_node2", END)subgraph = subgraph_builder.compile()# 定义父图
class ParentState(TypedDict):foo:str # 父图只定义了foo，没有bardef node_1(state:ParentState):return{"foo":" hello! "+ state["foo"]}# 构建父图
parent_graph = StateGraph(ParentState)
parent_graph.add_node("node_1", node_1)
parent_graph.add_node("node_2", subgraph)
parent_graph.add_edge(START, "node_1")
parent_graph.add_edge("node_1", "node_2")
parent_graph.add_edge("node_2", END)graph = parent_graph.compile()for chunk in graph.stream({"foo": "foo"},config):print(chunk)

{'node_1': {'foo': ' hello! foo'}}
{'node_2': {'foo': ' hello! foobar'}}

Different state schemas 不同的状态模式

        对于更复杂的系统，您可能希望定义与父图具有完全不同模式的子图（没有共享键）。例如，您可能希望为多代理系统中的每个代理保留一个私有消息历史记录。

        如果您的应用程序是这种情况，则需要定义一个调用子图的节点函数 。该函数需要在调用子图之前将输入（父）状态转换为子图状态，并在从节点返回状态更新之前将结果转换回父状态。

一层子图

from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START# 定义子图
class SubgraphState(TypedDict):# 这里定义的子图状态和父图的完全不同了bar: strbaz: strdef subgraph_node_1(state: SubgraphState):return {"baz": "baz"}def subgraph_node_2(state: SubgraphState):return {"bar": state["bar"] + state["baz"]}subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()# 定义父图
class ParentState(TypedDict):foo: strdef node_1(state: ParentState):return {"foo": "hi! " + state["foo"]}def node_2(state: ParentState):response = subgraph.invoke({"bar": state["foo"]})  # 在父图的节点函数内部调用子图，数据进行了转换才能调用return {"foo": response["bar"]}  builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()for chunk in graph.stream({"foo": "foo"}, subgraphs=True):print(chunk)

((), {'node_1': {'foo': 'hi! foo'}})
(('node_2:f916dc5a-e340-d03f-fde5-24825fb2c914',), {'subgraph_node_1': {'baz': 'baz'}})
(('node_2:f916dc5a-e340-d03f-fde5-24825fb2c914',), {'subgraph_node_2': {'bar': 'hi! foobaz'}})
((), {'node_2': {'foo': 'hi! foobaz'}})

二级子图

from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START, END
# 孙子图
class GrandChildState(TypedDict):my_grandchild_key: strdef grandchild_1(state: GrandChildState) -> GrandChildState:# NOTE: 这里是无法访问子图和父图的状态的return {"my_grandchild_key": state["my_grandchild_key"] + ", how are you"}grandchild = StateGraph(GrandChildState)
grandchild.add_node("grandchild_1", grandchild_1)grandchild.add_edge(START, "grandchild_1")
grandchild.add_edge("grandchild_1", END)grandchild_graph = grandchild.compile()# 子图
class ChildState(TypedDict):my_child_key: strdef call_grandchild_graph(state: ChildState) -> ChildState:# NOTE: 子图和孙子图的数据在这里进行交互grandchild_graph_input = {"my_grandchild_key": state["my_child_key"]}  grandchild_graph_output = grandchild_graph.invoke(grandchild_graph_input)return {"my_child_key": grandchild_graph_output["my_grandchild_key"] + " today?"}  child = StateGraph(ChildState)
child.add_node("child_1", call_grandchild_graph)  
child.add_edge(START, "child_1")
child.add_edge("child_1", END)
child_graph = child.compile()#父图
class ParentState(TypedDict):my_key: strdef parent_1(state: ParentState) -> ParentState:# NOTE: 父图不包含子图和父图return {"my_key": "hi " + state["my_key"]}def parent_2(state: ParentState) -> ParentState:return {"my_key": state["my_key"] + " bye!"}def call_child_graph(state: ParentState) -> ParentState:child_graph_input = {"my_child_key": state["my_key"]}  child_graph_output = child_graph.invoke(child_graph_input)return {"my_key": child_graph_output["my_child_key"]}  parent = StateGraph(ParentState)
parent.add_node("parent_1", parent_1)
parent.add_node("child", call_child_graph)  
parent.add_node("parent_2", parent_2)parent.add_edge(START, "parent_1")
parent.add_edge("parent_1", "child")
parent.add_edge("child", "parent_2")
parent.add_edge("parent_2", END)parent_graph = parent.compile()for chunk in parent_graph.stream({"my_key": "Bob"}, subgraphs=True):print(chunk)

((), {'parent_1': {'my_key': 'hi Bob'}})
(('child:991b3f8f-449a-e406-ef14-49951b36d4e6', 'child_1:62c9f423-eb44-d618-0693-fbc4489aef5f'), {'grandchild_1': {'my_grandchild_key': 'hi Bob, how are you'}})
(('child:991b3f8f-449a-e406-ef14-49951b36d4e6',), {'child_1': {'my_child_key': 'hi Bob, how are you today?'}})
((), {'child': {'my_key': 'hi Bob, how are you today?'}})
((), {'parent_2': {'my_key': 'hi Bob, how are you today? bye!'}})

添加持久性

你只需要在编译父图时提供checkpointer  。LangGraph 将自动将检查指针传播到子图。

from langgraph.graph import START, StateGraph
from langgraph.checkpoint.memory import InMemorySaver
from typing_extensions import TypedDictclass State(TypedDict):foo: str# Subgraphdef subgraph_node_1(state: State):return {"foo": state["foo"] + "bar"}subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph = subgraph_builder.compile()# Parent graphbuilder = StateGraph(State)
builder.add_node("node_1", subgraph)
builder.add_edge(START, "node_1")checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

如果你想让子图拥有自己的内存 ，你可以用 checkpointer=True 编译它。这在多代理系统中非常有用，如果您希望代理跟踪其内部消息历史记录：

subgraph_builder = StateGraph(...)
subgraph = subgraph_builder.compile(checkpointer=True)

查看子图状态

当您启用持久性时，您可以通过 graph.get_state（config） 检查图形状态 （检查点）。要查看子图状态，可以使用 graph.get_state(config, subgraphs=True) 。

子图状态只能在子图中断时查看。恢复图形后，将无法访问子图形状态。

from langgraph.graph import START, StateGraph
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
from typing_extensions import TypedDictclass State(TypedDict):foo: str# Subgraphdef subgraph_node_1(state: State):value = interrupt("Provide value:")return {"foo": state["foo"] + value}subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")subgraph = subgraph_builder.compile()# Parent graphbuilder = StateGraph(State)
builder.add_node("node_1", subgraph)
builder.add_edge(START, "node_1")checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)config = {"configurable": {"thread_id": "1"}}graph.invoke({"foo": ""}, config)
parent_state = graph.get_state(config)
subgraph_state = graph.get_state(config, subgraphs=True).tasks[0].state  # resume the subgraph
graph.invoke(Command(resume="bar"), config)

流子图输出

要在流输出中包含子图的输出，可以在父图的 .stream（） 方法中设置 subgraphs=True。这将从父图和任何子图流输出。

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