CHAPTER ONE: THE ILLUSION OF INTELLIGENCE
There is a seductive story told in boardrooms and on benchmark leaderboards. It goes like this: if we just get a smarter model, our agent will work better. If the model scores higher on MMLU, on HumanEval, on MATH, on ARC-Challenge, then the system built on top of it will be more reliable, more useful, more trustworthy. This story is wrong. Not partially wrong, not wrong in edge cases, but wrong in the specific way that matters most when you are an architect responsible for a system that touches real users, real data, and real consequences.
The confusion arises because intelligence and verifiability are orthogonal properties. A system can be extraordinarily capable and completely unauditable. It can produce brilliant outputs that you cannot trace, cannot verify, cannot replay, and cannot explain to a regulator, a customer, or your own engineering team when something goes wrong. Conversely, a system built on a modest model can be deeply trustworthy if it is instrumented, constrained, and observable at every step of its reasoning chain.
The shift that is now happening in the industry is a shift in what the word "quality" means when applied to an LLM-based agent. Quality used to mean: does it give good answers? Quality now means: can I prove it gave good answers, understand why it gave them, intervene when it is about to give bad ones, and reconstruct exactly what happened after the fact? This is not a soft, philosophical distinction. It is the difference between a system you can deploy in a regulated environment and one you cannot. It is the difference between a system that gets better over time because you can learn from its failures and one that stays mysterious forever.
This article is written for architects and technical decision-makers who are building or evaluating LLM-based agent systems. It will walk through the four pillars of verifiable agents: observability, verification, control, and auditability. For each pillar, it will explain not just what the concept means but why it matters, how it is implemented, and what it looks like in running code that works with both local models via Ollama and remote models via the OpenAI API. The code examples are not toys. They are the skeleton of a production-grade verifiable agent framework, and every line of them is chosen to make a specific architectural point.
Before running any of the code in this article, you will need Python 3.13 or higher and the requests library. If you want to use local models, you will need Ollama installed and running on your machine. If you want to use remote OpenAI models, you will need an OpenAI API key. The complete set of dependencies is small by design: the framework deliberately avoids heavy orchestration libraries so that every architectural decision is visible and explicit.
The file requirements.txt for this project contains exactly one third-party dependency:
requests>=2.32.5
You install it with:
pip install -r requirements.txt
To install Ollama, visit https://ollama.com and follow the instructions for your operating system. Once installed, pull the default model used in this article with:
ollama pull llama4
Llama 4 Scout, Meta's fourth-generation open-weight model released in April 2026, is the recommended local model. It uses a Mixture-of-Experts architecture with 17 billion active parameters and a 10-million-token context window, and it runs comfortably on a machine with 24 GB of VRAM. If your hardware is more constrained, Qwen 3.6 7B is an excellent alternative that runs on 8 GB of VRAM and can be pulled with ollama pull qwen3.6:7b. For the OpenAI backend, the default model in this article is gpt-5.6-terra, the balanced variant of OpenAI's July 2026 model family, which offers strong performance at a moderate price point. The flagship gpt-5.6-sol is available for tasks requiring maximum capability. Set your API key as an environment variable before running:
export OPENAI_API_KEY="your-key-here"
On Windows, use set OPENAI_API_KEY=your-key-here in the Command Prompt or $env:OPENAI_API_KEY="your-key-here" in PowerShell. With those prerequisites in place, every code example in this article will run as written.
CHAPTER TWO: WHAT MAKES AN AGENT "VERIFIABLE"
Before diving into implementation, it is worth being precise about what verifiability means in the context of an LLM agent. An agent, in the sense used here, is any system that uses a language model to take a sequence of actions toward a goal, where those actions may include calling tools, reading documents, writing code, querying databases, or interacting with external APIs. The agent is not just a chatbot that answers a single question. It is a system that reasons over multiple steps, accumulates context, and makes decisions that have downstream consequences.
Verifiability, then, is the property of a system that allows an external observer to answer four questions with confidence. The first question is: what did the system do, and in what order? The second question is: why did the system do each thing it did, in terms of the inputs it received and the reasoning it applied? The third question is: was each output correct, safe, and within the boundaries of what was intended? The fourth question is: if I replay the same inputs, will I get the same or equivalently correct outputs, and can I demonstrate that the system behaved consistently?
These four questions map directly onto the four pillars. Observability answers the first question by making every step of the agent's execution visible and structured. Verification answers the third question by applying explicit checks to outputs before they are acted upon. Control answers an implicit fifth question that architects often forget to ask: can I stop the agent, redirect it, or constrain it in real time? Auditability answers the fourth question by creating a durable, replayable record of everything that happened.
Notice what is not in this list. Model capability is not in this list. The number of parameters is not in this list. The benchmark score is not in this list. This is intentional. A verifiable agent built on a smaller model will outperform an unverifiable agent built on a larger model in any environment where accountability matters, because the verifiable agent can be debugged, improved, and trusted in ways the unverifiable one cannot.
CHAPTER THREE: THE ARCHITECTURE OF VERIFIABILITY
Before writing a single line of code, it helps to have a clear picture of what a verifiable agent architecture looks like. The diagram below represents the high-level structure. Each box is a component that will be implemented and explained in the chapters that follow.
+----------------------------------------------------------+
| VERIFIABLE AGENT SYSTEM |
| |
| +----------------+ +---------------------------+ |
| | LLM Adapter |<--->| Agent Execution Loop | |
| | (local/remote) | | (plan, act, observe) | |
| +----------------+ +---------------------------+ |
| | | |
| v v |
| +----------------+ +---------------------------+ |
| | Observability | | Verification Engine | |
| | Layer | | (output checking) | |
| | (span tracing) | +---------------------------+ |
| +----------------+ | |
| | v |
| v +---------------------------+ |
| +----------------+ | Control Plane | |
| | Audit Logger |<----->| (guardrails, policies) | |
| | (replay store) | +---------------------------+ |
| +----------------+ |
+----------------------------------------------------------+
The LLM Adapter sits at the foundation. It is the component that abstracts away the difference between a locally running model such as Llama 4 Scout served through Ollama and a remotely hosted model such as gpt-5.6-terra through the OpenAI API. Above it sits the Agent Execution Loop, which is the core reasoning engine: the component that decides what to do next, calls tools, and accumulates results. The Observability Layer wraps every interaction with the LLM and every tool call, recording structured spans that describe what happened. The Verification Engine inspects every output before it is used, applying rules and heuristics to catch problems early. The Control Plane enforces policies, applies guardrails, and provides the mechanism by which a human or automated system can intervene. The Audit Logger persists everything to a durable store and provides the replay capability that makes post-hoc analysis possible.
Each of these components is independently testable, independently deployable, and independently improvable. That is not an accident. It is the architectural consequence of taking verifiability seriously.
CHAPTER FOUR: THE LLM ADAPTER — ABSTRACTING AWAY THE MODEL
The first and most fundamental component is the LLM Adapter. Its job is to provide a uniform interface to language models regardless of where they run. This matters for verifiability because the observability and verification layers need to intercept every call to the model, and they can only do that reliably if all model calls go through a single, well-defined interface.
The adapter pattern used here is deliberately simple. It defines an abstract base class with a single method, complete, that takes a list of messages and a set of options and returns a structured response. Both the local Ollama and remote OpenAI implementations satisfy this interface. The calling code never needs to know which implementation it is using.
The entire framework lives in a single file called verifiable_agent.py. All imports are gathered at the top of that file. The from __future__ import annotations import enables postponed evaluation of annotations, which allows forward references in type hints without string quoting and is a best practice in Python 3.13:
"""
verifiable_agent.py
A production-skeleton framework for verifiable LLM agents.
Supports local inference via Ollama and remote inference via
the OpenAI-compatible Chat Completions API.
Requirements:
Python 3.13+
requests >= 2.32.5 (pip install requests)
Optional (local inference):
Ollama >= 0.32.0 (https://ollama.com)
Recommended model: ollama pull llama4
Optional (remote inference):
OPENAI_API_KEY environment variable set to a valid key.
Default model: gpt-5.6-terra
"""
from __future__ import annotations
import abc
import hashlib
import json
import logging
import os
import pathlib
import time
import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Dict, List, Optional
import requests
# Configure a module-level logger. Callers can adjust the level
# and handlers to suit their environment (console, file, OTLP, etc.).
logger = logging.getLogger(__name__)
With the imports established, the core data structures for the adapter layer can be defined. The Message dataclass represents a single turn in a conversation, and the LLMResponse dataclass carries not just the generated text but also the metadata that the observability layer will use to build spans:
@dataclass
class Message:
"""
Represents a single message in a conversation.
Attributes:
role : one of 'system', 'user', or 'assistant'
content : the text of the message
"""
role: str
content: str
@dataclass
class LLMResponse:
"""
A structured response from any LLM backend.
Carries not just the generated text but also metadata that
the observability layer uses to build spans. The latency_ms
field is always filled in by the base class after the call
completes, so subclasses should leave it at 0.0.
Attributes:
text : the model's generated text
model : the model identifier as reported by the backend
prompt_tokens : number of tokens in the input prompt
completion_tokens : number of tokens in the generated output
latency_ms : end-to-end call latency in milliseconds
raw : the full raw response dict from the backend
"""
text: str
model: str
prompt_tokens: int
completion_tokens: int
latency_ms: float
raw: Dict[str, Any] = field(default_factory=dict)
The abstract base class follows. The design decision to put timing and logging in the base class rather than in each subclass is important. It means that no matter which backend is used, every call is measured. A developer cannot add a new backend and accidentally skip the instrumentation. This is a small but concrete example of how architecture enforces verifiability: the structure of the code makes it impossible to do the wrong thing by accident:
class LLMAdapter(abc.ABC):
"""
Abstract base class for all LLM backends.
The public 'complete' method handles timing and debug logging,
then delegates to '_complete_impl', which subclasses must
implement. This guarantees that every call is instrumented
regardless of which backend is in use.
"""
def complete(
self,
messages: List[Message],
temperature: float = 0.0,
max_tokens: int = 1024,
) -> LLMResponse:
"""
Public entry point. Times the call, delegates to
_complete_impl, fills in latency_ms, and returns
a structured LLMResponse.
"""
start = time.monotonic()
response = self._complete_impl(messages, temperature, max_tokens)
elapsed_ms = (time.monotonic() - start) * 1000.0
response.latency_ms = elapsed_ms
logger.debug(
"LLM call completed",
extra={
"model": response.model,
"latency_ms": elapsed_ms,
"prompt_tokens": response.prompt_tokens,
"completion_tokens": response.completion_tokens,
},
)
return response
@abc.abstractmethod
def _complete_impl(
self,
messages: List[Message],
temperature: float,
max_tokens: int,
) -> LLMResponse:
"""
Subclasses implement this method to call their specific
backend. The base class handles timing; subclasses should
leave latency_ms at 0.0 in the returned LLMResponse.
"""
...
Now the two concrete implementations. The OllamaAdapter calls a locally running Ollama server, which as of version 0.32.0 (July 2026) can serve Llama 4 Scout, Llama 4 Maverick, Qwen 3.6, Gemma 3, Phi-4, and many other models. The OpenAIAdapter calls the OpenAI Chat Completions API. Both produce the same LLMResponse structure. Both use a requests.Session for connection pooling, which significantly reduces latency when the same adapter instance makes multiple calls in sequence, as it always does inside an agent loop:
class OllamaAdapter(LLMAdapter):
"""
Calls a locally running Ollama server (>= v0.32.0).
Ollama must be installed and running, with the desired model
already pulled. Examples:
ollama pull llama4 # Llama 4 Scout (recommended)
ollama pull qwen3.6:7b # Qwen 3.6 7B (low-VRAM option)
ollama pull phi4 # Phi-4 14B
ollama pull gemma3:12b # Gemma 3 12B
The default base URL assumes Ollama's standard local port.
Streaming is disabled so that the full response arrives in
one HTTP reply, which simplifies error handling and tracing.
"""
def __init__(
self,
model: str = "llama4",
base_url: str = "http://localhost:11434",
) -> None:
self.model = model
self.base_url = base_url.rstrip("/")
# A persistent session reuses the TCP connection across
# multiple calls, cutting per-call overhead significantly.
self._session = requests.Session()
def _complete_impl(
self,
messages: List[Message],
temperature: float,
max_tokens: int,
) -> LLMResponse:
"""
Calls the Ollama /api/chat endpoint with streaming disabled.
Ollama returns token counts in 'eval_count' (completion tokens)
and 'prompt_eval_count' (prompt tokens). These map directly
onto the LLMResponse fields.
"""
payload = {
"model": self.model,
"messages": [
{"role": m.role, "content": m.content}
for m in messages
],
"stream": False,
"options": {
"temperature": temperature,
"num_predict": max_tokens,
},
}
response = self._session.post(
f"{self.base_url}/api/chat",
json=payload,
timeout=120,
)
response.raise_for_status()
data = response.json()
return LLMResponse(
text=data["message"]["content"],
model=self.model,
prompt_tokens=data.get("prompt_eval_count", 0),
completion_tokens=data.get("eval_count", 0),
latency_ms=0.0, # filled in by LLMAdapter.complete
raw=data,
)
class OpenAIAdapter(LLMAdapter):
"""
Calls the OpenAI Chat Completions API (/v1/chat/completions).
Compatible with any OpenAI-API-compatible endpoint, including:
- OpenAI (gpt-5.6-sol / gpt-5.6-terra / gpt-5.6-luna)
- Azure OpenAI
- Together AI
- Groq
- vLLM (local, OpenAI-compatible server)
- LM Studio (local, OpenAI-compatible server)
The API key is read from the OPENAI_API_KEY environment variable
if not supplied explicitly. Never hardcode API keys in source files.
This adapter uses the raw requests library rather than the OpenAI
Python SDK. This makes it compatible with any provider that speaks
the OpenAI wire format, regardless of whether the SDK supports it.
The persistent session carries the Authorization header on every
request without repeating it in calling code.
"""
def __init__(
self,
model: str = "gpt-5.6-terra",
api_key: Optional[str] = None,
base_url: str = "https://api.openai.com/v1",
) -> None:
self.model = model
self.api_key = api_key or os.environ.get("OPENAI_API_KEY", "")
self.base_url = base_url.rstrip("/")
self._session = requests.Session()
self._session.headers.update(
{
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
)
def _complete_impl(
self,
messages: List[Message],
temperature: float,
max_tokens: int,
) -> LLMResponse:
"""
Calls the /chat/completions endpoint and parses the response.
The 'usage' field in the response contains prompt_tokens and
completion_tokens. The actual model name is read from the
response rather than from self.model, because the API may
resolve an alias (e.g., 'gpt-5.6-terra') to a versioned name.
"""
payload = {
"model": self.model,
"messages": [
{"role": m.role, "content": m.content}
for m in messages
],
"temperature": temperature,
"max_tokens": max_tokens,
}
response = self._session.post(
f"{self.base_url}/chat/completions",
json=payload,
timeout=120,
)
response.raise_for_status()
data = response.json()
usage = data.get("usage", {})
return LLMResponse(
text=data["choices"][0]["message"]["content"],
model=data.get("model", self.model),
prompt_tokens=usage.get("prompt_tokens", 0),
completion_tokens=usage.get("completion_tokens", 0),
latency_ms=0.0, # filled in by LLMAdapter.complete
raw=data,
)
One detail worth noting is the comment in OpenAIAdapter that says "compatible with any OpenAI-API-compatible endpoint." This is architecturally significant. The OpenAI Chat Completions wire format has become a de facto standard, and many providers — Together AI, Groq, Anyscale, and local servers like vLLM and LM Studio — expose the same interface. By implementing the adapter against the raw HTTP API rather than the OpenAI Python SDK, we gain the ability to point the same code at any of these providers simply by changing the base_url and api_key. This is a practical form of verifiability: the system is not locked to a single provider, which means you can switch backends for cost, latency, or compliance reasons without changing any other part of the system.
A factory function ties the two adapters together and makes it easy to select between them from configuration:
def create_adapter(config: Dict[str, Any]) -> LLMAdapter:
"""
Factory function that creates the appropriate LLM adapter
based on a configuration dictionary.
Expected config keys:
'backend' : 'ollama' or 'openai' (required)
'model' : model name string (optional, has defaults)
'base_url' : endpoint override (optional)
'api_key' : API key string (optional, env var fallback)
Raises ValueError for unknown backend values.
"""
backend = config.get("backend", "ollama").lower()
if backend == "ollama":
return OllamaAdapter(
model=config.get("model", "llama4"),
base_url=config.get("base_url", "http://localhost:11434"),
)
if backend == "openai":
return OpenAIAdapter(
model=config.get("model", "gpt-5.6-terra"),
api_key=config.get("api_key"),
base_url=config.get(
"base_url", "https://api.openai.com/v1"
),
)
raise ValueError(
f"Unknown backend '{backend}'. "
f"Supported backends: 'ollama', 'openai'."
)
With this factory in place, switching from a local Llama 4 Scout model to gpt-5.6-terra is a matter of changing a configuration dictionary, not changing any application code. This separation of configuration from logic is a prerequisite for the kind of systematic experimentation that makes agents improvable over time.
CHAPTER FIVE: OBSERVABILITY — MAKING THE INVISIBLE VISIBLE
The single most common failure mode in production LLM systems is not a bad model output. It is the inability to understand what happened when something went wrong. A user reports that the agent gave a wrong answer. The engineer looks at the logs and sees a request and a response. No intermediate steps. No tool calls. No reasoning chain. No indication of which part of the process failed. The engineer is left guessing, and guessing is not engineering.
Observability is the discipline of making the internal state of a system visible to external observers without changing the system's behavior. In the context of LLM agents, this means recording every step of the agent's reasoning as a structured "span," a concept borrowed from distributed systems tracing as standardized by OpenTelemetry. A span has a name, a start time, an end time, a set of attributes, and optionally a set of child spans. The collection of all spans for a single agent invocation forms a trace, which is a complete, hierarchical record of everything that happened.
The implementation below defines a lightweight span and trace system. It is intentionally not dependent on any external tracing library, so it can be understood and used without additional infrastructure. In a production system, you would replace the in-memory store with an OpenTelemetry exporter targeting Jaeger, Honeycomb, Datadog, or any OTLP-compatible backend, but the concepts and the interface remain identical:
@dataclass
class Span:
"""
A single unit of work within an agent trace.
Spans are hierarchical: a span can contain child spans,
forming a tree that represents the full execution. The
parent_id field links each span to its parent; the root
span has parent_id=None.
Attributes:
span_id : unique identifier for this span (UUID4)
parent_id : span_id of the parent, or None for the root
name : human-readable name of the operation
kind : 'llm', 'tool', 'agent', or 'verification'
start_time : Unix timestamp when the span began
end_time : Unix timestamp when the span ended (None if pending)
attributes : arbitrary key-value metadata attached to this span
events : timestamped log entries recorded within the span
status : 'ok', 'error', or 'pending'
error : error message string if status is 'error'
"""
span_id: str = field(default_factory=lambda: str(uuid.uuid4()))
parent_id: Optional[str] = None
name: str = ""
kind: str = "agent"
start_time: float = field(default_factory=time.time)
end_time: Optional[float] = None
attributes: Dict[str, Any] = field(default_factory=dict)
events: List[Dict[str, Any]] = field(default_factory=list)
status: str = "pending"
error: Optional[str] = None
def add_event(
self,
name: str,
attributes: Optional[Dict[str, Any]] = None,
) -> None:
"""Records a timestamped event within this span."""
self.events.append(
{
"name": name,
"timestamp": time.time(),
"attributes": attributes or {},
}
)
def finish(
self,
status: str = "ok",
error: Optional[str] = None,
) -> None:
"""Marks the span as complete and records the end time."""
self.end_time = time.time()
self.status = status
self.error = error
@property
def duration_ms(self) -> float:
"""Returns the span duration in milliseconds, or 0.0 if pending."""
if self.end_time is None:
return 0.0
return (self.end_time - self.start_time) * 1000.0
The Tracer class manages the creation and storage of spans for a single agent invocation. It maintains a stack of active spans so that child spans can automatically reference their parent. The context manager interface is the most important part of the Tracer design: it ensures that spans are always closed, even when exceptions occur, and that parent-child relationships are always recorded correctly:
class Tracer:
"""
Manages the creation and storage of spans for a single
agent invocation.
The tracer maintains a stack of active spans so that child
spans automatically reference their parent. The recommended
usage pattern is the context manager returned by 'span()':
tracer = Tracer(trace_id="my-run-001")
with tracer.span("agent.step", kind="agent") as step_span:
step_span.attributes["step_number"] = 1
with tracer.span("llm.call", kind="llm") as llm_span:
llm_span.attributes["model"] = "llama4"
Spans are stored in insertion order in self.spans.
"""
def __init__(self, trace_id: Optional[str] = None) -> None:
self.trace_id: str = trace_id or str(uuid.uuid4())
self.spans: List[Span] = []
self._active_stack: List[Span] = []
def start_span(self, name: str, kind: str = "agent") -> Span:
"""
Creates a new span and pushes it onto the active stack.
The new span's parent_id is set to the span_id of the
current top of the stack, or None if the stack is empty.
"""
parent_id = (
self._active_stack[-1].span_id
if self._active_stack
else None
)
span = Span(name=name, kind=kind, parent_id=parent_id)
self.spans.append(span)
self._active_stack.append(span)
return span
def end_span(
self,
span: Span,
status: str = "ok",
error: Optional[str] = None,
) -> None:
"""
Finishes a span and pops it from the active stack.
Raises ValueError if the given span is not the current
top of the stack, which would indicate a programming error
in the calling code (spans must be closed in LIFO order).
"""
if not self._active_stack or self._active_stack[-1] is not span:
raise ValueError(
"Span stack mismatch: the span being ended is not "
"the current active span. Spans must be closed in "
"last-in, first-out order."
)
span.finish(status=status, error=error)
self._active_stack.pop()
class _SpanContext:
"""
Context manager returned by Tracer.span().
On entry, starts a new span. On exit, finishes it with
status 'error' if an exception occurred, or 'ok' otherwise.
Exceptions are never suppressed.
"""
def __init__(
self,
tracer: Tracer,
name: str,
kind: str,
) -> None:
self._tracer = tracer
self._name = name
self._kind = kind
self._span: Optional[Span] = None
def __enter__(self) -> Span:
self._span = self._tracer.start_span(self._name, self._kind)
return self._span
def __exit__(
self,
exc_type: Any,
exc_val: Any,
exc_tb: Any,
) -> bool:
status = "error" if exc_type is not None else "ok"
error = str(exc_val) if exc_val is not None else None
self._tracer.end_span(self._span, status=status, error=error)
return False # never suppress exceptions
def span(self, name: str, kind: str = "agent") -> _SpanContext:
"""
Returns a context manager that creates a span on entry
and finishes it on exit, handling errors automatically.
"""
return self._SpanContext(self, name, kind)
def to_dict(self) -> Dict[str, Any]:
"""
Serializes the full trace to a dictionary.
This is the format that the AuditLogger persists to disk.
Every span is included with all its attributes and events.
"""
return {
"trace_id": self.trace_id,
"span_count": len(self.spans),
"spans": [
{
"span_id": s.span_id,
"parent_id": s.parent_id,
"name": s.name,
"kind": s.kind,
"start_time": s.start_time,
"end_time": s.end_time,
"duration_ms": s.duration_ms,
"attributes": s.attributes,
"events": s.events,
"status": s.status,
"error": s.error,
}
for s in self.spans
],
}
The context manager pattern used in the Tracer class is worth examining closely. When you write with tracer.span("llm.call", kind="llm") as span:, the span is automatically started when you enter the block and automatically finished when you exit it. If an exception is raised inside the block, the span is marked with status "error" and the exception message is recorded. This means that error information is captured automatically, without any explicit error-handling code in the calling code. The span stack ensures that parent-child relationships are recorded correctly even when spans are nested many levels deep.
The observability layer wraps the LLM adapter so that every call to the model creates a span automatically. This is the instrumented adapter pattern: a decorator that adds observability without changing the underlying implementation:
class ObservableLLMAdapter(LLMAdapter):
"""
Wraps any LLMAdapter and records a span for every call.
The span captures the full input messages, the response text,
token counts, and latency. This gives the observability layer
complete visibility into every model interaction without
requiring any changes to the underlying adapter.
Usage:
base = OllamaAdapter(model="llama4")
tracer = Tracer()
observable = ObservableLLMAdapter(base, tracer)
# Use 'observable' exactly like any other LLMAdapter.
"""
def __init__(self, adapter: LLMAdapter, tracer: Tracer) -> None:
self._adapter = adapter
self._tracer = tracer
def _complete_impl(
self,
messages: List[Message],
temperature: float,
max_tokens: int,
) -> LLMResponse:
"""
Opens a traced span, delegates to the wrapped adapter's
_complete_impl, records all relevant attributes, and
returns the response. The span is closed automatically
by the context manager, even if an exception occurs.
"""
with self._tracer.span("llm.complete", kind="llm") as span:
# Record the full input so the call can be replayed later.
span.attributes["messages"] = [
{"role": m.role, "content": m.content}
for m in messages
]
span.attributes["temperature"] = temperature
span.attributes["max_tokens"] = max_tokens
response = self._adapter._complete_impl(
messages, temperature, max_tokens
)
# Record the output and token usage for analysis.
span.attributes["response_text"] = response.text
span.attributes["model"] = response.model
span.attributes["prompt_tokens"] = response.prompt_tokens
span.attributes["completion_tokens"] = response.completion_tokens
return response
The ObservableLLMAdapter is a classic example of the decorator pattern applied to observability. The calling code uses it exactly like any other LLMAdapter. The difference is that every call now leaves a structured record in the tracer. When an engineer needs to understand what the agent did, they can look at the trace and see every message that was sent to the model, every response that came back, and exactly how long each call took. This is not logging in the traditional sense. It is structured, hierarchical, queryable data.
CHAPTER SIX: VERIFICATION — TRUSTING BUT CHECKING
The second pillar of verifiable agents is verification. Verification is the practice of applying explicit, programmatic checks to the outputs of the language model before those outputs are used to take actions. This is the "trust but verify" principle applied to AI systems, and it is more important than most architects initially realize.
The reason verification matters so much is that language models are probabilistic systems. Even a very capable model such as gpt-5.6-sol or Llama 4 Maverick will occasionally produce outputs that are syntactically malformed, semantically incorrect, outside the intended domain, or in violation of safety constraints. In a single-shot question-answering system, a bad output is annoying. In an agent system where the output of one step becomes the input to the next, a bad output can cascade through the entire reasoning chain, leading to compounding errors that are very difficult to diagnose after the fact.
Verification catches these problems at the point of production, before they propagate. The verification engine below implements a composable system of checks. Each check is a function that takes an LLMResponse and returns a VerificationResult. Multiple checks can be composed into a pipeline, and the pipeline can be configured differently for different kinds of agent steps:
@dataclass
class VerificationResult:
"""
The result of running a single verification check on an LLMResponse.
Attributes:
passed : True if the check passed, False otherwise
check_name : the name of the check that produced this result
message : human-readable explanation of the result
severity : 'info', 'warning', or 'error'
details : optional structured data for further analysis
"""
passed: bool
check_name: str
message: str
severity: str = "error"
details: Dict[str, Any] = field(default_factory=dict)
class VerificationCheck(abc.ABC):
"""
Abstract base class for a single verification check.
Subclasses implement the 'check' method. The 'name' property
must return a unique, stable string that identifies the check
in traces and audit logs.
"""
@property
@abc.abstractmethod
def name(self) -> str:
"""A unique, human-readable name for this check."""
...
@abc.abstractmethod
def check(self, response: LLMResponse) -> VerificationResult:
"""
Evaluates the response and returns a VerificationResult.
Implementations must not raise exceptions; any internal
errors should be returned as failed VerificationResults
so that the pipeline can continue and record the failure.
"""
...
class NotEmptyCheck(VerificationCheck):
"""
Verifies that the response text is not empty or whitespace-only.
This is the most basic check and should be included in every
verification pipeline. An empty response almost always indicates
a backend error or a misconfigured prompt.
"""
@property
def name(self) -> str:
return "not_empty"
def check(self, response: LLMResponse) -> VerificationResult:
passed = bool(response.text and response.text.strip())
return VerificationResult(
passed=passed,
check_name=self.name,
message=(
"Response is non-empty."
if passed
else "Response is empty or whitespace-only."
),
severity="error",
)
class MaxLengthCheck(VerificationCheck):
"""
Verifies that the response does not exceed a maximum character count.
Useful for preventing runaway generation in constrained contexts
and for catching cases where the model has entered a repetition loop.
The default limit of 4000 characters is generous for tool calls
and structured outputs; adjust it for your specific use case.
"""
def __init__(self, max_chars: int = 4000) -> None:
self._max_chars = max_chars
@property
def name(self) -> str:
return f"max_length_{self._max_chars}"
def check(self, response: LLMResponse) -> VerificationResult:
length = len(response.text)
passed = length <= self._max_chars
return VerificationResult(
passed=passed,
check_name=self.name,
message=(
f"Response length {length} is within limit {self._max_chars}."
if passed
else f"Response length {length} exceeds limit {self._max_chars}."
),
severity="warning",
details={
"actual_length": length,
"max_length": self._max_chars,
},
)
class ValidJSONCheck(VerificationCheck):
"""
Verifies that the response text is valid, parseable JSON.
Use this check for agent steps that are expected to return
structured data such as tool call arguments, planning outputs,
or structured reasoning chains. Invalid JSON at this stage
means the tool dispatch code will fail; catching it here
allows the agent to retry with explicit error feedback.
"""
@property
def name(self) -> str:
return "valid_json"
def check(self, response: LLMResponse) -> VerificationResult:
try:
parsed = json.loads(response.text.strip())
return VerificationResult(
passed=True,
check_name=self.name,
message="Response is valid JSON.",
severity="error",
details={"parsed_type": type(parsed).__name__},
)
except json.JSONDecodeError as exc:
return VerificationResult(
passed=False,
check_name=self.name,
message=f"Response is not valid JSON: {exc}",
severity="error",
details={"json_error": str(exc)},
)
class ForbiddenPhraseCheck(VerificationCheck):
"""
Verifies that the response does not contain any forbidden phrases.
This is a simple but effective guardrail for content safety and
domain restriction. The check is case-insensitive by default.
Common uses include blocking refusal phrases that indicate the
model has misunderstood its role, or blocking domain-specific
terms that should never appear in the output.
"""
def __init__(
self,
forbidden_phrases: List[str],
case_sensitive: bool = False,
) -> None:
self._phrases = forbidden_phrases
self._case_sensitive = case_sensitive
@property
def name(self) -> str:
return "forbidden_phrase"
def check(self, response: LLMResponse) -> VerificationResult:
text = response.text
if not self._case_sensitive:
text = text.lower()
found = [
phrase
for phrase in self._phrases
if (phrase if self._case_sensitive else phrase.lower()) in text
]
passed = len(found) == 0
return VerificationResult(
passed=passed,
check_name=self.name,
message=(
"No forbidden phrases found."
if passed
else f"Forbidden phrases found: {found}"
),
severity="error",
details={"found_phrases": found},
)
With the individual checks defined, the VerificationPipeline composes them into a sequence and integrates with the tracer. The pipeline runs all checks and aggregates the results, recording a verification span for every run so that the trace contains a complete record of what was checked and what passed or failed:
class VerificationPipeline:
"""
Runs a sequence of VerificationChecks against an LLMResponse
and aggregates the results.
The pipeline can be configured to fail fast (stop on first
failure) or to run all checks and collect all results. It
integrates with the Tracer to record a verification span for
every run, making the verification results visible in the trace.
Usage:
pipeline = VerificationPipeline(
checks=[NotEmptyCheck(), ValidJSONCheck()],
tracer=tracer,
fail_fast=False,
)
results = pipeline.run(response)
if not pipeline.all_passed(results):
# handle failure
"""
def __init__(
self,
checks: List[VerificationCheck],
tracer: Optional[Tracer] = None,
fail_fast: bool = False,
) -> None:
self._checks = checks
self._tracer = tracer
self._fail_fast = fail_fast
def run(self, response: LLMResponse) -> List[VerificationResult]:
"""
Runs all checks and returns the list of results.
If fail_fast is True, stops at the first failed check.
Records a 'verification.pipeline' span if a tracer is
provided, capturing which checks ran and which failed.
"""
results: List[VerificationResult] = []
def _execute_checks() -> None:
for check in self._checks:
result = check.check(response)
results.append(result)
if self._fail_fast and not result.passed:
break
if self._tracer is not None:
with self._tracer.span(
"verification.pipeline", kind="verification"
) as span:
_execute_checks()
failed = [r for r in results if not r.passed]
span.attributes["checks_run"] = len(results)
span.attributes["checks_failed"] = len(failed)
span.attributes["results"] = [
{
"check": r.check_name,
"passed": r.passed,
"message": r.message,
}
for r in results
]
if failed:
span.add_event(
"verification_failed",
{
"failed_checks": [
r.check_name for r in failed
]
},
)
else:
_execute_checks()
return results
def all_passed(self, results: List[VerificationResult]) -> bool:
"""Returns True only if every check in the results passed."""
return all(r.passed for r in results)
The composability of this verification system is its most important property. An architect can define different pipelines for different kinds of agent steps. A step that is expected to produce a JSON tool call gets the NotEmptyCheck, the ValidJSONCheck, and a ForbiddenPhraseCheck. A step that is expected to produce a natural language summary gets the NotEmptyCheck, the MaxLengthCheck, and a different ForbiddenPhraseCheck with domain-specific forbidden phrases. Each pipeline is independently testable, and the results are recorded in the trace, so you can see exactly which checks passed and failed for every step of every agent run.
The ValidJSONCheck deserves special attention because it addresses one of the most common failure modes in agentic systems. When an agent needs to call a tool, it typically does so by generating a JSON object that specifies the tool name and its arguments. If the model generates malformed JSON, the tool call fails. Without verification, this failure might propagate silently or produce a confusing error message deep in the tool execution code. With the ValidJSONCheck in place, the failure is caught immediately, labeled clearly, and recorded in the trace. The agent can then retry the step with an appropriate error message, rather than proceeding with bad data.
CHAPTER SEVEN: CONTROL — THE ABILITY TO INTERVENE
The third pillar is control. Control is the most underappreciated pillar of verifiable agents, and it is the one that most architects add last, when they should have added it first. Control is the property of a system that allows an external entity, whether a human operator, an automated policy engine, or another agent, to influence the behavior of a running agent. Without control, an agent is a black box that runs to completion regardless of what it does along the way. With control, an agent is a managed process that can be guided, constrained, paused, and redirected.
The control plane implementation below is built around the concept of a policy. A policy is a function that takes the current agent state and returns a control decision: allow the agent to continue, modify its behavior, or halt it entirely. Policies are composable: multiple policies can be applied in sequence, and the most restrictive decision wins:
class ControlDecision(Enum):
"""
The possible outcomes of a control policy evaluation.
ALLOW : the agent may proceed as planned
MODIFY : the agent should proceed but with modifications
specified in the ControlResult's 'modifications' field
HALT : the agent must stop immediately
ESCALATE : the agent must pause and wait for human review
"""
ALLOW = "allow"
MODIFY = "modify"
HALT = "halt"
ESCALATE = "escalate"
@dataclass
class ControlResult:
"""
The result of evaluating a single control policy.
Attributes:
decision : one of the ControlDecision values
policy_name : the name of the policy that produced this result
reason : human-readable explanation of the decision
modifications : optional dict of modifications to apply
(relevant when decision is MODIFY)
"""
decision: ControlDecision
policy_name: str
reason: str
modifications: Dict[str, Any] = field(default_factory=dict)
@dataclass
class AgentState:
"""
A snapshot of the agent's current state, passed to control
policies for evaluation at the start of each agent step.
Attributes:
step_number : how many steps the agent has completed so far
messages : the current conversation history
last_response : the most recent LLM response, or None at step 0
tool_calls : list of tool call records accumulated so far
metadata : arbitrary additional context from the caller
"""
step_number: int
messages: List[Message]
last_response: Optional[LLMResponse]
tool_calls: List[Dict[str, Any]]
metadata: Dict[str, Any] = field(default_factory=dict)
class ControlPolicy(abc.ABC):
"""
Abstract base class for a control policy.
Subclasses implement the 'evaluate' method, which inspects
the current AgentState and returns a ControlResult.
"""
@property
@abc.abstractmethod
def name(self) -> str:
"""A unique, human-readable name for this policy."""
...
@abc.abstractmethod
def evaluate(self, state: AgentState) -> ControlResult:
"""
Evaluates the current agent state and returns a ControlResult.
Implementations must not raise exceptions; any internal errors
should result in a HALT decision with an explanatory reason.
"""
...
class MaxStepsPolicy(ControlPolicy):
"""
Halts the agent if it has taken more than a maximum number of steps.
This prevents infinite loops and runaway agents that consume
unbounded resources. The default of 10 steps is conservative;
adjust it based on the complexity of the tasks your agent handles.
"""
def __init__(self, max_steps: int = 10) -> None:
self._max_steps = max_steps
@property
def name(self) -> str:
return f"max_steps_{self._max_steps}"
def evaluate(self, state: AgentState) -> ControlResult:
if state.step_number >= self._max_steps:
return ControlResult(
decision=ControlDecision.HALT,
policy_name=self.name,
reason=(
f"Agent has reached the maximum step limit of "
f"{self._max_steps}. Halting to prevent runaway "
f"execution."
),
)
return ControlResult(
decision=ControlDecision.ALLOW,
policy_name=self.name,
reason=f"Step {state.step_number} is within the limit.",
)
class TokenBudgetPolicy(ControlPolicy):
"""
Halts the agent if total token consumption exceeds a budget.
This provides a cost control mechanism for remote API calls
and a resource control mechanism for local inference. Token
counts are accumulated from the tool_calls log and from the
most recent LLM response.
The default budget of 50,000 tokens is suitable for most
short-to-medium agent tasks. Adjust it based on your cost
tolerance and the typical complexity of your workload.
"""
def __init__(self, max_tokens: int = 50000) -> None:
self._max_tokens = max_tokens
@property
def name(self) -> str:
return f"token_budget_{self._max_tokens}"
def evaluate(self, state: AgentState) -> ControlResult:
# Accumulate tokens from all completed tool call steps.
total_tokens = sum(
tc.get("tokens_used", 0) for tc in state.tool_calls
)
# Add tokens from the most recent response if available.
if state.last_response is not None:
total_tokens += (
state.last_response.prompt_tokens
+ state.last_response.completion_tokens
)
if total_tokens > self._max_tokens:
return ControlResult(
decision=ControlDecision.HALT,
policy_name=self.name,
reason=(
f"Token budget of {self._max_tokens} exceeded. "
f"Total tokens used: {total_tokens}."
),
)
return ControlResult(
decision=ControlDecision.ALLOW,
policy_name=self.name,
reason=f"Token usage {total_tokens} is within budget.",
)
The ControlPlane evaluates all policies and returns the most restrictive decision. The precedence ordering is a deliberate design choice: when multiple policies are evaluated, the most restrictive decision always wins. If one policy says ALLOW and another says HALT, the agent halts. This is the "fail safe" principle: in the presence of uncertainty or conflict, the system defaults to the more conservative behavior. Policy evaluation happens inside the traced span so that the span's duration accurately reflects the time spent evaluating policies:
class ControlPlane:
"""
Evaluates a sequence of control policies against the current
agent state and returns the most restrictive decision.
The order of precedence is:
HALT (3) > ESCALATE (2) > MODIFY (1) > ALLOW (0)
The control plane records its decisions in the tracer so that
every policy evaluation is visible in the trace. Policy
evaluation happens inside the traced span so that the span's
duration accurately reflects the evaluation time.
"""
_PRECEDENCE: Dict[ControlDecision, int] = {
ControlDecision.HALT: 3,
ControlDecision.ESCALATE: 2,
ControlDecision.MODIFY: 1,
ControlDecision.ALLOW: 0,
}
def __init__(
self,
policies: List[ControlPolicy],
tracer: Optional[Tracer] = None,
) -> None:
self._policies = policies
self._tracer = tracer
def evaluate(self, state: AgentState) -> ControlResult:
"""
Runs all policies and returns the most restrictive result.
All policy results are recorded in a 'control.evaluate' span
if a tracer is provided. The winning policy and its reason
are recorded as span attributes.
"""
if self._tracer is not None:
with self._tracer.span(
"control.evaluate", kind="agent"
) as span:
return self._run_policies(state, span)
return self._run_policies(state, span=None)
def _run_policies(
self,
state: AgentState,
span: Optional[Span],
) -> ControlResult:
"""
Internal helper that runs all policies, selects the most
restrictive result, and optionally records it on a span.
"""
results = [policy.evaluate(state) for policy in self._policies]
winning_result = max(
results,
key=lambda r: self._PRECEDENCE[r.decision],
)
if span is not None:
span.attributes["step_number"] = state.step_number
span.attributes["decision"] = winning_result.decision.value
span.attributes["winning_policy"] = winning_result.policy_name
span.attributes["all_results"] = [
{
"policy": r.policy_name,
"decision": r.decision.value,
"reason": r.reason,
}
for r in results
]
return winning_result
The MaxStepsPolicy and TokenBudgetPolicy are examples of resource control policies. They prevent the agent from consuming unbounded resources, which is one of the most common failure modes in production agentic systems. An agent that gets stuck in a loop, or that encounters an unexpectedly complex problem, can make hundreds of LLM calls and spend significant money before a human notices. These policies make such runaway behavior impossible by design.
CHAPTER EIGHT: THE AGENT EXECUTION LOOP
With the adapter, the observability layer, the verification engine, and the control plane in place, the agent execution loop can be assembled. This is the component that actually does the reasoning: it takes a goal, generates a plan, executes steps, observes results, and iterates until the goal is achieved or a control policy halts it.
The loop below implements a simplified but realistic ReAct-style (Reasoning + Acting) agent. In each step, the agent generates a response that either contains a tool call formatted as JSON or a final answer. The verification pipeline checks the response, the control plane evaluates the state, and the tracer records everything. The Tool dataclass represents a callable tool that the agent can invoke:
@dataclass
class Tool:
"""
Represents a tool that the agent can call.
Attributes:
name : the tool's identifier, used in JSON tool calls
description : a natural language description for the system prompt
function : the Python callable that implements the tool;
it is called with keyword arguments matching the
'input' dict from the agent's JSON tool call
"""
name: str
description: str
function: Callable[..., str]
The VerifiableAgent class assembles all components into a coherent execution loop. The system prompt template instructs the model to respond exclusively with JSON, which is what makes the ValidJSONCheck effective: the prompt establishes a contract, and the check enforces it:
class VerifiableAgent:
"""
A ReAct-style agent that integrates observability, verification,
and control at every step of its execution loop.
The agent operates as follows:
1. Build a system prompt that describes available tools.
2. At the start of each step, evaluate the control plane.
3. Call the LLM to get a response.
4. Run the verification pipeline on the response.
5. If verification fails, add error feedback and retry.
6. Parse the response as a JSON action.
7. If the action is 'final_answer', return the result.
8. Otherwise, execute the named tool and add the result
to the conversation, then repeat from step 2.
All steps are recorded as spans in the tracer.
"""
SYSTEM_PROMPT_TEMPLATE = (
"You are a helpful assistant with access to the following tools. "
"To use a tool, respond with a JSON object in this exact format:\n\n"
'{{"action": "tool_name", "input": {{"arg1": "value1"}}}}\n\n'
"To provide a final answer, respond with:\n\n"
'{{"action": "final_answer", "input": {{"answer": "your answer here"}}}}\n\n'
"Available tools:\n"
"{tool_descriptions}\n\n"
"Always respond with valid JSON. "
"Do not include any text outside the JSON object."
)
def __init__(
self,
adapter: LLMAdapter,
tools: List[Tool],
verification_pipeline: VerificationPipeline,
control_plane: ControlPlane,
tracer: Tracer,
max_retries: int = 2,
) -> None:
# Wrap the adapter with observability before storing it.
self._adapter = ObservableLLMAdapter(adapter, tracer)
self._tools: Dict[str, Tool] = {t.name: t for t in tools}
self._verification = verification_pipeline
self._control = control_plane
self._tracer = tracer
self._max_retries = max_retries
def _build_system_prompt(self) -> str:
"""Constructs the system prompt with tool descriptions."""
tool_descriptions = "\n".join(
f"- {t.name}: {t.description}"
for t in self._tools.values()
)
return self.SYSTEM_PROMPT_TEMPLATE.format(
tool_descriptions=tool_descriptions
)
def _parse_action(self, text: str) -> Optional[Dict[str, Any]]:
"""
Attempts to parse the LLM response as a JSON action dict.
Returns the parsed dict on success, or None if the text
is not valid JSON. The ValidJSONCheck should have caught
invalid JSON before this method is called, so None here
indicates an unexpected inconsistency.
"""
try:
return json.loads(text.strip())
except json.JSONDecodeError:
return None
def _execute_tool(
self,
action: Dict[str, Any],
span: Span,
) -> str:
"""
Executes a tool call and returns the result as a string.
Records the tool name, input, and result on the provided span.
If the tool is not found or raises an exception, returns an
error string rather than propagating the exception, so that
the agent can handle the error gracefully.
"""
tool_name = action.get("action", "")
tool_input = action.get("input", {})
span.attributes["tool_name"] = tool_name
span.attributes["tool_input"] = tool_input
if tool_name not in self._tools:
error_msg = (
f"Tool '{tool_name}' is not available. "
f"Available tools: {list(self._tools.keys())}"
)
span.add_event("tool_not_found", {"tool_name": tool_name})
return error_msg
try:
result = self._tools[tool_name].function(**tool_input)
result_str = str(result)
span.attributes["tool_result"] = result_str
span.add_event(
"tool_succeeded",
{"result_length": len(result_str)},
)
return result_str
except Exception as exc:
error_msg = f"Tool '{tool_name}' raised an error: {exc}"
span.add_event("tool_error", {"error": str(exc)})
return error_msg
def run(self, user_query: str) -> Dict[str, Any]:
"""
Runs the agent on a user query and returns a result dict.
The result dict contains:
'final_answer' : the agent's answer string, or None
'halt_reason' : why the agent stopped, if not by answer
'tool_calls' : list of tool call records
'trace' : the full serialized trace dict
'success' : True if a final_answer was produced
"""
with self._tracer.span("agent.run", kind="agent") as root_span:
root_span.attributes["user_query"] = user_query
messages: List[Message] = [
Message(
role="system",
content=self._build_system_prompt(),
),
Message(role="user", content=user_query),
]
tool_calls_log: List[Dict[str, Any]] = []
final_answer: Optional[str] = None
halt_reason: Optional[str] = None
step_number = 0
# The hard upper bound of 100 ensures the loop always
# terminates even if all control policies are misconfigured.
for step_number in range(100):
with self._tracer.span(
f"agent.step.{step_number}", kind="agent"
) as step_span:
step_span.attributes["step_number"] = step_number
# --- CONTROL CHECK (before any LLM call) ---
state = AgentState(
step_number=step_number,
messages=messages,
last_response=None,
tool_calls=tool_calls_log,
)
control_result = self._control.evaluate(state)
step_span.attributes["control_decision"] = (
control_result.decision.value
)
if control_result.decision == ControlDecision.HALT:
halt_reason = control_result.reason
step_span.add_event(
"agent_halted",
{"reason": halt_reason},
)
break
# --- LLM CALL WITH VERIFICATION AND RETRY ---
response: Optional[LLMResponse] = None
verification_passed = False
for attempt in range(self._max_retries + 1):
response = self._adapter.complete(
messages, temperature=0.0
)
v_results = self._verification.run(response)
verification_passed = self._verification.all_passed(
v_results
)
if verification_passed:
break
# Verification failed: add explicit error feedback
# so the model knows exactly what to correct.
failed = [r for r in v_results if not r.passed]
error_feedback = (
"Your previous response failed validation: "
+ "; ".join(r.message for r in failed)
+ ". Please respond again with a valid JSON object."
)
messages.append(
Message(
role="assistant",
content=response.text,
)
)
messages.append(
Message(role="user", content=error_feedback)
)
step_span.add_event(
"verification_retry",
{
"attempt": attempt,
"failed_checks": [
r.check_name for r in failed
],
},
)
if not verification_passed:
halt_reason = (
f"Verification failed after "
f"{self._max_retries + 1} attempt(s)."
)
break
# --- ACTION PARSING ---
action = self._parse_action(response.text)
if action is None:
# This should not happen if ValidJSONCheck passed,
# but we handle it defensively.
halt_reason = (
"Could not parse action from response after "
"successful verification. This indicates a "
"bug in the verification pipeline."
)
break
# --- DISPATCH: final answer or tool call ---
action_type = action.get("action", "")
if action_type == "final_answer":
final_answer = (
action.get("input", {}).get("answer", "")
)
step_span.attributes["final_answer"] = final_answer
break
# It is a tool call: execute and feed result back.
with self._tracer.span(
"tool.execute", kind="tool"
) as tool_span:
tool_result = self._execute_tool(action, tool_span)
tool_calls_log.append(
{
"step": step_number,
"action": action,
"result": tool_result,
"tokens_used": (
response.prompt_tokens
+ response.completion_tokens
),
}
)
messages.append(
Message(role="assistant", content=response.text)
)
messages.append(
Message(
role="user",
content=f"Tool result: {tool_result}",
)
)
root_span.attributes["total_steps"] = step_number
root_span.attributes["tool_calls_made"] = len(tool_calls_log)
root_span.attributes["final_answer"] = final_answer
root_span.attributes["halt_reason"] = halt_reason
return {
"final_answer": final_answer,
"halt_reason": halt_reason,
"tool_calls": tool_calls_log,
"trace": self._tracer.to_dict(),
"success": final_answer is not None,
}
The retry mechanism inside the step loop is a particularly important design decision. When verification fails, the agent does not simply give up or proceed with a bad response. Instead, it adds the error feedback as a new user message and asks the model to try again. This is a form of self-correction that is grounded in explicit, programmatic feedback rather than vague hope that the model will do better on the next attempt. The error message tells the model exactly what went wrong, which gives it the information it needs to correct its output.
The control check happens at the beginning of each step, before the LLM is called. This is intentional. If the agent has already exceeded its step limit or token budget, there is no point in making another LLM call. The control plane acts as a gate that prevents the agent from taking actions it is not authorized to take, rather than as a cleanup mechanism that runs after the fact.
CHAPTER NINE: AUDITABILITY — THE DURABLE RECORD
The fourth and final pillar is auditability. Auditability is the property of a system that allows its behavior to be reconstructed, analyzed, and explained after the fact. It is what makes it possible to answer the question "what exactly happened during run X?" days, weeks, or months after the run completed. It is what makes it possible to demonstrate to a regulator that the system behaved correctly. It is what makes it possible to learn from failures in a systematic way.
Auditability requires two things: a durable store that persists the trace data, and a replay mechanism that can reconstruct the agent's execution from the stored data. The AuditLogger below implements both. It also computes a SHA-256 content hash of each stored record, making the audit log tamper-evident: any modification to the stored file will cause the hash check to fail when the record is loaded:
class AuditLogger:
"""
Persists agent traces to a durable store and provides a replay
mechanism for post-hoc analysis.
In this implementation, traces are stored as JSON files in a
local directory. In a production system, this storage backend
would be replaced with a database, an object store (S3, GCS,
Azure Blob), or a dedicated observability platform such as
Datadog, Honeycomb, or an OTLP-compatible backend.
Each stored record includes a SHA-256 content hash that makes
the record tamper-evident: any modification to the stored file
will cause the hash check to fail on load. For cryptographic
guarantees, replace the hash with a digital signature.
Usage:
audit = AuditLogger(storage_dir="./audit_logs")
path = audit.log(trace=result["trace"], run_metadata={...})
record = audit.load(trace_id)
llm_calls = audit.replay_llm_calls(trace_id)
"""
# Schema version allows forward-compatible changes to the
# stored format without breaking existing readers.
SCHEMA_VERSION = "1.0"
def __init__(self, storage_dir: str = "./audit_logs") -> None:
self._storage_dir = pathlib.Path(storage_dir)
self._storage_dir.mkdir(parents=True, exist_ok=True)
def _compute_hash(self, data: str) -> str:
"""Returns the SHA-256 hex digest of the given string."""
return hashlib.sha256(data.encode("utf-8")).hexdigest()
def log(
self,
trace: Dict[str, Any],
run_metadata: Optional[Dict[str, Any]] = None,
) -> str:
"""
Persists a trace to the audit store.
The stored record contains the full trace, the run metadata,
a schema version, a log timestamp, and a content hash. The
hash is computed over the record without the hash field, then
the hash is added and the record is re-serialized and written.
Returns the filesystem path of the stored file.
"""
record: Dict[str, Any] = {
"schema_version": self.SCHEMA_VERSION,
"logged_at": time.time(),
"run_metadata": run_metadata or {},
"trace": trace,
}
# Compute hash over the record before adding the hash field,
# so that the hash can be verified by re-serializing the record
# after removing the hash field on load.
record_json_for_hashing = json.dumps(
record, indent=2, default=str
)
content_hash = self._compute_hash(record_json_for_hashing)
# Add the hash and write the final record to disk.
record["content_hash"] = content_hash
final_json = json.dumps(record, indent=2, default=str)
trace_id = trace.get("trace_id", str(uuid.uuid4()))
filepath = self._storage_dir / f"{trace_id}.json"
filepath.write_text(final_json, encoding="utf-8")
logger.info(
"Audit log written",
extra={
"trace_id": trace_id,
"filepath": str(filepath),
"content_hash": content_hash,
},
)
return str(filepath)
def load(self, trace_id: str) -> Dict[str, Any]:
"""
Loads a stored trace by its trace_id.
Verifies the content hash to detect tampering or corruption.
Raises FileNotFoundError if the trace does not exist.
Raises ValueError if the content hash does not match.
"""
filepath = self._storage_dir / f"{trace_id}.json"
if not filepath.exists():
raise FileNotFoundError(
f"No audit log found for trace_id '{trace_id}'."
)
raw_json = filepath.read_text(encoding="utf-8")
record = json.loads(raw_json)
# Remove the stored hash, re-serialize the remaining record,
# and recompute the hash to verify integrity.
stored_hash = record.pop("content_hash", None)
recomputed_json = json.dumps(record, indent=2, default=str)
recomputed_hash = self._compute_hash(recomputed_json)
if stored_hash != recomputed_hash:
raise ValueError(
f"Content hash mismatch for trace '{trace_id}'. "
f"The audit log may have been tampered with or corrupted.\n"
f"Stored hash: {stored_hash}\n"
f"Recomputed hash: {recomputed_hash}"
)
return record
def list_traces(self) -> List[Dict[str, Any]]:
"""
Returns a summary list of all stored traces, sorted by
log time in descending order (most recent first).
Unreadable files are skipped with a warning rather than
raising an exception, so that a single corrupted file does
not prevent access to all other traces.
"""
summaries: List[Dict[str, Any]] = []
for filepath in self._storage_dir.glob("*.json"):
try:
record = json.loads(filepath.read_text(encoding="utf-8"))
summaries.append(
{
"trace_id": record.get("trace", {}).get(
"trace_id"
),
"logged_at": record.get("logged_at"),
"span_count": record.get("trace", {}).get(
"span_count", 0
),
"filepath": str(filepath),
}
)
except Exception as exc:
logger.warning(
f"Could not read audit log {filepath}: {exc}"
)
return sorted(
summaries,
key=lambda s: s.get("logged_at", 0.0),
reverse=True,
)
def replay_llm_calls(self, trace_id: str) -> List[Dict[str, Any]]:
"""
Extracts all LLM call spans from a stored trace and returns
them as a list of replay-ready dictionaries.
Each dictionary contains the input messages, the response text,
and the model parameters, so that the call can be re-run against
any adapter for comparison. This is the foundation of
counterfactual analysis: 'what would have happened if we had
used a different model?'
"""
record = self.load(trace_id)
spans = record.get("trace", {}).get("spans", [])
llm_spans = [s for s in spans if s.get("kind") == "llm"]
return [
{
"span_id": s["span_id"],
"messages": s["attributes"].get("messages", []),
"temperature": s["attributes"].get("temperature", 0.0),
"max_tokens": s["attributes"].get("max_tokens", 1024),
"original_response": s["attributes"].get(
"response_text", ""
),
"original_model": s["attributes"].get("model", ""),
"original_latency_ms": s.get("duration_ms", 0.0),
}
for s in llm_spans
]
The content hash in the audit logger is a small but significant detail. It means that a stored audit log is tamper-evident: if anyone modifies the stored JSON file, the hash will no longer match when the file is loaded, and the system will raise an error. This is not cryptographic proof of authenticity — for that, you would need digital signatures — but it is sufficient to detect accidental corruption and most forms of deliberate tampering. In regulated environments, this kind of integrity checking is often a compliance requirement.
The replay_llm_calls method is the foundation of what might be called "counterfactual analysis." Once you have a stored trace, you can extract all the LLM calls that were made during the run and re-run them against a different model. This allows you to answer questions like: "If we had used Llama 4 Scout instead of gpt-5.6-terra for this run, would the outputs have been different? Would they have been correct?" This kind of systematic comparison is how you make evidence-based decisions about model selection, rather than relying on benchmark scores that may not reflect your actual use case.
CHAPTER TEN: PUTTING IT ALL TOGETHER
With all four pillars implemented, it is time to assemble them into a working system and demonstrate how they interact. The following code shows how to configure and run a VerifiableAgent on a concrete task, using either a local Ollama model or a remote OpenAI model depending on the configuration.
The example tools below are intentionally simple: a calculator and a simulated web search. In a real system, these would be database queries, API calls, code execution environments, or document retrievers. The calculator uses Python's built-in eval with a heavily restricted execution context. This is adequate for a demonstration but should be replaced with a proper expression parser in production:
def build_example_tools() -> List[Tool]:
"""
Creates a small set of example tools for demonstration purposes.
In a production system, these would be replaced with tools that
call real APIs, query databases, or execute code in sandboxed
environments. The calculator here uses a restricted eval context
for demonstration only; use a proper expression parser such as
the 'simpleeval' library in production.
"""
def calculator(expression: str) -> str:
"""
Evaluates a mathematical expression in a restricted context.
Only a small set of safe built-in functions is available.
This prevents arbitrary code execution while still supporting
common mathematical operations.
NOTE: For production use, replace eval with a dedicated
expression parser such as the 'simpleeval' library.
"""
allowed_names: Dict[str, Any] = {
"abs": abs,
"round": round,
"min": min,
"max": max,
"sum": sum,
"pow": pow,
}
try:
result = eval(
expression,
{"__builtins__": {}},
allowed_names,
)
return str(result)
except Exception as exc:
return f"Error evaluating '{expression}': {exc}"
def web_search(query: str) -> str:
"""
Simulates a web search with a small set of canned responses.
In production, replace this with a call to a real search API
such as Brave Search, Bing Web Search, or SerpAPI.
"""
simulated_results: Dict[str, str] = {
"population of germany": (
"Germany has a population of approximately "
"84 million people as of 2025."
),
"capital of france": "The capital of France is Paris.",
"capital of germany": "The capital of Germany is Berlin.",
}
query_lower = query.lower()
for key, value in simulated_results.items():
if key in query_lower:
return value
return (
f"No simulated results found for '{query}'. "
f"This is a demonstration search tool."
)
return [
Tool(
name="calculator",
description=(
"Evaluates a mathematical expression and returns the result. "
'Input format: {"expression": "math expression as string"}. '
'Example: {"expression": "84000000 / 1000"}'
),
function=calculator,
),
Tool(
name="web_search",
description=(
"Searches the web for factual information. "
'Input format: {"query": "search query string"}. '
'Example: {"query": "population of Germany"}'
),
function=web_search,
),
]
def run_verifiable_agent_demo(
backend: str = "ollama",
model: str = "llama4",
query: str = "What is the population of Germany divided by 1000?",
) -> None:
"""
Demonstrates the full verifiable agent system end-to-end.
Configures all components, runs the agent on a query, writes
the audit log, and prints a structured summary of the results
including the span tree and replay-ready LLM call records.
Arguments:
backend : 'ollama' for local inference, 'openai' for remote
model : the model name to use with the chosen backend
query : the natural language question for the agent to answer
Examples:
# Local inference with Llama 4 Scout (requires Ollama):
run_verifiable_agent_demo(backend="ollama", model="llama4")
# Local inference with Qwen 3.6 7B (low-VRAM option):
run_verifiable_agent_demo(backend="ollama", model="qwen3.6:7b")
# Remote inference with gpt-5.6-terra (requires OPENAI_API_KEY):
run_verifiable_agent_demo(backend="openai", model="gpt-5.6-terra")
# Remote inference with gpt-5.6-sol (highest capability):
run_verifiable_agent_demo(backend="openai", model="gpt-5.6-sol")
"""
# --- COMPONENT ASSEMBLY ---
config: Dict[str, Any] = {"backend": backend, "model": model}
tracer = Tracer()
base_adapter = create_adapter(config)
verification_pipeline = VerificationPipeline(
checks=[
NotEmptyCheck(),
MaxLengthCheck(max_chars=2000),
ValidJSONCheck(),
ForbiddenPhraseCheck(
forbidden_phrases=["I cannot", "I'm sorry, I can't"],
case_sensitive=False,
),
],
tracer=tracer,
fail_fast=False,
)
control_plane = ControlPlane(
policies=[
MaxStepsPolicy(max_steps=8),
TokenBudgetPolicy(max_tokens=20000),
],
tracer=tracer,
)
tools = build_example_tools()
agent = VerifiableAgent(
adapter=base_adapter,
tools=tools,
verification_pipeline=verification_pipeline,
control_plane=control_plane,
tracer=tracer,
max_retries=2,
)
audit_logger = AuditLogger(storage_dir="./audit_logs")
# --- EXECUTION ---
print(f"Backend : {backend}")
print(f"Model : {model}")
print(f"Query : {query}")
print("-" * 60)
result = agent.run(user_query=query)
# --- AUDIT LOGGING ---
run_metadata: Dict[str, Any] = {
"backend": backend,
"model": model,
"query": query,
"user": "demo_user",
"timestamp": time.time(),
}
log_path = audit_logger.log(
trace=result["trace"],
run_metadata=run_metadata,
)
# --- RESULTS DISPLAY ---
print(f"Success : {result['success']}")
print(f"Final answer : {result['final_answer']}")
if result["halt_reason"]:
print(f"Halt reason : {result['halt_reason']}")
print(f"Tool calls : {len(result['tool_calls'])}")
print(f"Spans recorded: {result['trace']['span_count']}")
print(f"Audit log : {log_path}")
print("-" * 60)
# --- SPAN TREE DISPLAY ---
print("Span tree:")
for span in result["trace"]["spans"]:
indent = " " if span["parent_id"] else ""
status_marker = "OK " if span["status"] == "ok" else "ERR"
print(
f" {indent}[{status_marker}] {span['name']} "
f"({span['duration_ms']:.1f} ms)"
)
# --- REPLAY DEMONSTRATION ---
trace_id = result["trace"]["trace_id"]
llm_calls = audit_logger.replay_llm_calls(trace_id)
print(f"\nReplay-ready LLM calls: {len(llm_calls)}")
for i, call in enumerate(llm_calls):
print(
f" Call {i + 1}: model={call['original_model']}, "
f"latency={call['original_latency_ms']:.1f} ms, "
f"messages={len(call['messages'])}"
)
if __name__ == "__main__":
# Configure basic logging so info messages are visible.
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
# Run with a local Llama 4 Scout model via Ollama.
# Requires: ollama pull llama4
run_verifiable_agent_demo(
backend="ollama",
model="llama4",
query="What is the population of Germany divided by 1000?",
)
# To run with gpt-5.6-terra via OpenAI, uncomment the block below.
# Requires: export OPENAI_API_KEY="your-key-here"
#
# run_verifiable_agent_demo(
# backend="openai",
# model="gpt-5.6-terra",
# query="What is the population of Germany divided by 1000?",
# )
The assembly of components in run_verifiable_agent_demo reveals something important about the architecture: each component is independently configurable. The verification pipeline can be made stricter or more lenient by adding or removing checks. The control plane can be made more or less restrictive by adjusting the policy parameters. The tracer can be swapped for an OpenTelemetry-based tracer without changing any other component. The adapter can be switched between local and remote backends by changing a single configuration value. This independence is not just a software engineering nicety. It is the property that makes the system improvable over time, because you can change one component and observe the effect without worrying about unintended interactions with the others.
CHAPTER ELEVEN: THE "AHA" MOMENT FOR ARCHITECTS
At this point, a thoughtful architect might ask: "This is all well and good, but isn't it a lot of overhead? Wouldn't it be simpler to just use a better model and trust it?" This question deserves a direct answer, and the answer is no, for three reasons that become clear only when you have operated an agent system in production.
The first reason is that model capability and system reliability are not the same thing. A more capable model will produce better outputs on average, but it will not eliminate the tail of bad outputs. In a system that runs thousands of agent invocations per day, even a 0.1% failure rate means dozens of failures per day. Without verification, those failures propagate silently. With verification, they are caught, logged, and handled. The verification layer is what converts a probabilistic system into a reliable one.
The second reason is that auditability is not optional in regulated environments. Healthcare, finance, legal, and government applications all have requirements around explainability and record-keeping that cannot be satisfied by a system that produces outputs without a trace. A verifiable agent can be deployed in these environments. An unverifiable one cannot, regardless of how capable its underlying model is.
The third reason is that the ability to improve a system depends on the ability to understand it. A team that operates an unverifiable agent can only improve it by changing the model or the prompt and hoping for the best. A team that operates a verifiable agent can look at the traces of failed runs, identify exactly where the failure occurred, understand why it occurred, and make a targeted fix. This is the difference between engineering and guessing.
The architectural insight that ties all of this together is that verifiability is a property of the system, not of the model. You cannot make a system verifiable by choosing a better model. You make it verifiable by designing it with observability, verification, control, and auditability as first-class concerns, from the very beginning. These are not features you add later. They are the foundation on which everything else is built.
CHAPTER TWELVE: WHAT COMES NEXT — THE FRONTIER OF VERIFIABLE AGENTS
The framework described in this article represents the current state of the art in verifiable agent design, but the field is moving quickly, and several important developments are on the horizon.
The first development is formal verification of reasoning chains. Current verification approaches check the format and content of outputs, but they do not verify the logical correctness of the reasoning that produced them. Researchers are working on methods that use symbolic reasoning, constraint satisfaction, and formal logic to verify that an agent's reasoning chain is internally consistent and leads to a correct conclusion. This is a much harder problem than format checking, but it is the next frontier.
The second development is cryptographic auditability. The content hashing used in the audit logger in this article is tamper-evident but not cryptographically secure. The next generation of audit systems will use digital signatures and zero-knowledge proofs to provide cryptographic guarantees that a trace was produced by a specific model at a specific time and has not been modified. This will be essential for applications where the audit log itself is a legal document.
The third development is multi-agent verifiability. The framework described here applies to a single agent. As systems evolve toward networks of agents that communicate and coordinate with each other, the verifiability problem becomes dramatically more complex. How do you trace a reasoning chain that spans multiple agents? How do you verify an output that was produced by the collaboration of three different models? These are open research questions, and the answers will define the next generation of agent architectures.
The fourth development is continuous verification. Current verification runs at the end of each agent step. Future systems will apply verification continuously, monitoring the agent's internal state in real time and intervening before a bad output is produced rather than after. This requires much deeper integration between the verification layer and the model itself, and it is one of the reasons why interpretability research is so important for the practical deployment of agent systems.
CONCLUSION: JUDGE AGENTS BY WHAT YOU CAN PROVE
The central argument of this article is simple: the quality of an LLM-based agent system should be measured not by the intelligence of its underlying model but by the verifiability of its behavior. A verifiable agent is one that you can observe, check, control, and audit. It is one that gets better over time because you can learn from its failures. It is one that can be deployed in regulated environments because it produces a durable, tamper-evident record of everything it does. It is one that your engineering team can maintain and improve because its behavior is transparent and its components are independently testable.
The four pillars of verifiability — observability, verification, control, and auditability — are not advanced features to be added after the system is working. They are the foundation of a system that is worth building. The code in this article is a starting point, not an endpoint. Take the adapter pattern and extend it to your specific backends. Take the verification pipeline and add checks that are specific to your domain. Take the control plane and add policies that reflect your organization's risk tolerance. Take the audit logger and connect it to your existing observability infrastructure.
The benchmark leaderboards will continue to advance. Models will continue to get smarter. But the agents that will be trusted with real work, in real organizations, with real consequences, will be the ones that can prove what they did, why they did it, and that they did it correctly. That proof is not in the model weights. It is in the architecture. Build accordingly.
No comments:
Post a Comment