<metadata>
purpose: Comprehensive guide to agent design patterns and architectures
type: reference-documentation
topic: agent-architecture-patterns
last-updated: 2025-09-30
</metadata>

<overview>
This document provides architectural patterns for designing AI agents, covering single-purpose agents, multi-agent systems, orchestration patterns, communication strategies, state management, and error handling approaches.
</overview>

# Agent Design Patterns

## Core Architectural Patterns

### Pattern 1: Single-Purpose Agent

**Intent**: Create an agent focused on one specific task with clear boundaries and responsibilities.

**Structure**:
```
┌─────────────────────────────────────┐
│     Single-Purpose Agent            │
├─────────────────────────────────────┤
│  Identity:                          │
│  - Clear, focused purpose           │
│  - Defined expertise domain         │
│  - Explicit capabilities            │
├─────────────────────────────────────┤
│  Inputs:                            │
│  - Well-defined input schema        │
│  - Validation rules                 │
│  - Context requirements             │
├─────────────────────────────────────┤
│  Processing:                        │
│  - Single responsibility            │
│  - Domain-specific logic            │
│  - Deterministic behavior           │
├─────────────────────────────────────┤
│  Outputs:                           │
│  - Structured result format         │
│  - Error handling                   │
│  - Confidence scores                │
└─────────────────────────────────────┘
```

**Implementation**:
```python
class SecurityReviewAgent:
    """Single-purpose agent for security review"""

    def __init__(self):
        self.purpose = "Security vulnerability detection"
        self.expertise = ["OWASP", "secure-coding", "crypto"]
        self.capabilities = [
            "detect_sql_injection",
            "detect_xss",
            "check_authentication",
            "verify_encryption"
        ]

    def execute(self, input_data: dict) -> AgentResult:
        """Single execution method with clear contract"""

        # Validate input
        self._validate_input(input_data)

        # Execute core responsibility
        findings = self._perform_security_review(input_data)

        # Format output
        return AgentResult(
            success=True,
            findings=findings,
            confidence=self._calculate_confidence(findings),
            metadata=self._generate_metadata()
        )

    def _validate_input(self, data: dict):
        """Ensure input meets requirements"""
        required_fields = ["code_path", "language", "target"]

        for field in required_fields:
            if field not in data:
                raise InvalidInputError(f"Missing required field: {field}")

    def _perform_security_review(self, data: dict) -> list:
        """Core single-purpose logic"""
        # Focused security analysis only
        pass
```

**When to Use**:
- Task has clear, narrow scope
- Requirements are well-defined
- Need predictable, consistent behavior
- Task executed frequently

**Benefits**:
- Easy to test
- Simple to maintain
- Predictable behavior
- Clear responsibility

**Drawbacks**:
- Limited flexibility
- May need coordination with other agents
- Cannot handle adjacent tasks

---

### Pattern 2: Multi-Agent Coordinator

**Intent**: Orchestrate multiple specialized agents to accomplish complex tasks requiring diverse expertise.

**Structure**:
```
                 ┌─────────────────┐
                 │   Coordinator   │
                 │     Agent       │
                 └────────┬────────┘
                          │
          ┌───────────────┼───────────────┐
          │               │               │
    ┌─────▼─────┐   ┌────▼────┐   ┌─────▼─────┐
    │  Agent A  │   │ Agent B │   │  Agent C  │
    │ (Security)│   │(Quality)│   │  (Perf)   │
    └───────────┘   └─────────┘   └───────────┘
          │               │               │
          └───────────────┼───────────────┘
                          │
                    ┌─────▼─────┐
                    │ Aggregator│
                    │   Agent   │
                    └───────────┘
```

