LangGraph.

Why a Graph Instead of a Loop

State, Nodes, Edges

pip install langgraph

Building a Simple Graph

from typing import TypedDict
from langgraph.graph import StateGraph, END

class State(TypedDict):
    question: str
    answer: str

def generate_answer(state: State) -> dict:
    response = client.chat.completions.create(
        model="gpt-4.1-mini",
        messages=[{"role": "user", "content": state["question"]}],
    )
    return {"answer": response.choices[0].message.content}

builder = StateGraph(State)
builder.add_node("generate", generate_answer)
builder.set_entry_point("generate")
builder.add_edge("generate", END)

graph = builder.compile()
result = graph.invoke({"question": "What is a vector embedding?", "answer": ""})
print(result["answer"])

Conditional Edges

A conditional edge is a function that inspects the current state and returns the name of the next node — this is where branching logic that would be nested ifs in Topic 8's loop becomes a named, visible part of the graph's shape.

class State(TypedDict):
    question: str
    answer: str
    needs_escalation: bool

def route_after_answer(state: State) -> str:
    return "escalate" if state["needs_escalation"] else END

builder = StateGraph(State)
builder.add_node("generate", generate_answer)
builder.add_node("escalate", escalate_to_human)
builder.set_entry_point("generate")
builder.add_conditional_edges("generate", route_after_answer, {
    "escalate": "escalate",
    END: END,
})

Cycles for Agent Loops

Topic 8's while loop becomes a cycle: an edge that points back to an earlier node. This is exactly how LangGraph represents the tool-calling loop from Topic 5 and Topic 8 — "call the model, maybe call a tool, go back and call the model again" is a two-node cycle with a conditional exit.

def route_after_model(state: State) -> str:
    last_message = state["messages"][-1]
    return "tools" if last_message.tool_calls else END

builder = StateGraph(AgentState)
builder.add_node("model", call_model)
builder.add_node("tools", execute_tools)
builder.set_entry_point("model")
builder.add_conditional_edges("model", route_after_model, {"tools": "tools", END: END})
builder.add_edge("tools", "model")  # the cycle: tool result goes back to the model

Persistence & Checkpointing

LangGraph can save state after every node runs, keyed by a thread ID — the mechanism that makes "pause this agent run and resume it tomorrow" or "keep memory across separate conversations" possible without hand-rolled serialization.

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()  # swap for a Postgres-backed saver in production (Topic 12)
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "user-42-conversation-7"}}
graph.invoke({"question": "What's pgvector?"}, config=config)
# ... later, possibly a different process entirely ...
graph.invoke({"question": "How does that relate to what I asked before?"}, config=config)
# resumes with full prior state, because it's keyed by the same thread_id

Human-in-the-Loop

Combine persistence with an explicit interrupt point: the graph pauses before a specific node — typically the one Topic 5 Section 6 flagged as destructive — and waits for a human decision before continuing.

graph = builder.compile(checkpointer=checkpointer, interrupt_before=["send_email"])

state = graph.invoke({"draft": "..."}, config=config)
# graph paused right before send_email — a human reviews state["draft"] here
# ... after approval ...
graph.invoke(None, config=config)  # resumes from exactly where it paused

Capstone: Rebuild Topic 8's Agent as a Graph

Take the two-tool research agent from Topic 8's checkpoint and redraw its loop as a LangGraph graph with an explicit cycle and exit condition.

import json
from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import StateGraph, END
from openai import OpenAI

client = OpenAI()

class AgentState(TypedDict):
    messages: Annotated[list, add]  # accumulates across nodes instead of overwriting

TOOLS = {"web_search": fake_web_search, "search_notes": search_notes}  # from Topic 8

def call_model(state: AgentState) -> dict:
    response = client.chat.completions.create(
        model="gpt-4.1-mini", messages=state["messages"], tools=TOOL_SCHEMAS,
    )
    return {"messages": [response.choices[0].message]}

def execute_tools(state: AgentState) -> dict:
    last_message = state["messages"][-1]
    results = []
    for call in last_message.tool_calls:
        args = json.loads(call.function.arguments)
        result = TOOLS[call.function.name](**args)
        results.append({"role": "tool", "tool_call_id": call.id,
                         "content": json.dumps(result)})
    return {"messages": results}

def route(state: AgentState) -> str:
    return "tools" if state["messages"][-1].tool_calls else END

builder = StateGraph(AgentState)
builder.add_node("model", call_model)
builder.add_node("tools", execute_tools)
builder.set_entry_point("model")
builder.add_conditional_edges("model", route, {"tools": "tools", END: END})
builder.add_edge("tools", "model")
graph = builder.compile()

if __name__ == "__main__":
    result = graph.invoke(
        {"messages": [
            {"role": "system", "content": "Answer using both tools as needed."},
            {"role": "user", "content":
                "How long had pgvector existed before our team started using it?"},
        ]},
        config={"recursion_limit": 12},  # the graph's equivalent of Topic 8's max_steps
    )
    print(result["messages"][-1].content)