LangGraph Basics: Part 6 — Subgraphs & Human-in-the-Loop

Author Photo
LangGraph Basics: Part 6 — Subgraphs & Human-in-the-Loop

📚 Table of Contents

  1. 1. Why Graphs Need Subgraphs & Human Control
  2. 2. Installation & Setup
  3. 3. Subgraphs: Concept & API
  4. 4. Human-in-the-Loop: interrupt() and Command
  5. 5. Complete Example: AI Travel Planner
  6. 6. Web Interface
  7. 7. Conclusion

🗺️ 1. Why Graphs Need Subgraphs & Human Control

The first five parts of this series built every foundational piece of LangGraph: nodes, edges, state, reducers, conditional routing, checkpointers, streaming, and tool calling. You can now assemble a working AI agent in a single graph file. But what happens when that graph grows? A real-world application — say, one that plans a trip, routes customer tickets, or generates a research report — quickly becomes a tangle of dozens of nodes, complex routing, and mixed responsibilities. Two features address this directly: subgraphs let you split a large graph into composable, independently testable modules, and Human-in-the-Loop lets you pause execution mid-run so a person can review, correct, or approve before the graph continues.

1.1

🧩 Subgraphs: Nesting Graphs Inside Graphs

A subgraph is a compiled StateGraph that is used as a node inside another graph. From the outside, the parent graph sees it as just another callable — it receives a state dict, runs its internal pipeline, and returns an updated state dict. The internal nodes, edges, and state keys are completely invisible to the parent. This is the same modular thinking you use when breaking a large Python program into functions and classes: hide the complexity, expose only the interface.

🔒

Private State

Each subgraph has its own TypedDict. Its internal keys never pollute the parent's state namespace.

♻️

Reusable Module

The same compiled subgraph can be called multiple times within one parent graph or shared across different parent graphs.

🧪

Independently Testable

Because it is a compiled graph, you can invoke a subgraph directly in isolation — no need to run the full parent to test it.

🏗️

Composable Architecture

Large multi-agent systems (supervisor + workers) are built by nesting multiple subgraphs under a coordinator parent graph.

1.2

🛑 Human-in-the-Loop: Keeping Humans in Charge

Human-in-the-Loop (HITL) is the ability to pause a running graph, show something to a human, and wait for their response before continuing. You might need this when the stakes are too high to let an AI act alone — booking a flight, sending an email, deleting data, committing a purchase. In LangGraph, HITL is implemented with two primitives: interrupt() suspends the graph at a specific node and captures an optional payload to show the user, and Command(resume=...) resumes the graph from where it stopped, injecting the human's response back into state.

🔗 Key distinction: interrupt() is not an exception or an error — it is a controlled pause. The graph's full state (including conversation history and intermediate results) is preserved in the checkpointer. When the human responds, execution resumes from the exact node that called interrupt(), as if nothing happened.
1.3

✈️ Our Scenario: AI Travel Planner

Picture a traveler — let's call her Mia — who wants an AI to build her a complete trip plan. She types: "4-day cultural trip to Kyoto, Japan — budget $1,500." The AI needs to research the destination, build a day-by-day itinerary, and estimate costs. That is three distinct LLM tasks, each depending on the previous result — a perfect job for a subgraph. But before anything is booked, Mia needs to review and approve (or revise) the plan. That requires Human-in-the-Loop.

The application is structured as two graphs. The PlannerSubGraph handles the AI work: research the destination → create the itinerary → estimate the budget, all with their own private state. The TravelGraph (the parent) calls the subgraph as a single node, then pauses at human_review for Mia's feedback. If she approves, the graph writes a final confirmation. If she asks for changes, the graph loops back and re-runs the subgraph with her revision notes. This scenario requires both concepts taught in this post and would be impossible without them.

⚙️ 2. Installation & Setup

The Travel Planner shares the same environment as every other post in this series. All packages live in the root langgraph/ folder — one virtual environment, one requirements.txt, one .env file.

Step 1 — Python version. This project requires Python 3.12.

python --version # Python 3.12.x

Step 2 — Virtual environment.

python -m venv langgraph source langgraph/bin/activate # macOS / Linux langgraph\Scripts\activate # Windows

Step 3 — Install dependencies. The requirements.txt is shared at the langgraph/ folder level:

langchain==1.2.17 langgraph==1.1.10 langchain-google-genai==4.2.2 python-dotenv==1.2.2 gradio==6.14.0
pip install -r requirements.txt

Step 4 — Gemini API key. Get your free key from Google AI Studio. Create a .env file inside langgraph/:

GEMINI_API_KEY=your_api_key_here GEMINI_MODEL_NAME=gemini-2.0-flash GEMINI_TEMPERATURE=0.7 GEMINI_MAX_RETRIES=2
⚠️ Add .env to your .gitignore — never commit API keys to a public repository.
2.1

🤖 Configuring the LLM

config.py loads the .env file and exposes settings as class attributes. llm.py wraps ChatGoogleGenerativeAI in a GeminiLLM class that every node imports.

import os from dotenv import load_dotenv load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), "..", ".env")) class Config: MODEL_NAME = os.getenv("GEMINI_MODEL_NAME", "gemini-2.0-flash") TEMPERATURE = float(os.getenv("GEMINI_TEMPERATURE", 0.7)) MAX_RETRIES = int(os.getenv("GEMINI_MAX_RETRIES", 2))
from langchain_google_genai import ChatGoogleGenerativeAI from config import Config class GeminiLLM: def __init__(self): self.llm = ChatGoogleGenerativeAI( model=Config.MODEL_NAME, temperature=Config.TEMPERATURE, max_retries=Config.MAX_RETRIES, ) def get_llm(self): return self.llm

Every node that needs the LLM calls GeminiLLM().get_llm() once in __init__ and stores the result. This keeps the model instantiation in one place and makes it easy to swap providers later.

🧩 3. Subgraphs: Concept & API

Before writing a subgraph, it helps to understand exactly how it differs from a regular node and why state isolation matters.

3.1

⚖️ What Makes a Subgraph Different from a Regular Node

A regular node is a Python function: it receives the current state, does something, and returns a partial update. A subgraph is a compiled graph used where a function would normally go. When the parent graph reaches the subgraph node, LangGraph invokes the compiled subgraph with the supplied input state, runs its full internal pipeline, and merges the result back into the parent's state.

AspectRegular NodeSubgraph Node
What it isA Python callableA compiled StateGraph
Internal structureFlat — single function bodyMultiple nodes & edges inside
StateReads/writes parent state keysHas its own private TypedDict
TestabilityUnit-test the function directlyInvoke the subgraph independently
ReusabilityCall the function from anywhereCompile once, nest anywhere
3.2

🔒 State Isolation: Private vs Shared State

The most important rule of subgraphs is this: the subgraph's state keys never appear in the parent's state. The parent hands the subgraph an input, the subgraph runs its pipeline internally, and the parent reads back only the keys it cares about from the result. Neither side can accidentally read or overwrite the other's private keys.

In the Travel Planner, the subgraph uses PlannerState with four keys: trip_request, destination_info, itinerary, and budget_estimate. These represent the pipeline's intermediate and final outputs. The parent graph uses TravelState, which also has itinerary and budget_estimate — but those are copied over from the subgraph result by the run_planner node, not shared directly.

📌 Why does isolation matter? Imagine a subgraph that uses a key called messages to track intermediate LLM conversation steps. Without isolation, those internal messages would accumulate in the parent's messages field alongside the main conversation — corrupting the chat history. Private state prevents this class of bug entirely.
3.3

🔧 Building, Compiling & Calling a Subgraph

The three steps to use a subgraph are: define its state, build and compile it (just like any graph), then call .invoke() on the compiled object from inside a parent node. Here is the minimal pattern:

from langgraph.graph import StateGraph, START, END from typing_extensions import TypedDict # Step 1: Define the subgraph's private state class InnerState(TypedDict): input_data: str result: str # Step 2: Build and compile the subgraph def inner_node(state: InnerState) -> dict: return {"result": f"processed: {state['input_data']}"} inner_graph = StateGraph(InnerState) inner_graph.add_node("inner", inner_node) inner_graph.add_edge(START, "inner") inner_graph.add_edge("inner", END) compiled_inner = inner_graph.compile() # ← compile returns a CompiledStateGraph # Step 3: Call it from a parent node def outer_node(state) -> dict: result = compiled_inner.invoke({ "input_data": state["user_query"], "result": "", }) return {"processed_result": result["result"]} # merge only what the parent needs

Notice that outer_node builds the input dict manually, passing only the keys InnerState expects. It then reads only result from the subgraph's output and maps it into the parent's own key. This explicit mapping is the interface contract between parent and subgraph.

✅ Best practice: Compile the subgraph once in __init__ (not on every call). Compilation is expensive — each StateGraph.compile() call validates the graph structure and builds internal routing tables. Store the compiled result and reuse it.

🛑 4. Human-in-the-Loop: interrupt() and Command

Now that subgraphs handle modularity, let's tackle the second concept: pausing a graph mid-run for human input. This section explains the two primitives — interrupt() and Command — and why a checkpointer is non-negotiable.

4.1

⏸️ How interrupt() Pauses Execution

