6. Agents in LangGraph

Advanced · 12 min read

The LangGraph page covered the building blocks: state, nodes, edges, checkpoints and interrupts. This page applies them to agents: when the prebuilt agent is enough, and how to build a custom agent graph when you need control over every step.

6.1 Start with the prebuilt agent

For a standard tool-calling agent you don't need to draw a graph. LangChain's create_agent builds the loop on LangGraph, and middleware adds behaviour such as human approval:

# no-run — pip install langchain langgraph langchain-openai ; APIs as of LangChain/LangGraph 1.x
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[get_order_status, refund_order],
    system_prompt="You are Acme's support agent. Never guess order details: use the tools.",
    middleware=[HumanInTheLoopMiddleware(interrupt_on={"refund_order": True})],   # pause before refunds
    checkpointer=InMemorySaver(),                                                  # memory + resumable runs
)
config = {"configurable": {"thread_id": "customer-42"}}
result = agent.invoke({"messages": [{"role": "user", "content": "A100 arrived damaged, refund it"}]}, config)

Use the prebuilt agent until you need something it can't express: extra steps between the model and the tools, different models per step, custom routing, or state beyond the message list.

6.2 Design the state

State is everything the graph remembers during a run. For agents it's usually the message list plus a few structured fields. Each field has a reducer that says how a node's update merges in. The message reducer (add_messages) appends new messages and replaces ones with the same ID. Here's that behaviour in plain Python:

def add_messages(current: list[dict], update: list[dict]) -> list[dict]:
    """Append new messages; a message whose id already exists replaces the old one (e.g. an edited tool call)."""
    merged = {m["id"]: m for m in current}
    for m in update:
        merged[m["id"]] = m
    return list(merged.values())

state = {"messages": [{"id": "1", "role": "user", "content": "Refund A100, it's damaged"}], "steps": 0}
updates = [
    {"messages": [{"id": "2", "role": "assistant", "content": "", "tool_calls": [
        {"name": "refund_order", "args": {"order_id": "A100", "amount_inr": 1200}}]}], "steps": 1},
    {"messages": [{"id": "2", "role": "assistant", "content": "", "tool_calls": [            # same id: replaced
        {"name": "refund_order", "args": {"order_id": "A100", "amount_inr": 999}}]}], "steps": 1},
]
for update in updates:
    state = {"messages": add_messages(state["messages"], update["messages"]),
             "steps": state["steps"] + update["steps"]}            # a counter field: reducer = add
print(len(state["messages"]), "messages; last tool call:", state["messages"][-1]["tool_calls"][0]["args"],
      "| steps:", state["steps"])

Output

2 messages; last tool call: {'order_id': 'A100', 'amount_inr': 999} | steps: 2

The second update replaced the tool call instead of adding a duplicate. That's exactly how a human editing a proposed refund amount during an approval works.

Field	Reducer	Why it's in state
messages	add_messages	The conversation and tool calls
steps	add	Enforce a step budget in your own routing
order_id, user_id	replace (default)	Structured facts tools need; never parsed back out of text
approved	replace	Result of a human approval

6.3 A custom agent graph with a guard

The custom version of the agent adds a guard between the model and the tools. Safe tools run immediately, risky ones go to approval, and the step budget is enforced in code:

flowchart LR
    S((START)) --> M[model]
    M --> R{route}
    R -->|no tool calls| E((END))
    R -->|safe tools| T[tools] --> M
    R -->|risky tool| A[approval<br/>interrupt] --> T
    R -->|budget spent| X[give_up] --> E

The routing function is plain Python, so you can test it without LangGraph or a model:

RISKY_TOOLS = {"refund_order", "cancel_subscription", "send_email"}
MAX_STEPS = 8

def route(state: dict) -> str:
    last = state["messages"][-1]
    # LangGraph passes message objects (AIMessage.tool_calls); the tests below use plain dicts. Handle both.
    calls = (last.get("tool_calls") if isinstance(last, dict) else getattr(last, "tool_calls", None)) or []
    if not calls:
        return "end"
    if state["steps"] >= MAX_STEPS:
        return "give_up"
    if any(c["name"] in RISKY_TOOLS for c in calls):
        return "approval"
    return "tools"

def msg(*names):
    return {"role": "assistant", "content": "", "tool_calls": [{"name": n, "args": {}} for n in names]}

