#334: Create Subgraphs in LangGraph

The more complex our applications get, the harder it is to follow along our graph. Luckily for us, there is the concept of subgraphs that let us split our graph into parts that we can reuse.

For this post we create a minimalistic text writing pipeline that puts the quality checks into a subgraph. Let us see how we can do that.

Preparation

As with most LangGraph applications, we need an LLM, state and a few nodes. In this example we have two state objects, the ParentState is for the main workflow, while GateState is for the subgraph.

import re
import sys
from typing import TypedDict

from langchain_openai import ChatOpenAI
from langgraph.graph import START, END, StateGraph

sys.stdout.reconfigure(encoding="utf-8")

llm = ChatOpenAI(
    base_url="http://localhost:1234/v1",
    api_key="lm-studio",
    model="openai/gpt-oss-20b",
    temperature=0.1,
)


class ParentState(TypedDict):
    topic: str
    draft: str
    verdict: str


class GateState(TypedDict):
    draft: str
    tone_score: int
    length_score: int
    fact_score: int
    verdict: str


def parse_int_or_default(text: str, default: int) -> int:
    m = re.search(r"\d+", text)
    return int(m.group()) if m else default


def write_draft(state: ParentState) -> dict:
    print(f"[write_draft] Drafting blog post on: {state['topic']}")
    response = llm.invoke([
        ("system", "Write a 3-sentence opening paragraph for a blog post. Body only."),
        ("user", state["topic"]),
    ])
    return {"draft": response.content}


def check_tone(state: GateState) -> dict:
    print("[check_tone] LLM-rating draft tone ...")
    response = llm.invoke([
        ("system", "Rate the tone of the following text from 1 to 10. "
                   "Respond with ONLY a single integer."),
        ("user", state["draft"]),
    ])
    return {"tone_score": parse_int_or_default(response.content, default=5)}


def check_length(state: GateState) -> dict:
    print("[check_length] Counting words ...")
    words = len(state["draft"].split())
    if 30 <= words <= 80:
        score = 10
    elif words < 30:
        score = max(1, 10 - (30 - words))
    else:
        score = max(1, 10 - (words - 80) // 5)
    return {"length_score": score}


def check_facts(state: GateState) -> dict:
    print("[check_facts] LLM-rating factual plausibility (demo only, not real fact-checking) ...")
    response = llm.invoke([
        ("system", "Rate the factual plausibility of the following text from 1 to 10. "
                   "Respond with ONLY a single integer. (This is a demo check.)"),
        ("user", state["draft"]),
    ])
    return {"fact_score": parse_int_or_default(response.content, default=5)}


def aggregate_score(state: GateState) -> dict:
    tone = state["tone_score"]
    length = state["length_score"]
    fact = state["fact_score"]
    avg = (tone + length + fact) / 3
    print(f"[aggregate_score] tone={tone} length={length} fact={fact} avg={avg:.1f}")

    if avg >= 7:
        verdict = "PUBLISH"
    else:
        dim_name, dim_score = min(
            (("tone", tone), ("length", length), ("fact", fact)),
            key=lambda t: t[1],
        )
        verdict = f"REVISE: {dim_name} (score={dim_score}/10)"
    return {"verdict": verdict}


def publish_or_revise(state: ParentState) -> dict:
    print("\n--- VERDICT ---")
    print(state["verdict"])
    print("--- /VERDICT ---")
    return {}

Create the subgraph

We create our subgraph the same way we create a regular graph in LangChain. A subgraph is a fully functional graph that we then put into another graph:

gate = StateGraph(GateState)

gate.add_node("check_tone", check_tone)
gate.add_node("check_length", check_length)
gate.add_node("check_facts", check_facts)
gate.add_node("aggregate_score", aggregate_score)

gate.add_edge(START, "check_tone")
gate.add_edge("check_tone", "check_length")
gate.add_edge("check_length", "check_facts")
gate.add_edge("check_facts", "aggregate_score")
gate.add_edge("aggregate_score", END)

gate_subgraph = gate.compile()

Use the subgraph

We can now create our main graph and use our subgraph in the variable gate_subgraph like a regular node:

workflow = StateGraph(ParentState)

workflow.add_node("write_draft", write_draft)
workflow.add_node("quality_gate", gate_subgraph)
workflow.add_node("publish_or_revise", publish_or_revise)

workflow.add_edge(START, "write_draft")
workflow.add_edge("write_draft", "quality_gate")
workflow.add_edge("quality_gate", "publish_or_revise")
workflow.add_edge("publish_or_revise", END)

graph = workflow.compile()


def main() -> None:
    topic = sys.argv[1] if len(sys.argv) > 1 else "the productivity benefits of standing desks"


    graph.invoke({"topic": topic})


if __name__ == "__main__":
    main()

Run the graph and subgraph

We can now run our script that comes up with an idea and then runs it through the quality gate (the subgraph). At the end we see if this post is ready to publish or not:

$ python .\subgraphs.py

[write_draft] Drafting blog post on: the productivity benefits of standing desks
[check_tone] LLM-rating draft tone ...
[check_length] Counting words ...
[check_facts] LLM-rating factual plausibility (demo only, not real fact-checking) ...
[aggregate_score] tone=5 length=1 fact=1 avg=2.3

--- VERDICT ---
REVISE: length (score=1/10)
--- /VERDICT ---


$ python .\subgraphs.py

[write_draft] Drafting blog post on: the productivity benefits of standing desks
[check_tone] LLM-rating draft tone ...
[check_length] Counting words ...
[check_facts] LLM-rating factual plausibility (demo only, not real fact-checking) ...
[aggregate_score] tone=8 length=7 fact=7 avg=7.3

--- VERDICT ---
PUBLISH
--- /VERDICT ---

Visualising the subgraph

When we visualise our graphs as we did in post #327, we only get to see the nodes of our main graph:

We only see the 3 nodes of the main graph.

To see the subgraph, we need to pass the parameter xray=1 to the get_graph() function:

png_bytes = graph.get_graph(xray=1).draw_mermaid_png()
with open("subgraphs.png", "wb") as f:
    f.write(png_bytes)

With this little change we see our subgraph in its entirety:

We now see the main graph and the parts that make the subgraph.