LangGraph - Graph-Based Agent Workflows

The chain from the previous lessons always goes one way: researcher, analyst, writer, done. But what if the writer has to send work back to the researcher, or an agent should ask a human for approval before buying tickets? A real trail across the savanna has turnbacks and forks.

LangGraph is a framework for building complex, stateful AI applications based on graphs. It lets you create cycles, branches and advanced control flows for agents. Think of it as a trail map: nodes are camps, edges are paths, and the state is the backpack you carry from camp to camp.

What is LangGraph?

LangGraph extends LangChain's capabilities with:

• State Management - managing state between steps
• Cycles - the ability to create loops (unlike a DAG)
• Branching - conditional flow branches
• Human-in-the-loop - breakpoints for human interaction

The diagram shows a typical agent graph with tools, in which the flow can return to the agent many times:

1┌─────────────────────────────────────────────────────────────┐
2│                     LangGraph Flow                          │
3├─────────────────────────────────────────────────────────────┤
4│                                                             │
5│     ┌──────────┐                                            │
6│     │  START   │                                            │
7│     └────┬─────┘                                            │
8│          │                                                   │
9│          ▼                                                   │
10│     ┌──────────┐     ┌──────────────┐                       │
11│     │  Agent   │────▶│   Condition  │                       │
12│     │  Node    │     │   (Router)   │                       │
13│     └──────────┘     └──────┬───────┘                       │
14│          ▲                  │                               │
15│          │        ┌─────────┼─────────┐                     │
16│          │        ▼         ▼         ▼                     │
17│          │   ┌────────┐ ┌────────┐ ┌────────┐              │
18│          │   │ Tool A │ │ Tool B │ │  END   │              │
19│          │   └───┬────┘ └───┬────┘ └────────┘              │
20│          │       │          │                               │
21│          └───────┴──────────┘                               │
22│              (cycle back)                                    │
23│                                                             │
24└─────────────────────────────────────────────────────────────┘

The agent decides, the router picks the path to a tool or to the end, and after a tool is used the flow returns to the agent.

Installation and Basics

The most important element of LangGraph is the state. We define it as a TypedDict, a dictionary with described key types:

1# pip install langgraph langchain-openai
2
3from langgraph.graph import StateGraph, END
4from langgraph.checkpoint.memory import MemorySaver
5from typing import TypedDict, Annotated, Literal
6from langchain_openai import ChatOpenAI
7from langchain_core.messages import HumanMessage, AIMessage
8import operator
9
10# State definition - key element of LangGraph
11class AgentState(TypedDict):
12    messages: Annotated[list, operator.add]  # Message list
13    current_step: str
14    iteration_count: int
15
16# LLM model
17llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

Annotated[list, operator.add] tells LangGraph how to merge updates: new messages are appended to the list instead of replacing it. Keys without an annotation are overwritten. For chat there is also a ready-made add_messages reducer in the langgraph.graph.message module.

Basic Agent Graph

Building a graph always follows the same order: state definition, StateGraph, nodes (add_node), edges (add_edge) and compilation (compile). We start with the state and the agent node:

1from langgraph.graph import StateGraph, START, END
2
3# State definition
4class State(TypedDict):
5    messages: Annotated[list, operator.add]
6    next_action: str
7
8# Node functions
9def agent_node(state: State) -> State:
10    """Agent node - makes decisions."""
11    messages = state["messages"]
12
13    response = llm.invoke(messages)
14
15    return {
16        "messages": [response],
17        "next_action": "tool" if response.tool_calls else "end"
18    }

A node is an ordinary function: it receives the state and returns a dictionary with the keys it wants to change. For the model to return tool_calls at all, you have to attach the tools with llm.bind_tools(tools), otherwise the list will always be empty.

The second node executes tools, and the router picks the next step:

1def tool_node(state: State) -> State:
2    """Tool node - executes actions."""
3    last_message = state["messages"][-1]
4
5    # Execute tool calls
6    tool_results = []
7    for tool_call in last_message.tool_calls:
8        result = execute_tool(tool_call)
9        tool_results.append(result)
10
11    return {"messages": tool_results, "next_action": "agent"}
12
13def router(state: State) -> Literal["tool", "end"]:
14    """Router - decides the next step."""
15    return state["next_action"]

execute_tool is the place for your own logic. In practice you will replace this node with the ready-made ToolNode from langgraph.prebuilt.

Now we connect everything into a graph:

1# Build graph
2workflow = StateGraph(State)
3
4# Add nodes
5workflow.add_node("agent", agent_node)
6workflow.add_node("tool", tool_node)
7
8# Add edges
9workflow.add_edge(START, "agent")
10workflow.add_conditional_edges(
11    "agent",
12    router,
13    {
14        "tool": "tool",
15        "end": END
16    }
17)
18workflow.add_edge("tool", "agent")  # Cycle back to agent
19
20# Compile graph
21app = workflow.compile()
22
23# Run
24result = app.invoke({
25    "messages": [HumanMessage(content="What is the weather in the Serengeti?")],
26    "next_action": ""
27})

add_conditional_edges asks the router for a path name and translates it to a node with the dictionary. The edge from "tool" back to "agent" creates the cycle.

ReAct Agent with LangGraph

This pattern is so popular that there is a ready-made version. First the tools with the @tool decorator, known from the lesson on multi-agent systems:

1from langchain_core.tools import tool
2from langgraph.prebuilt import create_react_agent
3
4# Tool definitions
5@tool
6def get_weather(location: str) -> str:
7    """Gets weather for a Safari location."""
8    weather_data = {
9        "Serengeti": "28C, sunny",
10        "Masai Mara": "25C, cloudy",
11        "Kruger": "30C, hot"
12    }
13    return weather_data.get(location, f"No data for {location}")
14
15@tool
16def search_animals(query: str) -> str:
17    """Searches for information about Safari animals."""
18    return f"Information about {query}: A fascinating animal of the African savanna!"
19
20@tool
21def calculate_safari_cost(days: int, people: int) -> str:
22    """Calculates Safari cost."""
23    cost_per_day = 200
24    total = days * people * cost_per_day
25    return f"Safari cost: {total} USD ({days} days, {people} people)"

Every tool has a docstring, which the model reads as its description, and parameter types that tell it what to pass.

create_react_agent builds the whole agent-tools graph in a single line:

1# Create ReAct agent
2tools = [get_weather, search_animals, calculate_safari_cost]
3react_agent = create_react_agent(llm, tools)
4
5# Run
6result = react_agent.invoke({
7    "messages": [HumanMessage(content="Calculate the cost of a 5-day Safari for 4 people")]
8})
9print(result["messages"][-1].content)

The model will call calculate_safari_cost with days=5 and people=4 by itself. Note: in LangGraph 1.0 create_react_agent is marked as deprecated, and new code should use create_agent from the langchain.agents package.

Multi-Agent with LangGraph

A graph is a great fit for a team of agents. Each agent is a node, and the current_agent field in the state says who is next:

1from langgraph.graph import StateGraph, START, END
2from typing import Literal
3
4class MultiAgentState(TypedDict):
5    messages: Annotated[list, operator.add]
6    current_agent: str
7    task_complete: bool
8
9# Specialist agents
10def researcher_agent(state: MultiAgentState) -> MultiAgentState:
11    """Research agent - gathers information."""
12    system_prompt = "You are a Safari researcher. Gather information about animals and locations."
13
14    response = llm.invoke([
15        {"role": "system", "content": system_prompt},
16        *state["messages"]
17    ])
18
19    return {
20        "messages": [AIMessage(content=f"[Researcher]: {response.content}")],
21        "current_agent": "analyzer"
22    }

The researcher appends its answer to messages and names the analyzer as the next one.

The analyzer and the writer work the same way, only the writer finishes with the task_complete flag:

1def analyzer_agent(state: MultiAgentState) -> MultiAgentState:
2    """Analytical agent - analyzes data."""
3    system_prompt = "You are an analyst. Analyze gathered information and draw conclusions."
4
5    response = llm.invoke([
6        {"role": "system", "content": system_prompt},
7        *state["messages"]
8    ])
9
10    return {
11        "messages": [AIMessage(content=f"[Analyzer]: {response.content}")],
12        "current_agent": "writer"
13    }
14
15def writer_agent(state: MultiAgentState) -> MultiAgentState:
16    """Writer agent - creates the final report."""
17    system_prompt = "You are a writer. Create an engaging report based on the analysis."
18
19    response = llm.invoke([
20        {"role": "system", "content": system_prompt},
21        *state["messages"]
22    ])
23
24    return {
25        "messages": [AIMessage(content=f"[Writer]: {response.content}")],
26        "current_agent": "end",
27        "task_complete": True
28    }

Every agent sees the whole message history, because messages grows with each node.

What remains is the router and the graph:

1def router(state: MultiAgentState) -> Literal["analyzer", "writer", "end"]:
2    """Router between agents."""
3    return state["current_agent"]
4
5# Build multi-agent graph
6multi_agent_graph = StateGraph(MultiAgentState)
7
8multi_agent_graph.add_node("researcher", researcher_agent)
9multi_agent_graph.add_node("analyzer", analyzer_agent)
10multi_agent_graph.add_node("writer", writer_agent)
11
12multi_agent_graph.add_edge(START, "researcher")
13multi_agent_graph.add_conditional_edges(
14    "researcher",
15    router,
16    {"analyzer": "analyzer"}
17)
18multi_agent_graph.add_conditional_edges(
19    "analyzer",
20    router,
21    {"writer": "writer"}
22)
23multi_agent_graph.add_conditional_edges(
24    "writer",
25    router,
26    {"end": END}
27)
28
29multi_agent_app = multi_agent_graph.compile()
30
31# Run pipeline
32result = multi_agent_app.invoke({
33    "messages": [HumanMessage(content="Prepare a report about lions in the Serengeti")],
34    "current_agent": "researcher",
35    "task_complete": False
36})

The order is linear, but the router makes it easy to add a way back to the researcher.

Checkpointing and Persistence

By default the graph forgets everything after invoke. A checkpointer saves the state after every step, which lets you continue a conversation:

1from langgraph.checkpoint.memory import MemorySaver
2from langgraph.checkpoint.sqlite import SqliteSaver  # pip install langgraph-checkpoint-sqlite
3import sqlite3
4
5# Memory checkpoint (in-memory)
6memory_checkpointer = MemorySaver()
7
8# SQLite checkpoint (persistent)
9sqlite_checkpointer = SqliteSaver(sqlite3.connect("safari_agent.db", check_same_thread=False))
10
11# Compile with checkpointer
12app_with_memory = workflow.compile(checkpointer=memory_checkpointer)

MemorySaver keeps the state in memory and loses it on restart. SqliteSaver from the langgraph-checkpoint-sqlite package stores it in an SQLite database file.

Conversations are told apart by the thread_id in the config:

1# Run with thread_id for continuation
2config = {"configurable": {"thread_id": "safari-session-1"}}
3
4# First message
5result1 = app_with_memory.invoke(
6    {"messages": [HumanMessage(content="Tell me about lions")]},
7    config
8)
9
10# Continue conversation (same thread_id)
11result2 = app_with_memory.invoke(
12    {"messages": [HumanMessage(content="What about their hunting?")]},
13    config
14)
15
16# History is preserved!
17print(result2["messages"])

The second call with the same thread_id sees the first message, so the model knows that "their" means the lions.

Human-in-the-loop

Some decisions an agent should not make alone. First the state, the nodes and the router:

1from langgraph.graph import StateGraph, START, END
2from langgraph.checkpoint.memory import MemorySaver
3
4class HumanLoopState(TypedDict):
5    messages: Annotated[list, operator.add]
6    needs_approval: bool
7    approved: bool
8
9def agent_node(state: HumanLoopState) -> HumanLoopState:
10    """Agent proposes an action."""
11    response = llm.invoke(state["messages"])
12
13    # Check if action requires approval
14    needs_approval = "delete" in response.content.lower() or "buy" in response.content.lower()
15
16    return {
17        "messages": [response],
18        "needs_approval": needs_approval
19    }
20
21def human_approval_node(state: HumanLoopState) -> HumanLoopState:
22    """Human approval checkpoint."""
23    # This node stops execution
24    # User must approve before continuing
25    return {"approved": state.get("approved", False)}
26
27def execute_node(state: HumanLoopState) -> HumanLoopState:
28    """Executes the approved action."""
29    return {"messages": [AIMessage(content="Action executed!")]}
30
31def approval_router(state: HumanLoopState) -> Literal["human", "execute", "end"]:
32    """Router checking if approval is needed."""
33    if state.get("needs_approval") and not state.get("approved"):
34        return "human"
35    elif state.get("approved"):
36        return "execute"
37    return "end"

The human node computes nothing, it is only a stopping point. It returns one key unchanged, because returning the whole state would append all messages a second time through the operator.add reducer.

We compile the graph with a checkpointer and interrupt_before:

1# Graph with human-in-the-loop
2human_loop_graph = StateGraph(HumanLoopState)
3
4human_loop_graph.add_node("agent", agent_node)
5human_loop_graph.add_node("human", human_approval_node)
6human_loop_graph.add_node("execute", execute_node)
7
8human_loop_graph.add_edge(START, "agent")
9human_loop_graph.add_conditional_edges(
10    "agent",
11    approval_router,
12    {"human": "human", "execute": "execute", "end": END}
13)
14human_loop_graph.add_edge("human", "agent")  # After approval, return to agent
15human_loop_graph.add_edge("execute", END)
16
17# Compile with interrupt_before for human node
18app = human_loop_graph.compile(
19    checkpointer=MemorySaver(),
20    interrupt_before=["human"]  # Stop before human node
21)

The router returns "end", which is why the path dictionary translates that name to END. Without it, LangGraph raises an error about an unknown end node.

Running stops before the human node, and we approve and resume:

1# Run - will stop before human node
2config = {"configurable": {"thread_id": "approval-1"}}
3result = app.invoke(
4    {"messages": [HumanMessage(content="Delete all files")]},
5    config
6)
7
8# Check state
9state = app.get_state(config)
10print(f"Needs approval: {state.values.get('needs_approval')}")
11
12# User approves
13app.update_state(config, {"approved": True})
14
15# Continue execution
16final_result = app.invoke(None, config)

update_state records the approval, and invoke(None, config) resumes the graph while the router sends the flow to execute. In newer versions the interrupt() function called inside a node is more convenient.

Branching and Parallel Execution

Two independent branches can run at the same time. First the node functions:

1from langgraph.graph import StateGraph, START, END
2from typing import Literal
3
4class ParallelState(TypedDict):
5    query: str
6    weather_result: str
7    animal_result: str
8    combined_result: str
9
10def weather_branch(state: ParallelState) -> ParallelState:
11    """Fetches weather in parallel."""
12    # Simulated API call
13    return {"weather_result": "Weather: 28C, sunny"}
14
15def animal_branch(state: ParallelState) -> ParallelState:
16    """Fetches animal info in parallel."""
17    return {"animal_result": "Animals: Lions, elephants, giraffes"}
18
19def combine_results(state: ParallelState) -> ParallelState:
20    """Combines results from parallel branches."""
21    combined = f"{state['weather_result']}. {state['animal_result']}"
22    return {"combined_result": combined}

Each branch writes a different state key, so there is no conflict.

Two edges from START split the flow:

1# Graph with parallel execution
2parallel_graph = StateGraph(ParallelState)
3
4parallel_graph.add_node("weather", weather_branch)
5parallel_graph.add_node("animals", animal_branch)
6parallel_graph.add_node("combine", combine_results)
7
8# Parallel edges
9parallel_graph.add_edge(START, "weather")
10parallel_graph.add_edge(START, "animals")
11parallel_graph.add_edge("weather", "combine")
12parallel_graph.add_edge("animals", "combine")
13parallel_graph.add_edge("combine", END)
14
15parallel_app = parallel_graph.compile()
16
17result = parallel_app.invoke({"query": "Safari info"})
18print(result["combined_result"])