Calling interrupt(payload) inside a node tells LangGraph to stop graph execution immediately at that point and surface the payload to the caller. The caller (your application code) can then display the payload to the user and wait. The graph's full state — including every message, every intermediate result, every prior node output — is frozen in the checkpointer.

from langgraph.types import interrupt def human_review_node(state) -> dict: # Build a summary to show the human payload = { "plan": state["itinerary"], "instruction": "Type 'approved' to confirm, or describe your changes.", } # Execution stops here. The value returned by interrupt() is whatever # the caller later passes via Command(resume=...). feedback = interrupt(payload) approved = feedback.strip().lower() in {"approved", "approve", "yes", "ok"} return {"human_feedback": feedback, "approved": approved}

A few things to notice: interrupt() takes any serialisable Python value as its payload — a string, a dict, a list. That value is what your application code receives from the first invoke() call, which you can display in a UI. The return value of interrupt() — assigned to feedback — is not set yet. It will be filled in only when the graph is resumed with Command(resume=...).

4.2

▶️ Resuming with Command(resume=...)

After the human has responded, you resume the paused graph by passing a Command object to invoke() instead of a state dict. Command(resume=value) tells LangGraph: "wake up the paused graph, set the return value of interrupt() to value, and continue from where you left off."

from langgraph.types import Command # First call: start the graph and get the interrupted state result = app.invoke({"trip_request": "4-day Kyoto trip"}, config=config) # → graph runs to interrupt(), pauses; result holds state up to that point # The human reads result["itinerary"] and types their response human_response = "Add more traditional tea ceremony experiences." # Second call: resume the same thread with the human's response result = app.invoke(Command(resume=human_response), config=config) # → interrupt() returns human_response; graph continues from human_review node

The config dict must carry the same thread_id in both calls. LangGraph uses the thread ID to find the correct paused state in the checkpointer. If the thread ID does not match, the resume will fail or start a new conversation.

💡 Multiple revision rounds: If the human requests changes instead of approving, the graph can loop back (via a conditional edge) and call the subgraph again with the revision feedback. The same thread_id is reused across all rounds — the checkpointer accumulates the full history of each round.
4.3

💾 Why MemorySaver Is Required for HITL

Human-in-the-Loop requires a checkpointer. When interrupt() is called, LangGraph serialises the entire current state and saves it to the checkpointer keyed by thread_id. Without a checkpointer, there is nowhere to persist the paused state — the graph would have no way to resume after the human responds.

from langgraph.checkpoint.memory import MemorySaver # MemorySaver keeps state in process memory (good for development and demos). # For production, swap in SqliteSaver or a database-backed checkpointer. graph.compile(checkpointer=MemorySaver())

MemorySaver stores the state in a Python dict — fast and zero-configuration, but it is lost when the process restarts. For a production HITL system where the human might not respond for hours, use a persistent checkpointer like SqliteSaver (introduced in Part 4) so the paused state survives server restarts.

1️⃣
First invoke()
Graph runs until interrupt(). State is saved to checkpointer. Partial state dict returned to caller.
⏸️
Human Reviews
Application displays the payload. Human types a response. State remains frozen in checkpointer.
2️⃣
Second invoke() with Command
Graph resumes from the interrupted node. interrupt() returns the human's text. Execution continues.
✅
Conditional Edge
Approved → finalize. Not approved → re-run subgraph with feedback and pause again.

🏗️ 5. Complete Example: AI Travel Planner

With the concepts in place, let's walk through every file in the project. Each file maps to one concept from Sections 3 and 4.

5.1

🏛️ Architecture Overview

Two graphs work together. The PlannerSubGraph is a sequential pipeline: research the destination → build the itinerary → estimate the budget. It runs inside a dedicated PlannerState and is compiled independently. The TravelGraph (parent) treats the entire subgraph as a single node called run_planner. After the plan is ready, it pauses at human_review via interrupt(). A conditional edge then routes to finalize_plan (approved) or loops back to run_planner (revision requested).

🗺️

PlannerSubGraph

research_destination → create_itinerary → estimate_budget. All three nodes share PlannerState privately.

✈️

TravelGraph (parent)

run_planner → human_review → conditional → finalize_plan or back to run_planner.

🛑

interrupt() + MemorySaver

Graph pauses at human_review. State is frozen in MemorySaver. Resumes with Command(resume=feedback).

✅

Revision Loop

If not approved, traveler feedback is merged into trip_request and the subgraph re-runs until the plan is confirmed.

5.2

📁 Project Structure

