langgraph/how-tos/subgraph-persistence/

[giscus[bot]]

langgraph/how-tos/subgraph-persistence/

Build language agents as graphs

https://langchain-ai.github.io/langgraph/how-tos/subgraph-persistence/

[obradovicma]

Hi guys
I‘m having issues with the subgraph‘s state when I re-run the graph.

I have a multi-agent architecture with a supervisor and several subgraphs. Subgraphs are compiled and added directly as parent nodes.

When I run the graph for the first time, it works fine. But I want the subgraph‘s state to be present for upcoming runs, since the subgraphs don’t share all the keys with the parent state. When I run it again, the subgraph state is gone unfortunately. It only persists the parent‘s state. I use the MemorySaver as Checkpointer.

What did I miss? Thanks in advance

[obradovicma]

Note that I‘m passing a new User Message as input for every run. I don‘t use human in the loop because of the lack of support for asynchronous checkpointing -> updating the state before the run is not an option for me.

[crisbertal]

Hi! I'm having the exact same issue trying to maintain state in subgraphs. Have you found a solution?

Thanks in advance

[obradovicma]

Hi @crisbertal ,
Unfortunately, I couldn't solve it. I'm waiting for someone to assist.

But what i could find out: The subgraph's checkpoint is there after execution. When I inspect the state after invokation, it appears in the list of state snapshots. I guess the only problem is that it doesn't resume the state for the subgraph when invoked again. Maybe subgraph feature is only made for human-in-the-loop interactions with interrupts. I don't know.

@vbarda, could you perhaps help? I’d really appreciate it.

[eyurtsev]

@obradovicma could you provide a minimal reproducible example of what you're trying to do? It'll make it so much easier for a maintainer to respond quickly if we can actually run the code.

[obradovicma]

Sure, here is a minimal example.

After the first invoke, I expect to have state["bar"] in the subgraph's state already filled. But instead, it's not there anymore.

I don't use human-in-the-loop, or any fancy interrupt.

# Code demonstrating a parent and subgraph setup with LangGraph

from langgraph.graph import START, StateGraph
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict, Union, Optional, Any, Sequence

from langchain_core.runnables import RunnableConfig
from langgraph.types import All, StreamMode


# Subgraph

class SubgraphState(TypedDict):
    foo: str  # Note that this key is shared with the parent graph state
    bar: str


def subgraph_node_1(state: SubgraphState):
    return {"bar": "bar" + state.get("bar", "")}  # After the first run, "bar" is not there anymore


def subgraph_node_2(state: SubgraphState):
    # This node uses a state key ('bar') that is only available in the subgraph
    # and sends an update on the shared state key ('foo')
    return {"foo": state["foo"] + state["bar"]}


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()


# Parent graph

class State(TypedDict):
    foo: str


def node_1(state: State):
    return {"foo": "hi! " + state["foo"]}


builder = StateGraph(State)
builder.add_node("node_1", node_1)
# Adding the compiled subgraph as a node to the parent graph
builder.add_node("node_2", subgraph)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")

checkpointer = MemorySaver()
# Pass checkpointer when compiling the parent graph
# LangGraph will automatically propagate the checkpointer to child subgraphs
graph = builder.compile(checkpointer=checkpointer)


def invoke_graph_with_subgraph(
        graph,
        input: Union[dict[str, Any], Any],
        config: Optional[RunnableConfig] = None,
        *,
        stream_mode: StreamMode = "values",
        output_keys: Optional[Union[str, Sequence[str]]] = None,
        interrupt_before: Optional[Union[All, Sequence[str]]] = None,
        interrupt_after: Optional[Union[All, Sequence[str]]] = None,
        debug: Optional[bool] = None,
        **kwargs: Any,
) -> Union[dict[str, Any], Any]:
    """
    Run the graph with a single input and config.

    Args:
        input: Input data for the graph. Can be a dictionary or any other type.
        config: Optional configuration for the graph run.
        stream_mode: Optional[str]. Stream mode for the graph run. Default is "values".
        output_keys: Optional. Output keys to retrieve from the graph run.
        interrupt_before: Optional. Nodes to interrupt the graph run before.
        interrupt_after: Optional. Nodes to interrupt the graph run after.
        debug: Optional. Enable debug mode for the graph run.
        **kwargs: Additional keyword arguments to pass to the graph run.

    Returns:
        The output of the graph run. If stream_mode is "values", it returns the latest output.
        If stream_mode is not "values", it returns a list of output chunks.
    """
    output_keys = output_keys if output_keys is not None else graph.output_channels
    if stream_mode == "values":
        latest: Union[dict[str, Any], Any] = None
    else:
        chunks = []
    for chunk in graph.stream(
            input,
            config,
            stream_mode=stream_mode,
            output_keys=output_keys,
            interrupt_before=interrupt_before,
            interrupt_after=interrupt_after,
            debug=debug,
            subgraphs=True,
            **kwargs,
    ):
        if stream_mode == "values":
            latest = chunk
        else:
            chunks.append(chunk)
    if stream_mode == "values":
        return latest
    else:
        return chunks


config = {"configurable": {"thread_id": "1"}}
result = invoke_graph_with_subgraph(graph, {"foo": "foo"}, config)
print(graph.get_state(config).values)
state_with_subgraph = [s for s in graph.get_state_history(config) if s.next == ("node_2",)]
subgraph_config = state_with_subgraph[0].tasks[0].state
print(subgraph_config)
print(graph.get_state(subgraph_config).values)

result = invoke_graph_with_subgraph(graph, {"foo": "foo"}, config)
print(graph.get_state(config).values)
state_with_subgraph = [s for s in graph.get_state_history(config) if s.next == ("node_2",)]
subgraph_config = state_with_subgraph[0].tasks[0].state
print(subgraph_config)
print(graph.get_state(subgraph_config).values)

And there is the output

{'foo': 'hi! foobar'}
{'configurable': {'thread_id': '1', 'checkpoint_ns': 'node_2:8e6fcb4a-8e9e-f923-5370-80cbba0c9512'}}
{'foo': 'hi! foobar', 'bar': 'bar'}
{'foo': 'hi! foobar'}
{'configurable': {'thread_id': '1', 'checkpoint_ns': 'node_2:0a54c5a4-94c7-6722-7b53-96107bf7409c'}}
{'foo': 'hi! foobar', 'bar': 'bar'}

[vvvin333]

I was surprised with that we can have several STARTs (and ENDs). So every subgraph can have it's own START and END.

[vermapratyush]

When a subgraph is invoked within a node as mentioned in add a node function that invokes the subgraph, does the state update/interrupts work the same way as mentioned in: how to view and update state in subgraphs

[vermapratyush]

I am wondering if there any side-effect of doing it one way or the other.