weather and animals run in the same step, and combine starts only when both have finished.

Supervisor Pattern

In this pattern one agent manages the rest. The supervisor asks the model who should work:

1from langgraph.graph import StateGraph, START, END
2from typing import Literal
3
4class SupervisorState(TypedDict):
5    messages: Annotated[list, operator.add]
6    next_worker: str
7    completed_workers: list[str]
8
9def supervisor_node(state: SupervisorState) -> SupervisorState:
10    """Supervisor decides which worker should act."""
11    system_prompt = """You are a supervisor managing a team:
12    - researcher: gathers information
13    - analyst: analyzes data
14    - writer: creates reports
15
16    Based on the task and results so far, choose the next worker.
17    Reply with ONLY the worker name or 'FINISH' if the task is complete."""
18
19    completed = state.get("completed_workers", [])
20
21    response = llm.invoke([
22        {"role": "system", "content": system_prompt},
23        {"role": "user", "content": f"Completed: {completed}\nTask: {state['messages'][-1].content}"}
24    ])
25
26    next_worker = response.content.strip().lower()
27
28    return {"next_worker": next_worker}

The supervisor returns only the name of the next worker.

A factory creates the worker nodes, and the router translates the decision:

1def worker_node(worker_name: str):
2    """Factory for worker nodes."""
3    def node(state: SupervisorState) -> SupervisorState:
4        prompts = {
5            "researcher": "You are a researcher. Gather information.",
6            "analyst": "You are an analyst. Analyze the data.",
7            "writer": "You are a writer. Write the report."
8        }
9
10        response = llm.invoke([
11            {"role": "system", "content": prompts[worker_name]},
12            *state["messages"]
13        ])
14
15        completed = state.get("completed_workers", []) + [worker_name]
16
17        return {
18            "messages": [AIMessage(content=f"[{worker_name}]: {response.content}")],
19            "completed_workers": completed
20        }
21    return node
22
23def supervisor_router(state: SupervisorState) -> Literal["researcher", "analyst", "writer", "end"]:
24    """Supervisor router."""
25    next_worker = state.get("next_worker", "").lower()
26    if next_worker == "finish" or next_worker == "end":
27        return "end"
28    return next_worker

If the model answers with something outside the list, the router returns an unknown name and the graph crashes, so in production enforce the answer format.

Every worker hands control back to the supervisor:

1# Build supervisor graph
2supervisor_graph = StateGraph(SupervisorState)
3
4supervisor_graph.add_node("supervisor", supervisor_node)
5supervisor_graph.add_node("researcher", worker_node("researcher"))
6supervisor_graph.add_node("analyst", worker_node("analyst"))
7supervisor_graph.add_node("writer", worker_node("writer"))
8
9supervisor_graph.add_edge(START, "supervisor")
10
11# All edges lead back to the supervisor
12for worker in ["researcher", "analyst", "writer"]:
13    supervisor_graph.add_edge(worker, "supervisor")
14
15supervisor_graph.add_conditional_edges(
16    "supervisor",
17    supervisor_router,
18    {
19        "researcher": "researcher",
20        "analyst": "analyst",
21        "writer": "writer",
22        "end": END
23    }
24)
25
26supervisor_app = supervisor_graph.compile()
27
28# Run
29result = supervisor_app.invoke({
30    "messages": [HumanMessage(content="Prepare a complete report on Safari in the Serengeti")],
31    "completed_workers": []
32})

I recommend starting with a linear graph and adding cycles and a supervisor when you really need decisions along the way.

LangGraph is a powerful tool for building complex AI workflows. In the next lesson you will learn about subagents and agent hierarchies!

Remember: LangGraph is a trail map on which you may turn back, split up and wait for the guide's decision.