• Github Repository

LanGraph Beyond Basics!

In the previous example I explored LangGraph's basics. Now let's go a little bit deeper and explore more advanced workflows.

• LanGraph Beyond Basics!
  • Workflow :: Parallelization
  • Workflow :: Routing
  • Workflow :: Orchestrator-Worker
  • Workflow :: Generator-Evaluator
  • Memory
  • Human in the Loop

Start by creating your virtual environment using uv:

uv init
uv sync
git add .
git commit -m "Initial commit"
uv add langchain-core langchain-openai langchain-community langgraph

And make sure your terminal is using it:

which python
/usr/bin/python
source .venv/bin/activate
which python
~/Projects/rag-experiments/langgraph_intro/.venv/bin/python

Workflow :: Parallelization

Running nodes in parallel in LangGraph is just connecting the START of your StateGraph to mor than one functional nodes - just like that:

from typing import TypedDict
from pydantic import SecretStr
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

class graph_schema(TypedDict):
    topic: str
    ig: str
    x: str
    li: str

demo_query: graph_schema = {
    "topic": "a very concise description of the PC game Unreal Tournament from 1999 by the campany Epic",
    "ig": "",
    "x": "",
    "li": "",
}

client = ChatOpenAI(
    model="qwen3.8:latest",
    temperature=0.6,
    top_p=0.95,
    base_url="http://localhost:11434/v1",
    api_key=SecretStr("ollama"),
)

def create_post_ig(state: graph_schema) -> dict:

    topic = state["topic"]
    post = client.invoke(f"Write a LinkedIn post about {topic}. Keep the tone Gen-z casual and engaging").content
    state["ig"] = str(post)

    return {"ig": post}

def create_post_x(state: graph_schema) -> dict:

    topic = state["topic"]
    post = client.invoke(f"Write a LinkedIn post about {topic}. Keep the tone like a mid-nineties action movie.").content
    state["x"] = str(post)

    return {"x": post}

def create_post_li(state: graph_schema) -> dict:

    topic = state["topic"]
    post = client.invoke(f"Write a LinkedIn post about {topic}. Keep the tone smart and business-like").content
    state["li"] = str(post)

    return {"li": post}

def build_state_graph():
    graph = StateGraph(graph_schema)

    graph.add_node("create_post_ig", create_post_ig)
    graph.add_node("create_post_x", create_post_x)
    graph.add_node("create_post_li", create_post_li)

    graph.add_edge(START, "create_post_ig")
    graph.add_edge(START, "create_post_x")
    graph.add_edge(START, "create_post_li")

    return graph.compile()


if __name__ == '__main__':
    parallel_graph = build_state_graph()
    response = parallel_graph.invoke(demo_query)

    print(response["ig"])
    print(response["x"])
    print(response["li"])

And here the results:

python -m langgraph_intro.06_langgraph_parallelization

• Instagram

ok so imagine it's 1999 and Epic Games drops Unreal Tournament like "we're just gonna give you a full-blown FPS with a jetpack, a hammer that hits like a small car, and a tournament mode that makes you feel like the main character of your own anime."

no lobby queues. no 47-minute matches. just you, a shotgun, and the absolute *audacity* to blast someone through a wall at 60fps while everyone's on a 56k connection.

it's giving "video games used to be a *thing* you played for hours and your mom actually came to check on you." lowkey the most unhinged little gem of the late '90s. no cap.

who else still has that hammer combo muscle memory? 🏈🔨

#UnrealTournament #EpicGames #RetroGaming #FPS #GamingHistory #Throwback

• Twitter / X

🔥 **UNREAL TOURNAMENT. 1999. EPIC GAMES.**

Let me make this simple.

You get a gun. You get a vehicle. You get a server full of 40 players who *think* they're good.

And then it ends. Fast.

No loading screens to hide behind. No "I'll just reposition." You're in the arena, the rocket launcher's already in your hand, and that guy in the blue vest? He's already a shadow on the wall.

Epic built a gunfight and called it a *game.* I call it a 90s action movie where YOU'RE the one-liner. Where you don't wait for the third act. You *are* the third act. You're the explosion in the parking garage. You're the "I'm back" except you never left.