basics-6-subgraphs-interrupt-hitl/ ├── config.py # Config class — loads .env settings ├── llm.py # GeminiLLM wrapper class ├── state.py # PlannerState + TravelState TypedDicts ├── subgraph.py # PlannerSubGraph — 3 sequential LLM nodes ├── nodes.py # TravelNodes — run_planner, human_review, finalize_plan ├── graph.py # TravelGraph — builds and compiles the parent graph ├── travel_runner.py # TravelRunner class + demo entrypoint ├── app.py # Gradio web UI ├── prompts/ # LLM prompt templates, one file per node │ ├── research.txt # destination research prompt │ ├── itinerary.txt # day-by-day itinerary prompt │ ├── budget.txt # budget estimate prompt │ └── finalize.txt # trip confirmation prompt └── figure/ # auto-created at runtime by save_figure() ├── graph.mmd # Mermaid diagram source └── graph.png # graph PNG exported by LangGraph

Each file maps to one part of the implementation. state.py defines both state classes (Section 5.3). subgraph.py contains the inner graph (Section 5.4). nodes.py and graph.py build the outer graph (Sections 5.5–5.6). travel_runner.py runs the demo (Section 5.7). All LLM prompts live in prompts/ as plain text files — editing a prompt never requires touching Python code.

5.3

📦 State Design — state.py

Two TypedDict classes live in state.py. They share some key names (itinerary, budget_estimate) but are completely separate types — sharing a name does not create a shared reference. The parent copies values from the subgraph result explicitly in run_planner.

from typing_extensions import TypedDict class PlannerState(TypedDict): """Private state for PlannerSubGraph — never touches TravelState.""" trip_request: str # user's original request (+ revision notes on re-runs) destination_info: str # filled by research_destination itinerary: str # filled by create_itinerary budget_estimate: str # filled by estimate_budget class TravelState(TypedDict): """Shared state for the parent TravelGraph.""" trip_request: str # user's original request destination_info: str # merged from subgraph result itinerary: str # merged from subgraph result budget_estimate: str # merged from subgraph result human_feedback: str # set by human_review after interrupt() approved: bool # True when traveler types "approved" final_plan: str # set by finalize_plan node

