Post

The Architecture of Intelligent Systems: 15 Agentic AI Patterns for Production

Posted
By Shivraj Badu 11 min read

The Architecture of Intelligent Systems: 15 Agentic AI Patterns for Production

Introduction

The evolution from prompt-based AI to autonomous agents represents one of the most significant shifts in software architecture since microservices. Unlike traditional applications where logic flows through deterministic code paths, agentic systems make decisions, use tools, reason through problems, and adapt their behavior based on context.

Building production-ready AI agents isn’t about stringing together LLM calls—it’s about implementing proven architectural patterns that handle failures gracefully, scale efficiently, and provide predictable behavior despite the non-deterministic nature of language models. These patterns emerge from real production systems processing millions of requests, where reliability and cost efficiency aren’t optional.

This guide explores 15 battle-tested patterns for building intelligent systems that work in production environments where uptime, latency, and cost matter.

Pattern 1: ReAct (Reasoning + Acting)

The ReAct pattern interleaves reasoning and action, allowing agents to think through problems step-by-step while taking actions based on those thoughts.

Architecture:

1

Thought → Action → Observation → Thought → Action → ...

Implementation:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

class ReActAgent:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}
        self.max_iterations = 10
    
    def run(self, task: str) -> str:
        scratchpad = []
        
        for i in range(self.max_iterations):
            # Reasoning step
            prompt = self._build_prompt(task, scratchpad)
            response = self.llm.generate(prompt)
            
            thought = self._extract_thought(response)
            action = self._extract_action(response)
            
            scratchpad.append(f"Thought: {thought}")
            scratchpad.append(f"Action: {action}")
            
            # Acting step
            if action.startswith("Final Answer:"):
                return action.replace("Final Answer:", "").strip()
            
            # Execute tool
            tool_name, tool_input = self._parse_action(action)
            observation = self.tools[tool_name].execute(tool_input)
            scratchpad.append(f"Observation: {observation}")
        
        return "Agent exceeded maximum iterations"
    
    def _build_prompt(self, task: str, scratchpad: list) -> str:
        return f"""Answer the following question: {task}

You have access to these tools:
{self._format_tools()}

Use this format:
Thought: [your reasoning]
Action: [tool_name: input]
Observation: [result will be provided]
... (repeat Thought/Action/Observation as needed)
Thought: I now know the final answer
Action: Final Answer: [your answer]

Begin!

{chr(10).join(scratchpad)}
Thought:"""

When to Use:

  • Complex tasks requiring multi-step reasoning
  • Tool usage depends on intermediate results
  • Transparency in decision-making is important

Production Considerations:

  • Set maximum iterations to prevent infinite loops
  • Log all thoughts and actions for debugging
  • Implement timeout mechanisms
  • Cost: ~3-5x more expensive than single-shot inference

Pattern 2: Plan-and-Execute

Separate planning from execution—create a complete plan upfront, then execute steps sequentially.

Architecture:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

class PlanAndExecuteAgent:
    def __init__(self, planner_llm, executor_llm, tools):
        self.planner = planner_llm
        self.executor = executor_llm
        self.tools = tools
    
    def run(self, task: str) -> str:
        # Planning phase
        plan = self.planner.generate(f"""Create a step-by-step plan to: {task}
        
Available tools: {self._list_tools()}

Return a numbered list of steps.""")
        
        steps = self._parse_plan(plan)
        results = []
        
        # Execution phase
        for step in steps:
            result = self._execute_step(step, results)
            results.append(result)
            
            # Replanning if step fails
            if result.failed:
                plan = self._replan(task, steps, results)
                steps = self._parse_plan(plan)
        
        return self._synthesize_final_answer(results)

When to Use:

  • Tasks with clear sequential dependencies
  • Want to estimate cost/time before execution
  • Need to parallelize independent steps

Advantages:

  • Predictable execution flow
  • Can optimize plan before execution
  • Easier to cache and reuse plans

Pattern 3: Reflection and Self-Critique

Agent evaluates its own outputs and iterates until quality threshold is met.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

class ReflectiveAgent:
    def __init__(self, generator_llm, critic_llm):
        self.generator = generator_llm
        self.critic = critic_llm
        self.max_iterations = 3
    
    def generate_with_reflection(self, task: str) -> str:
        for iteration in range(self.max_iterations):
            # Generate output
            output = self.generator.generate(task)
            
            # Self-critique
            critique = self.critic.generate(f"""
Evaluate this response to: {task}

Response: {output}

Rate from 1-10 and provide specific improvements needed.
If score >= 8, respond with "APPROVED"
""")
            
            if "APPROVED" in critique:
                return output
            
            # Refine based on critique
            task = f"{task}\n\nPrevious attempt: {output}\nFeedback: {critique}\nImprove the response."
        
        return output  # Return best attempt

Production Use Cases:

  • Code generation (syntax validation)
  • Content creation (fact-checking, tone adjustment)
  • Data extraction (schema validation)

Cost Trade-offs:

  • 2-3x token usage vs single-pass
  • Significantly higher quality output
  • Fewer downstream errors