40 players. One map. Zero mercy.

That's the whole pitch. That's the whole movie.

🎬 *Smirks. Racks a shotgun. Walks toward the screen.*

"Who the hell do you think you're teaming up with, buddy?"

—

Still the fastest, most *violent* multiplayer I've ever touched. 25 years later and the server browser still gives me chills.

If you played this, you know. You don't explain it. You just nod.

#UnrealTournament #EpicGames #90sGaming #FPS #Throwback #GamingHistory

• LinkedIn

Unreal Tournament (1999). Epic Games.

A fast-paced, arena-style FPS that did two things simultaneously: it delivered a genuinely compelling multiplayer experience, and it shipped the Unreal Engine into the hands of the entire industry.

The game itself was lean in scope—bots, weapon-based power mechanics, a handful of arenas, and a multiplayer loop tight enough to hold a player's attention for hours. What mattered more, in hindsight, was the platform it rode on. Epic used UT as a live demonstration of their in-house engine, effectively turning a product launch into a developer relations play.

That single decision—shipping the engine alongside the game—set the trajectory for one of the most commercially significant software tools in gaming history.

Worth a moment of recognition, 25+ years later.

#GameDevelopment #UnrealEngine #EpicGames #GamingHistory #Tech

Workflow :: Routing

Decision models - models that return a simple structured response instead of a wall of text - just became very popular. But the idea is not new ~ your regular model can do that as well... only much slower. So let's take the example from above and use our model in a schema-restrained form to decide what platform a post should land on based on it's topic - instead of writing all three:

from pydantic import BaseModel, Field, SecretStr
from typing import Literal, TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

class model_schema(BaseModel):
    target: Literal['ig', 'x', 'li'] = Field(..., description="The best target social network to which to generate the post for based on the user query - Instagram, Twitter/X or LinkedIn")
    topic: str = Field(..., description="The topic for the post to generate.")

class graph_schema(TypedDict):
    target: str
    input: str
    topic: str
    post: str

client = ChatOpenAI(
    model="qwen3.8:latest",
    temperature=0.6,
    top_p=0.95,
    base_url="http://localhost:11434/v1",
    api_key=SecretStr("ollama"),
)

client_with_structured_output = client.with_structured_output(model_schema)

def decider_node(state: graph_schema) -> graph_schema:
    user_query = state["input"]

    target_decision = client_with_structured_output.invoke(user_query)

    state["target"] = target_decision.target
    state["topic"] = target_decision.topic

    return state

def condition(state: graph_schema) -> str:
    target = state["target"]

    if target == "ig":
        return 'trigger_create_post_ig'
    elif target == "x":
        return 'trigger_create_post_x'
    elif target == "li":
        return 'trigger_create_post_li'
    else:
        raise ValueError("Invalid target")

def create_post_ig(state: graph_schema) -> dict:

    input = state["input"]
    post = client.invoke(f"Write a LinkedIn post about {input}. Keep the tone Gen-z casual and engaging").content
    state["post"] = str(post)

    return {"post": post}

def create_post_x(state: graph_schema) -> dict:

    input = state["input"]
    post = client.invoke(f"Write a LinkedIn post about {input}. Keep the tone like a mid-nineties action movie.").content
    state["post"] = str(post)

    return {"post": post}

def create_post_li(state: graph_schema) -> dict:

    input = state["input"]
    post = client.invoke(f"Write a LinkedIn post about {input}. Keep the tone smart and business-like").content
    state["post"] = str(post)

    return {"post": post}

def build_state_graph():
    routed_graph = StateGraph(graph_schema)

    routed_graph.add_node("decider_node", decider_node)
    routed_graph.add_node("create_post_ig", create_post_ig)
    routed_graph.add_node("create_post_x", create_post_x)
    routed_graph.add_node("create_post_li", create_post_li)

    routed_graph.add_edge(START, "decider_node")
    routed_graph.add_conditional_edges("decider_node", condition, {
        "trigger_create_post_ig": "create_post_ig",
        "trigger_create_post_x": "create_post_x",
        "trigger_create_post_li": "create_post_li"
    })
    routed_graph.add_edge("create_post_ig", END)
    routed_graph.add_edge("create_post_x", END)
    routed_graph.add_edge("create_post_li", END)

    return routed_graph.compile()