TravelState has three extra keys that PlannerState does not need: human_feedback (the traveler's raw response), approved (the boolean flag the conditional edge reads), and final_plan (the polished confirmation written by finalize_plan). The subgraph never sees these keys.

5.4

🗺️ Planner Subgraph — subgraph.py

subgraph.py contains two classes: PlannerNodes (the three LLM node methods) and PlannerSubGraph (which builds, compiles, and exposes the inner graph). All three prompts are loaded from prompts/ in __init__ — only once, not on every node call.

import os from langchain_core.messages import HumanMessage from langgraph.graph import END, START, StateGraph from llm import GeminiLLM from state import PlannerState _PROMPTS_DIR = os.path.join(os.path.dirname(__file__), "prompts") def _load_prompt(filename: str) -> str: with open(os.path.join(_PROMPTS_DIR, filename), "r") as f: return f.read().strip() class PlannerNodes: def __init__(self): self.llm = GeminiLLM().get_llm() self.research_prompt = _load_prompt("research.txt") self.itinerary_prompt = _load_prompt("itinerary.txt") self.budget_prompt = _load_prompt("budget.txt") def research_destination(self, state: PlannerState) -> dict: prompt = self.research_prompt.format(trip_request=state["trip_request"]) response = self.llm.invoke([HumanMessage(content=prompt)]) return {"destination_info": response.content} def create_itinerary(self, state: PlannerState) -> dict: prompt = self.itinerary_prompt.format( destination_info=state["destination_info"], trip_request=state["trip_request"], ) response = self.llm.invoke([HumanMessage(content=prompt)]) return {"itinerary": response.content} def estimate_budget(self, state: PlannerState) -> dict: prompt = self.budget_prompt.format( trip_request=state["trip_request"], itinerary=state["itinerary"], ) response = self.llm.invoke([HumanMessage(content=prompt)]) return {"budget_estimate": response.content} class PlannerSubGraph: def __init__(self): self.nodes = PlannerNodes() self.compiled = self._build() def _build(self): graph = StateGraph(PlannerState) graph.add_node("research_destination", self.nodes.research_destination) graph.add_node("create_itinerary", self.nodes.create_itinerary) graph.add_node("estimate_budget", self.nodes.estimate_budget) graph.add_edge(START, "research_destination") graph.add_edge("research_destination", "create_itinerary") graph.add_edge("create_itinerary", "estimate_budget") graph.add_edge("estimate_budget", END) return graph.compile() # no checkpointer — the parent handles persistence def get_compiled(self): return self.compiled

Notice that PlannerSubGraph._build() calls graph.compile() with no checkpointer. The subgraph does not need one — it runs to completion synchronously each time the parent node calls self.planner.invoke(). The parent graph holds the checkpointer, and that is enough for the interrupt() in human_review to work.

5.5

⚙️ Parent Graph Nodes — nodes.py

TravelNodes defines the three nodes of the parent graph. The first calls the subgraph, the second uses interrupt(), and the third writes the final confirmation.

import os from langchain_core.messages import HumanMessage from langgraph.types import interrupt from llm import GeminiLLM from state import TravelState from subgraph import PlannerSubGraph _PROMPTS_DIR = os.path.join(os.path.dirname(__file__), "prompts") _APPROVED_KEYWORDS = {"approved", "approve", "yes", "ok", "looks good", "great", "perfect"} def _load_prompt(filename: str) -> str: with open(os.path.join(_PROMPTS_DIR, filename), "r") as f: return f.read().strip() class TravelNodes: def __init__(self): planner = PlannerSubGraph() self.planner = planner.get_compiled() self.llm = GeminiLLM().get_llm() self.finalize_prompt = _load_prompt("finalize.txt") # Node 1: invoke the PlannerSubGraph def run_planner(self, state: TravelState) -> dict: trip_request = state["trip_request"] feedback = state.get("human_feedback", "") # On a revision pass, append the traveler's notes to the request if feedback and not state.get("approved", True): trip_request = ( f"{state['trip_request']}\n\n" f"Revision feedback from traveler: {feedback}" ) result = self.planner.invoke({ "trip_request": trip_request, "destination_info": "", "itinerary": "", "budget_estimate": "", }) return { "destination_info": result["destination_info"], "itinerary": result["itinerary"], "budget_estimate": result["budget_estimate"], } # Node 2: pause and wait for the traveler's review def human_review(self, state: TravelState) -> dict: plan_summary = ( f"DESTINATION OVERVIEW:\n{state['destination_info']}\n\n" f"DAY-BY-DAY ITINERARY:\n{state['itinerary']}\n\n" f"ESTIMATED BUDGET:\n{state['budget_estimate']}" ) feedback = interrupt({ "plan": plan_summary, "instruction": "Type 'approved' to confirm, or describe the changes you'd like.", }) approved = feedback.strip().lower() in _APPROVED_KEYWORDS return {"human_feedback": feedback, "approved": approved} # Node 3: write the polished confirmation def finalize_plan(self, state: TravelState) -> dict: prompt = self.finalize_prompt.format( itinerary=state["itinerary"], budget_estimate=state["budget_estimate"], ) response = self.llm.invoke([HumanMessage(content=prompt)]) return {"final_plan": response.content}

The revision logic in run_planner is worth a close look. On the first pass, human_feedback is an empty string, so trip_request is used unchanged. On a second pass (after the traveler requests changes), the feedback is appended to the original request before it is handed to the subgraph. This means the subgraph always receives a complete, self-contained request — it never needs to know about the revision history or look at TravelState directly.

5.6

🔗 Graph Assembly — graph.py

graph.py wires everything together: registers the three nodes, adds the edges, attaches MemorySaver, and exports the figure. The conditional edge after human_review reads state["approved"] and routes accordingly.

import os from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import END, START, StateGraph from nodes import TravelNodes from state import TravelState FIGURE_DIR = os.path.join(os.path.dirname(__file__), "figure") def route_after_review(state: TravelState) -> str: """Approved → write confirmation. Not approved → re-run subgraph.""" return "finalize_plan" if state.get("approved") else "run_planner" class TravelGraph: def __init__(self): self.nodes = TravelNodes() self.compiled_graph = self._build() def _build(self): graph = StateGraph(TravelState) graph.add_node("run_planner", self.nodes.run_planner) graph.add_node("human_review", self.nodes.human_review) graph.add_node("finalize_plan", self.nodes.finalize_plan) graph.add_edge(START, "run_planner") graph.add_edge("run_planner", "human_review") graph.add_conditional_edges("human_review", route_after_review) graph.add_edge("finalize_plan", END) # MemorySaver is required: interrupt() must persist state between # the first invoke() (pause) and the second invoke() (resume). return graph.compile(checkpointer=MemorySaver()) def save_figure(self): os.makedirs(FIGURE_DIR, exist_ok=True) mmd_path = os.path.join(FIGURE_DIR, "graph.mmd") png_path = os.path.join(FIGURE_DIR, "graph.png") with open(mmd_path, "w") as f: f.write(self.compiled_graph.get_graph().draw_mermaid()) with open(png_path, "wb") as f: f.write(self.compiled_graph.get_graph().draw_mermaid_png()) print(f" Graph saved → {mmd_path}") print(f" Graph saved → {png_path}") def get_compiled_graph(self): return self.compiled_graph

The route_after_review function is a simple one-liner, but it is what makes the revision loop possible. When approved is False, the graph goes back to run_planner — which calls the subgraph again with the updated request — and then arrives back at human_review for another round. The loop repeats until the traveler approves.

5.7

🚀 Runner & Console Output — travel_runner.py

TravelRunner wraps the compiled graph and exposes two methods: start_planning() for the first invoke and resume() for subsequent invokes. Each method passes the same thread_id so the checkpointer can match the correct paused state.

from langgraph.types import Command from graph import TravelGraph class TravelRunner: def __init__(self): self.travel_graph = TravelGraph() self.app = self.travel_graph.get_compiled_graph() def save_figure(self): self.travel_graph.save_figure() def _config(self, thread_id: str) -> dict: return {"configurable": {"thread_id": thread_id}} def start_planning(self, trip_request: str, thread_id: str) -> dict: """First invoke — runs until interrupt() and returns current state.""" return self.app.invoke( { "trip_request": trip_request, "destination_info": "", "itinerary": "", "budget_estimate": "", "human_feedback": "", "approved": False, "final_plan": "", }, config=self._config(thread_id), ) def resume(self, feedback: str, thread_id: str) -> dict: """Resume a paused graph — passes the traveler's response via Command.""" return self.app.invoke( Command(resume=feedback), config=self._config(thread_id), ) if __name__ == "__main__": SEP = "=" * 60 print(SEP) print(" LangGraph Basics — AI Travel Planner Demo") print(SEP) runner = TravelRunner() print("\n Saving graph architecture...") runner.save_figure() # Demo 1: direct approval print("\n" + "─" * 60) print(" Demo 1: 4-day Kyoto trip (direct approval)") print("─" * 60) thread_1 = "kyoto-001" result = runner.start_planning( "4-day cultural trip to Kyoto, Japan — budget $1,500", thread_1 ) print("\n✈️ GENERATED PLAN:") print(result.get("itinerary", "")) print("\n💰 BUDGET ESTIMATE:") print(result.get("budget_estimate", "")) print("\n👤 Traveler: approved") result = runner.resume("approved", thread_1) print("\n✅ FINAL CONFIRMED PLAN:") print(result.get("final_plan", "")) # Demo 2: revision then approval print("\n" + "─" * 60) print(" Demo 2: 5-day Barcelona trip (revise, then approve)") print("─" * 60) thread_2 = "barcelona-001" result = runner.start_planning( "5-day beach & culture trip to Barcelona, Spain — budget $2,000", thread_2 ) print("\n✈️ GENERATED PLAN:") print(result.get("itinerary", "")) feedback = "Please add more beach time on days 3 and 4, and swap the Picasso Museum for a tapas food tour." print(f"\n👤 Traveler: {feedback}") result = runner.resume(feedback, thread_2) print("\n✈️ REVISED PLAN:") print(result.get("itinerary", "")) print("\n👤 Traveler: approved") result = runner.resume("approved", thread_2) print("\n✅ FINAL CONFIRMED PLAN:") print(result.get("final_plan", "")) print("\n" + SEP)

Running python travel_runner.py produces output like this (LLM text abbreviated for clarity):

============================================================ LangGraph Basics — AI Travel Planner Demo ============================================================ Saving graph architecture... Graph saved → .../basics-6-subgraphs-interrupt-hitl/figure/graph.mmd Graph saved → .../basics-6-subgraphs-interrupt-hitl/figure/graph.png ──────────────────────────────────────────────────────────── Demo 1: 4-day Kyoto trip (direct approval) ──────────────────────────────────────────────────────────── ✈️ GENERATED PLAN: Day 1: Temples & Zen Gardens Morning: Fushimi Inari Shrine (arrive before 8 a.m. to beat the crowds) Afternoon: Arashiyama bamboo grove and Tenryu-ji garden Evening: Dinner at Nishiki Market food stalls Day 2: Imperial Kyoto ... 💰 BUDGET ESTIMATE: Accommodation: $75/night × 4 nights = $300 Food: $45/day × 4 days = $180 Local transportation: $35 Activities and entrance fees: $75 ───────────────────────────── Total estimated cost: $590 (well within the $1,500 budget) 👤 Traveler: approved ✅ FINAL CONFIRMED PLAN: ✅ Your trip is confirmed! Here's your final travel plan: Day 1: Temples & Zen Gardens ... ... Pack light and enjoy every moment — Kyoto awaits! ──────────────────────────────────────────────────────────── Demo 2: 5-day Barcelona trip (revise, then approve) ──────────────────────────────────────────────────────────── ✈️ GENERATED PLAN: Day 1: Gothic Quarter & Waterfront ... Day 3: Gaudí Day Morning: Sagrada Família tour Afternoon: Picasso Museum Evening: Tapas bar in El Born 👤 Traveler: Please add more beach time on days 3 and 4, and swap the Picasso Museum for a tapas food tour. ✈️ REVISED PLAN: Day 3: Gaudí & Food Tour Morning: Sagrada Família Afternoon: Barcelona tapas food tour through El Born & Barceloneta Evening: Sunset on Barceloneta beach Day 4: Beach & Relaxation Morning: Barceloneta beach swim Afternoon: Paddleboarding at Bogatell beach ... 👤 Traveler: approved ✅ FINAL CONFIRMED PLAN: ✅ Your trip is confirmed! Here's your final travel plan: ... Buen viaje — Barcelona is waiting! ============================================================
5.8

📊 Graph Diagram

The diagram below shows the parent TravelGraph. run_planner wraps the entire subgraph pipeline as a single node; human_review is where interrupt() fires. The conditional edge creates the revision loop.

flowchart TD S([__start__]) --> run_planner(run_planner) run_planner --> human_review(human_review) human_review -->|approved| finalize_plan(finalize_plan) human_review -->|revision| run_planner finalize_plan --> E([__end__]) style S fill:#e8f5e9,stroke:#43a047,color:#1b5e20 style E fill:#fce4ec,stroke:#e53935,color:#b71c1c style run_planner fill:#e3f2fd,stroke:#1e88e5,color:#0d47a1 style human_review fill:#fff3e0,stroke:#fb8c00,color:#e65100 style finalize_plan fill:#f3e5f5,stroke:#8e24aa,color:#4a148c

Figure 1: TravelGraph — run_planner invokes the PlannerSubGraph; human_review pauses with interrupt(); the conditional edge loops back for revisions.

The PlannerSubGraph is hidden inside run_planner — from the parent's perspective it is a single step. Its internal structure is a simple linear chain:

flowchart LR s([__start__]) --> A["research_destination"] A --> B["create_itinerary"] B --> C["estimate_budget"] C --> e([__end__]) style s fill:#e8f5e9,stroke:#43a047,color:#1b5e20 style e fill:#fce4ec,stroke:#e53935,color:#b71c1c style A fill:#e3f2fd,stroke:#1e88e5,color:#0d47a1 style B fill:#fff3e0,stroke:#fb8c00,color:#e65100 style C fill:#f3e5f5,stroke:#8e24aa,color:#4a148c

Figure 2: PlannerSubGraph — three sequential LLM nodes with their own private PlannerState.

🖥️ 6. Web Interface

The Gradio app in app.py brings the Travel Planner to the browser. Because HITL requires two separate invoke() calls, the UI needs a way to pass the same thread_id between the "Generate Plan" button click and the "Submit Response" button click. A hidden gr.State component solves this.

The UI has three phases: the traveler types their trip request and clicks "Generate My Plan" → the plan appears in a read-only textbox and an approval/feedback row becomes visible → the traveler types their response (or "approved") and clicks "Submit Response" → if approved, the final plan replaces the plan text and the feedback row hides; if revision, the revised plan shows and the row stays visible for another round.

import gradio as gr from travel_runner import TravelRunner class TravelApp: def __init__(self): self.runner = TravelRunner() self._counter = 0 def _new_thread_id(self) -> str: self._counter += 1 return f"trip-session-{self._counter}" def _format_plan(self, result: dict) -> str: parts = [] if result.get("destination_info"): parts.append(f"📍 DESTINATION OVERVIEW\n{result['destination_info']}") if result.get("itinerary"): parts.append(f"📅 DAY-BY-DAY ITINERARY\n{result['itinerary']}") if result.get("budget_estimate"): parts.append(f"💰 ESTIMATED BUDGET\n{result['budget_estimate']}") return "\n\n".join(parts) if parts else "" def plan_trip(self, trip_request: str, thread_id: str): if not trip_request.strip(): return "⚠️ Please describe your trip first.", thread_id, gr.update(visible=False) new_thread = self._new_thread_id() result = self.runner.start_planning(trip_request, new_thread) return self._format_plan(result), new_thread, gr.update(visible=True) def submit_feedback(self, feedback: str, thread_id: str): if not feedback.strip(): return "⚠️ Please enter your response.", thread_id, gr.update(visible=True) result = self.runner.resume(feedback, thread_id) if result.get("final_plan"): return result["final_plan"], thread_id, gr.update(visible=False) return self._format_plan(result), thread_id, gr.update(visible=True) def launch(self): with gr.Blocks(title="✈️ AI Travel Planner") as demo: thread_state = gr.State("") gr.Markdown("## ✈️ AI Travel Planner\n*Built with LangGraph Subgraphs + Human-in-the-Loop*") trip_input = gr.Textbox(label="Your Trip Request", lines=2, placeholder="e.g. 4-day cultural trip to Kyoto, Japan — budget $1,500") plan_btn = gr.Button("🗺️ Generate My Plan", variant="primary") plan_output = gr.Textbox(label="Generated Travel Plan", lines=18, interactive=False) with gr.Row(visible=False) as approval_section: feedback_input = gr.Textbox(label="Your Response", lines=2, placeholder="Type 'approved' to confirm, or describe your changes.") submit_btn = gr.Button("✅ Submit Response", variant="primary") plan_btn.click(fn=self.plan_trip, inputs=[trip_input, thread_state], outputs=[plan_output, thread_state, approval_section]) submit_btn.click(fn=self.submit_feedback, inputs=[feedback_input, thread_state], outputs=[plan_output, thread_state, approval_section]) demo.launch() if __name__ == "__main__": TravelApp().launch()

The approval_section row starts hidden (visible=False) and only appears after the first plan is generated. This keeps the UI clean — the traveler is not asked to approve something that does not exist yet. The thread_id in gr.State is set by plan_trip and reused unchanged by every subsequent submit_feedback call, ensuring all resume calls reach the correct paused state in the checkpointer.

AI Travel Planner Gradio web interface

Figure 3: AI Travel Planner web UI — trip request input, generated plan, and approval/feedback section.

✨ What to Try

Q "5-day cultural trip to Kyoto and Osaka, Japan — budget $2,000."
A The subgraph researches both cities, builds a combined day-by-day itinerary, and estimates the full budget. Review the plan and approve or request changes.
Q "7-day beach holiday in Bali, Indonesia — budget $1,800. Please add a cooking class."
A After reviewing the initial plan, type a revision request — the subgraph re-runs with your feedback and surfaces a revised itinerary for another round of approval.
Q "3-day food and architecture tour of Barcelona, Spain — budget $1,200."
A Type "approved" to accept the plan immediately and trigger the finalize node — which writes your polished, confirmed travel confirmation.

✅ 7. Conclusion

This post added two of the most powerful patterns in LangGraph to your toolkit. Subgraphs let you nest a compiled graph inside another as a reusable, black-box node with its own private state — each subgraph is independently testable, independently composable, and keeps its internals completely hidden from the parent. Human-in-the-Loop via interrupt() and Command(resume=...) lets you pause a running graph, surface information to a user, and continue only after receiving their input — enabling review-and-approve workflows that simply are not possible in a stateless pipeline.

Together, these two features address different sides of the same scaling problem. As graph complexity grows, subgraphs keep it manageable by decomposing the logic into focused, tested modules. As the stakes of automation grow, HITL keeps humans in control by inserting review checkpoints at exactly the right moments. The AI Travel Planner showed both working in tandem: a three-node subgraph handles all the LLM planning work privately, while the parent graph orchestrates the human review cycle and loops back for revisions until the traveler is satisfied.

  • Subgraphs — nested compiled graphs with private state, compiled once and reused as a black-box node
  • interrupt(payload) — pauses graph execution and surfaces a payload to the caller for review
  • Command(resume=value) — resumes a paused graph, injecting the human's response as the return value of interrupt()
  • MemorySaver — required checkpointer that persists state between the first invoke (pause) and the second invoke (resume)
  • Revision loop — conditional edge that re-runs the subgraph with traveler feedback until the plan is approved
🔗 LangGraph Basics Series — Complete!
This is Part 6 of 6 in the LangGraph Basics series. You now have every foundational building block: StateGraph, nodes, edges, state, reducers, conditional routing, checkpointers, streaming, tool calling, subgraphs, and Human-in-the-Loop. The LangGraph Advanced series (Parts 14–18 of this blog) puts all of these together in five production-grade applications.

Next up is the LangGraph Advanced series — five end-to-end projects that put everything together in production-grade applications:

  • Part 1 — Conversational Chatbot with Memory
  • Part 2 — Multi-Agent Supervisor Architecture
  • Part 3 — RAG Pipeline with Conditional Routing
  • Part 4 — ReAct Agent with Tool Calling
  • Part 5 — MCP Integration with LangGraph
Technical Stacks

Technical Stacks

Python Python
LangGraph LangGraph
LangChain LangChain
Gemini Gemini
Gradio Gradio
Download

Download Source Code

AI Travel Planner — LangGraph Basics Part 6

View on GitHub
📚

References

blog post

Related Blog Posts

1 / 1