The End-to-End Agentic AI Pipeline

A new architectural breakdown of production-grade agentic AI systems identifies seven essential components — perception, memory, reasoning and planning, tool execution, orchestration, guardrails, and observability — that separate a demo script from a survivable system, according to architectural surveys and production postmortems. The components form a closed feedback loop (Goal → Perception → Reasoning → Planning → Action → Observation → Memory Update → back to Reasoning) with guardrails and observability as cross-cutting wrapper layers.

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 https://arxiv.org/pdf/2601.06064 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" 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374 perception.py Prerequisites: none beyond Python's standard library Run: python perception.py from dataclasses import dataclass, fieldfrom typing import Anyfrom enum import Enumimport jsonfrom datetime import datetime, timezone class InputSource Enum : USER TEXT = "user text" WEBHOOK = "webhook" FILE UPLOAD = "file upload" @dataclassclass 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 https://machinelearningmastery.com/beyond-short-term-memory-the-3-types-of-long-term-memory-ai-agents-need/ — 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}" 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778 memory.py Prerequisites: none beyond Python's standard library Run: python memory.py from dataclasses import dataclass, fieldfrom datetime import datetime, timezone @dataclassclass 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 @dataclassclass 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 https://arxiv.org/pdf/2601.06064 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 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051 planning.py Prerequisites: none beyond Python's standard library Run: python planning.py from dataclasses import dataclass, fieldimport json @dataclassclass PlanStep: step number: int description: str tool tag: str which category of tool this step will need -- execution decides HOW @dataclassclass 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}'" 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980 tool execution.py Prerequisites: none beyond Python's standard library Run: python tool execution.py import timeimport hashlibfrom dataclasses import dataclassfrom typing import Callable, Any, Optional @dataclassclass 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 https://www.langchain.com/langgraph , CrewAI https://crewai.com/ , and AutoGen https://microsoft.github.io/autogen/stable//index.html 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}'" 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071 orchestrator.py Prerequisites: none beyond Python's standard library Run: python orchestrator.py from dataclasses import dataclass, field @dataclassclass PlanStep: step number: int description: str tool tag: str @dataclassclass Plan: goal: str steps: list PlanStep = field default factory=list @dataclassclass 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' 1234567 Executing step 1: Look up orderExecuting step 2: Check eligibilityStep 2 failed: Policy check failed: order too oldStopping run -- a failed step blocks the rest of this plan Total steps attempted: 2Final 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 https://andriifurmanets.com/blogs/ai-agents-2026-practical-architecture-tools-memory-evals-guardrails 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}" 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364 guardrails.py Prerequisites: none beyond Python's standard library Run: python guardrails.py from dataclasses import dataclassfrom enum import Enum class GuardrailVerdict Enum : ALLOW = "allow" DENY = "deny" REQUIRE APPROVAL = "require approval" @dataclassclass ProposedAction: tool name: str args: dict estimated cost: float irreversible: bool @dataclassclass 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 : php 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 1234 database lookup - allow: Passed all guardrail checksdelete customer account - deny: Tool 'delete customer account' is not on the allow-listpayment api - deny: Estimated cost $5000.0 exceeds ceiling $100.0payment 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 https://andriifurmanets.com/blogs/ai-agents-2026-practical-architecture-tools-memory-evals-guardrails — 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}" 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364 observability.py Prerequisites: none beyond Python's standard library Run: python observability.py from dataclasses import dataclass, fieldfrom datetime import datetime, timezonefrom typing import Anyimport json @dataclassclass 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.