if __name__ == '__main__':
    parallel_graph = build_state_graph()
    response = parallel_graph.invoke({
        "input": "Write a post about how sick 90s and early 2000s sitcoms were - contrast the differences between Love and a Mariage and the show Friends",
        "target": "",
        "topic": "",
        "post": ""
    })

    print(response["topic"])
    print(response["post"])

And the result ->

python -m langgraph_intro.07_langgraph_routing

90s-2000s sitcoms were sick compared to Friends, using Love and a Mariage as the contrast
Look.

I'm not here to do a "thought leadership" thing. I'm not here to "circle back" or "touch base." I'm here to tell you something, and I'm only gonna say it once.

The 90s and early 2000s sitcoms were *sick.*

Not "sick" like you got a tummy bug. SICK. Like a .44 in a rain-soaked alley. Like the second act of *Lethal Weapon* where Riggs just *stops being a man and becomes a force of nature.*

Let me paint you a picture.

**Love and a Marriage.** 1994. Canadian. Two people. A kitchen. A marriage that's *bleeding* and nobody's got a laugh track to patch it up. No central Perk. No six-person ensemble doing a synchronized "OH MY GOD" while a studio audience claps in 4/4 time. Just... two human beings in a room, saying the quiet, ugly, *real* thing. The kind of quiet that makes you put the glass down and stare at the wall. The kind of quiet that *hurts.*

That show didn't need a theme song. It didn't need a "PIVOT!" It needed a Tuesday. A *real* Tuesday. And it gave it to you. Cold. Unflinching. No safety net.

Now.

**Friends.** 1994. New York. Six people. A coffee shop that exists *solely* so they can sit in a circle and *perform* their lives to 800 people in a studio. "How you doin'?" "WE WERE ON A BREAK." A man in a *turtleneck* delivering a monologue about a *sandwich.* And the audience goes *WOOO.* And you *believe them.* And for twenty-two minutes, the whole goddamn world is a little less grey.

That show was a *nuke.* A full-spectrum, multi-cam, laugh-track-laced, "I'll have what she's having" nuke, and it hit you in the *chest* and you didn't even see it coming.

One was a knife in the dark. The other was a fireworks show over the Hudson.

And here's the thing nobody's gonna tell you on this platform with their "5 Tips for Synergizing Your Stakeholders" carousel:

*Both* of them were doing something the modern landscape can't. They were *committing.* To a tone. To a room. To a *feeling.* No jump cuts. No "wait for it" TikTok edit. No algorithm deciding your fourth episode is a 4K remaster of a show that was better in VHS grain. They just *played the scene.* Let it breathe. Let it *land.*

I've been in the room. I've watched both. I've sat in a dark apartment at 11 PM with a bowl of cereal and felt a *Canadian couple argue about a mortgage* hit me harder than a 90-second trailer for a Marvel film.

And I'm not mad. I'm just... *done* with the narrative that "content" is a noun. That a show is a "product." That you "consume" a 22-minute scene where a woman says, "I don't recognize you anymore," and walks to the window.

You don't *consume* that. You *survive* it.

So here's my ask, and I'm not gonna dress it up in a "call to action" box:

Go find an old show. A *real* old one. Not a reboot. Not a "reimagining." The 1996 one where the laugh track is half a second late and the lighting looks like a hardware store. Sit with it. Let it be *sick.* Let it be *quiet.* Let it be *loud.* Let it be a kitchen in Toronto or a couch in Manhattan.

You don't owe the algorithm your attention. You owe *yourself* the 22 minutes where someone just... *talks to another person* and the camera *stays on them.*

That's not nostalgia. That's a standard.

And I'm not lowering it.

—

#TV #Storytelling #Nostalgia #Creative #Content #Sitcoms #LoveAndAMarriage #Friends #1990s #Television #Craft #Writing #Media

Workflow :: Orchestrator-Worker

Let one orchestrator node delegate it's workload onto worker threads:

from pydantic import BaseModel, Field, SecretStr
from typing import TypedDict
from concurrent.futures import ThreadPoolExecutor

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END


class model_schema(BaseModel):
    tasks: list[str] = Field(..., description="Evaluate the user query to generate this list of invidiual tasks to be performed by the worker.")

class graph_schema(TypedDict):
    tasks: list[str]
    query: str
    results: list[str]
    summary: str

client = ChatOpenAI(
    model="qwen3.8:latest",
    temperature=0.6,
    top_p=0.95,
    base_url="http://localhost:11434/v1",
    api_key=SecretStr("ollama"),
)

client_with_structured_output = client.with_structured_output(model_schema)


def orchestrator_node(state: graph_schema) -> graph_schema:
    user_query = state["query"]

    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are an orchestrator that breaks down a user query into tasks for the worker."),
        ("user", f"User query: {user_query}. Please generate one prompt per task for the worker to complete.")
    ])

    chain = prompt | client_with_structured_output

    task_list = chain.invoke({"query": user_query})

    state["tasks"] = task_list.tasks

    return state


def execute(query: str):
    response = client.invoke(f"Please execute this task: {query}")

    return response.content


def worker_node(state: graph_schema) -> graph_schema:

    tasks = state["tasks"]
    results = []

    # spawn as many workers as there are tasks
    with ThreadPoolExecutor(max_workers=len(tasks)) as executer:
        # and divide the lists of tasks on your workers
        results_futures = (executer.map(execute, tasks))
        # once resolved append to results
        for result in results_futures:
            results.append(result)

    state["results"] = results

    return state


def collector_node(state: graph_schema):

    results = state["results"]

    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a collector that summarized the results from the worker."),
        ("user", f"Here are the results from the worker: {results}. Please provide an easy to read, concise and structured summary of the results.")
    ])

    chain = prompt | client

    summary = chain.invoke({"results": results})

    state["summary"] = str(summary.content)

    return state

def build_state_graph():
    orchestrator_graph = StateGraph(graph_schema)

    orchestrator_graph.add_node("orchestrator_node", orchestrator_node)
    orchestrator_graph.add_node("worker_node", worker_node)
    orchestrator_graph.add_node("collector_node", collector_node)

    orchestrator_graph.add_edge(START, "orchestrator_node")
    orchestrator_graph.add_edge("orchestrator_node", "worker_node")
    orchestrator_graph.add_edge("worker_node", "collector_node")
    orchestrator_graph.add_edge("collector_node", END)

    return orchestrator_graph.compile()


if __name__ == '__main__':
    orchestrator_graph = build_state_graph()
    response = orchestrator_graph.invoke({
        "query": "Give me the population numbers of the following cities: New York, Shanghai, Sydney, Hongkong, Berlin",
        "tasks": [],
        "results": [],
        "summary": ""
    })

    print(response["tasks"])
    print(response["summary"])

Let's check -> this is starting off 4 independent workers that return the answers to the orchestrator in parallel:

python -m langgraph_intro.08_langgraph_ochestrator_worker

['Find and report the current population of New York City, including the most recent reliable estimate and the source/year of the data.', 'Find and report the current population of Shanghai, including the most recent reliable estimate and the source/year of the data.', 'Find and report the current population of Sydney, including the most recent reliable estimate and the source/year of the data.', 'Find and report the current population of Hong Kong, including the most recent reliable estimate and the source/year of the data.', 'Find and report the current population of Berlin, including the most recent reliable estimate and the source/year of the data.']

# City Populations – Summary

| City | Most Recent Estimate | Baseline Census | Trend |
|---|---|---|---|
| **Shanghai, China** | **~24.87 M** (2023) | 24,870,895 (2020 7th Census) | Stable |
| **New York City, USA** | **~8.2–8.3 M** (2023 est.) | 8,804,190 (2020 Census) | Declining (pandemic out-migration) |
| **Hong Kong, SAR** | **~7.4–7.5 M** (end-2023) | 7,411,700 (2021 Census) | Flat / slightly declining |
| **Sydney, Australia** | **~5.3–5.4 M** (2024 est.) | 5,312,163 (2021 Census) | Growing |
| **Berlin, Germany** | **~3.81–3.82 M** (end-2023) | n/a (annual official count) | Growing (net migration) |