**Implementation**:
```python
class CodeReviewCoordinator:
    """Coordinates multiple specialized review agents"""

    def __init__(self):
        self.agents = {
            "security": SecurityReviewAgent(),
            "quality": QualityReviewAgent(),
            "performance": PerformanceReviewAgent(),
            "accessibility": AccessibilityReviewAgent()
        }
        self.aggregator = ResultAggregatorAgent()

    async def execute(self, code_review_request: dict) -> ReviewResult:
        """Coordinate multi-agent review process"""

        # Parallel execution of specialized agents
        review_tasks = []

        for agent_name, agent in self.agents.items():
            task = self._execute_agent(
                agent=agent,
                request=code_review_request,
                agent_name=agent_name
            )
            review_tasks.append(task)

        # Wait for all agents to complete
        results = await asyncio.gather(*review_tasks)

        # Aggregate results
        final_report = self.aggregator.aggregate(results)

        # Add coordinator metadata
        final_report.metadata = {
            "agents_used": list(self.agents.keys()),
            "coordination_strategy": "parallel",
            "total_time": self._calculate_total_time(results)
        }

        return final_report

    async def _execute_agent(self, agent, request, agent_name):
        """Execute single agent with error handling"""
        try:
            result = await agent.execute(request)
            return AgentExecutionResult(
                agent=agent_name,
                success=True,
                result=result
            )
        except Exception as e:
            return AgentExecutionResult(
                agent=agent_name,
                success=False,
                error=str(e)
            )
```

**When to Use**:
- Task requires multiple domains of expertise
- Subtasks can run in parallel
- Need to combine different perspectives
- Complex decision-making required

**Benefits**:
- Leverages specialized expertise
- Parallel execution possible
- Modular and extensible
- Clear separation of concerns

**Drawbacks**:
- Coordination overhead
- More complex error handling
- Result aggregation complexity
- Potential for conflicting recommendations

---

### Pattern 3: Sequential Pipeline

**Intent**: Process data through a series of agents in a defined order, where each agent builds upon the previous agent's output.

**Structure**:
```
Input → [Agent 1] → [Agent 2] → [Agent 3] → Output
         Parser      Analyzer     Reporter

Data flows left to right through pipeline
Each stage transforms and enriches the data
```

**Implementation**:
```python
class AnalysisPipeline:
    """Sequential pipeline for data analysis"""

    def __init__(self):
        self.stages = [
            DataParserAgent(),
            DataAnalyzerAgent(),
            InsightGeneratorAgent(),
            ReportGeneratorAgent()
        ]

    def execute(self, raw_data: dict) -> AnalysisReport:
        """Execute pipeline stages sequentially"""

        pipeline_state = PipelineState(raw_data)

        for stage in self.stages:
            try:
                # Execute stage
                result = stage.execute(pipeline_state.current_data)

                # Update pipeline state
                pipeline_state.advance_stage(
                    stage_name=stage.name,
                    result=result
                )

                # Check if should continue
                if not self._should_continue(result):
                    break

            except Exception as e:
                # Handle pipeline failure
                return self._handle_pipeline_failure(
                    stage=stage,
                    error=e,
                    state=pipeline_state
                )

        return pipeline_state.final_output

    def _should_continue(self, result) -> bool:
        """Determine if pipeline should continue"""
        return result.success and not result.terminal

class PipelineState:
    """Maintains state through pipeline execution"""

    def __init__(self, initial_data):
        self.stages_completed = []
        self.current_data = initial_data
        self.stage_outputs = {}

    def advance_stage(self, stage_name: str, result):
        """Move to next pipeline stage"""
        self.stages_completed.append(stage_name)
        self.stage_outputs[stage_name] = result
        self.current_data = result.output

    @property
    def final_output(self):
        """Get final pipeline output"""
        last_stage = self.stages_completed[-1]
        return self.stage_outputs[last_stage]
```

**When to Use**:
- Process requires specific ordering
- Each stage builds on previous results
- Clear transformation sequence
- Need intermediate outputs

**Benefits**:
- Clear execution flow
- Easy to debug (inspect each stage)
- Simple error isolation
- Reusable stages

**Drawbacks**:
- Cannot parallelize
- Slower than parallel approaches
- Failure in one stage stops pipeline
- Rigid execution order

---

### Pattern 4: Event-Driven Agents

**Intent**: Agents react to events and trigger other agents, creating a flexible, loosely-coupled system.

