Production-Ready AI Agent Design: Orchestration Patterns and Reliability

class AgentPipeline:
    def __init__(self, state_store):
        self.state_store = state_store
        self.phases = [
            self.phase_research,
            self.phase_analyze,
            self.phase_draft,
            self.phase_review,
            self.phase_finalize,
        ]
 
    async def run(self, task_id: str, input_data: dict):
        # Load existing checkpoint if available
        checkpoint = await self.state_store.get(task_id)
 
        start_phase = 0
        state = {"input": input_data, "results": {}}
 
        if checkpoint:
            start_phase = checkpoint["last_completed_phase"] + 1
            state = checkpoint["state"]
            print(f"Resuming from phase {start_phase}")
 
        for i, phase_fn in enumerate(self.phases[start_phase:], start=start_phase):
            try:
                result = await phase_fn(state)
                state["results"][phase_fn.__name__] = result
 
                # Save checkpoint
                await self.state_store.set(task_id, {
                    "last_completed_phase": i,
                    "state": state,
                    "timestamp": datetime.now().isoformat()
                })
 
            except Exception as e:
                print(f"Phase {phase_fn.__name__} failed: {e}")
                raise
 
        return state["results"]

class HITLAgent:
    def __init__(self, approval_service):
        self.approval_service = approval_service
 
    async def execute_with_approval(
        self,
        action: dict,
        risk_level: str  # "low", "medium", "high"
    ):
        if risk_level == "high":
            # Wait for human approval
            approval = await self.approval_service.request_approval(
                action=action,
                context=self.get_current_context(),
                timeout_seconds=3600  # 1-hour timeout
            )
 
            if not approval.approved:
                return {"status": "rejected", "reason": approval.reason}
 
        elif risk_level == "medium":
            # Async notification (no approval required, but logged)
            await self.approval_service.notify(action)
 
        # Execute the action
        return await self.execute_action(action)

import asyncio
import random
 
async def call_tool_with_retry(
    tool_fn,
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0
):
    last_error = None
 
    for attempt in range(max_retries + 1):
        try:
            return await tool_fn()
        except (TransientError, RateLimitError) as e:
            last_error = e
 
            if attempt == max_retries:
                break
 
            # Exponential backoff + jitter
            delay = min(base_delay * (2 ** attempt), max_delay)
            delay += random.uniform(0, delay * 0.1)  # 10% jitter
 
            print(f"Attempt {attempt + 1} failed. Retrying in {delay:.1f}s...")
            await asyncio.sleep(delay)
 
    raise last_error

async def search_with_fallback(query: str):
    """Try multiple search strategies in order."""
    strategies = [
        lambda: search_primary_api(query),
        lambda: search_secondary_api(query),
        lambda: search_cached_knowledge(query),
        lambda: ask_human_for_input(query),
    ]
 
    for strategy in strategies:
        try:
            result = await strategy()
            if result and result.get("confidence", 0) > 0.5:
                return result
        except Exception as e:
            print(f"Strategy failed: {e}, trying next...")
            continue
 
    return {"error": "All strategies exhausted", "query": query}

class ErrorClassifier:
    """Classify errors and determine the right response."""
 
    @staticmethod
    def classify(error: Exception) -> str:
        error_str = str(error).lower()
 
        if "rate limit" in error_str or "429" in error_str:
            return "rate_limit"      # Retry with backoff
        elif "timeout" in error_str or "504" in error_str:
            return "timeout"         # Retry immediately
        elif "not found" in error_str or "404" in error_str:
            return "not_found"       # Use fallback
        elif "permission" in error_str or "403" in error_str:
            return "permission"      # Notify human, stop
        elif "invalid" in error_str or "400" in error_str:
            return "invalid_input"   # Fix prompt, retry
        else:
            return "unknown"         # Log, notify human
 
    @staticmethod
    def should_retry(error_type: str) -> bool:
        return error_type in ["rate_limit", "timeout"]
 
    @staticmethod
    def should_escalate(error_type: str) -> bool:
        return error_type in ["permission", "unknown"]

# Bad: a super-agent with everything
all_tools = [
    read_file, write_file, delete_file,
    send_email, access_database,
    call_external_api, execute_code,
    access_filesystem, manage_users
]
 
# Good: minimal toolsets scoped to each task
research_agent_tools = [read_file, search_web, access_knowledge_base]
writing_agent_tools = [read_file, write_file, check_grammar]
notification_agent_tools = [send_email, send_slack_message]

class OutputValidator:
    def __init__(self, rules: list):
        self.rules = rules
 
    def validate(self, output: dict) -> tuple[bool, list[str]]:
        violations = []
 
        for rule in self.rules:
            if not rule.check(output):
                violations.append(rule.description)
 
        return len(violations) == 0, violations
 
# Example rules
class NoDeletionRule:
    description = "Deletion operations are not permitted"
 
    def check(self, output: dict) -> bool:
        actions = output.get("actions", [])
        return not any(a.get("type") == "delete" for a in actions)
 
class MaxActionsRule:
    description = "Maximum 10 actions per execution"
 
    def check(self, output: dict) -> bool:
        return len(output.get("actions", [])) <= 10

from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
from typing import Any
 
class AgentStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    WAITING_APPROVAL = "waiting_approval"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"
 
@dataclass
class AgentState:
    task_id: str
    status: AgentStatus
    current_step: int
    total_steps: int
    context: dict[str, Any]
    tool_calls: list[dict] = field(default_factory=list)
    errors: list[dict] = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)
 
    def record_tool_call(self, tool_name: str, input_data: dict, output: Any):
        self.tool_calls.append({
            "tool": tool_name,
            "input": input_data,
            "output": output,
            "timestamp": datetime.now().isoformat(),
            "step": self.current_step
        })
        self.updated_at = datetime.now()

import logging
import json
from datetime import datetime
 
class AgentLogger:
    def __init__(self, task_id: str):
        self.task_id = task_id
        self.logger = logging.getLogger(f"agent.{task_id}")
 
    def log_tool_call(self, tool_name: str, input_data: dict):
        self.logger.info(json.dumps({
            "event": "tool_call",
            "task_id": self.task_id,
            "tool": tool_name,
            "input_keys": list(input_data.keys()),
            "timestamp": datetime.now().isoformat()
        }))
 
    def log_decision(self, reasoning: str, chosen_action: str):
        self.logger.info(json.dumps({
            "event": "decision",
            "task_id": self.task_id,
            "reasoning_length": len(reasoning),
            "action": chosen_action,
            "timestamp": datetime.now().isoformat()
        }))
 
    def log_completion(self, result: dict, duration_seconds: float):
        self.logger.info(json.dumps({
            "event": "completion",
            "task_id": self.task_id,
            "success": True,
            "duration_seconds": duration_seconds,
            "timestamp": datetime.now().isoformat()
        }))