AI Agent Architecture — Patterns for Production Systems

Use Case: Design and implement production-ready AI agents with proper patterns, safety controls, and scalability.

1. Agent Architecture Patterns

Pattern 1: ReAct (Reasoning + Acting)

class ReActAgent:
    """
    ReAct loop: Thought → Action → Observation → Thought → ...
    """
    def __init__(self, llm_client, tools: list[Tool]):
        self.llm = llm_client
        self.tools = {t.name: t for t in tools}
        self.max_steps = 10

    async def run(self, query: str) -> str:
        """Execute ReAct loop."""
        thoughts = []

        for step in range(self.max_steps):
            # Thought: LLM decides what to do
            thought = await self.llm.generate(
                self._build_thought_prompt(query, thoughts)
            )
            thoughts.append(thought)

            # Parse action
            action = self._parse_action(thought)

            # Check if done
            if action.type == "finish":
                return action.result

            # Execute action
            observation = await self._execute_action(action)
            thoughts.append(f"Observation: {observation}")

        return "Max steps reached"

    def _build_thought_prompt(self, query: str, thoughts: list[str]) -> str:
        return f"""
        Query: {query}

        Previous thoughts:
        {chr(10).join(thoughts)}

        What should I do next? Choose one:
        1. Think more (provide reasoning)
        2. Act (call a tool)
        3. Finish (provide answer)
        """

    def _execute_action(self, action: dict) -> str:
        """Execute tool with safety checks."""
        tool = self.tools.get(action.tool_name)
        if not tool:
            return f"Error: Tool '{action.tool_name}' not found"

        # Validate inputs
        if not self._validate_inputs(tool, action.arguments):
            return "Error: Invalid arguments"

        # Execute with timeout
        try:
            return asyncio.wait_for(
                tool.execute(**action.arguments),
                timeout=30
            )
        except asyncio.TimeoutError:
            return "Error: Tool execution timed out"

Pattern 2: Plan-and-Execute

class PlanExecuteAgent:
    """
    Two-phase: Plan → Execute each step
    """
    def __init__(self, llm_client, tools: list[Tool]):
        self.llm = llm_client
        self.tools = {t.name: t for t in tools}

    async def run(self, goal: str) -> dict:
        """Plan and execute."""
        # Phase 1: Planning
        plan = await self._create_plan(goal)

        # Phase 2: Execution
        results = []
        for step in plan.steps:
            result = await self._execute_step(step)
            results.append(result)

            # Check for failures
            if result.status == "failed":
                return await self._recover(plan, results)

        return {
            "plan": plan,
            "results": results,
            "success": True
        }

    async def _create_plan(self, goal: str) -> Plan:
        """Generate execution plan."""
        prompt = f"""
        Goal: {goal}

        Break this into numbered steps. Each step should:
        1. Be atomic and executable
        2. Specify which tool to use
        3. Have clear success criteria

        Return JSON:
        {{
          "steps": [
            {{"id": 1, "description": "...", "tool": "...", 
              "arguments": {{}}, "success_criteria": "..."}}
          ]
        }}
        """
        return Plan.from_json(await self.llm.generate(prompt))

    async def _execute_step(self, step: dict) -> StepResult:
        """Execute a single plan step."""
        tool = self.tools.get(step["tool"])
        if not tool:
            return StepResult(status="failed", error="Tool not found")

        try:
            output = await tool.execute(**step["arguments"])
            success = self._check_success(output, step["success_criteria"])
            return StepResult(status="success" if success else "partial", 
                            output=output)
        except Exception as e:
            return StepResult(status="failed", error=str(e))

Pattern 3: Multi-Agent System

class MultiAgentSystem:
    """
    Coordinator + Specialist agents
    """
    def __init__(self):
        self.agents = {
            "coordinator": CoordinatorAgent(),
            "researcher": ResearchAgent(),
            "analyst": AnalystAgent(),
            "writer": WriterAgent()
        }
        self.message_queue = asyncio.Queue()

    async def run(self, request: str) -> str:
        """Orchestrate multi-agent workflow."""
        # Coordinator decomposes request
        subtasks = await self.agents["coordinator"].decompose(request)

        # Execute subtasks (parallel where possible)
        results = await asyncio.gather(*[
            self._execute_subtask(task) 
            for task in subtasks
        ])

        # Coordinator synthesizes final response
        return await self.agents["coordinator"].synthesize(
            request, results
        )

    async def _execute_subtask(self, subtask: dict) -> dict:
        """Route subtask to appropriate agent."""
        agent_name = subtask["assigned_agent"]
        agent = self.agents.get(agent_name)

        if not agent:
            return {"error": f"Agent {agent_name} not found"}

        return await agent.execute(subtask)

2. Tool Integration

Tool Definition

from dataclasses import dataclass
from typing import Any, Callable