Pattern 4: Multi-Agent Collaboration

Specialized agents work together, each handling specific domains.

Architecture Patterns:

A. Hierarchical (Manager-Worker)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

class ManagerAgent:
    def __init__(self, specialist_agents):
        self.specialists = specialist_agents
    
    def delegate(self, task: str):
        # Decompose task
        subtasks = self._decompose(task)
        
        # Assign to specialists
        results = {}
        for subtask in subtasks:
            agent = self._select_specialist(subtask)
            results[subtask.id] = agent.execute(subtask)
        
        # Synthesize results
        return self._synthesize(results)

B. Peer-to-Peer (Debate)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

class DebateSystem:
    def __init__(self, agents):
        self.agents = agents
        self.moderator = ModeratorAgent()
    
    def debate(self, question: str, rounds: int = 3):
        responses = []
        
        for round in range(rounds):
            for agent in self.agents:
                context = self._format_debate_history(responses)
                response = agent.argue(question, context)
                responses.append(response)
        
        return self.moderator.synthesize(responses)

When to Use:

  • Domain-specific expertise needed (legal + technical + financial)
  • Want diverse perspectives (multiple reasoning approaches)
  • Need verification (agent checks another agent’s work)

Pattern 5: Tool-Augmented Generation

Agents use external tools to extend their capabilities beyond text generation.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

from typing import Callable, Dict

class Tool:
    def __init__(self, name: str, description: str, func: Callable):
        self.name = name
        self.description = description
        self.func = func
    
    def execute(self, input: str) -> str:
        try:
            result = self.func(input)
            return f"Success: {result}"
        except Exception as e:
            return f"Error: {str(e)}"

# Example tools
def search_database(query: str) -> str:
    """Search PostgreSQL database"""
    # Implementation
    return results

def call_api(endpoint: str, params: dict) -> str:
    """Make HTTP API call"""
    # Implementation
    return response

def execute_python(code: str) -> str:
    """Execute Python in sandbox"""
    # Implementation with timeout and resource limits
    return output

tools = [
    Tool("search_db", "Search company database", search_database),
    Tool("api_call", "Call external API", call_api),
    Tool("python", "Execute Python code", execute_python),
]

Critical Production Patterns:

  • Sandboxing: Execute code in isolated containers
  • Timeouts: Prevent hanging tool calls
  • Rate limiting: Protect external APIs
  • Error handling: Graceful degradation when tools fail
  • Logging: Track tool usage for debugging and cost analysis

Pattern 6: Memory and Context Management

Short-term vs Long-term Memory:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

class AgentMemory:
    def __init__(self, vector_db, sql_db):
        self.working_memory = []  # Current conversation
        self.episodic_memory = vector_db  # Past interactions (embeddings)
        self.semantic_memory = sql_db  # Facts and knowledge
    
    def remember(self, interaction: dict):
        # Add to working memory
        self.working_memory.append(interaction)
        
        # Trim if exceeds context window
        if self._token_count() > self.max_tokens:
            self._summarize_old_context()
        
        # Store important interactions long-term
        if interaction['importance'] > 0.7:
            embedding = self._embed(interaction)
            self.episodic_memory.store(embedding, interaction)
    
    def recall(self, query: str, k: int = 5):
        # Semantic search for relevant past interactions
        query_embedding = self._embed(query)
        similar = self.episodic_memory.search(query_embedding, k)
        return similar

Context Window Management Strategies:

Strategy Use Case Cost Impact
Sliding Window Chatbots, conversations Low - drops old messages
Summarization Long documents Medium - periodic LLM calls
Retrieval (RAG) Knowledge bases Low - only relevant chunks
Hierarchical Multi-session tasks High - nested summaries

Pattern 7: Guardrails and Safety Layers

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

class SafeAgent:
    def __init__(self, agent, guardrails):
        self.agent = agent
        self.input_filter = guardrails['input']
        self.output_filter = guardrails['output']
        self.action_validator = guardrails['actions']
    
    def run(self, user_input: str) -> str:
        # Input validation
        if not self.input_filter.is_safe(user_input):
            return "I cannot process that request"
        
        # Agent execution with action validation
        response = self.agent.run(user_input, 
                                  action_callback=self.action_validator)
        
        # Output filtering
        if not self.output_filter.is_safe(response):
            return "I apologize, I cannot provide that information"
        
        return response

Essential Guardrails:

  1. Prompt Injection Detection: Identify attempts to override instructions
  2. PII Filtering: Remove sensitive information
  3. Action Whitelisting: Only allow approved tool usage
  4. Content Moderation: Block harmful outputs
  5. Rate Limiting: Prevent abuse

Pattern 8: Streaming and Progressive Response

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

import asyncio
from typing import AsyncIterator

class StreamingAgent:
    async def stream_response(self, query: str) -> AsyncIterator[str]:
        # Stream thinking process
        async for thought in self.agent.think(query):
            yield f"🤔 {thought}\n"
        
        # Stream tool usage
        async for action in self.agent.act():
            yield f"🔧 Using: {action.tool}\n"
            result = await action.execute()
            yield f"📊 Result: {result}\n"
        
        # Stream final answer
        async for chunk in self.agent.answer():
            yield chunk