**Structure**:
```
┌──────────────┐      Event        ┌──────────────┐
│  Agent A     │─────publish()────→│  Event Bus   │
└──────────────┘                   └──────┬───────┘
                                          │
                         ┌────────────────┼─────────────┐
                         │                │             │
                   ┌─────▼─────┐   ┌─────▼─────┐  ┌───▼────┐
                   │  Agent B  │   │  Agent C  │  │Agent D │
                   │(subscribed)│   │(subscribed)│  │(sub'd) │
                   └───────────┘   └───────────┘  └────────┘
```

**Implementation**:
```python
class EventBus:
    """Central event bus for agent communication"""

    def __init__(self):
        self.subscribers = defaultdict(list)

    def subscribe(self, event_type: str, agent: Agent):
        """Register agent for event type"""
        self.subscribers[event_type].append(agent)

    async def publish(self, event: Event):
        """Publish event to all subscribers"""
        subscribers = self.subscribers[event.type]

        tasks = []
        for agent in subscribers:
            task = agent.handle_event(event)
            tasks.append(task)

        # Execute all handlers in parallel
        await asyncio.gather(*tasks)

class EventDrivenAgent:
    """Base class for event-driven agents"""

    def __init__(self, event_bus: EventBus):
        self.event_bus = event_bus
        self._register_event_handlers()

    def _register_event_handlers(self):
        """Subscribe to relevant events"""
        for event_type in self.handles_events:
            self.event_bus.subscribe(event_type, self)

    async def handle_event(self, event: Event):
        """Handle incoming event"""
        handler = self._get_handler(event.type)
        result = await handler(event)

        # Publish result as new event if needed
        if result.should_publish:
            await self.event_bus.publish(result.new_event)

class CodeChangeAgent(EventDrivenAgent):
    """Reacts to code change events"""

    handles_events = ["code.changed", "code.committed"]

    async def on_code_changed(self, event: Event):
        """Handle code change event"""

        # Analyze the change
        analysis = await self._analyze_change(event.data)

        # Publish analysis result
        return EventResult(
            should_publish=True,
            new_event=Event(
                type="analysis.completed",
                data=analysis
            )
        )

class SecurityScanAgent(EventDrivenAgent):
    """Reacts to analysis completion"""

    handles_events = ["analysis.completed"]

    async def on_analysis_completed(self, event: Event):
        """Scan code when analysis completes"""

        if "security_relevant" in event.data.tags:
            scan_result = await self._security_scan(event.data)

            return EventResult(
                should_publish=True,
                new_event=Event(
                    type="security.scan.completed",
                    data=scan_result
                )
            )

# Usage
event_bus = EventBus()

code_agent = CodeChangeAgent(event_bus)
security_agent = SecurityScanAgent(event_bus)
notification_agent = NotificationAgent(event_bus)

# Trigger chain
await event_bus.publish(Event(
    type="code.changed",
    data={"file": "auth.py", "lines": [10, 20]}
))
# → CodeChangeAgent analyzes
# → SecurityScanAgent scans (if security-relevant)
# → NotificationAgent notifies (if issues found)
```

**When to Use**:
- Loosely coupled agents needed
- Dynamic agent composition
- Asynchronous processing preferred
- Complex event chains

**Benefits**:
- Loose coupling
- Easy to add new agents
- Flexible event routing
- Asynchronous by default

**Drawbacks**:
- Harder to debug
- Event ordering challenges
- Potential for event storms
- Complex error handling

---

### Pattern 5: Hierarchical Agents

**Intent**: Organize agents in a hierarchy where parent agents delegate to child agents and aggregate their results.

**Structure**:
```
                   ┌─────────────────┐
                   │  Root Agent     │
                   │  (Orchestrator) │
                   └────────┬────────┘
                            │
              ┌─────────────┼─────────────┐
              │             │             │
        ┌─────▼─────┐ ┌────▼────┐  ┌─────▼─────┐
        │  Level 1  │ │Level 1  │  │  Level 1  │
        │  Agent A  │ │Agent B  │  │  Agent C  │
        └─────┬─────┘ └─────────┘  └───────────┘
              │
     ┌────────┼────────┐
     │        │        │
┌────▼───┐┌──▼───┐┌───▼────┐
│Level 2 ││Level2││Level 2 │
│Agent 1 ││Agent2││Agent 3 │
└────────┘└──────┘└────────┘
```

