LangGraph - Graph-Based Agent Workflows

LangGraph is a framework for building complex, stateful AI applications based on graphs. It allows you to create cycles, branches, and advanced control flows for agents!

What is LangGraph?

LangGraph extends LangChain's capabilities with:

• State Management - managing state between steps
• Cycles - ability to create loops (unlike DAGs)
• Branching - conditional flow branching
• Human-in-the-loop - pause points for human interaction

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└─────────────────────────────────────────────────────────────┘

Installation and Basics

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)

Basic Agent Graph

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    }
19
20def tool_node(state: State) -> State:
21    """Tool node - executes actions."""
22    last_message = state["messages"][-1]
23
24    # Execute tool calls
25    tool_results = []
26    for tool_call in last_message.tool_calls:
27        result = execute_tool(tool_call)
28        tool_results.append(result)
29
30    return {"messages": tool_results, "next_action": "agent"}
31
32def router(state: State) -> Literal["tool", "end"]:
33    """Router - decides the next step."""
34    return state["next_action"]
35
36# Build graph
37workflow = StateGraph(State)
38
39# Add nodes
40workflow.add_node("agent", agent_node)
41workflow.add_node("tool", tool_node)
42
43# Add edges
44workflow.add_edge(START, "agent")
45workflow.add_conditional_edges(
46    "agent",
47    router,
48    {
49        "tool": "tool",
50        "end": END
51    }
52)
53workflow.add_edge("tool", "agent")  # Cycle back to agent
54
55# Compile graph
56app = workflow.compile()
57
58# Run
59result = app.invoke({
60    "messages": [HumanMessage(content="What is the weather in the Serengeti?")],
61    "next_action": ""
62})

ReAct Agent with LangGraph

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)"
26
27# Create ReAct agent
28tools = [get_weather, search_animals, calculate_safari_cost]
29react_agent = create_react_agent(llm, tools)
30
31# Run
32result = react_agent.invoke({
33    "messages": [HumanMessage(content="Calculate the cost of a 5-day Safari for 4 people")]
34})
35print(result["messages"][-1].content)

Multi-Agent with LangGraph

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    }
23
24def analyzer_agent(state: MultiAgentState) -> MultiAgentState:
25    """Analytical agent - analyzes data."""
26    system_prompt = "You are an analyst. Analyze gathered information and draw conclusions."
27
28    response = llm.invoke([
29        {"role": "system", "content": system_prompt},
30        *state["messages"]
31    ])
32
33    return {
34        "messages": [AIMessage(content=f"[Analyzer]: {response.content}")],
35        "current_agent": "writer"
36    }
37
38def writer_agent(state: MultiAgentState) -> MultiAgentState:
39    """Writer agent - creates the final report."""
40    system_prompt = "You are a writer. Create an engaging report based on the analysis."
41
42    response = llm.invoke([
43        {"role": "system", "content": system_prompt},
44        *state["messages"]
45    ])
46
47    return {
48        "messages": [AIMessage(content=f"[Writer]: {response.content}")],
49        "current_agent": "end",
50        "task_complete": True
51    }
52
53def router(state: MultiAgentState) -> Literal["analyzer", "writer", "end"]:
54    """Router between agents."""
55    return state["current_agent"]
56
57# Build multi-agent graph
58multi_agent_graph = StateGraph(MultiAgentState)
59
60multi_agent_graph.add_node("researcher", researcher_agent)
61multi_agent_graph.add_node("analyzer", analyzer_agent)
62multi_agent_graph.add_node("writer", writer_agent)
63
64multi_agent_graph.add_edge(START, "researcher")
65multi_agent_graph.add_conditional_edges(
66    "researcher",
67    router,
68    {"analyzer": "analyzer"}
69)
70multi_agent_graph.add_conditional_edges(
71    "analyzer",
72    router,
73    {"writer": "writer"}
74)
75multi_agent_graph.add_conditional_edges(
76    "writer",
77    router,
78    {"end": END}
79)
80
81multi_agent_app = multi_agent_graph.compile()
82
83# Run pipeline
84result = multi_agent_app.invoke({
85    "messages": [HumanMessage(content="Prepare a report about lions in the Serengeti")],
86    "current_agent": "researcher",
87    "task_complete": False
88})

Checkpointing and Persistence

