The End-to-End Agentic AI Pipeline

By Shittu Olumide on July 31, 2026 in Artificial Intelligence 0

In this article, you will learn the seven architectural components that separate a production-grade agentic AI system from a demo script, and how each one fits into the agent’s core feedback loop.

Topics we will cover include:

What each of the seven components — perception, memory, reasoning and planning, tool execution, orchestration, guardrails, and observability — is specifically responsible for.
Where each component tends to break in real systems, and why that component must be kept separate from the others.
Focused, runnable Python code illustrating the responsibility of each component in isolation.

Introduction

Most “build an AI agent” tutorials show a 40-line script that calls an LLM in a loop and calls it done. That script works fine for a demo. It does not survive a second concurrent user, a flaky third-party API, or a task that turns out to need twelve steps instead of two.

The gap between the demo and the production system isn’t clever prompting. It’s architecture. Production agentic systems are built from a consistent set of interconnected components: perception, reasoning, planning, memory, tool execution, orchestration, and guardrails. That same component breakdown shows up across nearly every serious architecture writeup, survey paper, and production postmortem published in the last year, regardless of which framework or vendor is doing the writing.

The loop underneath all of it is consistent: Goal → Perception → Reasoning → Planning → Action → Observation → Memory Update → back to Reasoning, repeating until the goal is met, a stop condition fires, or the agent decides it needs a human. This article walks through each piece of that loop as its own component — what it’s responsible for, where it tends to break, and a focused code excerpt that makes the responsibility concrete. Nothing here is wired into one running pipeline. Each piece is shown in isolation, which is also how you should reason about your own system when deciding what it needs.

The Seven Components, at a Glance

Architectural surveys converge on the same core set: Perception, Memory, Reasoning/Planning, Tool Execution, and Orchestration form a closed feedback loop — the cycle that actually runs, step after step. Guardrails and Observability wrap around that entire loop as cross-cutting concerns rather than steps inside the sequence. You don’t “do” guardrails at step 4; guardrails sit between every proposed action and the world, watching every step.

That distinction shapes the rest of this article. The first five sections walk through the loop in the order data actually flows through it. The last two sections cover the wrapper layers that make the loop survivable once real money, real customers, and real side effects are involved.

Turning Raw Input Into Something the Agent Can Reason About

Perception’s job is to transform raw inputs — text, voice, API payloads, sensor data, and file uploads — into a structured representation that the reasoning engine can actually work with. This is the component most tutorials skip entirely, because in a demo, “the user just types text” and there’s nothing to normalize. Real systems take input from webhooks, structured API calls, file uploads, and multiple channels simultaneously, and every one of those needs to land in the same shape before anything downstream can trust it.

# perception.py
# Prerequisites: none beyond Python's standard library
# Run: python perception.py

from dataclasses import dataclass, field
from typing import Any
from enum import Enum
import json
from datetime import datetime, timezone

class InputSource(Enum):
    USER_TEXT = "user_text"
    WEBHOOK = "webhook"
    FILE_UPLOAD = "file_upload"

@dataclass
class AgentInput:
    """
    The normalized internal shape every downstream component consumes,
    regardless of where the raw input actually came from. This is the
    entire point of a perception layer: everything past this point
    only ever sees this one structure.
    """
    source: InputSource
    content: str
    metadata: dict[str, Any] = field(default_factory=dict)
    received_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

def perceive_user_text(raw_text: str) -> AgentInput:
    """Raw chat input -- the simplest case, but it still needs normalization."""
    return AgentInput(
        source=InputSource.USER_TEXT,
        content=raw_text.strip(),
        metadata={"channel": "chat"},
    )

def perceive_webhook(raw_payload: str) -> AgentInput:
    """
    A webhook delivers structured JSON, not plain text. Perception extracts
    the part the agent should reason about and discards transport-level
    noise like headers and signatures.
    """
    payload = json.loads(raw_payload)
    event_type = payload.get("event_type", "unknown")
    description = payload.get("description", "")
    return AgentInput(
        source=InputSource.WEBHOOK,
        content=f"Event '{event_type}' received: {description}",
        metadata={"event_type": event_type, "raw_payload": payload},
    )

def perceive_file_upload(filename: str, file_size_bytes: int, mime_type: str) -> AgentInput:
    """
    A file upload event has no natural-language content at all -- perception
    has to construct something the reasoning engine can actually use.
    """
    return AgentInput(
        source=InputSource.FILE_UPLOAD,
        content=f"User uploaded file '{filename}' ({mime_type}, {file_size_bytes} bytes)",
        metadata={"filename": filename, "mime_type": mime_type, "size_bytes": file_size_bytes},
    )


if __name__ == "__main__":
    text_input = perceive_user_text("  What's the status of my refund?  ")
    webhook_input = perceive_webhook(json.dumps({
        "event_type": "payment_failed",
        "description": "Card declined for order #4821",
    }))
    file_input = perceive_file_upload("invoice_q3.pdf", 184320, "application/pdf")

    for inp in [text_input, webhook_input, file_input]:
        print(f"[{inp.source.value}] content='{inp.content}'")
        print(f"  metadata keys: {list(inp.metadata.keys())}\n")

# perception.py

# Prerequisites: none beyond Python's standard library

# Run: python perception.py

from dataclasses import dataclass, field

from typing import Any

from enum import Enum

import json

from datetime import datetime, timezone

class InputSource(Enum):

USER_TEXT = "user_text"

WEBHOOK = "webhook"

FILE_UPLOAD = "file_upload"

@dataclass

class AgentInput:

"""

The normalized internal shape every downstream component consumes,

regardless of where the raw input actually came from. This is the

entire point of a perception layer: everything past this point

only ever sees this one structure.

"""

source: InputSource

content: str

metadata: dict[str, Any] = field(default_factory=dict)

received_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

def perceive_user_text(raw_text: str) -> AgentInput:

"""Raw chat input -- the simplest case, but it still needs normalization."""

return AgentInput(

source=InputSource.USER_TEXT,

content=raw_text.strip(),

metadata={"channel": "chat"},

)

def perceive_webhook(raw_payload: str) -> AgentInput:

"""

A webhook delivers structured JSON, not plain text. Perception extracts

the part the agent should reason about and discards transport-level

noise like headers and signatures.

"""

payload = json.loads(raw_payload)

event_type = payload.get("event_type", "unknown")

description = payload.get("description", "")

return AgentInput(

source=InputSource.WEBHOOK,

content=f"Event '{event_type}' received: {description}",

metadata={"event_type": event_type, "raw_payload": payload},

)

def perceive_file_upload(filename: str, file_size_bytes: int, mime_type: str) -> AgentInput:

"""

A file upload event has no natural-language content at all -- perception

has to construct something the reasoning engine can actually use.

"""

return AgentInput(

source=InputSource.FILE_UPLOAD,

content=f"User uploaded file '{filename}' ({mime_type}, {file_size_bytes} bytes)",

metadata={"filename": filename, "mime_type": mime_type, "size_bytes": file_size_bytes},

)

if __name__ == "__main__":

text_input = perceive_user_text(" What's the status of my refund? ")

webhook_input = perceive_webhook(json.dumps({

"event_type": "payment_failed",

"description": "Card declined for order #4821",

}))

file_input = perceive_file_upload("invoice_q3.pdf", 184320, "application/pdf")

for inp in [text_input, webhook_input, file_input]:

print(f"[{inp.source.value}] content='{inp.content}'")

print(f" metadata keys: {list(inp.metadata.keys())}\n")

How to run: python perception.py, no dependencies required.

Three completely different raw shapes — plain text, a webhook JSON payload, and a file-upload event — all collapse into the same AgentInput structure. The reasoning component downstream never needs to know or care which channel something arrived through. That’s the entire value of treating perception as its own component rather than inlining ad hoc parsing wherever input happens to enter the system.

Working Context vs. What Actually Persists

This is the component with the most nuance, and the one demo code gets wrong most often by treating “memory” as just “the conversation so far.” Production memory architecture separates working memory — the immediate context window for the current task — from long-term memory, which itself splits into episodic memory (what happened), semantic memory (facts learned), and procedural memory (skills and how-to knowledge). Short-term memory lives in-context and is essentially free; long-term memory typically lives in a vector store, indexed for semantic retrieval rather than exact match.

The operational distinction matters: working memory is fast and disposable — it evaporates the moment the session ends. Episodic memory gives the agent something working memory structurally cannot provide: hindsight across sessions, the ability to recall “we handled something like this before, and here’s what happened.”

# memory.py
# Prerequisites: none beyond Python's standard library
# Run: python memory.py

from dataclasses import dataclass, field
from datetime import datetime, timezone

@dataclass
class WorkingMemory:
    """
    Working memory: the immediate context for the CURRENT task only.
    Lives in-process, bounded by a turn limit, and is gone the moment
    the session ends. This is what most demo code calls "memory" --
    but it's only one piece of the real picture.
    """
    max_turns: int = 10
    turns: list[dict] = field(default_factory=list)

    def add_turn(self, role: str, content: str) -> None:
        self.turns.append({"role": role, "content": content})
        if len(self.turns) > self.max_turns:
            self.turns.pop(0)   # Oldest turn drops off once the limit is hit

    def as_context(self) -> str:
        return "\n".join(f"{t['role']}: {t['content']}" for t in self.turns)


@dataclass
class EpisodicMemoryEntry:
    """A single stored episode -- what happened, when, and its embedding for later recall."""
    timestamp: str
    summary: str
    embedding: list[float]   # In production this comes from a real embedding model

class EpisodicMemory:
    """
    Episodic memory: persists ACROSS sessions, stored externally
    (a vector store in production), and retrieved by semantic similarity
    rather than recency. This is what gives an agent "hindsight" -- a
    capability working memory structurally cannot have, since it's gone
    the instant the session ends.
    """
    def __init__(self):
        self._store: list[EpisodicMemoryEntry] = []

    def record_episode(self, summary: str, embedding: list[float]) -> None:
        self._store.append(EpisodicMemoryEntry(
            timestamp=datetime.now(timezone.utc).isoformat(),
            summary=summary,
            embedding=embedding,
        ))

    def retrieve_similar(self, query_embedding: list[float], top_k: int = 2) -> list[EpisodicMemoryEntry]:
        """Real implementations do cosine similarity against a vector index."""
        def dot(a, b): return sum(x * y for x, y in zip(a, b))
        ranked = sorted(self._store, key=lambda e: dot(e.embedding, query_embedding), reverse=True)
        return ranked[:top_k]


if __name__ == "__main__":
    wm = WorkingMemory(max_turns=3)
    wm.add_turn("user", "What's my refund status?")
    wm.add_turn("agent", "Let me check that for you.")
    wm.add_turn("user", "It's order 4821")
    wm.add_turn("agent", "Found it -- refund is processing")  # pushes the first turn out

    print("Working memory (bounded to last 3 turns):")
    print(wm.as_context())

    em = EpisodicMemory()
    em.record_episode("User asked about refund for order 4821, resolved successfully", [0.9, 0.1, 0.0])
    em.record_episode("User asked about shipping delay for order 1190", [0.1, 0.9, 0.0])
    em.record_episode("User asked about refund eligibility for order 7734", [0.85, 0.15, 0.0])

    similar = em.retrieve_similar([0.88, 0.12, 0.0], top_k=2)
    print("\nEpisodic memory -- retrieved by similarity to a new refund query:")
    for entry in similar:
        print(f"  {entry.summary}")