## Key Takeaways

- **Shanghai** is by far the largest, nearly 3× the size of NYC.
- **NYC** is the only city showing a notable **decline** (≈ 500 K below its 2020 Census peak), largely due to 2020–2022 out-migration.
- **Sydney** and **Berlin** are on **growth trajectories**, driven by migration and natural increase.
- **Hong Kong** has been roughly **flat or slightly down** since 2020, reflecting low birth rates and net emigration.

## Authoritative Sources for Each City

| City | Primary Source |
|---|---|
| Shanghai | Shanghai Municipal Bureau of Statistics / National Bureau of Statistics of China |
| New York City | U.S. Census Bureau (Population Estimates Program) |
| Hong Kong | HK Census & Statistics Department (C&SD) |
| Sydney | Australian Bureau of Statistics (ABS) |
| Berlin | Statistisches Landesamt Berlin / Destatis |

## Caveat

All figures reflect data available up to the model's training cutoff. For the **very latest** numbers (mid-2025), check each city's official statistics agency directly (links above).

Workflow :: Generator-Evaluator

Next - let's let an evaluator node critique the output of a generator node until a standard of quality is reached:

from pydantic import BaseModel, Field, SecretStr
from typing import TypedDict
from concurrent.futures import ThreadPoolExecutor

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END


class model_schema(BaseModel):
    funny: bool = Field(..., description="True if the generated joke is funny, False if it is not.")
    feedback: str = Field(default="", description="Concrete feedback on how to make the joke funnier, or an empty string if it is already funny.")

class graph_schema(TypedDict):
    topic: str
    joke: str
    funny: bool
    feedback: str
    iterations: int

client = ChatOpenAI(
    model="qwen3.8:latest",
    temperature=0.6,
    top_p=0.95,
    base_url="http://localhost:11434/v1",
    api_key=SecretStr("ollama"),
)

client_with_structured_output = client.with_structured_output(model_schema)


def generator_node(state: graph_schema) -> graph_schema:

    topic = state["topic"]

    if state['feedback']:
        response = client.invoke(f"Please modify the following joke: {state['joke']} based on the evaluators feedback: {state['feedback']}")
    else:
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a joke creator and will be evaluated on your joke quality. You will receive a topic and then generate a joke around it."),
            ("user", f"Here is the topic: {topic}. Please create a funny joke about it.")
        ])

        chain = prompt | client
        response = chain.invoke({"topic": topic})

    state["joke"] = str(response.content)

    return state

def evaluator_node(state: graph_schema) -> graph_schema:

    joke = state["joke"]
    iteration = state['iterations']
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a comedy critic. Your job is to evaluat the following joke and provide feedback on how to make it funnier."),
        ("user", f"Evaluate the following joke: {joke}. Respond with True if it is funny and False if it is not and provide feedback if it's not funny.")
    ])

    chain = prompt | client_with_structured_output
    evaluation = chain.invoke({"joke": joke})
    state["funny"] = evaluation.funny
    state["feedback"] = evaluation.feedback
    state["iterations"] = iteration + 1

    return state

def stop(state: graph_schema) -> str:
    if state["funny"] or state["iterations"] >=5:
        return "end"
    else:
        return "evaluator_node"


def build_state_graph():
    evaluator_graph = StateGraph(graph_schema)

    evaluator_graph.add_node("generator_node", generator_node)
    evaluator_graph.add_node("evaluator_node", evaluator_node)

    evaluator_graph.add_edge(START, "generator_node")
    evaluator_graph.add_conditional_edges("generator_node", stop, {
        "evaluator_node": "evaluator_node",
        "end": END
    })
    evaluator_graph.add_edge("evaluator_node", "generator_node")
    evaluator_graph.add_edge("generator_node", END)

    return evaluator_graph.compile()


if __name__ == '__main__':
    evaluator_graph = build_state_graph()
    response = evaluator_graph.invoke({
        "topic": "The Rust programming language",
        "joke": "",
        "funny": False,
        "feedback": "",
        "iterations": 0,
    })

    print(response["funny"])
    print(response["joke"])

