LangGraph: Testing Your Agent, Not Just Your Prompts

Lesson 7: LangGraph — Testing Your Agent, Not Just Your Prompts

LangGraph is a framework for building stateful, multi-step agents: a graph of nodes (each a function that transforms state) with conditional edges, tool calls, loops, memory via a checkpointer, and human-in-the-loop interruptions. Testing an agent is harder than testing a chain, because the failure surface is bigger: not just "wrong answer" but wrong tool calls, infinite loops, lost state, and unapproved side effects. This lesson is the testing discipline for graphs — which you then feed straight into the LangSmith eval loop from Lesson 6.

A Minimal Graph to Talk About

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class AgentState(TypedDict):
    question: str
    retrieved: list
    answer: str
    tool_calls: list

def retrieve(state: AgentState) -> dict:
    docs = vector_store.search(state["question"], top_k=3)
    return {"retrieved": docs, "tool_calls": state.get("tool_calls", []) + ["retrieve"]}

def answer(state: AgentState) -> dict:
    return {"answer": generator.invoke(state["question"], context=state["retrieved"])}

builder = StateGraph(AgentState)
builder.add_node("retrieve", retrieve)
builder.add_node("answer", answer)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "answer")
builder.add_edge("answer", END)

app = builder.compile(checkpointer=MemorySaver())

Level 1 — Unit Tests: Nodes in Isolation

Every node is a plain function: state in, state-delta out. Test it directly with a crafted state — no graph, no model cost beyond the node itself:

def test_retrieve_returns_top_k():
    result = retrieve({"question": "how to reset password?", "tool_calls": []})
    assert len(result["retrieved"]) == 3
    assert result["tool_calls"] == ["retrieve"]

def test_answer_is_grounded():
    state = {"question": "what is the refund policy?",
             "retrieved": [fake_policy_doc], "tool_calls": []}
    result = answer(state)
    # your rubric check: every claim in result["answer"] appears in fake_policy_doc
    assert claims_supported_by(result["answer"], fake_policy_doc)

Node unit tests catch the deterministic bugs (wrong keys, missing fields, bad plumbing) fast, without burning tokens on the whole graph.

Level 2 — Integration Tests: The Whole Graph

def test_happy_path():
    final = app.invoke({"question": "how to reset password?", "tool_calls": []})
    assert final["tool_calls"] == ["retrieve"]      # expected tool sequence
    assert final["answer"]                           # produced an answer
    assert "settings" in final["answer"].lower()

def test_loops_terminate():
    # graph with a retry loop: assert max iterations not exceeded
    final = app.invoke({"question": "search again and again..."}, 
                       config={"recursion_limit": 10})

Key integration assertions: final state shape, the sequence of tool calls, no crash, termination within the recursion limit, and for RAG agents, that the answer cites only retrieved context.

Level 3 — Threads, Replay & Time Travel

The checkpointer stores every state transition per thread_id, which unlocks the agent-specific superpowers:

config = {"configurable": {"thread_id": "user-42"}}

# Threads: multi-turn conversations with memory
app.invoke({"question": "hi"}, config)
app.invoke({"question": "what did I just ask?"}, config)   # remembers

# Inspect current state
snapshot = app.get_state(config)

# Time travel: walk history
history = [s for s in app.get_state_history(config)]

# Fork: rewind to a checkpoint, patch state, replay forward
app.update_state(config, {"answer": None})
final = app.invoke(None, config)   # resumes from the patched checkpoint

Testing applications: replay a real production thread (fetched from your trace store) against a new graph version and diff the outcomes — the ultimate regression test for memory bugs. Fork+patch lets you test "what would the agent have done if the tool had returned X" without rerunning the world.

Level 4 — Human-in-the-Loop & Guardrail Nodes

High-stakes agents interrupt for approval. Test the interruption contract like any other state machine:

# Graph with interrupt_before=["send_email"] and a Command(resume=...) resume path
def test_approval_required():
    result = app.invoke({"question": "send the email"}, config, interrupt_before=["send_email"])
    assert result is None or "send_email" not in result  # halted before side effect

def test_approval_granted_resumes():
    app.invoke(None, config, Command(resume="approved"))
    final = app.get_state(config)
    assert final.values["email_sent"] is True

Guardrails are nodes too: a validate_tool_call node between the model and the tools, a check_permissions node before any action, an input-scrub node at the start. They get the same unit/integration treatment — assert the injection input is blocked, the legit input passes, and the permission check rejects out-of-scope tools (Lesson 5's two-layer rule applied to code).

Level 5 — The LangSmith Loop for Agents

The graph is just a callable — it plugs directly into the eval machinery from Lesson 6:

results = evaluate(
    target=lambda inputs: app.invoke(
        {"question": inputs["question"]},
        config={"configurable": {"thread_id": "eval-fresh"}},
    ),
    data="agent-golden-v1",
    evaluators=[tool_sequence_check, rubric_judge, groundedness],
)

Tracing captures every node, tool call, and token along the way, so when a regression appears you click into the exact failing step. Dataset examples for agents should record the expected tool sequence and expected final answer — because the sequence is usually where agents fail first.

Further Reading

• LangGraph documentation
• LangGraph: time travel (replay, fork, resume)
• LangSmith evaluation docs (agents as targets)