langgraph/how-tos/subgraphs-manage-state/

[giscus[bot]]

langgraph/how-tos/subgraphs-manage-state/

Build language agents as graphs

https://langchain-ai.github.io/langgraph/how-tos/subgraphs-manage-state/

[apoorvkk]

Is it possible to replay from a node within a subggraph like shown here for top level nodes? https://langchain-ai.github.io/langgraph/how-tos/human_in_the_loop/time-travel/#replay-a-state

[apoorvkk]

In addition, is it possible to replay between two different sessions?

[isahers1]

I am not 100% sure what you mean by replay between two different sessions - but replaying from a node inside a subgraph is possible, we just added a PR to add this information to the docs: #1759

[apoorvkk]

ah perfect, your PR helped significantly! The second question was more to do with replaying between different script runs but that was easily solvable with external persistence like sqlite saver

[sivaTwks010928]

@isahers1
Let’s say I’m using interrupt from langgraph.types inside some nodes of a subgraph. I have two subgraphs, both of which contain nodes that use interrupt, and I’ve combined them into a parent graph (each subgraph is compiled into a node in the parent graph).

Now, suppose I want to "travel back in time" — meaning I want the user to be able to select a checkpoint to return to, and then resume execution from that point, potentially following a different path than the original one.

How can I implement this kind of behavior, especially considering that the subgraphs are already compiled into the parent graph?

[sivaTwks010928]

@apoorvkk can you help me? did you find the answer for your question in that PR?

[wangxinzhang]

Does the subgraph have to be pre-compiled to support being suspended and resumed，Can the subgraph as a node method support it?

[ChatGPD]

So far as I tested, a node method will not be able to get the subgraph state and config. I tried to use checkpointer to get last config but don't work out.

[gssci]

Shouldn't there be a way to set a max_recursion limit to subgraphs?

[LiuXinke75]

I encountered the following error. How can I fix it?

TypeError Traceback (most recent call last)
Cell In[25], line 1
----> 1 state = graph.get_state(config, subgraphs=True)
2 state.values['messages']

TypeError: Pregel.get_state() got an unexpected keyword argument 'subgraphs'

[matteociccozzi-ah]

I'm also getting the same error

[vbarda]

which langgraph version are you using? i would recommend updating to the latest

[jhachirag7]

below is the latest graph response:

(('GenerateKeywordsIdeas:b458982c-c182-854a-44a9-f0bcdecba8e8',), {'generator': {'messages': [AIMessage(content=[{'type': 'text', 'text': "Certainly! I'll follow the steps you've outlined to find seed keywords for neoleads.com. Let's begin with scraping the website content and then use the search tools to expand on the keywords.\n\nStep 1: Let's use the scraping tool to extract data from neoleads.com.", 'index': 0}, {'type': 'tool_use', 'id': 'toolu_bdrk_01Fzc3TnbAfxkwwbvVBVNMn6', 'name': 'apify_website_content_crawler', 'input': {}, 'index': 1, 'partial_json': '{"query": "https://neoleads.com"}'}], additional_kwargs={}, response_metadata={'stop_reason': 'tool_use', 'stop_sequence': None}, id='run-3f3f97fa-11a1-48d7-b8f7-57bb9bf94ecd-0', tool_calls=[{'name': 'apify_website_content_crawler', 'args': {'query': 'https://neoleads.com'}, 'id': 'toolu_bdrk_01Fzc3TnbAfxkwwbvVBVNMn6', 'type': 'tool_call'}], usage_metadata={'input_tokens': 1709, 'output_tokens': 115, 'total_tokens': 1824})], 'sender': 'generator', 'limit': 0, 'generation': AIMessage(content=[{'type': 'text', 'text': "Certainly! I'll follow the steps you've outlined to find seed keywords for neoleads.com. Let's begin with scraping the website content and then use the search tools to expand on the keywords.\n\nStep 1: Let's use the scraping tool to extract data from neoleads.com.", 'index': 0}, {'type': 'tool_use', 'id': 'toolu_bdrk_01Fzc3TnbAfxkwwbvVBVNMn6', 'name': 'apify_website_content_crawler', 'input': {}, 'index': 1, 'partial_json': '{"query": "https://neoleads.com"}'}], additional_kwargs={}, response_metadata={'stop_reason': 'tool_use', 'stop_sequence': None}, id='run-3f3f97fa-11a1-48d7-b8f7-57bb9bf94ecd-0', tool_calls=[{'name': 'apify_website_content_crawler', 'args': {'query': 'https://neoleads.com'}, 'id': 'toolu_bdrk_01Fzc3TnbAfxkwwbvVBVNMn6', 'type': 'tool_call'}], usage_metadata={'input_tokens': 1709, 'output_tokens': 115, 'total_tokens': 1824})}}))

And when i am getting graph state, in tasks i didn't get subgraph tasks even though i have mentioned subgraphs to True:

StateSnapshot(values={'messages': [HumanMessage(content='find seed keywords for neoleads.com', additional_kwargs={}, response_metadata={}, id='1590f929-1b02-4fbf-8bc1-4c490ac1c3f8')], 'next': 'GenerateKeywordsIdeas', 'steps': "Step 1: Use the scraping tool to extract data from neoleads.com.\nStep 2: Analyze the extracted data to identify potential seed keywords related to the website's content and services.\nStep 3: Use Google Search, Bing Search, and YouTube Search tools to find related keywords and searches based on the identified seed keywords.\nStep 4: Compile a comprehensive list of seed keywords and related searches relevant to neoleads.com."}, next=('GenerateKeywordsIdeas',), config={'configurable': {'thread_id': 'gwZPugyWbAA008weA5ZT', 'checkpoint_ns': '', 'checkpoint_id': '1ef8df14-2442-6a18-8001-ed017ce3e52a'}}, metadata={'step': 1, 'source': 'loop', 'writes': {'supervisor': {'next': 'GenerateKeywordsIdeas', 'steps': "Step 1: Use the scraping tool to extract data from neoleads.com.\nStep 2: Analyze the extracted data to identify potential seed keywords related to the website's content and services.\nStep 3: Use Google Search, Bing Search, and YouTube Search tools to find related keywords and searches based on the identified seed keywords.\nStep 4: Compile a comprehensive list of seed keywords and related searches relevant to neoleads.com."}}, 'parents': {}}, created_at='2024-10-19T08:08:01.017277+00:00', parent_config={'configurable': {'thread_id': 'gwZPugyWbAA008weA5ZT', 'checkpoint_ns': '', 'checkpoint_id': '1ef8df13-f1ef-6b51-8000-9e465076db1d'}}, tasks=(PregelTask(id='b458982c-c182-854a-44a9-f0bcdecba8e8', name='GenerateKeywordsIdeas', path=('__pregel_pull', 'GenerateKeywordsIdeas'), error=None, interrupts=(), state=None, result=None),))

code:

def run(self, input_message: str):
        self.compile()
        with ConnectionPool(
            conninfo=self.DB_URI,
            kwargs=self.connection_kwargs,
        ) as pool:
            checkpointer = PostgresSaver(pool)
            # memory = MemorySaver()
            # checkpointer.setup()
            print(self.config)
            graph = self.workflow.compile(checkpointer = checkpointer, interrupt_before=["human_feedback"])
            # print(graph.get_graph().draw_mermaid())
            self.config['recursion_limit'] = 150
            self.config['callbacks'] = self.callbacks
            # self.config["configurable"] = {"thread_id": "1"}
            for s in graph.stream({"messages": [HumanMessage(content=input_message)],"user_feedback": "No Feedback till now, so continue your work",}, 
                                  config=self.config,subgraphs=True):
                print(s)
                # if "__end__" not in s:
                #     print(s)
                #     if "FINISH" in s:
                #         self.final_response = s["FINISH"]['messages'][-1].content
                #     print("----")
            print(graph.get_state(self.config, subgraphs=True))

[cocoza4]

For, double nested subgraphs, let's say we have the following routes

class RouterState(MessagesState):
    route: Literal["weather", "sql", "other"]


class Router(TypedDict):
    route: Literal["weather", "sq", "other"]

How can we implement a graph such that it supports question that requires multiple routes e.g. both weather and other route. For example, questions like what is the weather in SF and show me 3 countries with highest sales.

[atanudasgupta]

How to display subgraphs within graphs, IPython.display does not show subgraphs the way you would want to visualise/

[ficzusistvan]

I have the same question.

[jhachirag7]

graph.get_graph(xray=True).draw_mermaid()

use this you will get code that starts with graph TD;

paste that code in mermaid.live website

[ClaireJL]

Is there a way to stack & restore a subgraph's private state history? I'm developing a multi-agent system, where each agent is defined as a subgraph. A subgraph can be visited multiple times during a session, and I want it to keep all of its own state memory. In the earlier Langgraph version I tried to customize a "SubgraphStateManager" and use it as a context object in parent state. Sincecontext is no longer supported in later versions, I'm wondering if there's any easy way for me to do this.

[rushab-agi]

did you find the solution for this ?

[lixxvsky]

For the second case "Add a node function that invokes the subgraph"
If we set parameter config when invoke subgraph,

response = subgraph.invoke({"bar": state["foo"]}, config={"configurable": {"thread_id": "1233"}})

Parent graph fails to steam the events of subgraph. The output will be

((), {'node_1': {'foo': 'hi! foo'}})
((), {'node_2': {'foo': 'hi! foobaz'}})

May I know is there a way to stream subgraph when set config parameter when invoking subgraph?

[atanudasgupta]

While streaming. Use subgraphs=True, in graph.stream()

[lixxvsky]

Hi atanudasgupta, I add subgraphs=True, but it still cannot stream subgraph when config is added in subgraph.invoke() function

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

I use the sample code shown in Add a node function that invokes the subgraph for test, the only one difference is just adding config in subgraph.invoke() function.

[atanudasgupta]

you need to use as follows, with config.

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

see the output below.

