Step-by-Step Teaching Guide

STEP 1: Foundation — What Is an Agentic System?

Concept

An agentic system is an AI-powered pipeline where a model (agent) autonomously decides what actions to take, which tools to call, and in what sequence — to accomplish a goal.

User Goal → Agent (LLM) → Decides Tool → Executes Tool → Returns Result → Next Decision

Real-World Analogy (from the article context)

Think of Amazon's fulfillment system:

• Goal: Ship package to customer by promised date
• Agent: Optimization system
• Tools: Assign warehouse, route truck, allocate labor
• Validation: Before assigning a truck, check capacity exists

---

STEP 2: What Is Tool Calling (Programmatic)?

Concept

Tool calling means the agent invokes external functions/APIs programmatically — not just generating text, but executing real actions.

Basic Structure

# Define a tool
def assign_fulfillment_center(order_id: str, region: str) -> dict:
    """Assigns an order to the optimal fulfillment center"""
    return {"center_id": "FC_BARCELONA_01", "eta_hours": 24}

# Tool registry
TOOLS = {
    "assign_fulfillment_center": assign_fulfillment_center,
    "route_truck": route_truck,
    "check_capacity": check_capacity
}

How the Agent Calls It

# Agent output (structured)
agent_decision = {
    "tool": "assign_fulfillment_center",
    "parameters": {
        "order_id": "ORD-12345",
        "region": "EU-WEST"
    }
}

# Programmatic execution
tool_name = agent_decision["tool"]
params = agent_decision["parameters"]
result = TOOLS[tool_name](**params)

---

STEP 3: Why Pre-Execution Validation Is Critical

The Problem Without Validation

Agent calls tool → Tool executes → FAILURE (wrong params, bad state, cascade error)
                                         ↓
                              Millions of orders affected

The Solution: Validate BEFORE executing

Agent calls tool → VALIDATE → Safe? → Execute → Result
                      ↓
                   Unsafe? → Reject + Explain → Agent retries

Three Layers of Validation

---

STEP 4: Implementing Pre-Execution Validation

Step 4a: Schema Validation with Pydantic

from pydantic import BaseModel, validator
from typing import Literal

class AssignFulfillmentCenterInput(BaseModel):
    order_id: str
    region: Literal["EU-WEST", "EU-EAST", "EU-NORTH"]
    priority: int
    
    @validator("order_id")
    def order_id_format(cls, v):
        if not v.startswith("ORD-"):
            raise ValueError("order_id must start with 'ORD-'")
        return v
    
    @validator("priority")
    def priority_range(cls, v):
        if not 1 <= v <= 5:
            raise ValueError("Priority must be between 1 and 5")
        return v

Step 4b: Business Logic Validation

class BusinessValidator:
    def __init__(self, network_state):
        self.network_state = network_state
    
    def validate_assignment(self, order_id: str, region: str) -> tuple[bool, str]:
        # Rule 1: Order must not already be assigned
        if self.network_state.is_order_assigned(order_id):
            return False, f"Order {order_id} already assigned"
        
        # Rule 2: Region must have active centers
        if not self.network_state.has_active_centers(region):
            return False, f"No active centers in {region}"
        
        # Rule 3: Check delivery promise feasibility
        if not self.network_state.can_meet_promise(order_id, region):
            return False, "Cannot meet delivery promise from this region"
        
        return True, "Valid"

Step 4c: State Validation

class StateValidator:
    def validate_capacity(self, region: str, order_volume: int) -> tuple[bool, str]:
        current_load = get_current_load(region)
        max_capacity = get_max_capacity(region)
        
        if current_load + order_volume > max_capacity * 0.95:  # 95% threshold
            return False, f"Region {region} near capacity ({current_load}/{max_capacity})"
        
        return True, "Capacity available"

---

STEP 5: Building the Validation Pipeline

Combining All Validators

class PreExecutionValidator:
    def __init__(self, schema_model, business_validator, state_validator):
        self.schema_model = schema_model
        self.business_validator = business_validator
        self.state_validator = state_validator
    
    def validate(self, tool_name: str, params: dict) -> ValidationResult:
        errors = []
        
        # Layer 1: Schema
        try:
            validated_params = self.schema_model(**params)
        except ValidationError as e:
            return ValidationResult(passed=False, errors=str(e), layer="schema")
        
        # Layer 2: Business Logic
        is_valid, message = self.business_validator.validate_assignment(
            validated_params.order_id, 
            validated_params.region
        )
        if not is_valid:
            return ValidationResult(passed=False, errors=message, layer="business")
        
        # Layer 3: State
        is_valid, message = self.state_validator.validate_capacity(
            validated_params.region, 
            order_volume=1
        )
        if not is_valid:
            return ValidationResult(passed=False, errors=message, layer="state")
        
        return ValidationResult(passed=True, validated_params=validated_params)

---

STEP 6: The Complete Tool Execution Framework

