Streaming state changes from subgraphs

[Blake-Martin-code]

I have a main graph which is creating parallel send calls to a subgraph. The main graph has a different graph state than the subgraphs, but the subgraph updates the main graph state using a custom operator. When I astream the main graph with subgraphs=True and stream_mode=["updates", "values"] I can see the subgraph nodes being executed, but the subgraphs state is not showing up. Is there a way to show subgraph state? My goal is to stream all these updates so I can send them to the front end with detailed info about the state. A api post call is what is activating these graphs which is why I need to to stream instead of webhooks.

`
class MainGraph:

def __init__(self, llm: ChatOpenAI, section_builder: StateGraph, verbose: bool = False):
    self.llm = llm
    self.section_builder_subagent = section_builder
    self.verbose = verbose

    self.create_graph()
    self.add_nodes_and_edges()


async def section_subgraph_launcher(self, state: Section):
    # Generate unique identifiers for this subgraph run
    unique_name = f"section-{state.name}"
    # ‘section_builder_subagent’ is the compiled subgraph’s runnable
    return await self.section_builder_subagent.ainvoke(
        state,
        config={"run_name": unique_name}
    )

def parallelize_section_writing(self, state: Report):
    """ This is the "map" step when we kick off web research for some sections of the report in parallel and then write the section"""

    if self.verbose:
        logger.info('--- Parallelizing Section Writing ---')

    # Kick off section writing in parallel via Send() API for any sections that require research
    return [
        Send("research_section_builder", # name of the subagent node
            s)
            for s in state.sections
            if s.research
    ]

def create_graph(self):
    """ Create the report builder """
    self.graph = StateGraph(Report, input=Report, output=Report)

def add_nodes_and_edges(self):
    """ Add nodes to the report builder """
    self.graph.add_node("parallelize_section_writing", self.parallelize_section_writing)
    self.graph.add_node("research_section_builder", self.section_subgraph_launcher)

    self.graph.add_conditional_edges(START,
                                self.parallelize_section_writing,
                                ["parallelize_section_writing"])

    self.graph.add_edge("parallelize_section_writing", "research_section_builder")
    self.graph.add_edge("research_section_builder", END)

def compile_graph(self):
    """ Compile the report builder """
    return self.graph.compile()`

Here is the last node in the subgraph. This updates the Maingraph state with the custom function.

def return_section(self, state: Section): return {"sections": [state]}

`
custom operator

def replace_section(
old: List[Section],
new: Union[Section, List[Section]],
) -> List[Section]:

# Normalize to list for uniform handling

incoming = new if isinstance(new, list) else [new]

# Build a lookup by section name

by_name = {sec.name: sec for sec in old}

# Replace or add each incoming section

for sec in incoming:

    by_name[sec.name] = sec

# Reconstruct in original order, then append brand-new ones

original_names = [sec.name for sec in old]

merged = [by_name[name] for name in original_names]

for name, sec in by_name.items():

    if name not in original_names:

        merged.append(sec)

return merged`

Here is a sample output:

(('research_section_builder:b4bf5a7b-06f7-275f-b6d7-863adfacf01f',), 'updates', {'search_web': None})

(('research_section_builder:d809d84d-af64-2751-784f-e45022bc2360',), 'updates', {'reflect': None})
(('research_section_builder:d809d84d-af64-2751-784f-e45022bc2360',), 'updates', {'search_web': None})
(('research_section_builder:a23aaffa-84c5-8daa-9cf8-4ff71523ae14',), 'updates', {'reflect': None})
(('research_section_builder:a23aaffa-84c5-8daa-9cf8-4ff71523ae14',), 'updates', {'search_web': None})
(('research_section_builder:e200d250-6e7f-2526-19f3-a45a74d18d6b',), 'updates', {'reflect': None})
(('research_section_builder:a55f1f67-c983-f73c-142e-771da25b2ea6',), 'updates', {'reflect': None})
(('research_section_builder:b4bf5a7b-06f7-275f-b6d7-863adfacf01f',), 'updates', {'reflect': None})
(('research_section_builder:d809d84d-af64-2751-784f-e45022bc2360',), 'updates', {'reflect': None})
(('research_section_builder:a23aaffa-84c5-8daa-9cf8-4ff71523ae14',), 'updates', {'reflect': None})
(('research_section_builder:e200d250-6e7f-2526-19f3-a45a74d18d6b',), 'updates', {'write_section': None})
(('research_section_builder:d809d84d-af64-2751-784f-e45022bc2360',), 'updates', {'return_section': {'sections': [Section(name='Applications of LangGraph', description='This section will explore real-world applications of LangGraph in various domains, including web development, data science, and machine learning.', research=True, content="Content Gets written here I removed because it is long", queries=['LangGraph real-world applications 2024 web development comparison with other graph databases', 'LangGraph data science use cases 2023-2024 performance benchmarks graph neural networks', 'LangGraph machine learning implementation examples 2024 technical architecture comparison with TensorFlow', 'LangGraph core features API documentation practical examples for knowledge graph construction', 'LangGraph integration with existing web frameworks 2024 Python implementation tutorials comparison with Neo4j', 'LangGraph use cases in natural language processing 2023-2024 comparison with other NLP tools', 'LangGraph deployment and scalability benchmarks 2024 cloud platforms comparison with Amazon SageMaker', 'LangGraph use cases in finance 2023-2024 comparison with other graph databases', 'LangGraph limitations and potential drawbacks in different application domains'], urls=[], context='', source_filters=['source_1', 'source_2', 'source_3', 'source_4', 'source_5'], reflection_count=3, reflection_needed=True, reflection_queries=['LangGraph case studies in web development 2023-2024', 'LangGraph performance comparison with other graph databases on specific data science tasks'])]}})

As you can see the subgraph nodes are showing (search_web, reflect, and write_section) but not the subgraph states. Only the subgraph state shows when it returns a state that will update the maingraph state.