langchain-ai
Make state access for subgraphs easier from parent graph
Checked other resources
- [x] This is a bug, not a usage question. For questions, please use GitHub Discussions.
- [x] I added a clear and detailed title that summarizes the issue.
- [x] I read what a minimal reproducible example is (https://stackoverflow.com/help/minimal-reproducible-example).
- [x] I included a self-contained, minimal example that demonstrates the issue INCLUDING all the relevant imports. The code run AS IS to reproduce the issue.
Example Code
from typing import Literal
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.types import Command, interrupt
from pydantic import BaseModel
from rich import print
class State(BaseModel):
messages: list[str] = []
def node_builder(name: str, type: Literal["normal", "interrupt"]):
def node(state: State):
print(f"{name} node", state)
return {"messages": state.messages + [f"{name} node"]}
def node_interrupt(state: State):
print(f"{name} interrupt", state)
value = interrupt("enter interrupt")
return {"messages": state.messages + [f"{name} interrupt"]}
if type == "normal":
return node
elif type == "interrupt":
return node_interrupt
else:
raise ValueError(f"Invalid type: {type}")
def subgraph_builder(name: str):
def build():
graph = StateGraph(State)
graph.add_node(f"{name}1", node_builder(f"{name}1", "normal"))
graph.add_node(f"{name}2", node_builder(f"{name}2", "interrupt"))
graph.add_edge(START, f"{name}1")
graph.add_edge(f"{name}1", f"{name}2")
graph.add_edge(f"{name}2", END)
return graph
return build
def main():
memory = MemorySaver()
thread_id = "123"
config = RunnableConfig(configurable={"thread_id": thread_id})
graph = StateGraph(State)
graph.add_node("node1", node_builder("node1", "normal"))
subgraph = subgraph_builder("node2")()
compiled_subgraph = subgraph.compile(checkpointer=memory)
graph.add_node("node2", compiled_subgraph)
graph.add_edge(START, "node1")
graph.add_edge("node1", "node2")
graph.add_edge("node2", END)
graph.compile()
compiled_graph = graph.compile(checkpointer=memory)
for chunk in compiled_graph.stream({"messages": []}, config=config, subgraphs=True):
# print(chunk)
pass
state = compiled_graph.get_state(config=config, subgraphs=True)
print("state", state.values)
compiled_graph.invoke(Command(resume=True), config=config, subgraphs=True)
print("resume")
state = compiled_graph.get_state(config=config, subgraphs=True)
print("after resume state", state.values)
if __name__ == "__main__":
main()
Error Message and Stack Trace (if applicable)
Description
I am using subgraphs and interrupts. When my interrupt happens inside a subgraph, I can't get latest state by get_state API. Only after interrupt is resumed, I can get full state.
In reproduce code, I get
state {'messages': ['node1 node']}
resume
after resume state {'messages': ['node1 node', 'node21 node', 'node22 interrupt']}
I hope I can get
state {'messages': ['node1 node', 'node21 node']}
resume
after resume state {'messages': ['node1 node', 'node21 node', 'node22 interrupt']}
System Info
System Information
OS: Darwin OS Version: Darwin Kernel Version 23.5.0: Wed May 1 20:19:05 PDT 2024; root:xnu-10063.121.3~5/RELEASE_ARM64_T8112 Python Version: 3.12.4 | packaged by Anaconda, Inc. | (main, Jun 18 2024, 10:07:17) [Clang 14.0.6 ]
Package Information
langchain_core: 0.3.56 langchain: 0.3.9 langsmith: 0.1.147 langchain_mcp_adapters: 0.1.1 langchain_openai: 0.3.9 langchain_text_splitters: 0.3.8 langgraph_sdk: 0.1.63
Optional packages not installed
langserve
Other Dependencies
aiohttp: 3.11.18 async-timeout: Installed. No version info available. httpx: 0.28.1 jsonpatch<2.0,>=1.33: Installed. No version info available. langchain-core<0.4,>=0.3.36: Installed. No version info available. langchain-core<1.0.0,>=0.3.45: Installed. No version info available. langchain-core<1.0.0,>=0.3.51: Installed. No version info available. langsmith-pyo3: Installed. No version info available. langsmith<0.4,>=0.1.125: Installed. No version info available. mcp>=1.7: Installed. No version info available. numpy: 2.2.5 openai<2.0.0,>=1.66.3: Installed. No version info available. orjson: 3.10.16 packaging<25,>=23.2: Installed. No version info available. pydantic: 2.10.6 pydantic<3.0.0,>=2.5.2;: Installed. No version info available. pydantic<3.0.0,>=2.7.4;: Installed. No version info available. PyYAML: 6.0.2 PyYAML>=5.3: Installed. No version info available. requests: 2.32.3 requests-toolbelt: 1.0.0 SQLAlchemy: 2.0.40 tenacity: 9.1.2 tenacity!=8.4.0,<10.0.0,>=8.1.0: Installed. No version info available. tiktoken<1,>=0.7: Installed. No version info available. typing-extensions>=4.7: Installed. No version info available.
From the docs
You only need to provide the checkpointer when compiling the parent graph. LangGraph will automatically propagate the checkpointer to the child subgraphs.
Looking into this now!
Hey folks! This is actually the expected behavior. You're getting the state off of the parent graph, so the state hasn't yet been updated because the subgraph node (your subgraph) hasn't completed execution.
In the short term, you can access the subgraph state via something like:
parent_state = parent_graph.get_state(config=config)
sub_state = top_state.tasks[0].state
Can I update the state of a subgraph using the update_state method? (The content to be updated has no channel between it and the parent graph.)
Can I update the state of a subgraph using the update_state method? (The content to be updated has no channel between it and the parent graph.)
same question