1from langgraph.checkpoint.memory import MemorySaver
2from langgraph.checkpoint.sqlite import SqliteSaver
3
4# Memory checkpoint (in-memory)
5memory_checkpointer = MemorySaver()
6
7# SQLite checkpoint (persistent)
8sqlite_checkpointer = SqliteSaver.from_conn_string("safari_agent.db")
9
10# Compile with checkpointer
11app_with_memory = workflow.compile(checkpointer=memory_checkpointer)
12
13# Run with thread_id for continuation
14config = {"configurable": {"thread_id": "safari-session-1"}}
15
16# First message
17result1 = app_with_memory.invoke(
18    {"messages": [HumanMessage(content="Tell me about lions")]},
19    config
20)
21
22# Continue conversation (same thread_id)
23result2 = app_with_memory.invoke(
24    {"messages": [HumanMessage(content="What about their hunting?")]},
25    config
26)
27
28# History is preserved!
29print(result2["messages"])

Human-in-the-loop

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 state
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"
38
39# Graph with human-in-the-loop
40human_loop_graph = StateGraph(HumanLoopState)
41
42human_loop_graph.add_node("agent", agent_node)
43human_loop_graph.add_node("human", human_approval_node)
44human_loop_graph.add_node("execute", execute_node)
45
46human_loop_graph.add_edge(START, "agent")
47human_loop_graph.add_conditional_edges("agent", approval_router)
48human_loop_graph.add_edge("human", "agent")  # After approval, return to agent
49human_loop_graph.add_edge("execute", END)
50
51# Compile with interrupt_before for human node
52app = human_loop_graph.compile(
53    checkpointer=MemorySaver(),
54    interrupt_before=["human"]  # Stop before human node
55)
56
57# Run - will stop before human node
58config = {"configurable": {"thread_id": "approval-1"}}
59result = app.invoke(
60    {"messages": [HumanMessage(content="Delete all files")]},
61    config
62)
63
64# Check state
65state = app.get_state(config)
66print(f"Needs approval: {state.values.get('needs_approval')}")
67
68# User approves
69app.update_state(config, {"approved": True})
70
71# Continue execution
72final_result = app.invoke(None, config)

Branching and Parallel Execution

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}
23
24# Graph with parallel execution
25parallel_graph = StateGraph(ParallelState)
26
27parallel_graph.add_node("weather", weather_branch)
28parallel_graph.add_node("animals", animal_branch)
29parallel_graph.add_node("combine", combine_results)
30
31# Parallel edges
32parallel_graph.add_edge(START, "weather")
33parallel_graph.add_edge(START, "animals")
34parallel_graph.add_edge("weather", "combine")
35parallel_graph.add_edge("animals", "combine")
36parallel_graph.add_edge("combine", END)
37
38parallel_app = parallel_graph.compile()
39
40result = parallel_app.invoke({"query": "Safari info"})
41print(result["combined_result"])

Supervisor Pattern

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}
29
30def worker_node(worker_name: str):
31    """Factory for worker nodes."""
32    def node(state: SupervisorState) -> SupervisorState:
33        prompts = {
34            "researcher": "You are a researcher. Gather information.",
35            "analyst": "You are an analyst. Analyze the data.",
36            "writer": "You are a writer. Write the report."
37        }
38
39        response = llm.invoke([
40            {"role": "system", "content": prompts[worker_name]},
41            *state["messages"]
42        ])
43
44        completed = state.get("completed_workers", []) + [worker_name]
45
46        return {
47            "messages": [AIMessage(content=f"[{worker_name}]: {response.content}")],
48            "completed_workers": completed
49        }
50    return node
51
52def supervisor_router(state: SupervisorState) -> Literal["researcher", "analyst", "writer", "end"]:
53    """Supervisor router."""
54    next_worker = state.get("next_worker", "").lower()
55    if next_worker == "finish" or next_worker == "end":
56        return "end"
57    return next_worker
58
59# Build supervisor graph
60supervisor_graph = StateGraph(SupervisorState)
61
62supervisor_graph.add_node("supervisor", supervisor_node)
63supervisor_graph.add_node("researcher", worker_node("researcher"))
64supervisor_graph.add_node("analyst", worker_node("analyst"))
65supervisor_graph.add_node("writer", worker_node("writer"))
66
67supervisor_graph.add_edge(START, "supervisor")
68
69# All edges lead back to the supervisor
70for worker in ["researcher", "analyst", "writer"]:
71    supervisor_graph.add_edge(worker, "supervisor")
72
73supervisor_graph.add_conditional_edges(
74    "supervisor",
75    supervisor_router,
76    {
77        "researcher": "researcher",
78        "analyst": "analyst",
79        "writer": "writer",
80        "end": END
81    }
82)
83
84supervisor_app = supervisor_graph.compile()
85
86# Run
87result = supervisor_app.invoke({
88    "messages": [HumanMessage(content="Prepare a complete report on Safari in the Serengeti")],
89    "completed_workers": []
90})

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