# memory.py

# Prerequisites: none beyond Python's standard library

# Run: python memory.py

from dataclasses import dataclass, field

from datetime import datetime, timezone

@dataclass

class WorkingMemory:

"""

Working memory: the immediate context for the CURRENT task only.

Lives in-process, bounded by a turn limit, and is gone the moment

the session ends. This is what most demo code calls "memory" --

but it's only one piece of the real picture.

"""

max_turns: int = 10

turns: list[dict] = field(default_factory=list)

def add_turn(self, role: str, content: str) -> None:

self.turns.append({"role": role, "content": content})

if len(self.turns) > self.max_turns:

self.turns.pop(0) # Oldest turn drops off once the limit is hit

def as_context(self) -> str:

return "\n".join(f"{t['role']}: {t['content']}" for t in self.turns)

@dataclass

class EpisodicMemoryEntry:

"""A single stored episode -- what happened, when, and its embedding for later recall."""

timestamp: str

summary: str

embedding: list[float] # In production this comes from a real embedding model

class EpisodicMemory:

"""

Episodic memory: persists ACROSS sessions, stored externally

(a vector store in production), and retrieved by semantic similarity

rather than recency. This is what gives an agent "hindsight" -- a

capability working memory structurally cannot have, since it's gone

the instant the session ends.

"""

def __init__(self):

self._store: list[EpisodicMemoryEntry] = []

def record_episode(self, summary: str, embedding: list[float]) -> None:

self._store.append(EpisodicMemoryEntry(

timestamp=datetime.now(timezone.utc).isoformat(),

summary=summary,

embedding=embedding,

))

def retrieve_similar(self, query_embedding: list[float], top_k: int = 2) -> list[EpisodicMemoryEntry]:

"""Real implementations do cosine similarity against a vector index."""

def dot(a, b): return sum(x * y for x, y in zip(a, b))

ranked = sorted(self._store, key=lambda e: dot(e.embedding, query_embedding), reverse=True)

return ranked[:top_k]

if __name__ == "__main__":

wm = WorkingMemory(max_turns=3)

wm.add_turn("user", "What's my refund status?")

wm.add_turn("agent", "Let me check that for you.")

wm.add_turn("user", "It's order 4821")

wm.add_turn("agent", "Found it -- refund is processing") # pushes the first turn out

print("Working memory (bounded to last 3 turns):")

print(wm.as_context())

em = EpisodicMemory()

em.record_episode("User asked about refund for order 4821, resolved successfully", [0.9, 0.1, 0.0])

em.record_episode("User asked about shipping delay for order 1190", [0.1, 0.9, 0.0])

em.record_episode("User asked about refund eligibility for order 7734", [0.85, 0.15, 0.0])

similar = em.retrieve_similar([0.88, 0.12, 0.0], top_k=2)

print("\nEpisodic memory -- retrieved by similarity to a new refund query:")

for entry in similar:

print(f" {entry.summary}")

How to run: python memory.py, no dependencies required.

Working memory drops its oldest turn once the limit is hit, and the first exchange about checking the status is gone by the end of the session. Episodic memory does the opposite: it surfaces the two refund-related episodes out of three stored entries, ranked by meaning, not by when they happened. That’s the structural line between the two — one is a sliding window, the other is a searchable archive.

Reasoning and Planning (Deciding What to Do Next)

Reasoning and planning take the current goal, the perceived input, and whatever memory was retrieved, and produce a plan — sometimes a single next action, sometimes a multi-step decomposition. This is the agent’s cognitive core, consulting memory and knowledge resources to synthesize action plans that get handed off to the execution module.

The critical design point, easy to miss: planning’s responsibility ends at producing the plan. It does not call a tool, touch an API, or have any side effects. That separation is deliberate, and it’s what makes the next component — tool execution — independently testable and independently guardable.

# planning.py
# Prerequisites: none beyond Python's standard library
# Run: python planning.py

from dataclasses import dataclass, field
import json

@dataclass
class PlanStep:
    step_number: int
    description: str
    tool_tag: str   # which category of tool this step will need -- execution decides HOW

@dataclass
class Plan:
    goal: str
    steps: list[PlanStep] = field(default_factory=list)

def mock_llm_plan(goal: str) -> str:
    """
    Mock LLM call standing in for a real planning call. The point being
    demonstrated is structural: planning PRODUCES a plan object -- it does
    not execute anything. Execution is a separate component (next section)
    that the plan gets handed off to.
    """
    if "refund" in goal.lower():
        return json.dumps({
            "steps": [
                {"step_number": 1, "description": "Look up the order by ID", "tool_tag": "database_lookup"},
                {"step_number": 2, "description": "Check refund eligibility against policy", "tool_tag": "policy_check"},
                {"step_number": 3, "description": "Issue the refund if eligible", "tool_tag": "payment_api"},
                {"step_number": 4, "description": "Notify the customer of the outcome", "tool_tag": "email"},
            ]
        })
    return json.dumps({"steps": [
        {"step_number": 1, "description": "Search knowledge base for the answer", "tool_tag": "search"},
    ]})