((), {'node_1': {'foo': 'hi! foo'}})
(('node_2:a143cd80-3d42-40c1-02a8-99add744ed8b',), {'subgraph_node_1': {'baz': 'baz'}})
(('node_2:a143cd80-3d42-40c1-02a8-99add744ed8b',), {'subgraph_node_2': {'bar': 'hi! foobaz'}})
((), {'node_2': {'foo': 'hi! foobaz'}})

[youdar]

Hi.
How to create edges between a subgraph and the parent graph?

When creating an edge from parent to the subgraph, the subgraph need to be compiled first.
When creating an edge from the subgraph back to the parent, the parent needs to be compiled first.

So this method does not appear to work, since it a circular logic.
I assume that going to END in the subgraph, keeps us in the subgraph, state changes to the START if the subgraph.

If I want to convert Customer support bot to a graph implementation,
how to I leave the flight reservation subgraph and go back to the main assistant parent graph?

Any advice/thoughts ?
thank you

[youdar]

Looking at the code above it looks like when the interaction get to subgraph_node_2 it automatically goes back to the parent start .

Or maybe I am not understanding it correctly...

It looks to me like the one of the main points of using subgraphs architecture is to allow to remain in the subgraph until routed beck to the parent, avoiding always going to the parent start automatically.

[weatherman85]

I was trying to get this to work where there are two subgraphs running in parallel with a breakpoint in one of them. It always throws a GraphInterrupt error though. Using the latest lang graph. Can’t share the code since it’s at work.

[atanudasgupta]

Use add interrupt Yahoo Mail: Search, Organize, Conquer On Sat, Nov 16, 2024 at 7:43 PM, ***@***.***> wrote: I was trying to get this to work where there are two subgraphs running in parallel with a breakpoint in one of them. It always throws a GraphInterrupt error though. Using the latest lang graph. Can’t share the code since it’s at work. — Reply to this email directly, view it on GitHub, or unsubscribe. You are receiving this because you commented.Message ID: ***@***.***>

[weatherman85]

I have an interrupt in the one subgraph. Does it need to go into the primary graph somewhere?

[padamson]

Is it possible to interrupt after a subgraph node by passing "interrupt_after" to CompiledGraph.stream()? This would be useful for testing, but it seems like it's not possible since the subgraph namespace includes a dynamic runtime UID (subgraph:UID.node_name).

[vigneshmj1997]

Wont it be a better design if langgraph its self knows where to continue from, irrespective of graph being a subgraph or parent graph ..?

[david101-hunter]

Hi
I'm in To get the state history of the subgraph, we need to first pass in stage

after run

parent_graph_state_before_subgraph = next(
    h
    for h in graph.get_state_history(config)
    if h.next == ("weather_graph",)
)

subgraph_state_before_model_node = next(
    h
    for h in graph.get_state_history(parent_graph_state_before_subgraph.tasks[0].state)
    if h.next == ("model_node",)
)

it raise

Traceback (most recent call last):
  File "../view-and-update-state-graph.py", line 155, in <module>
    parent_graph_state_before_subgraph = next(
                                         ^^^^^
StopIteration

[sivaTwks010928]

class SubgraphState(TypedDict):
    foo: str
    bar: str


def subgraph_node_1(state: SubgraphState):
    return {"bar": "subgraph_node_1:bar"}


def subgraph_node_2(state: SubgraphState):
    return {"foo": state["foo"] + " + subgraph_node_2(" + state["bar"] + ")"}


def subgraph_node_3(state: SubgraphState):
    return {"bar": state["bar"] + " + subgraph_node_3"}


def subgraph_node_4(state: SubgraphState):
    return {"foo": state["foo"] + " + subgraph_node_4 FINAL"}


subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node("subgraph_node_1", subgraph_node_1)
subgraph_builder.add_node("subgraph_node_2", subgraph_node_2)
subgraph_builder.add_node("subgraph_node_3", subgraph_node_3)
subgraph_builder.add_node("subgraph_node_4", subgraph_node_4)

subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph_builder.add_edge("subgraph_node_2", "subgraph_node_3")
subgraph_builder.add_edge("subgraph_node_3", "subgraph_node_4")

subgraph = subgraph_builder.compile(checkpointer=True)


class ParentState(TypedDict):
    foo: str


def node_1(state: ParentState):
    return {"foo": "parent_node_1(" + state["foo"] + ")"}


builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", subgraph)

builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_1", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "1"}}

print("\n--- Streamed Updates ---")

for event in graph.stream({"foo": "foo"}, config=config, stream_mode="updates", subgraphs=True):
    for key, value in event[1].items():
        print(key, value)

print("\n--- Final Parent Graph State ---")
pprint(graph.get_state(config).values)

print("\n--- Parent Graph State History ---")
snapshots_parent = graph.get_state_history(config=config)


state_with_subgraph = [s for s in snapshots_parent if s.next == ('node_2',)][0]
subgraph_config =state_with_subgraph.tasks[0].state

subgraph_state_histories = graph.get_state_history(config=subgraph_config)

subgraph_histories = []
for h in subgraph_state_histories:
    subgraph_histories.append(h)

I have tried this, but I got subgraph_state_histories as empty list... do u have any idea?

[sivaTwks010928]

anyone have any idea why we can't get the subgraph state history?