And here is the outcome:

python -m  langgraph_intro.09_langgraph_generator_evaluator

True
My Rust compiler is like that one ex — the one who still has your toothbrush, the one who "helps" you rewrite your LinkedIn profile at 2 a.m. while you're trying to sleep.

It remembers every mistake I've ever made.
It won't let me move on.
And it will absolutely let me know — in 200 lines of error messages, each one a little more condescending than the last, all in monospace, all in that specific shade of red that says *I've been through this before and I am not doing it again* — exactly why I'm not good enough.

The real problem, though…

…is that it's right.

And it still leaves your code on the counter. Like a passive-aggressive note.

*I refactored this for you. Don't make me do it again.*

Memory

For Chatbot interactions it is nice to have the agent remember the conversation in between prompts. This can be done manually by appending every conversation to the following prompt. But LangGraph also adds a CheckPoint function that does it for you:

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
from pydantic import SecretStr
from langchain_openai import ChatOpenAI

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

client = ChatOpenAI(
    model="qwen3.8:latest",
    temperature=0.6,
    top_p=0.95,
    base_url="http://localhost:11434/v1",
    api_key=SecretStr("ollama")
)

# Define the schema

class graph_shema(TypedDict):
    messages: Annotated[list[str], add_messages] # <- HAS TO BE CALLED MESSAGES, otherwise memory does not work...

# Create Node functions

def welcome(state: graph_shema) -> graph_shema:
    current_message = state['messages']

    resonse = client.invoke(current_message).content

    state['messages'] = [f"You asked {current_message}. Here is the response: {resonse}."]

    return state

# Create the State Graph with Memory Checkpoint

memory_graph = StateGraph(graph_shema)
checkpoint = InMemorySaver() # <- initialize the CheckPoint system

memory_graph.add_node("welcome", welcome)

memory_graph.add_edge(START, "welcome")
memory_graph.add_edge("welcome", END)
graph = memory_graph.compile(checkpointer=checkpoint) # <- Compile the graph with CheckPoints

# Run Graph with configuration

response = graph.invoke(
    {"messages": ["Hey there! My name is Mike. Who are you?"]},
    {"configurable":  {"thread_id": "this-is-a-memory-test"}} # <- Make sure that all requests with shared memory use the same thread ID
)

for message in response['messages']:
    message.pretty_print()

response2 = graph.invoke(
    {"messages": ["Hey there! Do you know who I am?"]},
    {"configurable":  {"thread_id": "this-is-a-memory-test"}} # <- Make sure that all requests with shared memory use the same thread ID
)

for message in response2['messages']:
    message.pretty_print()

python -m  langgraph_intro.10_langgraph_memory
================================ Human Message =================================

Hey there! My name is Mike. Who are you?
================================ Human Message =================================

You asked [HumanMessage(content='Hey there! My name is Mike. Who are you?', additional_kwargs={}, response_metadata={}, id='8210da7d-1858-447e-86dc-acfe8fb17a83')]. Here is the response: Hey Mike! Nice to meet you. I'm Qwen (Tongyi Qianwen), a large language model developed by Alibaba Group's Tongyi Lab. I'm here to help with all sorts of things — whether that's answering questions, brainstorming ideas, writing, coding, analyzing problems, or just having a good conversation.

How can I help you today?.
================================ Human Message =================================

Hey there! My name is Mike. Who are you?
================================ Human Message =================================

You asked [HumanMessage(content='Hey there! My name is Mike. Who are you?', additional_kwargs={}, response_metadata={}, id='8210da7d-1858-447e-86dc-acfe8fb17a83')]. Here is the response: Hey Mike! Nice to meet you. I'm Qwen (Tongyi Qianwen), a large language model developed by Alibaba Group's Tongyi Lab. I'm here to help with all sorts of things — whether that's answering questions, brainstorming ideas, writing, coding, analyzing problems, or just having a good conversation.

How can I help you today?.
================================ Human Message =================================

Hey there! Do you know who I am?
================================ Human Message =================================