**Implementation**:
```python
class HierarchicalAgent:
    """Base class for hierarchical agents"""

    def __init__(self, name: str, parent=None):
        self.name = name
        self.parent = parent
        self.children = []

    def add_child(self, child: 'HierarchicalAgent'):
        """Add child agent"""
        self.children.append(child)
        child.parent = self

    async def execute(self, task: dict) -> AgentResult:
        """Execute with child delegation"""

        # Can this agent handle directly?
        if self._can_handle_directly(task):
            return await self._execute_directly(task)

        # Delegate to appropriate child
        child = self._select_child(task)
        if child:
            child_result = await child.execute(task)
            return self._process_child_result(child_result)

        # No child can handle - escalate to parent
        if self.parent:
            return await self.parent.execute(task)

        # Cannot be handled
        raise NoHandlerError(f"No agent can handle task: {task}")

    def _select_child(self, task: dict):
        """Select most appropriate child agent"""
        for child in self.children:
            if child.can_handle(task):
                return child
        return None

class CodeReviewRootAgent(HierarchicalAgent):
    """Root agent for code review hierarchy"""

    def __init__(self):
        super().__init__("CodeReviewRoot")

        # Create language-specific child agents
        self.add_child(PythonReviewAgent())
        self.add_child(JavaScriptReviewAgent())
        self.add_child(GoReviewAgent())

    def _can_handle_directly(self, task):
        """Root handles orchestration only"""
        return False

class PythonReviewAgent(HierarchicalAgent):
    """Python-specific review agent"""

    def __init__(self):
        super().__init__("PythonReview")

        # Add specialized Python sub-agents
        self.add_child(PythonSecurityAgent())
        self.add_child(PythonPerformanceAgent())
        self.add_child(PythonStyleAgent())

    def can_handle(self, task):
        """Check if task is Python-related"""
        return task.get("language") == "python"

    async def _execute_directly(self, task):
        """Coordinate Python-specific reviews"""

        # Delegate to all specialized children
        results = []
        for child in self.children:
            result = await child.execute(task)
            results.append(result)

        # Aggregate results
        return self._aggregate_python_reviews(results)

# Usage
root = CodeReviewRootAgent()

result = await root.execute({
    "language": "python",
    "file": "auth.py",
    "review_type": "security"
})
# Flows: Root → PythonReview → PythonSecurityAgent
```

**When to Use**:
- Clear domain hierarchy exists
- Need delegation with escalation
- Specialized agents at different levels
- Task routing required

**Benefits**:
- Natural task delegation
- Escalation support
- Clear responsibility hierarchy
- Reusable specialized agents

**Drawbacks**:
- Can become complex
- Potential performance overhead
- Difficult to reorganize
- Rigid hierarchy

---

## State Management Patterns

### Pattern: Stateless Agents

**Intent**: Agents with no persistent state between executions.

```python
class StatelessAgent:
    """Purely functional agent with no state"""

    def execute(self, input_data: dict) -> dict:
        """Process input and return output without side effects"""

        # All data comes from input
        result = self._process(input_data)

        # Return result without storing anything
        return result

    def _process(self, data):
        """Pure function - same input always produces same output"""
        # No instance variables modified
        # No external state accessed
        # Deterministic behavior
        return transform(data)
```

**Benefits**:
- Easy to test
- No concurrency issues
- Horizontally scalable
- Predictable behavior

**Use When**:
- No context needed between calls
- Stateless operations
- High concurrency required

---

### Pattern: Stateful Agents with Memory

**Intent**: Agents that maintain state across executions to provide context-aware behavior.