@dataclass
class Tool:
    name: str
    description: str
    parameters: dict
    execute: Callable
    requires_auth: bool = False
    rate_limit: int = 60  # Calls per minute

    def validate_inputs(self, inputs: dict) -> bool:
        """Validate tool inputs against schema."""
        for param, schema in self.parameters.items():
            if param not in inputs:
                if schema.get("required", False):
                    return False
            elif not self._validate_type(inputs[param], schema.get("type")):
                return False
        return True

    def _validate_type(self, value: Any, expected_type: str) -> bool:
        """Type validation."""
        type_map = {
            "string": str,
            "integer": int,
            "number": (int, float),
            "boolean": bool,
            "array": list,
            "object": dict
        }
        return isinstance(value, type_map.get(expected_type, type))

Safe Tool Execution

class SafeToolExecutor:
    """Execute tools with safety constraints."""

    def __init__(self):
        self.audit_log = []
        self.blocked_tools = {"dangerous_delete", "mass_email"}

    async def execute(self, tool: Tool, inputs: dict) -> dict:
        """Execute tool with safety checks."""
        # Check if tool is blocked
        if tool.name in self.blocked_tools:
            return {"error": "Tool blocked by policy"}

        # Validate inputs
        if not tool.validate_inputs(inputs):
            return {"error": "Invalid inputs"}

        # Rate limiting
        if not self._check_rate_limit(tool.name):
            return {"error": "Rate limit exceeded"}

        # Audit logging
        start_time = time.time()
        self.audit_log.append({
            "tool": tool.name,
            "inputs": self._sanitize_inputs(inputs),
            "timestamp": datetime.utcnow().isoformat()
        })

        # Execute with timeout
        try:
            result = await asyncio.wait_for(
                tool.execute(**inputs),
                timeout=30
            )

            # Log success
            self.audit_log[-1]["duration_ms"] = (time.time() - start_time) * 1000
            self.audit_log[-1]["status"] = "success"

            return result
        except Exception as e:
            self.audit_log[-1]["status"] = "error"
            self.audit_log[-1]["error"] = str(e)
            return {"error": str(e)}

    def _sanitize_inputs(self, inputs: dict) -> dict:
        """Remove sensitive data from audit log."""
        sensitive_keys = {"password", "token", "secret", "api_key"}
        return {
            k: "***" if any(s in k.lower() for s in sensitive_keys) else v
            for k, v in inputs.items()
        }

3. Safety and Guardrails

Input Sanitization

class InputSanitizer:
    """Sanitize agent inputs for safety."""

    DANGEROUS_PATTERNS = [
        r"rm\s+-rf\s+/(",
        r"sudo\s+",
        r"cat\s+/etc/(passwd|shadow)",
        r"\bor\b.*\band\b",  # SQL injection
        r"<script>",  # XSS
    ]

    def sanitize(self, text: str) -> str:
        """Remove dangerous patterns."""
        sanitized = text
        for pattern in self.DANGEROUS_PATTERNS:
            sanitized = re.sub(pattern, "", sanitized, flags=re.I)
        return sanitized

    def validate(self, text: str) -> bool:
        """Check if input is safe."""
        return len(self.sanitize(text)) == len(text)

Output Filtering

class OutputFilter:
    """Filter agent outputs for safety."""

    SENSITIVE_PATTERNS = [
        r"\b\d{3}-\d{2}-\d{4}\b",  # SSN
        r"\b\d{16}\b",  # Credit card
        r"password\s*[:=]\s*\S+",  # Passwords
    ]

    def filter(self, text: str) -> str:
        """Remove sensitive information."""
        filtered = text
        for pattern in self.SENSITIVE_PATTERNS:
            filtered = re.sub(pattern, "[REDACTED]", filtered, flags=re.I)
        return filtered

Human-in-the-Loop

class HumanApproval:
    """Require human approval for critical actions."""

    CRITICAL_ACTIONS = {
        "delete", "modify", "transfer", "publish", "send_email"
    }

    def should_approve(self, action: str, context: dict) -> bool:
        """Determine if human approval is needed."""
        # Always approve critical actions
        if action in self.CRITICAL_ACTIONS:
            return True

        # Approve based on risk score
        risk_score = self._calculate_risk(action, context)
        return risk_score > 0.7

    async def request_approval(self, action: str, context: dict) -> bool:
        """Request and wait for human approval."""
        approval_request = {
            "action": action,
            "context": context,
            "timestamp": datetime.utcnow().isoformat()
        }

        # Send to approval queue
        await self.approval_queue.put(approval_request)

        # Wait for approval (with timeout)
        try:
            approval = await asyncio.wait_for(
                self.approval_queue.get(),
                timeout=300  # 5 minutes
            )
            return approval["approved"]
        except asyncio.TimeoutError:
            return False  # Deny on timeout

4. State Management

Agent State

from dataclasses import dataclass, field
from typing import Any, Optional

