You've built a chatbot that works great for single-turn Q&A. Then someone asks you to make it "actually do things" — pull data from a database, generate a report, validate the output, email a stakeholder if something looks wrong, and retry if the LLM hallucinates a function call. Suddenly, you're not building a prompt anymore. You're building a system. And systems fail in ways that prompts don't.
Agentic loops — LLM pipelines where the model reasons, acts, observes results, and decides what to do next — are one of the most powerful patterns in modern AI engineering. They're also one of the most brittle if you don't design them carefully. An agent that silently retries a failed database write three times without telling anyone has just corrupted your data three times. An agent that loops forever because it can't parse its own tool output will burn through your API budget while you sleep. The gap between a demo that impresses stakeholders and a production system that earns trust comes down almost entirely to the engineering discipline you apply to the loop itself.
By the end of this lesson, you'll be able to design and implement production-grade agentic loops from scratch. You'll understand the state machine model underlying most agent frameworks, how to write retry logic that distinguishes between recoverable and unrecoverable errors, how to persist agent state so runs can resume after failure, and when and how to pause execution and hand control to a human.
What you'll learn:
This lesson assumes you're comfortable with:
You don't need prior experience with LangChain, LlamaIndex, or any agent framework — we're building from primitives on purpose. Understanding what's happening underneath frameworks makes you a much more effective user of them.
Before writing a single line of code, let's get the mental model right, because most agent bugs are really modeling bugs in disguise.
An agentic loop has a finite set of states, a set of transitions between those states, and actions that happen during or between transitions. Sound familiar? That's a state machine. When you accept this framing, a lot of questions answer themselves: What happens when a tool call fails? That's a transition to an error state. What triggers human review? That's a transition to a waiting state. Can we retry from here? That depends on whether the current state is idempotent.
Here's the state diagram for a moderately realistic agent — one that processes financial reports:
IDLE
│ (task received)
▼
PLANNING
│ (plan generated)
▼
EXECUTING ◄────────────────────────┐
│ │
├─ (tool call succeeds) ─────► OBSERVING
│ │
├─ (tool call fails, retryable) ─┘ (retry counter incremented)
│
├─ (tool call fails, unrecoverable) ──► ERROR
│
├─ (confidence below threshold) ──► AWAITING_HUMAN
│ │
◄────────────────────────────────────────┘ (human responds)
│
├─ (all steps complete) ──► COMPLETING
│
└─ (max steps exceeded) ──► ABORTED
Now translate this to code. We'll represent states as an enum and the agent's current position as a field on a persistent state object:
from enum import Enum
from datetime import datetime
from typing import Any, Optional
from pydantic import BaseModel, Field
import uuid
class AgentState(str, Enum):
IDLE = "idle"
PLANNING = "planning"
EXECUTING = "executing"
OBSERVING = "observing"
AWAITING_HUMAN = "awaiting_human"
COMPLETING = "completing"
ERROR = "error"
ABORTED = "aborted"
class ToolCall(BaseModel):
tool_name: str
arguments: dict[str, Any]
result: Optional[Any] = None
error: Optional[str] = None
attempt_count: int = 0
succeeded: bool = False
class AgentRunState(BaseModel):
run_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
task: str
state: AgentState = AgentState.IDLE
plan: Optional[list[str]] = None
tool_call_history: list[ToolCall] = Field(default_factory=list)
current_step_index: int = 0
retry_count: int = 0
max_retries: int = 3
max_steps: int = 20
created_at: datetime = Field(default_factory=datetime.utcnow)
updated_at: datetime = Field(default_factory=datetime.utcnow)
human_input_request: Optional[str] = None
human_input_response: Optional[str] = None
final_output: Optional[str] = None
error_message: Optional[str] = None
def transition_to(self, new_state: AgentState) -> None:
"""Explicit state transitions with logging."""
print(f"[{self.run_id}] {self.state} → {new_state}")
self.state = new_state
self.updated_at = datetime.utcnow()
The transition_to method is deliberately simple. In production you'd add validation logic here — for example, raising an exception if a transition is illegal (you should never go from COMPLETING back to EXECUTING). Making transitions explicit rather than just reassigning self.state = something everywhere means you have one place to add logging, validation, and side effects.
Why this matters in practice: When an agent fails at 2 AM and you look at the logs, the state transition history is the single most valuable debugging artifact you have. If your state changes are scattered across 15 functions with no centralized logging, you'll spend hours reconstructing what happened. Centralize your transitions.
An agent that loses all its progress when a process crashes is not a production agent — it's an expensive demo. State persistence is what separates the two.
The requirements for agent state storage are different from typical application storage. You need:
SQLite works well for local development and moderate scale. Postgres is better for production multi-agent systems. We'll use SQLite with a simple wrapper so the interface is backend-agnostic.
import sqlite3
import json
from pathlib import Path
class AgentStateStore:
def __init__(self, db_path: str = "agent_state.db"):
self.db_path = db_path
self._initialize_db()
def _initialize_db(self):
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS agent_runs (
run_id TEXT PRIMARY KEY,
state TEXT NOT NULL,
task TEXT NOT NULL,
full_state_json TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_state
ON agent_runs(state)
""")
conn.commit()
def save(self, agent_run: AgentRunState) -> None:
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
INSERT INTO agent_runs
(run_id, state, task, full_state_json, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(run_id) DO UPDATE SET
state = excluded.state,
full_state_json = excluded.full_state_json,
updated_at = excluded.updated_at
""", (
agent_run.run_id,
agent_run.state.value,
agent_run.task,
agent_run.model_dump_json(),
agent_run.created_at.isoformat(),
agent_run.updated_at.isoformat()
))
conn.commit()
def load(self, run_id: str) -> Optional[AgentRunState]:
with sqlite3.connect(self.db_path) as conn:
row = conn.execute(
"SELECT full_state_json FROM agent_runs WHERE run_id = ?",
(run_id,)
).fetchone()
if row is None:
return None
return AgentRunState.model_validate_json(row[0])
def find_by_state(self, state: AgentState) -> list[AgentRunState]:
with sqlite3.connect(self.db_path) as conn:
rows = conn.execute(
"SELECT full_state_json FROM agent_runs WHERE state = ?",
(state.value,)
).fetchall()
return [AgentRunState.model_validate_json(row[0]) for row in rows]
The key design decision here is storing full_state_json alongside indexed columns. The indexed columns let you query efficiently ("give me all runs awaiting human review") while the JSON blob lets you reconstruct the full agent state without schema migrations every time you add a field.
Warning: The UPSERT pattern here (
ON CONFLICT DO UPDATE) is intentional. You wantsave()to be idempotent — calling it twice with the same state should be safe. This matters because your retry logic will sometimes callsave()more than once for a given state if an exception occurs mid-step.
Now, every step of your agent loop ends with a store.save(state) call. If the process dies, the next invocation loads the state and resumes. This is the foundation of reliable agentic systems.
The most dangerous part of an agentic loop is tool execution — this is where the agent reaches out and touches the real world. A poor tool execution layer treats all errors the same way. A good one classifies errors and responds appropriately.
There are three broad categories of tool call errors:
1. Transient errors — the tool is temporarily unavailable. The right response is retry with backoff. Examples: API rate limits (HTTP 429), network timeouts, database connection pool exhaustion.
2. Recoverable logic errors — the LLM called the tool with bad arguments. The right response is retry with corrective context fed back to the LLM. Examples: invalid SQL syntax, a date range where start > end, a missing required field.
3. Unrecoverable errors — retrying won't help. The right response is escalation (human review or abort). Examples: authentication failures, attempting to write to a read-only resource, a tool call that would cause data loss with no reversibility.
from enum import Enum
import asyncio
import random
class ErrorCategory(str, Enum):
TRANSIENT = "transient"
RECOVERABLE_LOGIC = "recoverable_logic"
UNRECOVERABLE = "unrecoverable"
class ToolExecutionError(Exception):
def __init__(self, message: str, category: ErrorCategory, original: Exception = None):
super().__init__(message)
self.category = category
self.original = original
def classify_error(exc: Exception) -> ErrorCategory:
"""
Map exceptions to error categories.
You'll expand this based on the tools you're wrapping.
"""
err_str = str(exc).lower()
# Transient: worth retrying automatically
if any(kw in err_str for kw in ["timeout", "rate limit", "429", "503", "connection"]):
return ErrorCategory.TRANSIENT
# Recoverable logic: the LLM needs corrective feedback
if any(kw in err_str for kw in ["invalid", "syntax error", "validation", "parse error"]):
return ErrorCategory.RECOVERABLE_LOGIC
# Everything else: escalate
return ErrorCategory.UNRECOVERABLE
Now, here's the retry wrapper with exponential backoff and jitter. The jitter is not optional — without it, a fleet of agents that all fail at the same moment will all retry at the same moment, creating a thundering herd that takes down your backend:
async def execute_tool_with_retry(
tool_fn,
arguments: dict,
max_transient_retries: int = 4,
base_delay: float = 1.0,
) -> Any:
"""
Execute a tool with layered retry logic.
Returns the result or raises a ToolExecutionError.
"""
last_exception = None
for attempt in range(max_transient_retries + 1):
try:
result = await tool_fn(**arguments)
return result
except Exception as exc:
last_exception = exc
category = classify_error(exc)
if category == ErrorCategory.UNRECOVERABLE:
raise ToolExecutionError(
f"Unrecoverable error in tool: {exc}",
category=ErrorCategory.UNRECOVERABLE,
original=exc
)
if category == ErrorCategory.RECOVERABLE_LOGIC:
# Don't retry at the tool level — surface to the LLM
raise ToolExecutionError(
f"Logic error (feed back to LLM): {exc}",
category=ErrorCategory.RECOVERABLE_LOGIC,
original=exc
)
# Transient: retry with exponential backoff + jitter
if attempt < max_transient_retries:
delay = base_delay * (2 ** attempt) + random.uniform(0, 1.0)
print(f"Transient error (attempt {attempt + 1}), retrying in {delay:.2f}s: {exc}")
await asyncio.sleep(delay)
else:
raise ToolExecutionError(
f"Max transient retries exceeded: {exc}",
category=ErrorCategory.TRANSIENT,
original=exc
)
raise ToolExecutionError(
f"Exhausted retries: {last_exception}",
category=ErrorCategory.TRANSIENT
)
Design note: Notice that
RECOVERABLE_LOGICerrors are raised immediately rather than retried at the tool level. That's intentional. The fix for a logic error isn't "try the same thing again" — it's "go back to the LLM with information about what went wrong." We handle that in the main loop.
Now we bring it all together. Here's the core loop for an agent that uses OpenAI tool calling to accomplish multi-step tasks. This is intentionally written to be readable and hackable, not maximally concise:
import openai
from openai import AsyncOpenAI
client = AsyncOpenAI()
# Define tools the agent can call
TOOLS = [
{
"type": "function",
"function": {
"name": "query_database",
"description": "Run a read-only SQL query against the analytics database.",
"parameters": {
"type": "object",
"properties": {
"sql": {
"type": "string",
"description": "Valid SQL SELECT statement"
}
},
"required": ["sql"]
}
}
},
{
"type": "function",
"function": {
"name": "send_email_report",
"description": "Send a formatted report to a stakeholder email address.",
"parameters": {
"type": "object",
"properties": {
"recipient_email": {"type": "string"},
"subject": {"type": "string"},
"body_markdown": {"type": "string"}
},
"required": ["recipient_email", "subject", "body_markdown"]
}
}
}
]
async def run_agent_loop(
task: str,
store: AgentStateStore,
tool_registry: dict,
run_id: Optional[str] = None,
) -> AgentRunState:
# Resume existing run or start new one
if run_id:
state = store.load(run_id)
if state is None:
raise ValueError(f"No run found with id {run_id}")
print(f"Resuming run {run_id} from state {state.state}")
else:
state = AgentRunState(task=task)
state.transition_to(AgentState.PLANNING)
store.save(state)
# Build message history from persisted state
messages = build_message_history(state)
while state.state not in {
AgentState.COMPLETING,
AgentState.ERROR,
AgentState.ABORTED,
AgentState.AWAITING_HUMAN
}:
# Guard: max steps
if state.current_step_index >= state.max_steps:
state.transition_to(AgentState.ABORTED)
state.error_message = f"Exceeded max steps ({state.max_steps})"
store.save(state)
break
state.transition_to(AgentState.EXECUTING)
store.save(state)
# Call the LLM
try:
response = await client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=TOOLS,
tool_choice="auto",
)
except openai.RateLimitError as e:
# LLM API itself rate-limited — treat as transient
delay = 10 + random.uniform(0, 5)
print(f"LLM rate limited, waiting {delay:.1f}s")
await asyncio.sleep(delay)
continue
except openai.APIError as e:
state.error_message = f"LLM API error: {e}"
state.transition_to(AgentState.ERROR)
store.save(state)
break
choice = response.choices[0]
messages.append(choice.message)
state.current_step_index += 1
# Agent decided it's done
if choice.finish_reason == "stop":
state.final_output = choice.message.content
state.transition_to(AgentState.COMPLETING)
store.save(state)
break
# Agent wants to call a tool
if choice.finish_reason == "tool_calls":
for tool_call in choice.message.tool_calls:
tool_call_record = ToolCall(
tool_name=tool_call.function.name,
arguments=json.loads(tool_call.function.arguments)
)
try:
tool_fn = tool_registry[tool_call.function.name]
result = await execute_tool_with_retry(
tool_fn,
tool_call_record.arguments
)
tool_call_record.result = result
tool_call_record.succeeded = True
# Feed result back to conversation
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result)
})
except ToolExecutionError as e:
tool_call_record.error = str(e)
tool_call_record.attempt_count += 1
if e.category == ErrorCategory.RECOVERABLE_LOGIC:
# Give the LLM a chance to self-correct
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": f"ERROR: {e}. Please review your arguments and try again."
})
state.retry_count += 1
if state.retry_count >= state.max_retries:
await escalate_to_human(
state,
store,
reason=f"Agent failed to self-correct after {state.retry_count} attempts. Last error: {e}"
)
return state
elif e.category in {ErrorCategory.UNRECOVERABLE, ErrorCategory.TRANSIENT}:
await escalate_to_human(
state,
store,
reason=f"Tool '{tool_call.function.name}' failed with unrecoverable error: {e}"
)
return state
state.tool_call_history.append(tool_call_record)
state.transition_to(AgentState.OBSERVING)
store.save(state)
return state
def build_message_history(state: AgentRunState) -> list[dict]:
"""Reconstruct message history from persisted tool call history."""
messages = [
{
"role": "system",
"content": (
"You are a data analyst agent. Complete the assigned task step by step, "
"using the provided tools. Be methodical. If a tool returns an error, "
"analyze it carefully before retrying."
)
},
{
"role": "user",
"content": state.task
}
]
# Replay tool call history into message format
for tc in state.tool_call_history:
if tc.succeeded:
messages.append({
"role": "assistant",
"content": None,
"tool_calls": [{
"id": f"call_{tc.tool_name}_{id(tc)}",
"type": "function",
"function": {
"name": tc.tool_name,
"arguments": json.dumps(tc.arguments)
}
}]
})
messages.append({
"role": "tool",
"tool_call_id": f"call_{tc.tool_name}_{id(tc)}",
"content": json.dumps(tc.result)
})
return messages
A note on context window management: As your tool call history grows, you'll eventually exceed the context window. The
build_message_historyfunction above is naive — it replays everything. In production, you need a context management strategy: summarize old tool results, drop the least-relevant steps, or use a retrieval layer. We cover this in depth in the context management lesson, but be aware the problem exists.
Human escalation is where a lot of agent designs fall down. The naive approach is to just raise an exception and let an on-call engineer figure it out. The better approach is to pause the agent in a well-defined state, package the context a human needs to make a decision, and resume automatically once they respond.
async def escalate_to_human(
state: AgentRunState,
store: AgentStateStore,
reason: str
) -> None:
"""
Pause the agent and notify a human reviewer.
The agent run remains in AWAITING_HUMAN state until
human_input_response is populated.
"""
state.human_input_request = reason
state.transition_to(AgentState.AWAITING_HUMAN)
store.save(state)
# In production: send a Slack message, PagerDuty alert, email, etc.
# The notification should include a link to a review UI with the run_id.
await notify_human_reviewer(
run_id=state.run_id,
task=state.task,
reason=reason,
tool_history_summary=summarize_tool_history(state.tool_call_history)
)
print(f"Run {state.run_id} escalated to human review: {reason}")
async def notify_human_reviewer(
run_id: str,
task: str,
reason: str,
tool_history_summary: str
) -> None:
"""
Stub for your notification system.
Replace with your Slack/PagerDuty/email integration.
"""
print(f"""
=== HUMAN REVIEW REQUIRED ===
Run ID: {run_id}
Task: {task}
Reason: {reason}
Tool History:
{tool_history_summary}
To resume: POST /agent/resume with run_id and response
============================
""")
def summarize_tool_history(tool_calls: list[ToolCall]) -> str:
lines = []
for i, tc in enumerate(tool_calls, 1):
status = "✓" if tc.succeeded else "✗"
lines.append(f"{i}. [{status}] {tc.tool_name}({json.dumps(tc.arguments)[:80]}...)")
if tc.error:
lines.append(f" Error: {tc.error}")
return "\n".join(lines) if lines else "No tool calls made yet."
Now, the resume path. This is the part most tutorials skip. A human has reviewed the situation and sent a response — how does the agent pick up where it left off?
async def resume_from_human_input(
run_id: str,
human_response: str,
store: AgentStateStore,
tool_registry: dict,
) -> AgentRunState:
"""
Resume an agent run that was awaiting human input.
The human's response is injected into the conversation as context.
"""
state = store.load(run_id)
if state is None:
raise ValueError(f"Run {run_id} not found")
if state.state != AgentState.AWAITING_HUMAN:
raise ValueError(f"Run {run_id} is in state {state.state}, not AWAITING_HUMAN")
state.human_input_response = human_response
state.retry_count = 0 # Reset retry counter after human intervention
# We'll pass the human's guidance as a system message on resume
# The run_agent_loop will call build_message_history which needs to include it
state.tool_call_history.append(
ToolCall(
tool_name="__human_input__",
arguments={"question": state.human_input_request},
result={"human_response": human_response},
succeeded=True
)
)
# Resume the loop
return await run_agent_loop(
task=state.task,
store=store,
tool_registry=tool_registry,
run_id=run_id
)
You'll notice the human input is inserted as a synthetic tool call in the history. This is a practical trick: it means build_message_history naturally includes the human's guidance when reconstructing the conversation, without needing special-case logic for "was there a human intervention?"
So far we've discussed reactive escalation — the agent escalates because something broke. But you also want proactive escalation — the agent escalates because it's uncertain, even if nothing technically failed.
One practical approach is to ask the LLM to self-assess confidence as part of its structured output before taking consequential actions:
from pydantic import BaseModel
class PreActionCheck(BaseModel):
action_description: str
confidence_score: float # 0.0 to 1.0
confidence_rationale: str
is_reversible: bool
potential_side_effects: list[str]
async def pre_action_confidence_check(
client: AsyncOpenAI,
proposed_action: str,
context: str
) -> PreActionCheck:
"""
Ask the LLM to evaluate its own confidence before a consequential action.
This is a separate, lightweight LLM call — not part of the main loop.
"""
response = await client.beta.chat.completions.parse(
model="gpt-4o-mini", # Use a cheaper model for the check
messages=[
{
"role": "system",
"content": "Evaluate the proposed action and return a structured confidence assessment."
},
{
"role": "user",
"content": f"Context: {context}\n\nProposed action: {proposed_action}"
}
],
response_format=PreActionCheck,
)
return response.choices[0].message.parsed
CONFIDENCE_THRESHOLD = 0.75
IRREVERSIBLE_ACTIONS = {"send_email_report", "write_database", "delete_record"}
async def should_escalate_before_action(
tool_name: str,
arguments: dict,
context: str,
client: AsyncOpenAI
) -> tuple[bool, str]:
"""
Returns (should_escalate, reason).
"""
if tool_name not in IRREVERSIBLE_ACTIONS:
return False, ""
check = await pre_action_confidence_check(
client,
proposed_action=f"Call {tool_name} with {json.dumps(arguments)[:200]}",
context=context
)
if check.confidence_score < CONFIDENCE_THRESHOLD:
reason = (
f"Confidence {check.confidence_score:.0%} below threshold for "
f"irreversible action '{tool_name}'. "
f"Rationale: {check.confidence_rationale}"
)
return True, reason
return False, ""
Calibration warning: LLM self-reported confidence scores are notoriously poorly calibrated. The model may say "95% confidence" and be wrong, or "40% confidence" and be right. Treat these scores as heuristics that trigger human review for genuinely ambiguous cases, not as precise probability estimates. Over time, tune your threshold based on observed outcomes in your specific domain.
One of the subtlest problems in agentic systems is side-effect management. If your agent retries a step, does it do the thing twice? Sending an email twice, charging a credit card twice, or inserting a row twice are all real failure modes.
The solution is idempotency keys. Before calling any tool with side effects, generate an idempotency key tied to the run_id and step:
import hashlib
def generate_idempotency_key(run_id: str, tool_name: str, arguments: dict) -> str:
"""
Deterministic idempotency key based on run context.
Same run + same tool + same arguments = same key.
"""
content = f"{run_id}:{tool_name}:{json.dumps(arguments, sort_keys=True)}"
return hashlib.sha256(content.encode()).hexdigest()[:16]
# Usage in tool wrapper:
async def send_email_report_idempotent(
recipient_email: str,
subject: str,
body_markdown: str,
idempotency_key: str
) -> dict:
"""
Email sender that respects idempotency keys.
In practice, you'd check a sent_emails table before sending.
"""
# Check if we already sent this
already_sent = await check_email_sent(idempotency_key)
if already_sent:
return {"status": "already_sent", "idempotency_key": idempotency_key}
# Actually send
result = await _do_send_email(recipient_email, subject, body_markdown)
# Record that we sent it
await record_email_sent(idempotency_key, recipient_email, subject)
return {"status": "sent", "idempotency_key": idempotency_key}
Not every tool you wrap will support idempotency keys natively. For those that don't, maintain an idempotency table in your state store:
class IdempotencyStore:
def __init__(self, db_path: str = "agent_state.db"):
self.db_path = db_path
with sqlite3.connect(db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS idempotency_log (
key TEXT PRIMARY KEY,
tool_name TEXT,
result_json TEXT,
executed_at TEXT
)
""")
conn.commit()
def get_cached_result(self, key: str) -> Optional[Any]:
with sqlite3.connect(self.db_path) as conn:
row = conn.execute(
"SELECT result_json FROM idempotency_log WHERE key = ?", (key,)
).fetchone()
return json.loads(row[0]) if row else None
def record_result(self, key: str, tool_name: str, result: Any) -> None:
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
INSERT OR IGNORE INTO idempotency_log (key, tool_name, result_json, executed_at)
VALUES (?, ?, ?, ?)
""", (key, tool_name, json.dumps(result), datetime.utcnow().isoformat()))
conn.commit()
You cannot debug what you cannot observe. An agentic loop that runs silently and fails silently is a production nightmare. Here's a minimal but effective observability layer:
import time
from dataclasses import dataclass, field
from typing import Callable
@dataclass
class AgentMetrics:
run_id: str
total_llm_calls: int = 0
total_tool_calls: int = 0
total_tool_errors: int = 0
total_human_escalations: int = 0
total_tokens_used: int = 0
loop_start_time: float = field(default_factory=time.time)
step_durations: list[float] = field(default_factory=list)
def record_step(self, duration: float):
self.step_durations.append(duration)
def summary(self) -> dict:
elapsed = time.time() - self.loop_start_time
return {
"run_id": self.run_id,
"elapsed_seconds": round(elapsed, 2),
"total_llm_calls": self.total_llm_calls,
"total_tool_calls": self.total_tool_calls,
"error_rate": round(
self.total_tool_errors / max(self.total_tool_calls, 1), 3
),
"total_tokens_used": self.total_tokens_used,
"avg_step_duration_ms": round(
sum(self.step_durations) / max(len(self.step_durations), 1) * 1000, 1
),
"human_escalations": self.total_human_escalations
}
In production, you'd push these metrics to Prometheus, Datadog, or OpenTelemetry. The fields above give you the minimum viable dashboard:
Structured logging tip: Every log line from your agent loop should include the
run_id. This lets you grep or query for the complete trace of any single run. Without this, correlating log lines from a long-running, concurrent agent system becomes nearly impossible.
You'll build a complete agentic loop for a realistic data task: a "quarterly metrics agent" that queries a SQLite database for Q3 revenue data, calculates month-over-month growth, and emails a formatted summary to a stakeholder — with full retry logic, state persistence, and human escalation if the growth rate exceeds a threshold (because anomalous numbers deserve human eyes).
Setup:
# Create a test database
import sqlite3
def setup_test_database(db_path: str = "metrics.db"):
with sqlite3.connect(db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS monthly_revenue (
month TEXT PRIMARY KEY,
revenue_usd REAL,
region TEXT
)
""")
conn.executemany(
"INSERT OR REPLACE INTO monthly_revenue VALUES (?, ?, ?)",
[
("2024-07", 1_250_000, "North America"),
("2024-08", 1_380_000, "North America"),
("2024-09", 1_895_000, "North America"), # Big jump — triggers escalation
]
)
conn.commit()
Your task:
Implement query_database as an async function that runs SELECT queries against metrics.db, returning results as a list of dicts. Add error handling that raises ToolExecutionError(ErrorCategory.RECOVERABLE_LOGIC) for SQL syntax errors.
Implement send_email_report as a stub that prints the email content and records the send in an idempotency table. Use the idempotency pattern from the lesson to ensure it never sends twice.
Wire up run_agent_loop with these two tools, using this task string:
Query the monthly_revenue table for Q3 2024 (July, August, September).
Calculate month-over-month growth rates. If any month shows growth > 30%,
flag it as anomalous. Generate a markdown-formatted summary and send it to
analytics@company.com with subject "Q3 2024 Revenue Summary".
Add the should_escalate_before_action check before send_email_report calls, with a confidence threshold of 0.80. Since September shows 37% growth, the agent should flag the anomaly in its report and the confidence check should trigger human review before sending.
Run the loop, observe the escalation, then use resume_from_human_input to inject the human response: "Confirmed — the September spike is accurate due to enterprise deal close. Proceed with sending."
Verify the final state shows COMPLETING, that the email was only sent once (even if you run the loop twice), and that AgentMetrics.summary() shows exactly one human escalation.
Mistake 1: Treating all retry logic as identical
The most common mistake is a single retry loop that doesn't distinguish between error types. If the LLM generates invalid SQL and you just retry the same tool call with the same SQL, you'll exhaust your retries without ever solving the problem. The fix is the three-category classification system from this lesson. Always ask: "Is this error transient, a logic error, or unrecoverable?"
Mistake 2: State drift — saving state before confirming success
If you save state.transition_to(OBSERVING) before the tool result is confirmed durable, you can end up with a state that says "tool succeeded" when it actually didn't. Save state after the tool result is in hand, not before. The pattern is: execute → record result → save state.
Mistake 3: Missing the max_steps guard
An agent that loops forever is not theoretical — it happens when the LLM gets into a pattern where each tool result looks like it should trigger another tool call. The max_steps guard is not a suggestion. Set it conservatively (20-30 steps is plenty for most tasks), and treat ABORTED as a signal to tune your prompt rather than to increase the limit.
Mistake 4: Escalating without context
A human reviewer who receives "the agent failed" is useless. A human reviewer who receives the full tool call history, the last error message, and the agent's current reasoning has what they need to make a decision in seconds. Always include summarize_tool_history and the human_input_request context in your escalation notifications.
Mistake 5: Not testing the resume path
The happy path gets tested constantly. The resume-from-human-input path almost never does, until it breaks in production at the worst possible moment. Write an integration test that explicitly runs a task to escalation, calls resume_from_human_input, and verifies the final state. Make it part of your CI pipeline.
Mistake 6: Infinite context accumulation
Every tool result you feed back into the conversation grows the context window. For long-running agents, you'll eventually hit the model's context limit. Watch your total_tokens_used metric. When it approaches 80% of your model's limit, implement context compression: summarize old tool results, drop intermediate reasoning steps, or use a RAG layer to retrieve relevant history rather than including everything.
Troubleshooting: The agent loops on OBSERVING → EXECUTING without making progress
This usually means the tool results are being appended to the message list correctly, but the model isn't recognizing them as resolved. Double-check that your tool_call_id values match between the assistant message (where the tool call is declared) and the tool message (where the result is provided). A mismatch causes OpenAI's API to silently treat the result as malformed, and the model re-requests the same tool.
Troubleshooting: Human escalation triggers but resume never works
The most common cause is that build_message_history doesn't handle the __human_input__ synthetic tool call correctly. When replaying history, the human input needs to surface as a user or system message, not as a tool result (since the LLM wasn't the one who called a tool). Consider a different representation: a special assistant message summarizing what the human said, inserted at the correct point in the conversation.
You've now built the core engineering infrastructure for production-grade agentic loops from the ground up. Let's consolidate the key design principles:
State machines, not implicit flow. Modeling your agent as an explicit state machine with typed states and logged transitions turns debugging from archaeology into structured investigation. The question "what happened?" has a clear answer.
Error classification is non-negotiable. Three categories — transient, recoverable logic, unrecoverable — determine whether you retry silently, surface to the LLM, or escalate to a human. Conflating these categories is the source of most reliability failures.
Persistence enables recovery. Every state transition should be persisted before the agent moves on. This isn't overhead — it's the foundation of a system that stakeholders can trust to run unattended.
Human escalation is a feature, not a failure. Well-designed escalation with rich context, a clean resume path, and idempotent tool execution means humans and agents can collaborate reliably rather than the agent trying to handle everything alone.
Observe everything. Token counts, step durations, error rates, and escalation frequencies are your telemetry. Without them, you're flying blind.
Where to go next:
The gap between an agentic prototype and an agentic system that earns production trust is precisely the engineering discipline covered in this lesson. Build it right from the start.
Learning Path: Building with LLMs