Why It Matters:

  • Perceived performance improvement
  • User engagement (see agent thinking)
  • Early cancellation (stop bad responses)
  • Lower time-to-first-token

Pattern 9: Fallback and Error Recovery

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

class ResilientAgent:
    def __init__(self, primary_llm, fallback_llm, cache):
        self.primary = primary_llm
        self.fallback = fallback_llm
        self.cache = cache
    
    def generate(self, prompt: str, max_retries: int = 3):
        # Check cache first
        if cached := self.cache.get(prompt):
            return cached
        
        for attempt in range(max_retries):
            try:
                response = self.primary.generate(prompt)
                self.cache.set(prompt, response)
                return response
            
            except RateLimitError:
                # Exponential backoff
                await asyncio.sleep(2 ** attempt)
            
            except ModelOverloadedError:
                # Switch to fallback
                response = self.fallback.generate(prompt)
                return response
            
            except ContextLengthError:
                # Truncate and retry
                prompt = self._truncate_prompt(prompt)
        
        raise AgentFailureError("All retry attempts failed")

Pattern 10: Evaluation-Driven Development

Continuous Evaluation Loop:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

class AgentEvaluator:
    def __init__(self, test_cases, metrics):
        self.test_cases = test_cases
        self.metrics = metrics
    
    def evaluate(self, agent_version: str) -> dict:
        results = {
            'accuracy': [],
            'latency': [],
            'cost': [],
            'tool_success_rate': []
        }
        
        for test in self.test_cases:
            start = time.time()
            response = agent.run(test.input)
            latency = time.time() - start
            
            results['accuracy'].append(
                self._score_response(response, test.expected)
            )
            results['latency'].append(latency)
            results['cost'].append(response.tokens * COST_PER_TOKEN)
        
        return {
            metric: np.mean(values) 
            for metric, values in results.items()
        }

Key Metrics to Track:

  • Task completion rate
  • Average cost per task
  • P95 latency
  • Tool usage patterns
  • Error rates by type

Patterns 11-15: Advanced Techniques

Pattern 11: Constitutional AI (Principle-Based Behavior)

Define principles that guide agent behavior, self-critique against those principles.

Pattern 12: Chain-of-Thought Prompting

Explicit step-by-step reasoning before answering, improves accuracy on complex tasks.

Pattern 13: Few-Shot Learning with Examples

Provide examples of desired behavior in the prompt, especially for structured outputs.

Pattern 14: Retrieval-Augmented Generation (RAG)

Fetch relevant context from external knowledge base before generating response.

Pattern 15: Function Calling with Schema Validation

Use structured outputs with JSON Schema validation to ensure reliable tool calls.

Production Deployment Checklist

Before deploying agents to production:

✅ Observability

  • Log all LLM calls with latency and cost
  • Track tool usage patterns
  • Monitor error rates by failure type
  • Set up alerts for cost anomalies

✅ Cost Management

  • Implement caching for repeated queries
  • Use cheaper models for simple tasks
  • Set per-user budget limits
  • Cache embeddings and tool results

✅ Safety

  • Content filtering on inputs and outputs
  • Action whitelisting for tools
  • PII detection and redaction
  • Rate limiting per user

✅ Testing

  • Regression test suite with expected outputs
  • Load testing for concurrent users
  • Chaos testing (API failures, timeouts)
  • Red team testing for prompt injection

✅ User Experience

  • Streaming for long responses
  • Progress indicators during tool usage
  • Graceful error messages
  • Ability to cancel long-running tasks

Common Pitfalls to Avoid

❌ Over-Engineering Early Start simple (single agent, few tools), add complexity only when needed.

❌ Ignoring Costs LLM calls add up fast. Track costs per user, per task type. Cache aggressively.

❌ No Evaluation Framework You can’t improve what you don’t measure. Build eval harness from day one.

❌ Deterministic Expectations Agents will surprise you. Design for non-determinism with retries and validation.

❌ Unbounded Tool Access Every tool is a potential failure point and cost center. Start with minimal tools.

Conclusion

Agentic AI patterns transform how we build software, shifting from deterministic control flow to systems that reason, adapt, and use tools autonomously. The patterns outlined here emerge from production systems processing millions of agent requests, where reliability and cost efficiency determine success.

The key insight: successful agent systems aren’t about the latest model or framework—they’re about architectural discipline. Start simple with ReAct or Plan-and-Execute, add memory when context matters, layer in safety guardrails, and continuously evaluate. The most sophisticated pattern is useless without proper observability, cost tracking, and error handling.

As agent capabilities expand, these patterns will evolve, but the fundamentals remain: clear separation of concerns, graceful failure handling, and metrics-driven iteration. Build systems that work reliably today, architect them to adapt tomorrow.

Suggested Reading

AI, Backend
ai-agents llm system-design architecture production langchain openai agent-patterns
This post is licensed under CC BY 4.0 by the author.