def create_plan(goal: str) -> Plan:
    """Take a goal, produce a structured plan. Nothing here has a side effect."""
    parsed = json.loads(mock_llm_plan(goal))
    return Plan(goal=goal, steps=[PlanStep(**s) for s in parsed["steps"]])


if __name__ == "__main__":
    for goal in ["Process a refund for order 4821", "What are your business hours?"]:
        plan = create_plan(goal)
        print(f"Goal: {plan.goal}")
        for step in plan.steps:
            print(f"  Step {step.step_number}: {step.description} [tool_tag={step.tool_tag}]")
        print()

# planning.py

# Prerequisites: none beyond Python's standard library

# Run: python planning.py

from dataclasses import dataclass, field

import json

@dataclass

class PlanStep:

step_number: int

description: str

tool_tag: str # which category of tool this step will need -- execution decides HOW

@dataclass

class Plan:

goal: str

steps: list[PlanStep] = field(default_factory=list)

def mock_llm_plan(goal: str) -> str:

"""

Mock LLM call standing in for a real planning call. The point being

demonstrated is structural: planning PRODUCES a plan object -- it does

not execute anything. Execution is a separate component (next section)

that the plan gets handed off to.

"""

if "refund" in goal.lower():

return json.dumps({

"steps": [

{"step_number": 1, "description": "Look up the order by ID", "tool_tag": "database_lookup"},

{"step_number": 2, "description": "Check refund eligibility against policy", "tool_tag": "policy_check"},

{"step_number": 3, "description": "Issue the refund if eligible", "tool_tag": "payment_api"},

{"step_number": 4, "description": "Notify the customer of the outcome", "tool_tag": "email"},

]

})

return json.dumps({"steps": [

{"step_number": 1, "description": "Search knowledge base for the answer", "tool_tag": "search"},

]})

def create_plan(goal: str) -> Plan:

"""Take a goal, produce a structured plan. Nothing here has a side effect."""

parsed = json.loads(mock_llm_plan(goal))

return Plan(goal=goal, steps=[PlanStep(**s) for s in parsed["steps"]])

if __name__ == "__main__":

for goal in ["Process a refund for order 4821", "What are your business hours?"]:

plan = create_plan(goal)

print(f"Goal: {plan.goal}")

for step in plan.steps:

print(f" Step {step.step_number}: {step.description} [tool_tag={step.tool_tag}]")

print()

How to run: python planning.py, no dependencies required.

The refund goal produces a four-step plan; the business-hours question produces one. Neither call executed a single tool — both just returned a Plan object describing what should happen next. That object is the handoff artifact between reasoning and the rest of the pipeline, which is exactly why orchestration (covered later) can choose to pause, modify, or reject a plan before anything in it actually runs.

Tool Execution

Tool execution connects agents to external systems — APIs, databases, and services — handling the mechanics of invoking a capability and feeding the result back into the reasoning process. It’s also where most production incidents actually originate, because it’s the only component in the loop with real, external side effects.

The constraint is worth stating in plain numbers: at a 5% per-action failure rate, an agent taking 20 actions in a run will fail often enough to be unusable without guardrails. That single statistic is why tool execution can’t just be “call the API and hope” — it needs validation, a timeout, and idempotency as baseline requirements, not nice-to-haves.

# tool_execution.py
# Prerequisites: none beyond Python's standard library
# Run: python tool_execution.py

import time
import hashlib
from dataclasses import dataclass
from typing import Callable, Any, Optional

@dataclass
class ToolResult:
    success: bool
    output: Any = None
    error: Optional[str] = None
    idempotency_key: Optional[str] = None
    from_cache: bool = False

class ToolExecutor:
    """
    The non-negotiable basics for tool execution: validate inputs before
    calling anything, enforce a timeout so a slow call can't hang the
    whole run, and make side-effecting calls idempotent so a retry
    doesn't double-charge a customer or double-send an email.
    """
    def __init__(self, timeout_seconds: float = 5.0):
        self.timeout_seconds = timeout_seconds
        self._idempotency_cache: dict[str, ToolResult] = {}

    def _make_idempotency_key(self, tool_name: str, args: dict) -> str:
        raw = f"{tool_name}:{sorted(args.items())}"
        return hashlib.sha256(raw.encode()).hexdigest()[:16]

    def execute(self, tool_name: str, tool_fn: Callable, args: dict,
                required_args: list[str], idempotent: bool = False) -> ToolResult:
        # 1. Validate before touching anything external
        missing = [a for a in required_args if a not in args]
        if missing:
            return ToolResult(success=False, error=f"Missing required args: {missing}")

        # 2. Idempotency -- a retried call with identical args returns the cached result
        idem_key = self._make_idempotency_key(tool_name, args) if idempotent else None
        if idem_key and idem_key in self._idempotency_cache:
            cached = self._idempotency_cache[idem_key]
            return ToolResult(success=cached.success, output=cached.output,
                              idempotency_key=idem_key, from_cache=True)

        # 3. Execute against a timeout budget
        start = time.monotonic()
        try:
            output = tool_fn(**args)
        except Exception as e:
            return ToolResult(success=False, error=str(e), idempotency_key=idem_key)

        if (time.monotonic() - start) > self.timeout_seconds:
            return ToolResult(success=False, error=f"Tool exceeded {self.timeout_seconds}s timeout",
                              idempotency_key=idem_key)

        result = ToolResult(success=True, output=output, idempotency_key=idem_key)
        if idem_key:
            self._idempotency_cache[idem_key] = result
        return result


def issue_refund(order_id: str, amount: float) -> str:
    return f"Refunded ${amount} for order {order_id}"