class ProductionToolExecutor:
    def __init__(self, tools: dict, validator: PreExecutionValidator):
        self.tools = tools
        self.validator = validator
        self.audit_log = AuditLogger()
    
    def execute(self, agent_decision: dict) -> ExecutionResult:
        tool_name = agent_decision.get("tool")
        params = agent_decision.get("parameters", {})
        
        # Step 1: Tool exists?
        if tool_name not in self.tools:
            return ExecutionResult(
                success=False, 
                error=f"Unknown tool: {tool_name}"
            )
        
        # Step 2: Pre-execution validation
        validation = self.validator.validate(tool_name, params)
        
        if not validation.passed:
            self.audit_log.log_rejection(tool_name, params, validation.errors)
            return ExecutionResult(
                success=False,
                error=validation.errors,
                layer_failed=validation.layer
            )
        
        # Step 3: Execute (only if validated)
        try:
            result = self.tools[tool_name](**validation.validated_params.dict())
            self.audit_log.log_success(tool_name, params, result)
            return ExecutionResult(success=True, result=result)
            
        except Exception as e:
            self.audit_log.log_error(tool_name, params, str(e))
            return ExecutionResult(success=False, error=str(e))

---

STEP 7: Production Considerations

7a: Retry Logic with Exponential Backoff

import time

def execute_with_retry(executor, decision, max_retries=3):
    for attempt in range(max_retries):
        result = executor.execute(decision)
        
        if result.success:
            return result
        
        # Don't retry schema/business errors (agent must fix params)
        if result.layer_failed in ["schema", "business"]:
            return result  # Return error for agent to correct
        
        # Retry state errors (transient)
        wait_time = 2 ** attempt  # 1s, 2s, 4s
        time.sleep(wait_time)
    
    return result

7b: Circuit Breaker Pattern

class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failures = 0
        self.threshold = failure_threshold
        self.timeout = timeout
        self.state = "CLOSED"  # CLOSED=normal, OPEN=blocking, HALF_OPEN=testing
        self.last_failure_time = None
    
    def call(self, func, *args, **kwargs):
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.timeout:
                self.state = "HALF_OPEN"
            else:
                raise Exception("Circuit OPEN: Tool temporarily unavailable")
        
        try:
            result = func(*args, **kwargs)
            self.reset()
            return result
        except Exception as e:
            self.record_failure()
            raise
    
    def record_failure(self):
        self.failures += 1
        self.last_failure_time = time.time()
        if self.failures >= self.threshold:
            self.state = "OPEN"
    
    def reset(self):
        self.failures = 0
        self.state = "CLOSED"

7c: Observability & Monitoring

class AuditLogger:
    def log_rejection(self, tool: str, params: dict, reason: str):
        print({
            "event": "TOOL_REJECTED",
            "tool": tool,
            "params": params,
            "reason": reason,
            "timestamp": time.time()
        })
        # → Send to CloudWatch / Datadog / your monitoring system
    
    def log_success(self, tool: str, params: dict, result: dict):
        print({
            "event": "TOOL_SUCCESS",
            "tool": tool,
            "latency_ms": ...,
            "timestamp": time.time()
        })

---

STEP 8: Full System Architecture

┌─────────────────────────────────────────────────────────┐
│                    AGENTIC SYSTEM                        │
│                                                          │
│  User Goal → LLM Agent → Tool Decision                  │
│                              ↓                           │
│              ┌───────────────────────────┐               │
│              │   PRE-EXECUTION VALIDATOR  │               │
│              │  1. Schema Check           │               │
│              │  2. Business Logic Check   │               │
│              │  3. State/Capacity Check   │               │
│              └───────────┬───────────────┘               │
│                    Pass? │                               │
│                   ┌──────┴──────┐                        │
│                  YES            NO                        │
│                   ↓             ↓                        │
│            Execute Tool    Return Error                  │
│                   ↓        to Agent                      │
│            Audit Log +     (Agent retries               │
│            Return Result    with fix)                    │
│                   ↓                                      │
│            Next Agent Step                               │
└─────────────────────────────────────────────────────────┘

---

STEP 9: Key Principles Summary

---

Quick Knowledge Check

Q1: Why should schema validation happen before business logic validation?

A: Schema errors are cheapest to detect and don't require database/network calls. Catch them first.

Q2: An agent calls a tool with a valid schema but the fulfillment center is at 98% capacity. What should happen?

A: State validation rejects the call, returns a clear error, agent selects a different region.

Q3: When should you NOT retry a failed tool call?

A: When the failure is a schema or business logic error — the agent must fix its parameters first, not just retry.

---

What You Should Now Be Able To Do ✅

• Define tools with clear input schemas using Pydantic
• Build a three-layer validation pipeline (schema, business, state)
• Implement a production tool executor with audit logging
• Apply circuit breaker and retry patterns appropriately
• Design the full agentic tool-calling architecture for production systems