You asked [HumanMessage(content='Hey there! My name is Mike. Who are you?', additional_kwargs={}, response_metadata={}, id='8210da7d-1858-447e-86dc-acfe8fb17a83'), HumanMessage(content="You asked [HumanMessage(content='Hey there! My name is Mike. Who are you?', additional_kwargs={}, response_metadata={}, id='8210da7d-1858-447e-86dc-acfe8fb17a83')]. Here is the response: Hey Mike! Nice to meet you. I'm Qwen (Tongyi Qianwen), a large language model developed by Alibaba Group's Tongyi Lab. I'm here to help with all sorts of things — whether that's answering questions, brainstorming ideas, writing, coding, analyzing problems, or just having a good conversation.\n\nHow can I help you today?.", additional_kwargs={}, response_metadata={}, id='9f88a5f3-3ce5-43bc-9d40-468c9a3c1724'), HumanMessage(content='Hey there! Do you know who I am?', additional_kwargs={}, response_metadata={}, id='b7016876-a090-4fe7-961d-d00f09fa548a')]. Here is the response: Hey Mike! Yes, you told me in your first message — your name is Mike. Nice to have a name to work with! 😄

Is there something I can help you with today?.

Human in the Loop

Add a node that forces the agent to seek user approval before continuing with a task. To avoid having to re-run the entire graph this feature is build on the Memory feature above - allowing the agent to resume operation after the approval was received.

from typing import Annotated, Literal, Optional, TypedDict
from langgraph.graph.message import add_messages
from pydantic import SecretStr
from langchain_openai import ChatOpenAI

from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import MemorySaver

client = ChatOpenAI(
    model="qwen3.8:latest",
    temperature=0.6,
    top_p=0.95,
    base_url="http://localhost:11434/v1",
    api_key=SecretStr("ollama")
)

# Define the schema

class graph_shema(TypedDict):
    messages: Annotated[list[str], add_messages] # <- HAS TO BE CALLED MESSAGES, otherwise memory does not work...

class ApprovalState(TypedDict):
    action_details: str
    status: Optional[Literal["pending", "approved", "rejected"]]

# Create Node functions

def approval_node(state: ApprovalState) -> Command[Literal["proceed", "cancel"]]:
   decision = interrupt({
   "question": "Do you approve this action?",
   "details": state["action_details"],
   })

   return Command(goto="proceed" if decision else "cancel")

def proceed_node(state: ApprovalState):
    return {"status": "approved"}

def cancel_node(state: ApprovalState):
    return {"status": "rejected"}

approval_graph = StateGraph(ApprovalState)
checkpoint = MemorySaver()

approval_graph.add_node("approval", approval_node)
approval_graph.add_node("proceed", proceed_node)
approval_graph.add_node("cancel", cancel_node)

approval_graph.add_edge(START, "approval")
approval_graph.add_edge("proceed", END)
approval_graph.add_edge("cancel", END)

graph = approval_graph.compile(checkpointer=checkpoint)
config = {"configurable": {"thread_id": "approval-flow"}}

# Run Graph with configuration

initial = graph.invoke(
    {"action_details": "Send sumary by Email?", "status": "pending"},
    config=config,
)

# The graph is now PAUSED at the interrupt: nothing after "approval" has run.
# The pending question lives in state, keyed by "__interrupt__":
pending = initial["__interrupt__"][0].value
print(pending)  # {'question': 'Do you approve this action?', 'details': 'Send sumary by Email?'}

# --- THIS is the human in the loop: stop the process and wait for a real answer.
answer = input(f"{pending['question']} ({pending['details']}) [y/n] ").strip().lower()
resume_value = answer in ("y", "yes")

# Resuming on the SAME thread_id replays the interrupted node;
# interrupt() returns whatever we resume with, so the branch follows the answer.
resumed = graph.invoke(
    Command(resume=resume_value),
    config=config,
)

print(resumed["status"])

python -m  langgraph_intro.11_langgraph_human-in-the-loop

{'question': 'Do you approve this action?', 'details': 'Send sumary by Email?'}
Do you approve this action? (Send sumary by Email?) [y/n] y

approved