```python
class StatefulAgent:
    """Agent with persistent memory"""

    def __init__(self, memory_store: MemoryStore):
        self.memory = memory_store
        self.conversation_history = []

    async def execute(self, input_data: dict) -> dict:
        """Execute with access to historical context"""

        # Retrieve relevant context
        context = await self.memory.retrieve_context(
            query=input_data,
            limit=5
        )

        # Process with context
        result = await self._process_with_context(
            input_data=input_data,
            context=context
        )

        # Update memory
        await self.memory.store(
            interaction={
                "input": input_data,
                "output": result,
                "timestamp": datetime.now()
            }
        )

        # Update conversation history
        self.conversation_history.append({
            "input": input_data,
            "output": result
        })

        return result

class MemoryStore:
    """Persistent memory for agent state"""

    def __init__(self):
        self.short_term = []  # Recent interactions
        self.long_term = {}   # Persistent facts
        self.vector_db = VectorDatabase()

    async def retrieve_context(self, query: dict, limit: int):
        """Retrieve relevant context for query"""

        # Search vector database for semantic similarity
        similar = await self.vector_db.search(
            query=query,
            limit=limit
        )

        # Combine with recent short-term memory
        recent = self.short_term[-limit:]

        return {
            "similar_interactions": similar,
            "recent_interactions": recent,
            "persistent_facts": self._get_relevant_facts(query)
        }

    async def store(self, interaction: dict):
        """Store interaction in memory"""

        # Add to short-term memory
        self.short_term.append(interaction)

        # Keep only recent items in short-term
        if len(self.short_term) > 100:
            self.short_term = self.short_term[-100:]

        # Store embedding for semantic search
        await self.vector_db.store_embedding(interaction)

        # Extract and store facts
        facts = self._extract_facts(interaction)
        self.long_term.update(facts)
```

**Use When**:
- Context improves decision quality
- Learning from past interactions
- Personalization needed
- Conversational agents

---

### Pattern: Shared State with Blackboard

**Intent**: Multiple agents collaborate by reading and writing to a shared knowledge base.

```python
class Blackboard:
    """Shared knowledge space for agent collaboration"""

    def __init__(self):
        self.knowledge = {}
        self.subscribers = defaultdict(list)
        self.lock = asyncio.Lock()

    async def write(self, key: str, value: any, agent_id: str):
        """Write knowledge to blackboard"""
        async with self.lock:
            self.knowledge[key] = {
                "value": value,
                "author": agent_id,
                "timestamp": datetime.now()
            }

            # Notify subscribers
            await self._notify_subscribers(key, value)

    async def read(self, key: str) -> any:
        """Read knowledge from blackboard"""
        async with self.lock:
            return self.knowledge.get(key, {}).get("value")

    async def query(self, pattern: str) -> dict:
        """Query blackboard with pattern"""
        async with self.lock:
            return {
                k: v["value"]
                for k, v in self.knowledge.items()
                if self._matches_pattern(k, pattern)
            }

    def subscribe(self, key: str, agent: Agent):
        """Subscribe to blackboard changes"""
        self.subscribers[key].append(agent)

class BlackboardAgent:
    """Agent that interacts with shared blackboard"""

    def __init__(self, agent_id: str, blackboard: Blackboard):
        self.agent_id = agent_id
        self.blackboard = blackboard

    async def execute(self, task: dict):
        """Execute using blackboard for coordination"""

        # Read current state from blackboard
        current_state = await self.blackboard.query("state.*")

        # Process based on shared knowledge
        result = self._process(task, current_state)

        # Write findings back to blackboard
        await self.blackboard.write(
            key=f"result.{self.agent_id}",
            value=result,
            agent_id=self.agent_id
        )

        return result

# Usage: Multiple agents collaborating
blackboard = Blackboard()

agent1 = AnalysisAgent("agent1", blackboard)
agent2 = SynthesisAgent("agent2", blackboard)
agent3 = ValidationAgent("agent3", blackboard)

# Agents collaborate through blackboard
await agent1.execute(task)  # Writes analysis
await agent2.execute(task)  # Reads analysis, writes synthesis
await agent3.execute(task)  # Reads both, writes validation
```

**Use When**:
- Multiple agents need shared context
- Complex problem requires collaboration
- Dynamic agent coordination needed
- Incremental problem solving

---

## Error Handling Patterns

### Pattern: Graceful Degradation

**Intent**: Agent provides partial results when full execution fails.

