The Testing Pyramid for AI Agents
graph TB
E2E["E2E Evals · LangSmith\nFull pipeline, real inputs\nMost expensive, run weekly"]
INT["Integration Tests\nTools + nodes together\nMock LLM, real tool logic"]
UNIT["Unit Tests · fast\nIndividual tools and nodes\nFully mocked, run on every PR"]
E2E --- INT --- UNIT
style E2E fill:#EEF0F7,stroke:#6366F1,color:#0F172A
style INT fill:#EEF0F7,stroke:#818CF8,color:#0F172A
style UNIT fill:#f0fdf9,stroke:#0D9488,color:#0F172AEach layer catches different bugs. Don’t skip the bottom: unit tests for tools are cheap, fast, and catch the most common failures before they reach production.
Unit Testing Tools
Unit tests for agent tools mock all external calls so tests run fast and offline, focusing purely on your tool’s input validation and error-handling logic.
# tests/test_tools.py
import pytest
from unittest.mock import AsyncMock, patch
from src.agents.tools import search_web, get_stock_price, OrderSearchArgs
class TestWebSearch:
@pytest.mark.asyncio # required to run async test functions with pytest-asyncio
async def test_returns_list_of_results(self, mock_httpx):
# mock_httpx replaces the real HTTP call so this test runs without network access
mock_httpx.return_value = {
"results": [
{"title": "Test", "url": "https://example.com", "snippet": "Content"}
]
}
results = await search_web("test query", num_results=3)
assert isinstance(results, list)
assert len(results) <= 3 # verify the num_results cap is enforced
assert all("title" in r for r in results) # every result must have a title key
@pytest.mark.asyncio
async def test_handles_api_timeout(self, mock_httpx):
# simulate a network timeout: the tool should return an error dict, not raise
mock_httpx.side_effect = TimeoutError("Request timed out")
result = await search_web("test query")
assert result["success"] == False
assert result["error"]["code"] == "timeout" # structured error codes aid debugging
@pytest.mark.asyncio
async def test_query_cannot_be_empty(self):
# empty queries should be rejected before hitting the network
result = await search_web("")
assert result["success"] == False
class TestOrderSearchArgs:
def test_valid_args_pass(self):
# Pydantic model validates at construction time: no need to call a validator manually
args = OrderSearchArgs(
customer_email="[email protected]",
date_from="2024-01-01",
date_to="2024-12-31",
)
assert args.customer_email == "[email protected]"
def test_invalid_date_range_raises(self):
# pytest.raises asserts that a specific exception is thrown with a matching message
with pytest.raises(ValueError, match="date_to.*must be.*>="):
OrderSearchArgs(
customer_email="[email protected]",
date_from="2024-12-31",
date_to="2024-01-01", # before date_from, should be rejected by validator
)
def test_invalid_email_raises(self):
# Pydantic's EmailStr type validates the format: no custom validator needed
with pytest.raises(ValueError):
OrderSearchArgs(customer_email="not-an-email")Unit Testing LangGraph Nodes
Node tests isolate a single graph node by patching the LLM, letting you verify state transitions without spending tokens.
# tests/test_nodes.py
import pytest
from unittest.mock import MagicMock, patch
from langchain_core.messages import HumanMessage, AIMessage
from src.agents.react_agent import call_llm, AgentState
class TestCallLLMNode:
def make_state(self, messages, iterations=0) -> AgentState:
# helper that builds a minimal AgentState dict for use in each test
return {
"messages": messages,
"iterations": iterations,
"final_answer": None,
}
@patch("src.agents.react_agent.llm_with_tools") # replace the real LLM with a mock object
def test_increments_iterations(self, mock_llm):
# mock_llm.invoke returns a fake AIMessage so no real API call is made
mock_response = AIMessage(content="I'll help you with that.", tool_calls=[])
mock_llm.invoke.return_value = mock_response
state = self.make_state([HumanMessage(content="Hello")], iterations=3)
result = call_llm(state)
assert result["iterations"] == 4 # node must increment the counter by 1
assert len(result["messages"]) == 1 # node should append only the AI response
@patch("src.agents.react_agent.llm_with_tools")
def test_hits_max_iterations_guard(self, mock_llm):
# at iterations=15 the guard should fire before calling the LLM
state = self.make_state([HumanMessage(content="Hello")], iterations=15)
result = call_llm(state)
# Should not call the LLM: guard should have kicked in
mock_llm.invoke.assert_not_called() # assert_not_called() fails if the LLM was invoked
assert result["final_answer"] == "max_iterations_reached"LLM-as-Judge Evaluator
When there is no single correct answer to assert against, a second LLM grades the agent’s output using a rubric, this is the LLM-as-judge pattern.
# src/evals/judge.py
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from typing import Literal
# Each dimension gets a score and a written justification: the reasoning is as useful as the score
class EvaluationScore(BaseModel):
dimension: str
score: Literal[1, 2, 3, 4, 5] # Literal enforces only valid integer scores 1–5
reasoning: str
class AgentEvaluation(BaseModel):
factual_accuracy: EvaluationScore
completeness: EvaluationScore
hallucination_risk: EvaluationScore
overall: int = Field(ge=1, le=5) # ge/le are Pydantic validators for min/max value
notes: str
# temperature=0 makes the judge deterministic: same input always produces the same score
# with_structured_output forces the response to parse into the AgentEvaluation Pydantic model
judge_llm = ChatOpenAI(model="gpt-5.6-sol", temperature=0).with_structured_output(AgentEvaluation)
async def evaluate_response(
question: str,
agent_response: str,
reference_answer: str = "",
context_docs: list[str] = None,
) -> AgentEvaluation:
"""Evaluate agent response across multiple dimensions."""
context_str = ""
if context_docs:
# limit to 3 source docs so the judge prompt doesn't exceed context limits
context_str = "Source documents:\n" + "\n---\n".join(context_docs[:3])
reference_str = ""
if reference_answer:
reference_str = f"\nExpected answer (for reference):\n{reference_answer}"
# structured rubric in the prompt produces more consistent scores than open-ended grading
prompt = f"""Evaluate this AI agent response:
Question: {question}
Agent Response: {agent_response}
{reference_str}
{context_str}
Score each dimension 1-5:
factual_accuracy (1=many errors, 5=all facts correct):
- Are claims supported by sources?
- Are there factual errors?
completeness (1=major gaps, 5=fully addresses question):
- Does it cover all parts of the question?
- Are important caveats included?
hallucination_risk (1=clearly hallucinating, 5=no hallucinations):
- Are specific claims verifiable?
- Are there unsupported invented details?"""
return await judge_llm.ainvoke([HumanMessage(content=prompt)])
# Usage in tests
@pytest.mark.asyncio
async def test_research_agent_quality():
result = await research_agent.ainvoke({
"messages": [HumanMessage(content="What are the main features of LangGraph?")]
})
response = result["messages"][-1].content
eval_result = await evaluate_response(
question="What are the main features of LangGraph?",
agent_response=response,
reference_answer="LangGraph features: typed state, nodes, edges, checkpointing, streaming...",
)
# score >= 3 is a passing threshold: adjust based on your quality bar
assert eval_result.factual_accuracy.score >= 3, f"Low accuracy: {eval_result.factual_accuracy.reasoning}"
assert eval_result.hallucination_risk.score >= 3, f"Hallucination: {eval_result.hallucination_risk.reasoning}"Trajectory Evaluation
Trajectory evaluation asks a judge LLM to assess whether the agent took sensible steps, not just whether the final answer was correct.
# src/evals/trajectory.py
from pydantic import BaseModel
# each step in the agent's execution gets its own pass/fail judgment with reasoning
class StepEvaluation(BaseModel):
step: int
node: str
tool_used: str | None # None means the step was an LLM reasoning call, not a tool call
was_appropriate: bool
reasoning: str
class TrajectoryEvaluation(BaseModel):
steps: list[StepEvaluation]
overall_quality: Literal[1, 2, 3, 4, 5]
issues_found: list[str] # list of specific problems like "redundant tool call" or "skipped verification"
# separate judge model so trajectory evaluation can use a different (possibly cheaper) model
trajectory_judge = ChatOpenAI(model="gpt-5.6-sol").with_structured_output(TrajectoryEvaluation)
async def evaluate_agent_trajectory(
task: str,
execution_steps: list[dict], # captured from LangSmith trace
) -> TrajectoryEvaluation:
# format each step as a single readable line for the judge prompt
steps_str = "\n".join([
f"Step {i+1}: {s['node']} → {s.get('tool', 'LLM call')} → {str(s.get('output', ''))[:100]}"
for i, s in enumerate(execution_steps)
])
return await trajectory_judge.ainvoke([HumanMessage(content=f"""
Evaluate whether this agent took appropriate steps to complete the task:
Task: {task}
Execution trajectory:
{steps_str}
For each step, assess:
- Was this tool/action appropriate given what the agent knew at this point?
- Were there unnecessary or redundant steps?
- Did the agent miss any obvious approaches?
""")])Building a Regression Test Dataset
A golden dataset stores known-good examples so you can detect regressions whenever you change the model, prompt, or tools.
# src/evals/dataset.py
from langsmith import Client
from typing import Optional
# LangSmith Client connects to the LangSmith API for dataset and run management
ls_client = Client()
def add_to_golden_dataset(
dataset_name: str,
inputs: dict,
expected_outputs: dict,
metadata: dict = {},
):
"""Add a curated example to the regression dataset."""
# read_dataset looks up the dataset by name: create it in LangSmith UI first
dataset = ls_client.read_dataset(dataset_name=dataset_name)
ls_client.create_example(
inputs=inputs,
outputs=expected_outputs,
metadata=metadata, # metadata lets you filter by category, date, or source
dataset_id=dataset.id,
)
# Curate examples from production runs
def curate_from_production_run(run_id: str, is_good: bool, notes: str):
"""Promote a production run to the eval dataset."""
# read_run fetches the full inputs and outputs from a LangSmith trace
run = ls_client.read_run(run_id)
if is_good:
add_to_golden_dataset(
dataset_name="agent-golden-set",
inputs={"query": run.inputs["messages"][0]["content"]},
expected_outputs={"answer": run.outputs["messages"][-1]["content"]},
metadata={"notes": notes, "promoted_from": run_id}, # track provenance
)
# Run the full evaluation suite
async def run_regression_suite(agent, dataset_name: str = "agent-golden-set"):
from langsmith.evaluation import evaluate
# evaluate() runs every example in the dataset and scores it with the provided evaluators
results = evaluate(
lambda inputs: agent.invoke({"messages": [HumanMessage(content=inputs["query"])]}),
data=dataset_name,
evaluators=[accuracy_evaluator, hallucination_evaluator],
experiment_prefix=f"regression-{datetime.now().strftime('%Y%m%d')}", # timestamp groups runs by date
)
print(f"Accuracy: {results['accuracy']:.2%}")
print(f"Hallucination-free: {results['hallucination_risk']:.2%}")
return resultsThe investment in evaluation infrastructure pays back every time you change your agent’s model, prompt, or tools. Without it, you’re deploying blind.