andreasbloomquist Agent evals
Overview
Two layers, deterministic first. Layer 1 runs on every push: scripted fake model + recorded tool results, driving the production tools and graph, asserting the trajectory (tools, args, order), the payload contracts (every custom event validates), and the render (the consuming component mounts with it). Layer 2 is opt-in (nightly / on demand): real model, LLM-as-judge — that's deepeval's job (boundary at the end). Layer 1 catches regressions; layer 2 measures judgment. A layer-1 test that swaps the production tools for doubles verifies only the doubles.
Quick reference
| Verify | How |
|---|---|
| Right tools, order, args | Script the model; capture on_tool_start from astream_events(version="v2"); assert [(name, input)]. No instrumenting tools, no patching internals. |
| Production tool bodies run | Tools reach the network through one injectable seam (ToolRuntime.context.client or a module-level client). Tests swap the seam for replay, never the tool. |
| Payload contract | Capture on_custom_event; Model.model_validate(ev["data"]), one pydantic model per event name, imported from the producer. |
| The panel renders it | A vitest mount test per consuming component, fed the same recorded payload. An SSE-bytes test is not a render test — Ferdinand shipped a panel that received its event and rendered nothing (early-return on an unrelated empty prop). |
| Clean termination | Last on_chat_model_end output has tool_calls == [] — the loop stopped because the model stopped, not because a limit hit. |
| Multi-turn | Same thread_id across two runs sharing one InMemorySaver; assert turn 2's history via aget_state. |
| Fast path / short-circuit | Script zero responses; if the model is called, IndexError — or assert no on_chat_model_start. |
| Error path | Replay returns the tool's error-as-data shape (see agent-tool-design); assert the model text and the {error, retryable} panel. |
| Budgets | Tokens from on_chat_model_end usage metadata; time-to-first on_chat_model_stream. Ceilings, not exact values. |
The fake model (verified against installed source)
langchain_core.language_models.fake_chat_models ships four fakes. Use FakeMessagesListChatModel with one override — the others don't survive create_agent + tools:
from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
class ScriptedModel(FakeMessagesListChatModel):
def bind_tools(self, tools, **kwargs): # BaseChatModel.bind_tools raises NotImplementedError;
return self # create_agent calls it whenever tools are present
responses = one AIMessage per model call, in order: tool-call messages, then a final one with no tool_calls. Exactly as many as the model will be called — it cycles when exhausted (Traps).
Recording tool results
A committed dict, not a framework: RECORDED = {(tool_name, json.dumps(args, sort_keys=True)): result} in tests/replay.py. replay() raises KeyError on a miss so a new arg shape fails loudly instead of returning []. Re-record behind an explicit --record flag against the real API; never on a normal run.
Worked example
import pytest
from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
from langchain_core.messages import AIMessage, HumanMessage
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from myapp.agent.tools import search_flights, search_hotels # production tools; client seam → replay
from myapp.agent.payloads import PanelState # pydantic model the producer also uses
class ScriptedModel(FakeMessagesListChatModel):
def bind_tools(self, tools, **kwargs):
return self
def tool_call(name, **args):
return AIMessage(content="", tool_calls=[{"id": f"call_{name}", "name": name, "args": args}])
async def collect(agent, prompt, thread="t1"):
tools, panels, final = [], [], None
cfg = {"configurable": {"thread_id": thread}}
async for ev in agent.astream_events({"messages": [HumanMessage(prompt)]}, config=cfg, version="v2"):
if ev["event"] == "on_tool_start":
tools.append((ev["name"], ev["data"]["input"]))
elif ev["event"] == "on_custom_event" and ev["name"] == "panel_state":
panels.append(PanelState.model_validate(ev["data"]))
elif ev["event"] == "on_chat_model_end":
final = ev["data"]["output"]
return tools, panels, final
async def test_flight_then_hotel_trajectory(replay_client): # fixture swaps the client seam
model = ScriptedModel(responses=[
tool_call("search_flights", destination="TYO", month="2027-05"),
tool_call("search_hotels", area="Shibuya"),
AIMessage(content="Here are your options."),
])
agent = create_agent(model, [search_flights, search_hotels], checkpointer=InMemorySaver())
tools, panels, final = await collect(agent, "find me a flight to Tokyo in May and a hotel near Shibuya")
assert [t[0] for t in tools] == ["search_flights", "search_hotels"]
assert tools[0][1] == {"destination": "TYO", "month": "2027-05"}
assert [p.kind for p in panels] == ["flights", "hotels"]
assert final.tool_calls == [] and final.text == "Here are your options."
The checkpointer + thread_id are there because production requires them (the AG-UI adapter calls aget_state per run) — build the test graph the way production builds it.
Traps
| Symptom | Cause |
|---|---|
Tool ran, on_custom_event never fired, browser blank |
Tool emits with get_stream_writer(). That's stream_mode="custom", a different channel; the AG-UI adapter forwards only adispatch_custom_event. A test that listens on stream_mode="custom" passes anyway — listen on on_custom_event. |
NotImplementedError from bind_tools |
Stock fake with tools bound. Subclass as above. Don't hand-roll a BaseChatModel — the shipped one plus one method is the whole fake. |
ValueError: No generations found in stream |
GenericFakeChatModel + a tool-call message under astream_events. |
GraphRecursionError: Recursion limit of 25 |
Script never reaches a no-tool-call message. |
KeyError: 'model' from the router after N tool calls |
Script shorter than the number of model calls: it cycled and returned the same AIMessage object, now carrying the id the reducer assigned it, so add_messages replaced instead of appended. One fresh message per call. |
| Trajectory green, prod calls the wrong tool | Layer 1 proves the harness, not the model's choice. Layer-2 question (ToolCorrectnessMetric). |
Boundary with deepeval
This skill: the deterministic layer — fake model, replay, trajectory + contract assertions, mount tests, budgets. deepeval (installed plugin skill): the judge layer — TaskCompletionMetric, ToolCorrectnessMetric, ArgumentCorrectnessMetric, ConversationalGEval, goldens, deepeval test run, @observe. Run this on every push; run deepeval on a schedule against the real model.
Related: langgraph-patterns (event names, checkpointers) · agent-tool-design (what tools return vs. emit) · ag-ui-protocol (what the browser receives).