Testing LangGraph Agents with Pytest
LangGraph testing is different from testing a normal Python function because the thing under test is a stateful graph that calls a language model, branches on its output, and sometimes loops. This guide shows how to structure a pytest suite around LangGraph: unit tests for individual nodes, tests for conditional routing, full-graph integration tests with a fake chat model, and patterns for checkpointing and streaming. Every example below is runnable with pytest, langgraph, and langchain-core installed, no live API key required.
Why LangGraph testing needs its own approach
A LangGraph app is a graph of nodes that read and write a shared state object. Each node is usually a plain Python function, which means you can unit test it exactly like any other function: call it with a state dict, assert on the return value. The harder part is everything that depends on an LLM: conditional edges that route based on a model's tool call, agents that loop until the model stops calling tools, and nodes that parse a model's free-text output into structured state.
If your tests call a real model, you get three problems: the suite is slow, it costs money on every CI run, and it is flaky because LLM output is non-deterministic. The fix is the same one you'd use for any external API: put a fake or mock chat model behind the same interface LangGraph nodes already expect, and drive your tests off that. LangChain ships fake chat models for exactly this purpose, so you don't need to hand-roll a mock client.
Setting up the project
A minimal layout for LangGraph testing looks like this:
myagent/
graph.py
nodes.py
state.py
tests/
conftest.py
test_nodes.py
test_routing.py
test_graph.py
pytest.iniInstall the dependencies:
pip install langgraph langchain-core pytest pytest-asynciopytest.ini (or a [tool.pytest.ini_options] block in pyproject.toml) should register markers if you plan to separate fast unit tests from slower integration tests:
[pytest]
markers =
integration: tests that build and invoke a full graphDefining state and a simple graph
Here's a small support-ticket triage agent used throughout the examples. It classifies a ticket, then routes to either an auto-reply node or a human-escalation node.
# myagent/state.py
from typing import TypedDict, Literal
class TicketState(TypedDict):
ticket_text: str
category: str
priority: Literal["low", "high"]
reply: str# myagent/nodes.py
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import HumanMessage
from myagent.state import TicketState
def classify_ticket(llm: BaseChatModel):
def _node(state: TicketState) -> dict:
response = llm.invoke([
HumanMessage(content=f"Classify this ticket: {state['ticket_text']}")
])
content = response.content.strip().lower()
priority = "high" if "urgent" in content or "down" in content else "low"
return {"category": content, "priority": priority}
return _node
def auto_reply(state: TicketState) -> dict:
return {"reply": f"Thanks, we logged your {state['category']} ticket."}
def escalate(state: TicketState) -> dict:
return {"reply": "This has been escalated to a human agent."}
def route_by_priority(state: TicketState) -> str:
return "escalate" if state["priority"] == "high" else "auto_reply"# myagent/graph.py
from langgraph.graph import StateGraph, START, END
from myagent.state import TicketState
from myagent.nodes import classify_ticket, auto_reply, escalate, route_by_priority
def build_graph(llm):
graph = StateGraph(TicketState)
graph.add_node("classify", classify_ticket(llm))
graph.add_node("auto_reply", auto_reply)
graph.add_node("escalate", escalate)
graph.add_edge(START, "classify")
graph.add_conditional_edges(
"classify",
route_by_priority,
{"auto_reply": "auto_reply", "escalate": "escalate"},
)
graph.add_edge("auto_reply", END)
graph.add_edge("escalate", END)
return graph.compile()Note that classify_ticket takes the LLM as a constructor argument rather than importing a global client. This one decision is what makes LangGraph testing tractable: every node that touches a model becomes trivially injectable.
Unit testing individual nodes
Nodes that don't touch the model need no mocking at all. Test them as plain functions:
# tests/test_nodes.py
from myagent.nodes import auto_reply, escalate, route_by_priority
def test_auto_reply_includes_category():
state = {"ticket_text": "x", "category": "billing", "priority": "low", "reply": ""}
result = auto_reply(state)
assert "billing" in result["reply"]
def test_route_by_priority_high():
state = {"ticket_text": "x", "category": "outage", "priority": "high", "reply": ""}
assert route_by_priority(state) == "escalate"
def test_route_by_priority_low():
state = {"ticket_text": "x", "category": "question", "priority": "low", "reply": ""}
assert route_by_priority(state) == "auto_reply"These run in milliseconds and don't need pytest fixtures beyond plain dicts. Keep as many of your assertions here as possible; the fewer tests that need a graph invocation, the faster your suite stays.
Mocking the LLM with a fake chat model
For the classify_ticket node, use langchain_core.language_models.fake_chat_models.GenericFakeChatModel or FakeListChatModel. FakeListChatModel returns a fixed list of responses in order, which is enough for most LangGraph testing:
# tests/conftest.py
import pytest
from langchain_core.language_models.fake_chat_models import FakeListChatModel
@pytest.fixture
def fake_llm():
def _make(responses):
return FakeListChatModel(responses=responses)
return _make# tests/test_nodes_with_llm.py
from myagent.nodes import classify_ticket
def test_classify_ticket_marks_urgent_as_high(fake_llm):
llm = fake_llm(["server is down, urgent"])
node = classify_ticket(llm)
state = {"ticket_text": "our checkout is broken", "category": "", "priority": "low", "reply": ""}
result = node(state)
assert result["priority"] == "high"
assert "urgent" in result["category"]
def test_classify_ticket_marks_normal_as_low(fake_llm):
llm = fake_llm(["general billing question"])
node = classify_ticket(llm)
state = {"ticket_text": "how do I update my card", "category": "", "priority": "low", "reply": ""}
result = node(state)
assert result["priority"] == "low"This gives you full control over what the model "says" without a network call. If your node parses structured output (JSON, a tool call), have the fake model return exactly the malformed and well-formed variants you want to test, including edge cases like empty strings or truncated JSON, since that's where LangGraph agents actually break in production.
For agents built with create_react_agent or that expect tool-calling responses, use GenericFakeChatModel with AIMessage objects that include a tool_calls list, so the fake model's output shape matches what a real tool-calling model would send:
from langchain_core.messages import AIMessage
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
def test_tool_calling_node():
ai_message = AIMessage(
content="",
tool_calls=[{"name": "lookup_order", "args": {"order_id": "123"}, "id": "call_1"}],
)
llm = GenericFakeChatModel(messages=iter([ai_message]))
response = llm.invoke("look up order 123")
assert response.tool_calls[0]["name"] == "lookup_order"Testing conditional edges in isolation
Conditional edge functions in LangGraph are plain functions that take state and return a string key. Test them directly, without building the graph, exactly as shown above with route_by_priority. The only extra case worth adding is an unexpected or missing value, since that's what causes a KeyError at runtime when the routing map doesn't have a matching branch:
import pytest
def test_route_by_priority_rejects_unknown_value():
state = {"ticket_text": "x", "category": "x", "priority": "medium", "reply": ""}
with pytest.raises(KeyError):
{"escalate": "escalate", "auto_reply": "auto_reply"}[
"escalate" if state["priority"] == "high" else "auto_reply"
]In practice, guard against this in the node itself (default to a safe branch) rather than relying on tests to catch it after the fact, but a test like this documents the expectation and will fail loudly if someone changes priority to an open string later.
Full graph integration tests
Once individual nodes and routing are covered, test the compiled graph end to end. Build the graph with a fake LLM and call .invoke():
# tests/test_graph.py
import pytest
from myagent.graph import build_graph
@pytest.mark.integration
def test_high_priority_ticket_gets_escalated(fake_llm):
llm = fake_llm(["outage, urgent, server down"])
graph = build_graph(llm)
result = graph.invoke({
"ticket_text": "prod is down",
"category": "",
"priority": "low",
"reply": "",
})
assert result["priority"] == "high"
assert "escalated" in result["reply"]
@pytest.mark.integration
def test_low_priority_ticket_gets_auto_reply(fake_llm):
llm = fake_llm(["billing question"])
graph = build_graph(llm)
result = graph.invoke({
"ticket_text": "how do I cancel",
"category": "",
"priority": "low",
"reply": "",
})
assert "logged" in result["reply"]Run only the fast tests during local development and the full suite (including integration) in CI:
pytest -m "not integration"
pytestTesting loops and iteration limits
Agents that loop (call a tool, check the result, call the model again) are the riskiest part of any LangGraph app to leave untested, because an infinite loop in production means a runaway bill. Test two things: that the loop terminates on a normal path, and that it terminates under a recursion limit even when the model never produces a stop condition.
langgraph's .invoke() and .stream() accept a config dict with recursion_limit. Force a fake model to always request another tool call, then assert that the graph raises GraphRecursionError instead of hanging:
from langgraph.errors import GraphRecursionError
from langchain_core.messages import AIMessage
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
def test_agent_stops_at_recursion_limit(build_looping_agent):
always_calls_tool = AIMessage(
content="",
tool_calls=[{"name": "search", "args": {"q": "x"}, "id": "call_1"}],
)
llm = GenericFakeChatModel(messages=iter([always_calls_tool] * 50))
graph = build_looping_agent(llm)
with pytest.raises(GraphRecursionError):
graph.invoke({"messages": [("user", "find something")]}, config={"recursion_limit": 5})This test is cheap to write and catches a real production failure mode: a prompt change that removes the model's incentive to stop calling tools.
Testing checkpointing and persistence
If your graph uses a checkpointer (MemorySaver, SqliteSaver, or a Postgres-backed one) for multi-turn conversations or human-in-the-loop interrupts, test the checkpointer with the in-memory implementation rather than a real database:
from langgraph.checkpoint.memory import MemorySaver
def test_graph_resumes_from_checkpoint(fake_llm):
llm = fake_llm(["first response", "second response"])
checkpointer = MemorySaver()
graph = build_graph(llm).with_config(checkpointer=checkpointer)
thread = {"configurable": {"thread_id": "test-thread-1"}}
graph.invoke({"ticket_text": "first message", "category": "", "priority": "low", "reply": ""}, config=thread)
state = graph.get_state(thread)
assert state.values["reply"] != ""MemorySaver runs entirely in process memory, so these tests are as fast as any other unit test and don't need a Docker container or a test database. Save integration tests against SqliteSaver or PostgresSaver for a smaller set of contract tests that confirm your production checkpointer backend actually persists across process restarts.
Testing streaming output
If the graph is consumed via .stream() rather than .invoke(), iterate the generator in the test and assert on the sequence of events, not just the final state:
def test_stream_emits_classify_then_reply(fake_llm):
llm = fake_llm(["billing question"])
graph = build_graph(llm)
events = list(graph.stream(
{"ticket_text": "cancel my plan", "category": "", "priority": "low", "reply": ""},
stream_mode="updates",
))
node_names = [list(event.keys())[0] for event in events]
assert node_names == ["classify", "auto_reply"]stream_mode="updates" yields one dict per node as it completes, keyed by node name, which makes it straightforward to assert both order and content without parsing the full state object each time.
Snapshotting prompts sent to the model
A subtle bug in LangGraph apps is a prompt template that silently drops a variable or double-includes conversation history after a refactor. Catch this by asserting on what your node actually sends to the LLM, using a fake model that records its inputs:
from langchain_core.language_models.fake_chat_models import FakeListChatModel
class RecordingFakeChatModel(FakeListChatModel):
calls: list = []
def invoke(self, input, *args, **kwargs):
self.calls.append(input)
return super().invoke(input, *args, **kwargs)
def test_classify_prompt_includes_ticket_text():
llm = RecordingFakeChatModel(responses=["billing"])
node = classify_ticket(llm)
node({"ticket_text": "refund please", "category": "", "priority": "low", "reply": ""})
sent = llm.calls[0]
assert "refund please" in sent[0].contentThis kind of test catches prompt regressions that a pure output-based test would miss, because the output can still look reasonable even when the input was wrong.
Structuring tests by speed
A LangGraph test suite naturally splits into three tiers. Keep them separate with pytest markers so you can run the fast tier on every save and the slow tier in CI or before a release:
- Node-level unit tests: no LLM involved, pure functions, run in milliseconds.
- Node and routing tests with a fake chat model: no network, still fast, cover model-dependent logic.
- Full graph integration tests, including recursion limits and checkpointing: slightly slower, still no network, run on every CI push.
- A small number of live-model smoke tests, marked
@pytest.mark.liveand skipped by default, that hit a real model to catch drift in behavior after a model or prompt version bump. Gate these behind an environment variable so they never run accidentally in CI without a budget check.
import os
import pytest
live = pytest.mark.skipif(not os.getenv("RUN_LIVE_LLM_TESTS"), reason="live model tests disabled")
@live
def test_classify_ticket_against_real_model(real_llm):
node = classify_ticket(real_llm)
result = node({"ticket_text": "the app crashed on login", "category": "", "priority": "low", "reply": ""})
assert result["priority"] in {"low", "high"}FAQ
Do I need a real LLM API key to run LangGraph tests? No. FakeListChatModel and GenericFakeChatModel from langchain_core implement the same BaseChatModel interface as a real provider client, so any node or graph built against that interface works unchanged in tests. Reserve real API calls for a small, explicitly marked smoke-test tier.
How do I test a graph that uses `create_react_agent` instead of a hand-built `StateGraph`? The same fake-model approach works. Pass a GenericFakeChatModel configured with AIMessage objects that include tool_calls in the right sequence (tool call, then a plain text final answer), and assert on the final state or on the sequence of tool invocations captured by a recording tool wrapper.
How do I avoid infinite loops during test runs? Always pass a recursion_limit in the config when invoking a graph in a test, even for happy-path tests. If a routing bug creates an infinite loop, you want the test to fail fast with GraphRecursionError, not hang until the CI job times out.
Should I test the graph's visual structure (nodes and edges), not just its behavior? It's optional but cheap. graph.get_graph().nodes and .edges after .compile() let you assert that a specific node or edge exists, which catches accidental deletions during refactors. Keep this to one or two structural tests; behavior tests catch far more real bugs.
How do I test human-in-the-loop interrupts? Compile the graph with interrupt_before or interrupt_after set to the relevant node, invoke it, then assert graph.get_state(thread).next contains the expected paused node. Resume by calling .invoke(None, config=thread) and assert the final state, using MemorySaver as the checkpointer so the whole test stays in-process.
Can I use `pytest-asyncio` for async graphs? Yes. LangGraph exposes .ainvoke() and .astream() alongside the sync versions. Mark async test functions with @pytest.mark.asyncio (after enabling asyncio_mode = auto or the marker in your pytest config) and await graph.ainvoke(...) exactly as you would the sync call, using the same fake chat models, which support both sync and async invocation.
How much of the graph should be covered by integration tests versus unit tests? Favor unit tests for nodes and routing functions since they're fast and pinpoint failures precisely. Use integration tests to confirm the pieces are wired together correctly, meaning edges point where you think they do and state flows through as expected, not to re-verify logic already covered by unit tests.
BootcampA 30-day guided bootcamp: build, harden and ship a production autonomous agent from scratch.
AI AgentsUnderstand how AI agents really work: the loop, the tools, the memory, and why most agent projects fail.