if __name__ == "__main__":
    executor = ToolExecutor(timeout_seconds=1.0)

    # Missing arg -- caught before execution
    r1 = executor.execute("issue_refund", issue_refund, {"order_id": "4821"}, ["order_id", "amount"])
    print(f"Missing arg test:  success={r1.success}, error='{r1.error}'")

    # First call executes, retry with identical args hits the idempotency cache
    args = {"order_id": "4821", "amount": 49.99}
    r2a = executor.execute("issue_refund", issue_refund, args, ["order_id", "amount"], idempotent=True)
    r2b = executor.execute("issue_refund", issue_refund, args, ["order_id", "amount"], idempotent=True)
    print(f"\nFirst call:        from_cache={r2a.from_cache}, output='{r2a.output}'")
    print(f"Retry, same args:  from_cache={r2b.from_cache}, output='{r2b.output}'")

# tool_execution.py

# Prerequisites: none beyond Python's standard library

# Run: python tool_execution.py

import time

import hashlib

from dataclasses import dataclass

from typing import Callable, Any, Optional

@dataclass

class ToolResult:

success: bool

output: Any = None

error: Optional[str] = None

idempotency_key: Optional[str] = None

from_cache: bool = False

class ToolExecutor:

"""

The non-negotiable basics for tool execution: validate inputs before

calling anything, enforce a timeout so a slow call can't hang the

whole run, and make side-effecting calls idempotent so a retry

doesn't double-charge a customer or double-send an email.

"""

def __init__(self, timeout_seconds: float = 5.0):

self.timeout_seconds = timeout_seconds

self._idempotency_cache: dict[str, ToolResult] = {}

def _make_idempotency_key(self, tool_name: str, args: dict) -> str:

raw = f"{tool_name}:{sorted(args.items())}"

return hashlib.sha256(raw.encode()).hexdigest()[:16]

def execute(self, tool_name: str, tool_fn: Callable, args: dict,

required_args: list[str], idempotent: bool = False) -> ToolResult:

# 1. Validate before touching anything external

missing = [a for a in required_args if a not in args]

if missing:

return ToolResult(success=False, error=f"Missing required args: {missing}")

# 2. Idempotency -- a retried call with identical args returns the cached result

idem_key = self._make_idempotency_key(tool_name, args) if idempotent else None

if idem_key and idem_key in self._idempotency_cache:

cached = self._idempotency_cache[idem_key]

return ToolResult(success=cached.success, output=cached.output,

idempotency_key=idem_key, from_cache=True)

# 3. Execute against a timeout budget

start = time.monotonic()

try:

output = tool_fn(**args)

except Exception as e:

return ToolResult(success=False, error=str(e), idempotency_key=idem_key)

if (time.monotonic() - start) > self.timeout_seconds:

return ToolResult(success=False, error=f"Tool exceeded {self.timeout_seconds}s timeout",

idempotency_key=idem_key)

result = ToolResult(success=True, output=output, idempotency_key=idem_key)

if idem_key:

self._idempotency_cache[idem_key] = result

return result

def issue_refund(order_id: str, amount: float) -> str:

return f"Refunded ${amount} for order {order_id}"

if __name__ == "__main__":

executor = ToolExecutor(timeout_seconds=1.0)

# Missing arg -- caught before execution

r1 = executor.execute("issue_refund", issue_refund, {"order_id": "4821"}, ["order_id", "amount"])

print(f"Missing arg test: success={r1.success}, error='{r1.error}'")

# First call executes, retry with identical args hits the idempotency cache

args = {"order_id": "4821", "amount": 49.99}

r2a = executor.execute("issue_refund", issue_refund, args, ["order_id", "amount"], idempotent=True)

r2b = executor.execute("issue_refund", issue_refund, args, ["order_id", "amount"], idempotent=True)

print(f"\nFirst call: from_cache={r2a.from_cache}, output='{r2a.output}'")

print(f"Retry, same args: from_cache={r2b.from_cache}, output='{r2b.output}'")

How to run: python tool_execution.py, no dependencies required.

The retry with identical arguments returns the cached result instead of calling issue_refund a second time. The customer gets refunded once, not twice, even if the orchestrator above it retries the step after a transient network blip. That’s the entire purpose of building idempotency into the execution layer rather than hoping the orchestrator never retries.

Orchestration

Orchestration holds the loop together across multiple steps and, in multi-agent systems, across multiple agents — deciding when to continue, when a step’s outcome should change the path, and when the run is actually finished. This is the layer that has matured fastest recently, with LangGraph, CrewAI, and AutoGen now handling production-grade coordination rather than every team hand-rolling their own loop from scratch.

# orchestrator.py
# Prerequisites: none beyond Python's standard library
# Run: python orchestrator.py

from dataclasses import dataclass, field

@dataclass
class PlanStep:
    step_number: int
    description: str
    tool_tag: str

@dataclass
class Plan:
    goal: str
    steps: list[PlanStep] = field(default_factory=list)

@dataclass
class StepOutcome:
    step_number: int
    success: bool
    output: str

# Stand-ins for the planning and tool-execution components above --
# orchestration's job is to CALL these in sequence, not do their work itself.
def mock_create_plan(goal: str) -> Plan:
    return Plan(goal=goal, steps=[
        PlanStep(1, "Look up order", "database_lookup"),
        PlanStep(2, "Check eligibility", "policy_check"),
        PlanStep(3, "Issue refund", "payment_api"),
    ])

def mock_execute_tool(step: PlanStep) -> StepOutcome:
    if step.tool_tag == "policy_check":  # simulate a failure to test stop logic
        return StepOutcome(step.step_number, success=False, output="Policy check failed: order too old")
    return StepOutcome(step.step_number, success=True, output=f"Completed: {step.description}")

