How To Nested Subgraphs

How-To: Nested Subgraphs

AgentMesh compiled applications implement the standard callable interface (__call__), allowing them to be nested as child nodes within parent graphs.

---

Hierarchical Workflow Architecture

from typing import TypedDict, Annotated, List
import operator
from langgraph.graph import StateGraph
import agentmesh_adapter

class HierarchicalState(TypedDict):
    input_text: str
    sub_results: Annotated[List[str], operator.add]
    final_output: str

# 1. Define Child Subgraph
child_builder = StateGraph(HierarchicalState)
child_builder.add_node("child_node1", lambda s: {"sub_results": ["Child step 1"]})
child_builder.add_node("child_node2", lambda s: {"sub_results": ["Child step 2"]})
child_builder.add_edge("__start__", "child_node1")
child_builder.add_edge("child_node1", "child_node2")
child_builder.add_edge("child_node2", "__end__")

child_app = agentmesh_adapter.compile(child_builder)

# 2. Embed Child App into Parent Graph
parent_builder = StateGraph(HierarchicalState)
parent_builder.add_node("parent_preprocess", lambda s: {"sub_results": ["Parent Preprocess"]})
parent_builder.add_node("sub_flow", child_app)
parent_builder.add_node("parent_postprocess", lambda s: {"final_output": "Done: " + " -> ".join(s["sub_results"])})

parent_builder.add_edge("__start__", "parent_preprocess")
parent_builder.add_edge("parent_preprocess", "sub_flow")
parent_builder.add_edge("sub_flow", "parent_postprocess")
parent_builder.add_edge("parent_postprocess", "__end__")

parent_app = agentmesh_adapter.compile(parent_builder)

result = parent_app.invoke({"input_text": "Task", "sub_results": [], "final_output": ""})
print(result["final_output"])
```\n