cases = [("final answer", {"messages": [{"role": "assistant", "content": "Done!"}], "steps": 2}),
         ("lookup", {"messages": [msg("get_order_status")], "steps": 1}),
         ("refund", {"messages": [msg("get_order_status", "refund_order")], "steps": 2}),
         ("over budget", {"messages": [msg("get_order_status")], "steps": 8})]
for label, case in cases:
    print(f"{label:12} -> {route(case)}")

Output

final answer -> end
lookup       -> tools
refund       -> approval
over budget  -> give_up

The graph wires those routes to nodes:

# no-run
from typing import Annotated, TypedDict

from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langgraph.types import interrupt

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    steps: int

tools = [get_order_status, refund_order]
model = init_chat_model("openai:gpt-4o-mini", temperature=0).bind_tools(tools)

def call_model(state: AgentState):
    return {"messages": [model.invoke(state["messages"])], "steps": state["steps"] + 1}

def approval(state: AgentState):
    calls = state["messages"][-1].tool_calls
    decision = interrupt({"approve": [c for c in calls if c["name"] in RISKY_TOOLS]})   # pauses the run
    if decision != "approve":
        return {"messages": [{"role": "tool", "tool_call_id": c["id"], "content": "Rejected by a human reviewer."}
                             for c in calls]}
    return {}

def give_up(state: AgentState):
    return {"messages": [{"role": "assistant", "content": "I couldn't finish this. A person will follow up."}]}

builder = StateGraph(AgentState)
builder.add_node("model", call_model)
builder.add_node("tools", ToolNode(tools))
builder.add_node("approval", approval)
builder.add_node("give_up", give_up)
builder.add_edge(START, "model")
builder.add_conditional_edges("model", route, {"end": END, "tools": "tools", "approval": "approval", "give_up": "give_up"})
builder.add_conditional_edges("approval", lambda s: "model" if s["messages"][-1].type == "tool" else "tools")
builder.add_edge("tools", "model")
builder.add_edge("give_up", END)
agent = builder.compile(checkpointer=InMemorySaver())

6.4 Limits, streaming and progress

Need	LangGraph tool
Hard stop on runaway loops	config={"recursion_limit": 25} raises GraphRecursionError; your own steps budget ends gracefully first
Show progress ("Looking up your order…")	for update in agent.stream(inputs, config, stream_mode="updates") yields each node's update
Stream the answer token by token	stream_mode="messages"
Resume after approval or a crash	Same thread_id + a checkpointer; agent.invoke(Command(resume="approve"), config)
Inspect or debug a run	agent.get_state(config) and agent.get_state_history(config)

6.5 Subgraphs: agents inside agents

A compiled graph can be a node in another graph. A research sub-agent with its own tools and loop becomes one step in a bigger workflow, which is how hierarchical multi-agent systems are built:

# no-run
research_agent = research_builder.compile()                 # its own state, tools and loop
main = StateGraph(MainState)
main.add_node("research", research_agent)                   # used like any other node
main.add_node("write_report", write_report)
main.add_edge(START, "research")
main.add_edge("research", "write_report")

6.6 Test nodes as functions

Nodes and routers are plain functions of state, so most of an agent can be unit-tested with no model at all: routing (like 6.3), guards, state reducers and tool wrappers. Test the whole graph with a fake model that returns scripted tool calls (the technique from ReAct from scratch), and save real-model tests for the evaluation suite.

Interview questions

When would you build a custom LangGraph agent instead of using the prebuilt one?

When you need steps between the model and the tools (guards, validation, approvals), different models for different steps, custom routing or budgets, or structured state beyond the message list. Otherwise the prebuilt agent plus middleware is less code and less to maintain.

What is a reducer in LangGraph state?

A function that defines how a node's partial update merges into the existing state for a field. The default replaces the value; add_messages appends new messages and replaces messages with the same ID, which also allows editing a proposed tool call before it runs.

How do you stop an agent from looping forever in LangGraph?

Keep a step counter in state and route to a graceful "give up" node when the budget is reached, and set recursion_limit in the config as a hard backstop. Also cap retries per tool and alert on runs that hit the limit.

Practice

• Add a case to the router in 6.3 where the model calls send_email and a safe tool together.
• Change add_messages so an update with "delete": True removes the message with that ID.
• Sketch the graph for an agent that must look up the order before it may propose a refund.

Next: 7. Multi-agent patterns Back to: AI Agents overview · Notes overview