class Orchestrator:
    """
    Holds the sequence together across multiple steps, decides whether to
    continue or stop, and caps the run with an explicit step limit. It does
    not plan and it does not execute tools -- it calls the components that do.
    """
    def __init__(self, max_steps: int = 10):
        self.max_steps = max_steps

    def run(self, goal: str) -> list[StepOutcome]:
        plan = mock_create_plan(goal)
        outcomes: list[StepOutcome] = []

        for step in plan.steps:
            if len(outcomes) >= self.max_steps:
                print(f"  Step limit ({self.max_steps}) reached -- stopping.")
                break

            print(f"  Executing step {step.step_number}: {step.description}")
            outcome = mock_execute_tool(step)
            outcomes.append(outcome)

            if not outcome.success:
                print(f"  Step {step.step_number} failed: {outcome.output}")
                print(f"  Stopping run -- a failed step blocks the rest of this plan.")
                break

        return outcomes


if __name__ == "__main__":
    outcomes = Orchestrator(max_steps=10).run("Process a refund for order 4821")
    print(f"\nTotal steps attempted: {len(outcomes)}")
    print(f"Final outcome: success={outcomes[-1].success}, output='{outcomes[-1].output}'")

# orchestrator.py

# Prerequisites: none beyond Python's standard library

# Run: python orchestrator.py

from dataclasses import dataclass, field

@dataclass

class PlanStep:

step_number: int

description: str

tool_tag: str

@dataclass

class Plan:

goal: str

steps: list[PlanStep] = field(default_factory=list)

@dataclass

class StepOutcome:

step_number: int

success: bool

output: str

# Stand-ins for the planning and tool-execution components above --

# orchestration's job is to CALL these in sequence, not do their work itself.

def mock_create_plan(goal: str) -> Plan:

return Plan(goal=goal, steps=[

PlanStep(1, "Look up order", "database_lookup"),

PlanStep(2, "Check eligibility", "policy_check"),

PlanStep(3, "Issue refund", "payment_api"),

])

def mock_execute_tool(step: PlanStep) -> StepOutcome:

if step.tool_tag == "policy_check": # simulate a failure to test stop logic

return StepOutcome(step.step_number, success=False, output="Policy check failed: order too old")

return StepOutcome(step.step_number, success=True, output=f"Completed: {step.description}")

class Orchestrator:

"""

Holds the sequence together across multiple steps, decides whether to

continue or stop, and caps the run with an explicit step limit. It does

not plan and it does not execute tools -- it calls the components that do.

"""

def __init__(self, max_steps: int = 10):

self.max_steps = max_steps

def run(self, goal: str) -> list[StepOutcome]:

plan = mock_create_plan(goal)

outcomes: list[StepOutcome] = []

for step in plan.steps:

if len(outcomes) >= self.max_steps:

print(f" Step limit ({self.max_steps}) reached -- stopping.")

break

print(f" Executing step {step.step_number}: {step.description}")

outcome = mock_execute_tool(step)

outcomes.append(outcome)

if not outcome.success:

print(f" Step {step.step_number} failed: {outcome.output}")

print(f" Stopping run -- a failed step blocks the rest of this plan.")

break

return outcomes

if __name__ == "__main__":

outcomes = Orchestrator(max_steps=10).run("Process a refund for order 4821")

print(f"\nTotal steps attempted: {len(outcomes)}")

print(f"Final outcome: success={outcomes[-1].success}, output='{outcomes[-1].output}'")

How to run: python orchestrator.py, no dependencies required.

Output:

Executing step 1: Look up order
Executing step 2: Check eligibility
Step 2 failed: Policy check failed: order too old
Stopping run -- a failed step blocks the rest of this plan

Total steps attempted: 2
Final outcome: success=False, output='Policy check failed: order too old'

Executing step 1: Look up order

Executing step 2: Check eligibility

Step 2 failed: Policy check failed: order too old

Stopping run -- a failed step blocks the rest of this plan

Total steps attempted: 2

Final outcome: success=False, output='Policy check failed: order too old'

Step 3 — the actual refund — never ran. That’s not an accident of the mock; it’s the orchestrator doing its specific job. The planner produced a three-step plan with no knowledge of whether step 2 would succeed. The tool executor ran step 2 and reported failure. Deciding to stop there, rather than blindly continuing to issue a refund on an order that just failed eligibility, belongs to neither of those components — it belongs to orchestration.

Guardrails

Guardrails enforce the rules of the road: allow/deny lists for tools and domains, privacy and data-residency controls, cost ceilings, rate limits, and escalation paths for risky or irreversible actions. This isn’t a feature bolted on after launch; it’s the difference between an agent that’s impressive in a demo and one that’s safe to point at real customer accounts and real payment systems.

Current production guidance converges on the same core pattern: policy-as-code, mandatory approval gates for irreversible actions, and defenses against prompt injection where untrusted retrieved content could be mistaken for an instruction.

# guardrails.py
# Prerequisites: none beyond Python's standard library
# Run: python guardrails.py

from dataclasses import dataclass
from enum import Enum

class GuardrailVerdict(Enum):
    ALLOW = "allow"
    DENY = "deny"
    REQUIRE_APPROVAL = "require_approval"

@dataclass
class ProposedAction:
    tool_name: str
    args: dict
    estimated_cost: float
    irreversible: bool

@dataclass
class GuardrailResult:
    verdict: GuardrailVerdict
    reason: str

