Picture this: you need to analyze six months of customer support tickets, identify the top complaint categories, cross-reference them against your product changelog, write a root cause analysis report, and then draft a prioritized remediation plan for the engineering team. If you hand that to a standard LLM prompt, you'll get either a refusal or a hallucinated answer based on training data that knows nothing about your actual tickets. If you do it manually, you're blocking out two days of calendar time. What you actually need is a system that can break that problem down autonomously, execute steps in sequence, check its own work, recover from failures, and hand you a finished artifact.
That's what agentic AI workflows do. The shift from "prompt → response" to "goal → autonomous multi-step execution" is one of the most consequential architectural changes happening in applied AI right now. But most explanations of agentic systems either stay at a hand-wavy conceptual level ("the agent uses tools!") or dive straight into framework code without explaining the underlying design decisions that determine whether your pipeline is actually reliable. This lesson bridges that gap. We're going to build real understanding of how these systems work, where they break, and how to design them for production.
By the end of this lesson, you will be able to design, implement, and debug multi-step autonomous pipelines with confidence — not just copy-paste from a framework README.
What you'll learn:
This is an expert-level lesson. You should be comfortable with:
You do not need prior experience with agent frameworks like LangChain or AutoGen, though understanding them will help you see where the abstractions are leaky.
The word "agent" has been so thoroughly abused in AI marketing that it's worth establishing precise language before writing a single line of code.
An agentic workflow is a system where an LLM (or ensemble of LLMs) makes decisions about which actions to take next based on intermediate results, rather than following a fixed, pre-determined sequence of steps. The key properties are:
What an agentic workflow is not: a chain of LLM calls where you've hardcoded the sequence. If you always call "summarize," then "extract entities," then "classify," in that order with no branching, you have a pipeline — a useful thing, but not an agent. The distinction matters because agents require fundamentally different reliability strategies than pipelines.
The spectrum looks roughly like this:
Single Prompt → Prompt Chain → Conditional Pipeline → ReAct Agent → Multi-Agent System
↑ ↑
Simplest Most Complex
Most Reliable Most Capable
The right point on this spectrum depends on your task. A hard-coded pipeline with good error handling will outperform a poorly designed agent on most structured tasks. Only reach for a full agentic architecture when the task genuinely requires dynamic reasoning about what to do next.
The intellectual backbone of most agentic systems is a loop borrowed from cognitive science and robotics. You'll see it called ReAct (Reason + Act), PAOR, or the "agent loop." The names vary but the structure is consistent.
The agent receives a goal and produces a plan. This is where LLMs shine — decomposing ambiguous goals into concrete sub-tasks. But planning has a critical failure mode: overcommitment. If the agent plans all twelve steps upfront, those later steps are planned without the information that will only exist after steps 1-4 execute.
Good planning is therefore hierarchical and lazy: produce a high-level plan immediately, but defer detailed planning for later steps until earlier steps have completed. This is sometimes called "planning with replanning."
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
class StepStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
FAILED = "failed"
SKIPPED = "skipped"
@dataclass
class PlanStep:
step_id: str
description: str
tool_hint: Optional[str] # Which tool this step likely needs
depends_on: list[str] = field(default_factory=list)
status: StepStatus = StepStatus.PENDING
result: Optional[str] = None
attempt_count: int = 0
max_attempts: int = 3
@dataclass
class ExecutionPlan:
goal: str
steps: list[PlanStep]
context_summary: str = "" # Rolling summary of what we've learned
def next_executable_step(self) -> Optional[PlanStep]:
"""Return the next step whose dependencies are all completed."""
completed_ids = {s.step_id for s in self.steps if s.status == StepStatus.COMPLETED}
for step in self.steps:
if step.status == StepStatus.PENDING:
if all(dep in completed_ids for dep in step.depends_on):
return step
return None
def is_complete(self) -> bool:
return all(s.status in (StepStatus.COMPLETED, StepStatus.SKIPPED)
for s in self.steps)
def has_failed(self) -> bool:
return any(s.status == StepStatus.FAILED and s.attempt_count >= s.max_attempts
for s in self.steps)
The agent selects and invokes a tool. The mechanics here are well-documented in framework READMEs, but the design of tool interfaces is where most production systems have problems. We'll cover tool design in depth in its own section.
The agent receives the tool's output and adds it to context. This sounds trivial, but it's not. Raw tool output is often long, noisy, or formatted in ways that degrade LLM reasoning. Observation processing — filtering, summarizing, structuring tool outputs before feeding them back — is one of the highest-leverage things you can do to improve agent reliability.
def process_observation(raw_output: str, step: PlanStep, max_chars: int = 2000) -> str:
"""
Transform raw tool output into an observation suitable for the agent's context.
In production, this might be another LLM call for complex outputs.
"""
# Truncate excessively long outputs with a note
if len(raw_output) > max_chars:
truncated = raw_output[:max_chars]
observation = (
f"[Output truncated at {max_chars} chars. Full result stored separately.]\n"
f"{truncated}"
)
else:
observation = raw_output
return f"Step '{step.step_id}' result:\n{observation}"
After observing a result, the agent evaluates whether the step succeeded, whether the result changes the remaining plan, and whether any recovery is needed. Most beginner implementations skip this phase entirely — they just append the result to context and move on. That's why most beginner agent implementations produce garbage at step 6.
Reflection is where self-correction happens. We'll cover the patterns in detail below.
Tools are the agent's hands. A poorly designed tool interface will cause your agent to fail in ways that are extremely hard to debug because the failure looks like model reasoning failure when it's actually an interface design failure.
Every tool should expose:
search, but search_internal_kb vs. search_web)Here's the difference between a naive tool and a production tool:
# NAIVE: Vague, returns raw output, errors are unstructured
def search_tickets(query: str) -> str:
results = db.query(f"SELECT * FROM tickets WHERE body LIKE '%{query}%'")
return str(results)
# PRODUCTION: Explicit contract, structured output, diagnostic errors
from dataclasses import dataclass
from typing import Any
import json
@dataclass
class ToolResult:
success: bool
data: Any
error_code: Optional[str] = None
error_message: Optional[str] = None
metadata: dict = field(default_factory=dict)
def to_json(self) -> str:
return json.dumps({
"success": self.success,
"data": self.data,
"error_code": self.error_code,
"error_message": self.error_message,
"metadata": self.metadata
}, default=str)
def search_support_tickets(
query: str,
date_from: str, # ISO 8601 format: "2024-01-01"
date_to: str,
max_results: int = 50,
category: Optional[str] = None
) -> ToolResult:
"""
Search customer support tickets by semantic query within a date range.
Use this when you need to find tickets related to a specific issue or topic.
Returns ticket IDs, subjects, and relevant excerpts — NOT full ticket bodies.
To get the full body of a specific ticket, use get_ticket_detail().
"""
try:
validated_from = parse_date(date_from)
validated_to = parse_date(date_to)
except ValueError as e:
return ToolResult(
success=False,
data=None,
error_code="INVALID_DATE_FORMAT",
error_message=f"Date parsing failed: {e}. Use ISO 8601 format: YYYY-MM-DD"
)
results = ticket_db.semantic_search(
query=query,
date_range=(validated_from, validated_to),
limit=max_results,
category_filter=category
)
return ToolResult(
success=True,
data=[{"id": t.id, "subject": t.subject, "excerpt": t.excerpt[:200]}
for t in results],
metadata={"total_matching": results.total_count, "returned": len(results)}
)
The description clause "NOT full ticket bodies" and "use get_ticket_detail() for full content" is doing important work here. It teaches the agent the intended usage pattern, preventing it from trying to extract information from a summary that isn't there.
One of the most common architectural mistakes is making tools too coarse-grained. A tool called analyze_customer_sentiment_and_generate_report combines three distinct operations. When it fails, you can't tell which sub-operation failed. When you want to reuse components, you can't.
Design tools at the grain of a single, atomic operation. If you find yourself writing an and in a tool name, split it.
The flip side — tools that are too fine-grained — creates a different problem: the agent has to make too many tool calls to accomplish simple goals, consuming context budget and increasing latency. The right grain is roughly "one meaningful unit of external state change or retrieval."
Architecture Tip: Design your tool library so that each tool maps to one external system boundary crossing. Queries within the same database can be different tools, but a tool that hits both your database and an external API is a smell — it means error handling is going to be messy and retries will be non-idempotent.
Here's a failure mode that isn't in any framework's quickstart: your agent works perfectly on five-step tasks, fails inexplicably on fifteen-step tasks, and nobody can figure out why. The answer is almost always context management.
Every step adds content to the conversation context. By step 12, you might have:
That's 32,600 tokens, pushing against GPT-4's 128K context limit — not on raw length, but on attention quality. LLMs don't maintain uniform attention across long contexts; early content gets increasingly "forgotten" in a practical sense. Your agent will start making decisions that ignore work it did in steps 1-3.
The solution is active context management: don't let the raw transcript grow unbounded. Compress it.
class AgentContextManager:
def __init__(self, llm_client, max_raw_steps: int = 5, summary_model: str = "gpt-4o-mini"):
self.llm_client = llm_client
self.max_raw_steps = max_raw_steps
self.summary_model = summary_model
self.completed_steps_summary: str = ""
self.recent_raw_steps: list[dict] = []
self.goal: str = ""
self.plan: ExecutionPlan = None
def add_step_result(self, step: PlanStep, observation: str):
"""Add a completed step. If we exceed max_raw_steps, compress the oldest."""
self.recent_raw_steps.append({
"step_id": step.step_id,
"description": step.description,
"observation": observation
})
if len(self.recent_raw_steps) > self.max_raw_steps:
self._compress_oldest_steps()
def _compress_oldest_steps(self, compress_count: int = 3):
"""Summarize the oldest N steps into the running summary."""
steps_to_compress = self.recent_raw_steps[:compress_count]
self.recent_raw_steps = self.recent_raw_steps[compress_count:]
compress_prompt = f"""
You are maintaining a running summary of an AI agent's work.
The agent is working toward this goal: {self.goal}
EXISTING SUMMARY OF PRIOR WORK:
{self.completed_steps_summary or "None yet."}
STEPS TO ADD TO SUMMARY:
{json.dumps(steps_to_compress, indent=2)}
Write an updated, concise summary that:
1. Preserves all factual findings (numbers, entity names, specific results)
2. Notes any errors that occurred and how they were handled
3. Identifies what's been confirmed vs. what's still uncertain
4. Does NOT include step IDs or procedural details — just the substance
Keep the summary under 500 words.
"""
response = self.llm_client.complete(
model=self.summary_model,
messages=[{"role": "user", "content": compress_prompt}]
)
self.completed_steps_summary = response.content
def build_context_for_next_step(self) -> str:
"""Construct the context block to prepend to the next agent prompt."""
parts = [f"GOAL: {self.goal}\n"]
if self.completed_steps_summary:
parts.append(f"SUMMARY OF COMPLETED WORK:\n{self.completed_steps_summary}\n")
if self.recent_raw_steps:
parts.append("RECENT STEP DETAILS (last few steps):")
for step in self.recent_raw_steps:
parts.append(f"- {step['step_id']}: {step['observation'][:300]}")
parts.append(f"\nCURRENT PLAN STATUS:\n{self._format_plan_status()}")
return "\n".join(parts)
This pattern ensures that factual findings are preserved in compressed form even as raw detail is shed, while keeping the last few steps in full detail for immediate reasoning.
"Self-correction" in most tutorial code means: if the tool returns an error, tell the LLM the error message and ask it to try again. That works for about 40% of failures. Here's what works for the other 60%.
Before you can correct a failure, you need to know what kind of failure you have:
| Failure Type | Symptom | Correct Response |
|---|---|---|
| Tool parameter error | Malformed JSON, wrong types | Regenerate the tool call with corrected schema |
| Semantic error | Tool succeeded but result is wrong | Reframe the query or use a different tool |
| Logic error | Correct data, wrong conclusion | Reflection step + re-evaluation |
| Hallucination | Confident wrong answer without tool use | Force tool use, add verification step |
| Stuck loop | Same action repeated 3+ times | Escalate to different strategy or human |
| Plan invalidation | New info makes old plan wrong | Full replan |
class SelfCorrectionEngine:
def __init__(self, llm_client, max_retries_per_step: int = 3):
self.llm_client = llm_client
self.max_retries = max_retries_per_step
self.failure_history: list[dict] = []
def diagnose_failure(self, step: PlanStep, error: ToolResult) -> str:
"""Classify the failure to route to appropriate recovery strategy."""
if error.error_code in ("INVALID_DATE_FORMAT", "MISSING_REQUIRED_PARAM", "TYPE_ERROR"):
return "parameter_error"
elif error.error_code in ("NO_RESULTS", "EMPTY_RESPONSE"):
return "semantic_error"
elif error.error_code in ("RATE_LIMIT", "TIMEOUT", "SERVICE_UNAVAILABLE"):
return "transient_error"
elif step.attempt_count >= 2 and step.result is None:
return "stuck"
else:
return "unknown"
async def attempt_recovery(
self,
step: PlanStep,
failure: ToolResult,
context: AgentContextManager
) -> tuple[bool, str]:
"""
Attempt to recover from a failure.
Returns (should_retry, recovery_instruction).
"""
failure_type = self.diagnose_failure(step, failure)
self.failure_history.append({
"step_id": step.step_id,
"attempt": step.attempt_count,
"failure_type": failure_type,
"error_code": failure.error_code
})
if failure_type == "transient_error":
# Don't ask the LLM — just wait and retry
import asyncio
wait_seconds = 2 ** step.attempt_count # Exponential backoff
await asyncio.sleep(wait_seconds)
return True, "Retrying after transient error."
elif failure_type == "parameter_error":
# Give the model the exact error and schema
correction_prompt = f"""
The tool call for step '{step.step_id}' failed with a parameter error.
ERROR: {failure.error_message}
The tool's parameter schema requires:
{self._get_tool_schema(step.tool_hint)}
Rewrite the tool call with corrected parameters.
Only output the corrected JSON tool call, nothing else.
"""
return True, correction_prompt
elif failure_type == "semantic_error" and step.attempt_count < 2:
# Try rephrasing the query
rephrase_prompt = f"""
The search for step '{step.step_id}' returned no results.
Original query approach: {step.description}
Context of what we know so far: {context.completed_steps_summary[:500]}
Suggest an alternative search approach: broader terms, different time range,
or a different tool entirely. Output a revised plan for this step only.
"""
return True, rephrase_prompt
elif failure_type == "stuck" or step.attempt_count >= self.max_retries:
# Don't keep hammering — reassess the plan
return False, self._generate_plan_revision_prompt(step, context)
else:
return True, f"Previous attempt failed: {failure.error_message}. Try a different approach."
def _generate_plan_revision_prompt(
self, failed_step: PlanStep, context: AgentContextManager
) -> str:
return f"""
Step '{failed_step.step_id}' has failed {failed_step.attempt_count} times and cannot continue.
What we've accomplished so far:
{context.completed_steps_summary}
The stuck step was: {failed_step.description}
Given the current state of knowledge, revise the remaining plan:
1. Can we achieve the original goal without this step? If yes, describe how.
2. If not, what additional tool or information source would unblock us?
3. Is there a partial result we should return to the user with a clear statement of what's missing?
Output a revised plan as JSON matching the ExecutionPlan schema.
"""
The graduated nature of this is critical. Transient errors don't need LLM reasoning — they just need backoff. Parameter errors need schema-aware correction. Semantic dead-ends need strategy pivoting. Mixing these up wastes tokens and context budget.
A single agent has fundamental limitations. Its context window is finite. Its reasoning depth on any given sub-task is limited by how much of its attention budget that sub-task gets. And in practice, mixing many different kinds of expertise in a single agent prompt produces mediocre performance on all of them.
Multi-agent systems address this through specialization and parallelism, at the cost of coordination complexity.
The most reliable multi-agent architecture for production use is a clear hierarchy: one orchestrator agent that only does planning and delegation, and multiple specialized subagents that only do execution.
class OrchestratorAgent:
"""
High-level agent responsible for goal decomposition and delegation.
Has NO direct tool access — delegates all execution to subagents.
"""
def __init__(self, llm_client, subagents: dict[str, 'SubAgent']):
self.llm = llm_client
self.subagents = subagents # {"data_analyst": DataAnalystAgent, ...}
async def decompose_and_delegate(self, goal: str) -> list[dict]:
"""Break the goal into tasks and assign each to a subagent."""
available_agents = {
name: agent.capability_description
for name, agent in self.subagents.items()
}
decomposition_prompt = f"""
You are an orchestrator. Your job is to break down goals into tasks and assign
them to the right specialist agent. You do not execute tasks yourself.
GOAL: {goal}
AVAILABLE SPECIALIST AGENTS:
{json.dumps(available_agents, indent=2)}
Decompose this goal into a sequence of tasks. For each task, specify:
- task_id: unique identifier
- assigned_to: which agent handles this
- description: what the agent should do
- depends_on: list of task_ids that must complete first
- success_criteria: how to know this task succeeded
Output as JSON array.
"""
response = await self.llm.complete_async(messages=[
{"role": "system", "content": "You are a precise task decomposition specialist."},
{"role": "user", "content": decomposition_prompt}
])
return json.loads(extract_json(response.content))
async def run(self, goal: str) -> dict:
task_plan = await self.decompose_and_delegate(goal)
results = {}
for task in topological_sort(task_plan):
agent = self.subagents[task["assigned_to"]]
# Inject results from dependencies into this task's context
dependency_context = {
dep_id: results[dep_id]
for dep_id in task["depends_on"]
}
result = await agent.execute(
task=task["description"],
prior_results=dependency_context,
success_criteria=task["success_criteria"]
)
results[task["task_id"]] = result
# Orchestrator evaluates whether to continue or replan
if not result["success"]:
revised_plan = await self.handle_subagent_failure(task, result, task_plan)
# Insert revised tasks into remaining plan
task_plan = revised_plan
return results
For high-stakes decisions, one of the most powerful multi-agent patterns is having two agents independently solve the same problem, then a third agent reconcile the differences. This is especially effective for:
class VerificationPipeline:
async def run_with_verification(
self,
task: str,
executor_agent: 'SubAgent',
verifier_agent: 'SubAgent'
) -> dict:
# First pass: produce the result
initial_result = await executor_agent.execute(task=task, prior_results={})
# Second pass: independent verification
verification_task = f"""
Verify the following analysis result. Check for:
1. Logical consistency — does the conclusion follow from the data?
2. Completeness — were any important cases missed?
3. Factual accuracy — do cited numbers match the source data?
TASK THAT WAS PERFORMED: {task}
RESULT TO VERIFY: {json.dumps(initial_result['output'])}
Output: {{"verified": true/false, "issues": [...], "confidence": 0-1}}
"""
verification = await verifier_agent.execute(
task=verification_task,
prior_results={"original_task_data": initial_result.get("source_data", {})}
)
if not verification["output"]["verified"]:
# Revision round with specific feedback
revision_task = f"""
Your previous output had the following issues identified by a reviewer:
{json.dumps(verification["output"]["issues"])}
Original task: {task}
Your original output: {json.dumps(initial_result['output'])}
Produce a revised output that addresses all identified issues.
"""
return await executor_agent.execute(
task=revision_task,
prior_results={}
)
return initial_result
Warning: Multi-agent systems multiply your token costs and latency. A three-agent pipeline where each agent has a 32K context window and runs three steps each will easily consume 300K+ tokens per run. Profile before deploying, and make sure your business case justifies it.
Long-running agents need durable state. If your agent is doing a 45-minute analysis and the process crashes at step 28, you need to resume from step 28 — not restart from step 1.
import pickle
import hashlib
from pathlib import Path
class DurableAgentRunner:
def __init__(self, agent, checkpoint_dir: str = "/tmp/agent_checkpoints"):
self.agent = agent
self.checkpoint_dir = Path(checkpoint_dir)
self.checkpoint_dir.mkdir(exist_ok=True)
def _goal_hash(self, goal: str) -> str:
return hashlib.sha256(goal.encode()).hexdigest()[:12]
def _checkpoint_path(self, goal: str) -> Path:
return self.checkpoint_dir / f"checkpoint_{self._goal_hash(goal)}.pkl"
def save_checkpoint(self, goal: str, plan: ExecutionPlan, context: AgentContextManager):
path = self._checkpoint_path(goal)
with open(path, 'wb') as f:
pickle.dump({
"plan": plan,
"context_summary": context.completed_steps_summary,
"recent_steps": context.recent_raw_steps,
"timestamp": datetime.utcnow().isoformat()
}, f)
def load_checkpoint(self, goal: str) -> Optional[dict]:
path = self._checkpoint_path(goal)
if not path.exists():
return None
with open(path, 'rb') as f:
return pickle.load(f)
async def run_with_checkpointing(self, goal: str) -> dict:
checkpoint = self.load_checkpoint(goal)
if checkpoint:
print(f"Resuming from checkpoint: {checkpoint['timestamp']}")
plan = checkpoint["plan"]
context = AgentContextManager(self.agent.llm)
context.completed_steps_summary = checkpoint["context_summary"]
context.recent_raw_steps = checkpoint["recent_steps"]
else:
plan, context = await self.agent.initialize(goal)
while not plan.is_complete() and not plan.has_failed():
step = plan.next_executable_step()
result = await self.agent.execute_step(step, context)
# Save after every successful step
self.save_checkpoint(goal, plan, context)
# Clean up checkpoint on successful completion
if plan.is_complete():
self._checkpoint_path(goal).unlink(missing_ok=True)
return self.agent.compile_final_result(plan, context)
Security Note: Pickle files are not safe to load from untrusted sources — a malicious pickle file can execute arbitrary code on load. If your checkpoints will ever cross trust boundaries (e.g., stored in shared storage, passed between services), use JSON serialization with explicit deserialization logic instead.
This is the most underinvested area in most teams building agentic systems. "Did the agent produce an output?" is not evaluation. Here's a more rigorous framework.
1. Task Completion Rate What fraction of runs produce a complete, usable output? Measure this separately from "did the agent think it finished" — agents will confidently produce incomplete outputs. You need human or automated ground-truth comparison.
2. Step Efficiency How many tool calls and LLM calls did the agent use compared to a theoretical minimum? An agent that uses 40 tool calls to do what could be done in 12 is brittle, expensive, and will hit context limits in complex tasks.
3. Factual Faithfulness Does the final output accurately reflect the tool results? Generate test cases where you know the ground truth and measure how often the agent's conclusions match.
4. Recovery Rate When things go wrong (and you should deliberately inject failures in testing), what fraction of the time does the agent successfully recover vs. spiral into failure?
class AgentEvaluator:
def __init__(self, ground_truth_answers: dict):
self.ground_truth = ground_truth_answers
def evaluate_run(self, run_log: dict) -> dict:
metrics = {}
# Step efficiency
total_steps = len(run_log["steps"])
tool_calls = sum(1 for s in run_log["steps"] if s["type"] == "tool_call")
llm_calls = sum(1 for s in run_log["steps"] if s["type"] == "llm_call")
metrics["tool_calls"] = tool_calls
metrics["llm_calls"] = llm_calls
metrics["steps_per_tool_call"] = total_steps / max(tool_calls, 1)
# Recovery effectiveness
failures = [s for s in run_log["steps"] if s.get("failed")]
recoveries = [s for s in run_log["steps"] if s.get("recovered_from_failure")]
metrics["failure_count"] = len(failures)
metrics["recovery_rate"] = len(recoveries) / max(len(failures), 1)
# Context pressure (proxy for reliability risk)
peak_context_tokens = max(s.get("context_tokens", 0) for s in run_log["steps"])
metrics["peak_context_tokens"] = peak_context_tokens
metrics["context_budget_used"] = peak_context_tokens / 128000 # GPT-4 limit
# Faithfulness (requires ground truth)
task_id = run_log["task_id"]
if task_id in self.ground_truth:
metrics["faithfulness_score"] = self._score_faithfulness(
run_log["final_output"],
self.ground_truth[task_id]
)
return metrics
You're going to design and implement a small but realistic agentic pipeline. The scenario: an e-commerce company wants an agent that, given a product SKU, will research customer complaints about that product, identify the top 3 issue categories, check if those issues appear in the open bug tracker, and output a prioritized action memo.
Step 1: Define your tool library
Create Python stubs (no actual backend needed — return mock data) for exactly these tools:
get_reviews_by_sku(sku: str, sentiment: str, limit: int) — returns customer reviewscategorize_text_batch(texts: list[str], candidate_categories: list[str]) — LLM-powered batch classifiersearch_bug_tracker(query: str, status: str) — returns open/closed issuesget_bug_detail(bug_id: str) — returns full bug informationStep 2: Write the planning prompt
Write the system prompt that instructs an LLM to produce a JSON execution plan for the goal "Analyze complaints and bug coverage for SKU {sku} and produce a prioritized action memo." Include how the model should express dependencies between steps.
Step 3: Implement the agent loop
Using the patterns from this lesson, implement a basic PAOR loop that:
Step 4: Add a verification step
After the agent produces its action memo, add a verification pass that checks: do the priority rankings in the memo actually correspond to the issue frequency data from the reviews tool? If not, flag the discrepancy and ask the agent to revise.
Stretch goal: Add checkpointing so the pipeline can resume if interrupted after any tool call.
Symptom: The agent produces confident, fluent output that has nothing to do with your actual data.
Cause: You haven't made tool use mandatory. The model defaults to using its parametric knowledge.
Fix: Add explicit instructions: "You MUST use the available tools to answer this question. Do not use knowledge from your training data. If a tool is unavailable, say so explicitly rather than guessing." Also consider using your LLM provider's forced tool use / tool choice feature to require at least one tool call per step.
Symptom: The agent executes all steps even when step 3's result makes steps 7-9 obviously irrelevant.
Cause: Your reflection phase doesn't include a "should I revise the plan?" check.
Fix: After each step, include a brief evaluation: "Given this result, are any remaining plan steps now unnecessary, newly necessary, or should be reordered?" This adds tokens but dramatically improves plan quality on tasks where intermediate findings are informative.
Symptom: The agent hits a dead end and retries the exact same failing action 47 times until it exhausts your API budget.
Cause: No retry cap, or retry cap exists but the fallback path also loops.
Fix: Implement a hard cap (step.attempt_count >= max_attempts → mark as FAILED and route to plan revision, not more retries). Also log all attempts so you can detect when the same error code appears three times in a row.
Symptom: After a few steps, the agent starts making decisions that contradict earlier findings or ignores relevant context it already has.
Cause: Raw tool outputs are being appended to context without compression, and the relevant earlier findings are now "forgotten" in the long-tail attention.
Fix: Implement the rolling summary pattern. Treat context as a finite, managed resource, not an infinite transcript.
Symptom: The agent uses the wrong tool for a given step, or passes parameters in the wrong format.
Cause: One-line tool descriptions don't give the model enough signal about the tool's purpose, expected inputs, or relationship to other tools.
Fix: Write tool descriptions that include: what the tool does, when to use it (vs. its siblings), what parameters are required vs. optional, what the output format is, and any common errors to expect. Treat tool descriptions like API documentation for a junior developer.
Symptom: You built a 5-agent pipeline that takes 12 minutes and costs $3.50 per run when a carefully designed single-agent workflow would take 90 seconds and cost $0.30.
Cause: Multi-agent architectures feel sophisticated and are heavily promoted in framework demos, which are designed to showcase capability, not cost-efficiency.
Fix: Start with the simplest architecture that could possibly work. Add agents when you have a specific, measurable reason: context window exhaustion, specialization requirements, parallelism needs, or reliability requirements that justify the verification overhead.
You've covered a lot of ground. Let's consolidate the key architectural principles:
The PAOR loop is the foundation of reliable agentic behavior — but the Reflect phase is where most implementations cut corners, and it's where reliability actually lives. Invest in your reflection prompts.
Tool design is an interface design problem. Treat your tools like public APIs: explicit contracts, structured error responses, and descriptions that teach the agent when to use a tool, not just what it does.
Context is a managed resource, not a transcript. Long-running agents need active compression strategies or they'll degrade. The rolling summary pattern is your primary tool here.
Self-correction must be failure-type-aware. Generic "try again" retries work on 40% of failures. Diagnosing whether a failure is a parameter error, semantic dead-end, transient error, or plan invalidation lets you apply the right recovery strategy.
Multi-agent systems are powerful and expensive. The orchestrator-subagent pattern gives you clean separation between planning and execution. The debate/verification pattern gives you reliability for high-stakes outputs. Neither should be your default starting point.
Evaluation is non-negotiable for production. Task completion rate, step efficiency, factual faithfulness, and recovery rate are the four axes that actually tell you if your agent is working.
Learning Path: Intro to AI & Prompt Engineering