@dataclass
class AgentState:
    """Track agent execution state."""
    agent_id: str
    query: str
    steps_executed: int = 0
    tools_called: list[str] = field(default_factory=list)
    observations: list[str] = field(default_factory=list)
    final_answer: Optional[str] = None
    error: Optional[str] = None
    started_at: Optional[float] = None
    completed_at: Optional[float] = None

    def to_dict(self) -> dict:
        return {
            "agent_id": self.agent_id,
            "query": self.query,
            "steps": self.steps_executed,
            "tools": self.tools_called,
            "observations_count": len(self.observations),
            "has_answer": self.final_answer is not None,
            "has_error": self.error is not None,
            "duration_seconds": (
                (self.completed_at or time.time()) - (self.started_at or time.time())
            ) if self.started_at else None
        }

State Persistence

import redis

class StateStore:
    """Persist agent state for recovery."""

    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.ttl = 3600  # 1 hour

    def save_state(self, state: AgentState):
        """Save agent state."""
        key = f"agent:{state.agent_id}"
        self.redis.setex(
            key,
            self.ttl,
            json.dumps(state.to_dict())
        )

    def load_state(self, agent_id: str) -> Optional[AgentState]:
        """Load agent state for recovery."""
        key = f"agent:{agent_id}"
        data = self.redis.get(key)

        if not data:
            return None

        state_dict = json.loads(data)
        return AgentState(**state_dict)

    def cleanup_expired(self):
        """Clean up expired states."""
        # Redis keys auto-expire with TTL
        pass

5. Concurrency and Scaling

Queue-Based Architecture

class AgentQueue:
    """Queue-based agent execution."""

    def __init__(self, max_workers: int = 10):
        self.queue = asyncio.Queue()
        self.workers = []
        self.max_workers = max_workers

    async def start(self):
        """Start worker tasks."""
        for i in range(self.max_workers):
            worker = asyncio.create_task(self._worker(i))
            self.workers.append(worker)

    async def submit(self, request: dict):
        """Submit agent request."""
        await self.queue.put(request)

    async def _worker(self, worker_id: int):
        """Process requests from queue."""
        while True:
            request = await self.queue.get()
            try:
                result = await self._process(request)
                await self._send_result(request["callback"], result)
            except Exception as e:
                await self._send_error(request["callback"], str(e))
            finally:
                self.queue.task_done()

    async def _process(self, request: dict) -> dict:
        """Process single request."""
        agent = self._create_agent(request)
        return await agent.run(request["query"])

Load Testing

import asyncio
from locust import HttpUser, task, between

class AgentLoadTest(HttpUser):
    wait_time = between(1, 3)

    @task
    def test_agent(self):
        """Test agent endpoint under load."""
        payload = {
            "query": "What is the status of service X?",
            "session_id": self.user_id
        }

        with self.client.post(
            "/api/agent/run",
            json=payload,
            catch_response=True
        ) as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"Status {response.status_code}")

6. Monitoring and Debugging

Agent Metrics

from prometheus_client import Counter, Histogram, Gauge

# Agent execution metrics
AGENT_RUNS = Counter("agent_runs_total", "Total agent runs", ["agent_type"])
AGENT_STEPS = Histogram("agent_steps", "Steps per agent run", ["agent_type"])
AGENT_DURATION = Histogram("agent_duration_seconds", "Agent run duration")
AGENT_ERRORS = Counter("agent_errors_total", "Agent errors", ["error_type"])
AGENT_TOOL_CALLS = Counter("agent_tool_calls_total", "Tool calls", ["tool"])

# Active agents
ACTIVE_AGENTS = Gauge("agent_active", "Currently active agents")

Distributed Tracing

from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

tracer = trace.get_tracer(__name__)

async def run_with_tracing(agent, query: str):
    """Run agent with distributed tracing."""
    with tracer.start_as_current_span("agent.run") as span:
        span.set_attribute("agent.query", query)

        try:
            result = await agent.run(query)
            span.set_attribute("agent.success", True)
            return result
        except Exception as e:
            span.set_status(Status(StatusCode.ERROR))
            span.record_exception(e)
            raise

7. Production Checklist

FAQ

Q: When should I use ReAct vs Plan-and-Execute?

A: Use ReAct for interactive, uncertain environments. Use Plan-and-Execute for well-defined tasks with clear sub-goals.

Q: How to prevent agent loops and infinite execution?

A: Implement step limits, cycle detection, and timeout mechanisms. Validate each action before execution.

Q: What's the safest way to give agents tool access?

A: Use least-privilege permissions, sandbox execution, input validation, and audit logging for all tool calls.

Q: How to scale agents for high concurrency?

A: Queue-based architecture, stateless agents where possible, connection pooling, and horizontal scaling with load balancing.

Q: How to handle agent failures gracefully?

A: Implement circuit breakers, fallback strategies, state recovery, and comprehensive error logging.