class GuardrailEngine:
    """
    Sits between planning's proposed action and the tool executor.
    Enforces an allow-list of permitted tools, a cost ceiling per action,
    and mandatory human approval for anything irreversible -- regardless
    of how confident the planner was that it was the right move.
    """
    def __init__(self, allowed_tools: set[str], cost_ceiling: float):
        self.allowed_tools = allowed_tools
        self.cost_ceiling = cost_ceiling

    def check(self, action: ProposedAction) -> GuardrailResult:
        if action.tool_name not in self.allowed_tools:
            return GuardrailResult(GuardrailVerdict.DENY,
                                   f"Tool '{action.tool_name}' is not on the allow-list")
        if action.estimated_cost > self.cost_ceiling:
            return GuardrailResult(GuardrailVerdict.DENY,
                                   f"Estimated cost ${action.estimated_cost} exceeds ceiling ${self.cost_ceiling}")
        if action.irreversible:
            return GuardrailResult(GuardrailVerdict.REQUIRE_APPROVAL,
                                   "Action is irreversible -- requires human approval before execution")
        return GuardrailResult(GuardrailVerdict.ALLOW, "Passed all guardrail checks")


if __name__ == "__main__":
    guardrails = GuardrailEngine(
        allowed_tools={"database_lookup", "policy_check", "payment_api", "email"},
        cost_ceiling=100.0,
    )

    test_actions = [
        ProposedAction("database_lookup", {"order_id": "4821"}, 0.0, False),
        ProposedAction("delete_customer_account", {"id": "u_991"}, 0.0, True),
        ProposedAction("payment_api", {"order_id": "4821", "amount": 5000.0}, 5000.0, True),
        ProposedAction("payment_api", {"order_id": "4821", "amount": 49.99}, 49.99, True),
    ]

    for action in test_actions:
        result = guardrails.check(action)
        print(f"{action.tool_name} -> {result.verdict.value}: {result.reason}")

# guardrails.py

# Prerequisites: none beyond Python's standard library

# Run: python guardrails.py

from dataclasses import dataclass

from enum import Enum

class GuardrailVerdict(Enum):

ALLOW = "allow"

DENY = "deny"

REQUIRE_APPROVAL = "require_approval"

@dataclass

class ProposedAction:

tool_name: str

args: dict

estimated_cost: float

irreversible: bool

@dataclass

class GuardrailResult:

verdict: GuardrailVerdict

reason: str

class GuardrailEngine:

"""

Sits between planning's proposed action and the tool executor.

Enforces an allow-list of permitted tools, a cost ceiling per action,

and mandatory human approval for anything irreversible -- regardless

of how confident the planner was that it was the right move.

"""

def __init__(self, allowed_tools: set[str], cost_ceiling: float):

self.allowed_tools = allowed_tools

self.cost_ceiling = cost_ceiling

def check(self, action: ProposedAction) -> GuardrailResult:

if action.tool_name not in self.allowed_tools:

return GuardrailResult(GuardrailVerdict.DENY,

f"Tool '{action.tool_name}' is not on the allow-list")

if action.estimated_cost > self.cost_ceiling:

return GuardrailResult(GuardrailVerdict.DENY,

f"Estimated cost ${action.estimated_cost} exceeds ceiling ${self.cost_ceiling}")

if action.irreversible:

return GuardrailResult(GuardrailVerdict.REQUIRE_APPROVAL,

"Action is irreversible -- requires human approval before execution")

return GuardrailResult(GuardrailVerdict.ALLOW, "Passed all guardrail checks")

if __name__ == "__main__":

guardrails = GuardrailEngine(

allowed_tools={"database_lookup", "policy_check", "payment_api", "email"},

cost_ceiling=100.0,

)

test_actions = [

ProposedAction("database_lookup", {"order_id": "4821"}, 0.0, False),

ProposedAction("delete_customer_account", {"id": "u_991"}, 0.0, True),

ProposedAction("payment_api", {"order_id": "4821", "amount": 5000.0}, 5000.0, True),

ProposedAction("payment_api", {"order_id": "4821", "amount": 49.99}, 49.99, True),

]

for action in test_actions:

result = guardrails.check(action)

print(f"{action.tool_name} -> {result.verdict.value}: {result.reason}")

How to run: python guardrails.py, no dependencies required.

Output:

database_lookup -> allow: Passed all guardrail checks
delete_customer_account -> deny: Tool 'delete_customer_account' is not on the allow-list
payment_api -> deny: Estimated cost $5000.0 exceeds ceiling $100.0
payment_api -> require_approval: Action is irreversible -- requires human approval before execution

database_lookup -> allow: Passed all guardrail checks

delete_customer_account -> deny: Tool 'delete_customer_account' is not on the allow-list

payment_api -> deny: Estimated cost $5000.0 exceeds ceiling $100.0

payment_api -> require_approval: Action is irreversible -- requires human approval before execution

The last case is the one worth sitting with: a $49.99 refund, well within the $100 cost ceiling, still gets flagged for human approval because it’s irreversible — full stop. Being within budget doesn’t override that. This is exactly the kind of rule that’s trivial to write down and easy to skip if guardrails aren’t treated as their own component with their own checks, separate from whatever the planner decided was a good idea.

Observability

Observability means trace-level logging of every step — not just the final output, but tool choices and intermediate reasoning — because without traces you cannot debug or improve agent behavior. This is the component that turns “the agent did something wrong” into “the agent called policy_check at step 4, it returned success=False, and the orchestrator correctly stopped the run there.” That’s the difference between a system you can actually iterate on and one you can only restart and hope.

# observability.py
# Prerequisites: none beyond Python's standard library
# Run: python observability.py

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any
import json

