Agentic AI Workflows: Designing Multi-Step Autonomous Pipelines That Plan, Act, and Self-Correct

Introduction

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:

• How the Plan-Act-Observe-Reflect (PAOR) loop actually works under the hood and why each phase matters
• How to design tool interfaces and context management schemes that don't collapse under real-world complexity
• How to implement self-correction mechanisms that go beyond simple retries
• How to architect multi-agent systems with appropriate task decomposition and inter-agent communication
• How to evaluate, monitor, and harden agentic pipelines for production deployment, including failure modes you won't read about in the docs

---

Prerequisites

This is an expert-level lesson. You should be comfortable with:

• Python at an intermediate-to-advanced level (decorators, async, dataclasses)
• Core LLM concepts: tokens, context windows, temperature, function/tool calling
• At least one LLM API (OpenAI, Anthropic, or equivalent)
• Basic software architecture patterns (you should know what a message queue is, why idempotency matters)
• Familiarity with JSON schema and REST APIs

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.

---

What "Agentic" Actually Means (And What It Doesn't)

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:

1. Dynamic control flow: The path through the workflow is determined at runtime by the model's reasoning, not by a hardcoded DAG you wrote in advance.
2. Tool use: The agent can invoke external capabilities — APIs, code execution, database queries, file I/O — and incorporate results into subsequent reasoning.
3. State accumulation: Each step contributes to a growing context that informs future decisions.
4. Self-correction: The agent can detect when something went wrong and attempt a different approach.

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 Plan-Act-Observe-Reflect Loop

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.

Phase 1: Planning

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)

Phase 2: Acting

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.

Phase 3: Observing

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}"

Phase 4: Reflecting

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.

---

Designing Tool Interfaces That Don't Collapse

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.

The Tool Contract

Every tool should expose:

• A name that is unambiguous (not search, but search_internal_kb vs. search_web)
• A description that explains when to use this tool, not just what it does
• A parameter schema with types, constraints, and examples
• A structured return format the agent can parse reliably
• An error response that is itself parseable and diagnostic

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.

Tool Granularity and Composability

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.

---

Context Management: The Silent Killer of Long-Running Agents

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:

• The original goal (300 tokens)
• 12 tool call results (2000 tokens each = 24,000 tokens)
• 12 reflection summaries (500 tokens each = 6,000 tokens)
• The current plan state (1,500 tokens)
• System instructions (800 tokens)

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 Rolling Summary Pattern

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 Mechanisms That Actually Work

"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%.

Failure Taxonomy

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

Implementing Graduated Recovery

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.

---

Multi-Agent Architectures: When and How

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 Orchestrator-Subagent Pattern

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

The Debate/Verification Pattern

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:

• Code generation (one writes, one reviews)
• Data interpretation (one analyzes, one challenges assumptions)
• Report writing (one drafts, one fact-checks against source data)

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.

---

State Management and Persistence

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.

Checkpointing Pattern

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.

---

Evaluating Agentic Systems: Beyond "Did It Finish?"

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.

The Four-Axis Evaluation Model

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

---

Hands-On Exercise

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 reviews
• categorize_text_batch(texts: list[str], candidate_categories: list[str]) — LLM-powered batch classifier
• search_bug_tracker(query: str, status: str) — returns open/closed issues
• get_bug_detail(bug_id: str) — returns full bug information

Step 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:

• Takes the execution plan from Step 2
• Calls each tool in dependency order
• Maintains a rolling summary context
• Handles at least one type of tool failure with a meaningful recovery strategy

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.

---

Common Mistakes & Troubleshooting

Mistake 1: The agent "thinks" without looking anything up

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.

Mistake 2: The plan never changes even when it should

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.

Mistake 3: Infinite retry loops

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.

Mistake 4: Context poisoning

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.

Mistake 5: Tool descriptions that are too short

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.

Mistake 6: Treating multi-agent systems as a default solution

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.

---

Summary & Next Steps

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.

What to Study Next

• LangGraph and LlamaIndex Workflows: Both provide production-grade implementations of the patterns in this lesson. Now that you understand the underlying design, reading their source code will be much more illuminating than their documentation.
• Human-in-the-loop patterns: Production agentic systems almost always need interruption and approval mechanisms for high-stakes actions. Study how to design clean handoff points.
• Structured outputs and grammar-constrained decoding: Reliable JSON output from LLMs — critical for agent reliability — requires more than just asking nicely.
• Agent observability: Tools like LangSmith, Langfuse, and Weights & Biases' LLM tracking let you trace agent execution in production. Set this up before you deploy.
• Security hardening: Prompt injection attacks — where malicious content in tool outputs hijacks the agent — are a real production concern. Research output sanitization and sandboxed execution environments.