```python
class GracefulAgent:
    """Agent with graceful degradation"""

    def execute(self, input_data: dict) -> AgentResult:
        """Execute with fallback strategies"""

        result = AgentResult()

        # Try primary approach
        try:
            result.primary = self._primary_execution(input_data)
            result.strategy = "primary"
            return result

        except PrimaryFailure as e:
            result.errors.append(str(e))

        # Fallback to secondary approach
        try:
            result.secondary = self._secondary_execution(input_data)
            result.strategy = "secondary"
            result.degraded = True
            return result

        except SecondaryFailure as e:
            result.errors.append(str(e))

        # Last resort: minimal result
        result.minimal = self._minimal_execution(input_data)
        result.strategy = "minimal"
        result.degraded = True
        result.confidence = 0.3

        return result
```

---

### Pattern: Circuit Breaker

**Intent**: Prevent cascading failures by temporarily disabling failing agents.

```python
class CircuitBreaker:
    """Circuit breaker for agent execution"""

    def __init__(self, failure_threshold: int = 5, timeout: int = 60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.state = "closed"  # closed, open, half-open
        self.last_failure_time = None

    async def execute(self, agent: Agent, input_data: dict):
        """Execute agent with circuit breaker protection"""

        if self.state == "open":
            # Check if should try half-open
            if self._should_attempt_reset():
                self.state = "half-open"
            else:
                raise CircuitOpenError("Circuit breaker is open")

        try:
            result = await agent.execute(input_data)

            # Success - reset if in half-open
            if self.state == "half-open":
                self._reset()

            return result

        except Exception as e:
            self._record_failure()

            if self.failure_count >= self.failure_threshold:
                self._trip_circuit()

            raise

    def _trip_circuit(self):
        """Open circuit breaker"""
        self.state = "open"
        self.last_failure_time = datetime.now()

    def _reset(self):
        """Close circuit breaker"""
        self.state = "closed"
        self.failure_count = 0

# Usage
circuit_breaker = CircuitBreaker(failure_threshold=3)

try:
    result = await circuit_breaker.execute(agent, data)
except CircuitOpenError:
    # Use fallback agent
    result = await fallback_agent.execute(data)
```

---

### Pattern: Retry with Exponential Backoff

```python
class RetryableAgent:
    """Agent with retry logic"""

    async def execute_with_retry(
        self,
        input_data: dict,
        max_retries: int = 3,
        base_delay: float = 1.0
    ) -> AgentResult:
        """Execute with exponential backoff retry"""

        last_error = None

        for attempt in range(max_retries + 1):
            try:
                return await self.execute(input_data)

            except RetryableError as e:
                last_error = e

                if attempt < max_retries:
                    # Calculate delay: base_delay * (2 ^ attempt)
                    delay = base_delay * (2 ** attempt)

                    # Add jitter
                    delay += random.uniform(0, delay * 0.1)

                    await asyncio.sleep(delay)
                else:
                    # Max retries reached
                    raise MaxRetriesExceeded(
                        f"Failed after {max_retries} retries: {last_error}"
                    )

            except NonRetryableError as e:
                # Don't retry these errors
                raise
```

---

## Summary

**Key Patterns:**

1. **Single-Purpose**: Focused, specialized agents
2. **Multi-Agent Coordinator**: Parallel specialist execution
3. **Sequential Pipeline**: Ordered processing stages
4. **Event-Driven**: Reactive, loosely-coupled agents
5. **Hierarchical**: Delegation and escalation

**State Management:**
- Stateless: Simple, scalable
- Stateful: Context-aware, learning
- Shared State: Collaborative problem-solving

**Error Handling:**
- Graceful Degradation: Partial results
- Circuit Breaker: Prevent cascading failures
- Retry Logic: Handle transient failures

**Selection Criteria:**

| Pattern | Complexity | Scalability | Flexibility |
|---------|------------|-------------|-------------|
| Single-Purpose | Low | High | Low |
| Multi-Agent | Medium | High | High |
| Pipeline | Low | Medium | Low |
| Event-Driven | High | High | Very High |
| Hierarchical | High | Medium | Medium |

<references>
- lesson-06-review-documentation.md (Review systems)
- lesson-11-domain-agents.md (Domain specialization)
- specialization-strategy.md (When to specialize)
</references>