Building Multi‑Agent Systems Like Lego with LangGraph Subgraphs

This article explains why subgraphs are needed for complex tasks, shows how to define a subgraph as a node, use named channels for data flow, isolate namespaces, reuse compiled subgraphs, add checkpoints, and presents a complete multi‑agent orchestration example in Python using LangGraph.

Tech Ocean
Tech Ocean
Tech Ocean
Building Multi‑Agent Systems Like Lego with LangGraph Subgraphs
Single Agent has limited capability? How can multiple Agents cooperate?

1. Why Subgraphs Are Needed

Complex tasks often require several specialized agents to work together. A main planning agent can delegate to independent sub‑agents such as a search agent, a writing agent, and an audit agent. Each sub‑agent is encapsulated as a subgraph, which becomes a node in the main graph.

2. Subgraph as a Node

A subgraph can be added to the main graph just like any other node.

from langgraph.graph import StateGraph, START, END

# 1. Define the subgraph
class SubgraphState(TypedDict):
    task: str
    result: str

def subgraph_node(state: SubgraphState) -> dict:
    return {"result": f"处理: {state['task']}"}

subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node("task_handler", subgraph_node)
subgraph_builder.add_edge(START, "task_handler")
subgraph_builder.add_edge("task_handler", END)
subgraph = subgraph_builder.compile()

# 2. Use the subgraph in the main graph
class MainState(TypedDict):
    message: str
    subgraph_result: str

def main_node(state: MainState) -> dict:
    # Call the subgraph
    result = subgraph.invoke({"task": state["message"]})
    return {"subgraph_result": result["result"]}

main_builder = StateGraph(MainState)
main_builder.add_node("main", main_node)
main_builder.add_edge(START, "main")
main_builder.add_edge("main", END)
main_graph = main_builder.compile()

3. Named Channels: Data Flow Between Subgraph and Main Graph

Data is passed between the subgraph and the main graph through explicitly defined channels.

class SubgraphState(TypedDict):
    input_data: str
    output_data: str  # output to main graph

class MainState(TypedDict):
    main_data: str
    subgraph_output: str  # receives subgraph output

4. Namespace Isolation

The internal state of a subgraph is transparent to the main graph; communication occurs only via the defined output channel.

def main_node(state: MainState) -> dict:
    # Subgraph processing
    subgraph_result = subgraph.invoke({
        "input_data": state["main_data"]
    })
    # Only receive the subgraph's output_data
    return {"subgraph_output": subgraph_result["output_data"]}

5. Multi‑Agent Collaboration Example

A full example shows how a search subgraph and a writing subgraph are orchestrated by a main node.

from typing import TypedDict

# Search Agent subgraph
def search_node(state):
    return {"result": f"搜索: {state['query']}"}

# Writing Agent subgraph
def write_node(state):
    return {"content": f"写作: {state['result']}"}

class MainState(TypedDict):
    query: str
    search_result: str
    content: str

def orchestrator(state: MainState) -> dict:
    # Search
    search_result = search_subgraph.invoke({"query": state["query"]})
    # Write
    content = write_subgraph.invoke({"result": search_result["result"]})
    return {
        "search_result": search_result["result"],
        "content": content["content"]
    }

builder = StateGraph(MainState)
builder.add_node("orchestrator", orchestrator)
builder.add_edge(START, "orchestrator")
builder.add_edge("orchestrator", END)

6. Reusing a Compiled Subgraph

The same compiled subgraph can be attached to multiple nodes, each invocation maintaining its own state while the main graph reads and writes through channels.

# Reuse the compiled subgraph for different tasks
builder.add_node("task_1", subgraph)
builder.add_node("task_2", subgraph)
# Each call has independent state; main graph communicates via channels

7. Checkpoints for Subgraphs

Subgraphs can have their own checkpoints to enable state persistence.

from langgraph.checkpoint.memory import InMemorySaver

subgraph_memory = InMemorySaver()
subgraph = subgraph_builder.compile(checkpointer=subgraph_memory)

Key Concepts Recap

Subgraph : encapsulated independent workflow.

Named Channels : data pathways between subgraph and main graph.

Namespace Isolation : subgraph state is hidden from the main graph.

Subgraph as Node : compiled subgraph can be added directly with add_node.

Official documentation: https://langchain-ai.github.io/langgraph/concepts/low_level/#subgraphs

Multi‑Agent guide: https://langchain-ai.github.io/langgraph/how-tos/

Original Source

Signed-in readers can open the original source through BestHub's protected redirect.

Sign in to view source
Republication Notice

This article has been distilled and summarized from source material, then republished for learning and reference. If you believe it infringes your rights, please contactadmin@besthub.devand we will review it promptly.

PythonMulti-AgentcheckpointLangGraphNamespace IsolationStateGraphNamed ChannelsSubgraph
Tech Ocean
Written by

Tech Ocean

Focused on AI programming, sharing ready-to-use development efficiency solutions.

0 followers
Reader feedback

How this landed with the community

Sign in to like

Rate this article

Was this worth your time?

Sign in to rate
Discussion

0 Comments

Thoughtful readers leave field notes, pushback, and hard-won operational detail here.