@dataclass
class TraceEntry:
    step_number: int
    component: str
    event: str
    detail: dict[str, Any]
    timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

class TraceLogger:
    """
    Wraps each step of a run with a structured, timestamped log entry.
    Deliberately separate from the orchestrator: the orchestrator decides
    what happens, the trace logger just watches and records it -- which is
    what lets you point at the exact step, component, and input that
    caused a deviation, without re-running anything.
    """
    def __init__(self, run_id: str):
        self.run_id = run_id
        self.entries: list[TraceEntry] = []
        self._step_counter = 0

    def log(self, component: str, event: str, **detail) -> None:
        self._step_counter += 1
        self.entries.append(TraceEntry(self._step_counter, component, event, detail))

    def export_json(self) -> str:
        return json.dumps({
            "run_id": self.run_id,
            "trace": [
                {"step": e.step_number, "component": e.component, "event": e.event,
                 "detail": e.detail, "timestamp": e.timestamp}
                for e in self.entries
            ],
        }, indent=2)

    def find_failure_point(self) -> TraceEntry | None:
        """The single most useful query on a trace: where did it first go wrong."""
        for entry in self.entries:
            if entry.detail.get("success") is False:
                return entry
        return None


if __name__ == "__main__":
    tracer = TraceLogger(run_id="run_2026_06_20_001")
    tracer.log("perception", "input_normalized", source="user_text", content="Refund order 4821")
    tracer.log("planning", "plan_created", step_count=3, goal="Process a refund for order 4821")
    tracer.log("tool_execution", "tool_called", tool="database_lookup", success=True, output="Order found")
    tracer.log("tool_execution", "tool_called", tool="policy_check", success=False,
               output="Order too old", error="Policy check failed: order too old")
    tracer.log("orchestrator", "run_stopped", reason="step_failed", step_number=4)

    failure = tracer.find_failure_point()
    print(f"Failure point: component='{failure.component}', step={failure.step_number}")
    print(f"  detail: {failure.detail}")

# observability.py

# Prerequisites: none beyond Python's standard library

# Run: python observability.py

from dataclasses import dataclass, field

from datetime import datetime, timezone

from typing import Any

import json

@dataclass

class TraceEntry:

step_number: int

component: str

event: str

detail: dict[str, Any]

timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

class TraceLogger:

"""

Wraps each step of a run with a structured, timestamped log entry.

Deliberately separate from the orchestrator: the orchestrator decides

what happens, the trace logger just watches and records it -- which is

what lets you point at the exact step, component, and input that

caused a deviation, without re-running anything.

"""

def __init__(self, run_id: str):

self.run_id = run_id

self.entries: list[TraceEntry] = []

self._step_counter = 0

def log(self, component: str, event: str, **detail) -> None:

self._step_counter += 1

self.entries.append(TraceEntry(self._step_counter, component, event, detail))

def export_json(self) -> str:

return json.dumps({

"run_id": self.run_id,

"trace": [

{"step": e.step_number, "component": e.component, "event": e.event,

"detail": e.detail, "timestamp": e.timestamp}

for e in self.entries

}, indent=2)

def find_failure_point(self) -> TraceEntry | None:

"""The single most useful query on a trace: where did it first go wrong."""

for entry in self.entries:

if entry.detail.get("success") is False:

return entry

return None

if __name__ == "__main__":

tracer = TraceLogger(run_id="run_2026_06_20_001")

tracer.log("perception", "input_normalized", source="user_text", content="Refund order 4821")

tracer.log("planning", "plan_created", step_count=3, goal="Process a refund for order 4821")

tracer.log("tool_execution", "tool_called", tool="database_lookup", success=True, output="Order found")

tracer.log("tool_execution", "tool_called", tool="policy_check", success=False,

output="Order too old", error="Policy check failed: order too old")

tracer.log("orchestrator", "run_stopped", reason="step_failed", step_number=4)

failure = tracer.find_failure_point()

print(f"Failure point: component='{failure.component}', step={failure.step_number}")

print(f" detail: {failure.detail}")

How to run: python observability.py, no dependencies required.

find_failure_point() walks the trace and lands directly on the policy_check call at step 4 — the exact tool, the exact arguments, the exact reason it failed. No re-running the agent, no guessing which of five steps went sideways. That’s the practical payoff of treating observability as a structural component that wraps the loop, rather than scattering print() statements through the orchestrator and hoping they’re enough when something breaks at 2 AM.

Wrapping Up

None of these components is optional once a system leaves the demo stage, even though most tutorials only ever show two or three of them. Perception and memory feed the reasoning core. Reasoning and planning hand off a plan object to tool execution. Orchestration holds the whole sequence together across steps and decides when to stop. Guardrails and observability wrap the entire loop rather than sitting inside it — one constrains what the loop is allowed to do, the other records what it actually did.

The reason production agentic systems end up looking more like software architecture than prompt engineering is that, underneath the LLM calls, they genuinely are software architecture. The model is one component among seven that handles reasoning and planning — it’s not the entire system. Understanding each component’s specific, separate responsibility is what makes it possible to debug a failure, secure a risky action, and scale a pipeline past the second user, instead of just hoping the 40-line script keeps working.

Navigation

The End-to-End Agentic AI Pipeline

Introduction

The Seven Components, at a Glance

Turning Raw Input Into Something the Agent Can Reason About

Working Context vs. What Actually Persists

Reasoning and Planning (Deciding What to Do Next)

Tool Execution

Orchestration

Guardrails

Observability

Wrapping Up

More On This Topic

No comments yet.

Leave a Reply Click here to cancel reply.