agent-architecture-patterns

# Agent Architecture Patterns Implements structural design patterns for building robust, scalable AI agent systems. This skill guides the model in selecting and applying the right architecture pattern based on complexity, communication needs, and fault tolerance requirements of the agent system. Agent architecture is not a single choice — it is a layered decision spanning system topology (monolithic vs distributed), communication protocols (request-reply, pub-sub, event streaming), state management strategies, lifecycle management, and security primitives that together determine whether an agent system scales gracefully or collapses under load. ## TL;DR Checklist - [ ] Choose architecture topology based on task complexity and isolation requirements - [ ] Implement message routing with explicit handler registration for each agent type - [ ] Set up state manager with serialization before any cross-agent state access - [ ] Register lifecycle hooks (startup, shutdown, recovery) in correct priority order - [ ] Configure circuit breakers with domain-specific thresholds per external dependency - [ ] Validate security: authentication at entry points, authorization at action boundaries --- ## When to Use Use this skill when: - Designing the architecture for a new AI agent system from scratch - Refactoring an existing single-agent system into a multi-agent topology - Integrating third-party services or APIs that require fault-tolerant communication - Choosing between monolithic, hub-and-spoke, peer-to-peer, or event-driven architectures - Implementing state management across multiple collaborating agents - Setting up security boundaries between independent agent components ## When NOT to Use Avoid this skill for: - Simple scripting tasks with no agent collaboration — use a linear workflow instead (see `workflow-patterns`) - Building individual LLM prompts or system messages — focus on prompt engineering patterns - Network infrastructure setup — use CNCF networking skills like Kubernetes service mesh - As the only architectural consideration — always pair with reliability patterns from `agent-reliability-engineering` --- ## Core Workflow 1. **Analyze System Requirements** — Determine agent count, communication intensity, state sharing needs, and failure tolerance requirements. Classify: simple (1–2 agents, synchronous), moderate (3–5 agents, mixed sync/async), or complex (6+ agents, fully async with shared state). **Checkpoint:** If all agents can share a single process memory space and complete tasks sequentially, prefer monolithic architecture over distributed. 2. **Select Architecture Topology** — Match requirements to topology: - Monolithic: Single process, shared memory, simple coordination via function calls. Best for ≤3 tightly coupled agents. - Hub-and-Spoke: Central coordinator dispatches tasks to specialized worker agents. Best for predictable task routing with centralized oversight. - Peer-to-Peer: Agents communicate directly without central coordinator. Best for decentralized collaboration and fault tolerance. - Event-Driven: Agents publish events; interested agents subscribe via message bus. Best for decoupled systems with high scalability. 3. **Design Communication Layer** — Implement the chosen protocol: - Request-Reply: Synchronous call-and-response. Use `AgentRouter.call(target, method, args)` for direct agent calls. - Pub-Sub: Async publish/subscribe. Agents register handlers via `router.subscribe(event_type, handler)`. - Event Stream: Persistent event log (like Kafka). Agents append events and replay from arbitrary offsets. 4. **Implement State Management** — Choose state scope strategy: - Local State: Agent-private, no serialization needed. Store in agent instance attributes. - Shared State: Cross-agent via centralized store with typed accessors. Always serialize before cross-boundary transfer. - Eventual Consistency: Agents maintain local copies, reconcile via events. Acceptable for non-critical state. 5. **Configure Lifecycle Management** — Register hooks in execution order: - Startup: Load configurations, initialize connections, warm caches - Ready: Transition from initializing to operational - Shutdown: Flush pending operations, close connections, release resources - Recovery: Rehydrate state from checkpoint after crash 6. **Apply Security Primitives** — Enforce boundaries at every layer: - Authentication: Verify agent identity before any communication (API keys, mutual TLS) - Authorization: Check permissions per action using role-based access control - Input Validation: Sanitize all cross-agent message payloads against schemas - Audit Logging: Record every inter-agent call with caller, target, action, and result --- ## Implementation Patterns ### Pattern 1: Monolithic Agent Architecture ```python from dataclasses import dataclass, field from typing import Any, Callable import logging logger = logging.getLogger(__name__) @dataclass class AgentContext: """Shared context passed through all agents in a monolithic system.""" state: dict[str, Any] = field(default_factory=dict) events: list[dict[str, Any]] = field(default_factory=list) metadata: dict[str, str] = field(default_factory=lambda: { "agent_count": "0", "started_at": "", }) class AgentRegistry: """Central registry for monolithic agent coordination.""" def __init__(self, context: AgentContext): self.context = context self._handlers: dict[str, Callable] = {} self._hooks_startup: list[Callable] = [] self._hooks_shutdown: list[Callable] = [] def register(self, name: str, handler: Callable) -> None: """Register an agent handler by name.""" if name in self._handlers: raise ValueError(f"Agent '{name}' already registered") self._handlers[name] = handler self.context.metadata["agent_count"] = str(len(self._handlers)) logger.info("Registered agent '%s' (total: %d)", name, len(self._handlers)) def call(self, agent_name: str, **kwargs) -> Any: """Invoke a registered agent synchronously.""" handler = self._handlers.get(agent_name) if not handler: raise KeyError(f"Agent '{agent_name}' not found in registry") result = handler(**kwargs) self.context.events.append({ "type": "call", "agent": agent_name, "args": kwargs, "result": str(result)[:200], }) return result def on_startup(self, hook: Callable) -> None: self._hooks_startup.append(hook) def on_shutdown(self, hook: Callable) -> None: self._hooks_shutdown.append(hook) def run_lifecycle(self) -> dict[str, Any]: """Execute startup hooks, return status.""" results = {"startup": [], "shutdown": []} for hook in self._hooks_startup: try: results["startup"].append(hook()) except Exception as e: logger.error("Startup hook failed: %s", e) return results def shutdown(self) -> list[str]: """Execute all shutdown hooks in reverse order.""" errors = [] for hook in reversed(self._hooks_shutdown): try: hook() except Exception as e: logger.error("Shutdown hook failed: %s", e) errors.append(str(e)) return errors ``` ### Pattern 2: Event-Driven Agent Communication ```python import asyncio from typing import Any, Awaitable, Callable from dataclasses import dataclass, field @dataclass class Event: """Typed event for pub-sub agent communication.""" event_type: str payload: dict[str, Any] source_agent: str = "" timestamp: float = 0.0 def __post_init__(self): import time if self.timestamp == 0.0: self.timestamp = time.time() class EventRouter: """Async event router for pub-sub agent communication.""" def __init__(self) -> None: self._subscribers: dict[str, list[Callable[[Event], Awaitable[None]]]] = {} self._max_history: int = 10000 self._history: list[Event] = [] async def subscribe( self, event_type: str, handler: Callable[[Event], Awaitable[None]] ) -> None: """Register an async handler for a specific event type.""" if event_type not in self._subscribers: self._subscribers[event_type] = [] self._subscribers[event_type].append(handler) logger.info("EventRouter: subscribed '%s' to %s", id(handler), event_type) async def publish(self, event: Event) -> list[str]: """Publish an event to all subscribers. Returns list of handler names called.""" handlers = self._subscribers.get(event.event_type, []) called = [] for handler in handlers: try: await handler(event) called.append(handler.__name__ if hasattr(handler, "__name__") else "anonymous") except Exception as e: logger.error( "EventRouter: handler %s failed for event %s: %s", handler.__name__, event.event_type, e, ) self._history.append(event) if len(self._history) > self._max_history: self._history = self._history[-self._max_history:] return called async def publish_all(self, events: list[Event]) -> dict[str, list[str]]: """Batch-publish multiple events. Returns per-event handler results.""" results = {} for event in events: results[event.event_type] = await self.publish(event) return results def get_history( self, event_type: str | None = None, limit: int = 100 ) -> list[Event]: """Retrieve recent event history, optionally filtered by type.""" if event_type: filtered = [e for e in self._history if e.event_type == event_type] return filtered[-limit:] return self._history[-limit:] # Example usage: routing events between specialized agents async def handle_trade_event(event: Event) -> None: """Trade agent processes trade execution events.""" payload = event.payload logger.info("TradeAgent: processing %s for %s at %.2f", payload.get("action"), payload.get("symbol"), payload.get("price")) async def handle_risk_event(event: Event) -> None: """Risk agent monitors and validates risk constraints.""" payload = event.payload position_value = payload.get("position_value", 0) if position_value > 100_000: logger.warning("RiskAgent: position %.2f exceeds threshold", position_value) async def main(): router = EventRouter() await router.subscribe("trade.execution", handle_trade_event) await router.subscribe("trade.execution", handle_risk_event) event = Event( event_type="trade.execution", payload={"action": "BUY", "symbol": "AAPL", "price": 178.50}, source_agent="strategy_engine", ) called = await router.publish(event) assert len(called) == 2, f"Expected 2 handlers called, got {len(called)}" ``` ### Pattern 3: State Management with Serialization ```python import json import hashlib from typing import Any from dataclasses import dataclass, field @dataclass class StateSnapshot: """Serializable state snapshot for cross-agent state sharing.""" agent_id: str version: int = 0 data: dict[str, Any] = field(default_factory=dict) checksum: str = "" def compute_checksum(self) -> str: content = f"{self.agent_id}:{self.version}:{json.dumps(self.data, sort_keys=True)}" self.checksum = hashlib.sha256(content.encode()).hexdigest()[:16] return self.checksum def is_intact(self) -> bool: return self.compute_checksum() == self.checksum class StateManager: """Centralized state manager with versioning and checksum validation.""" def __init__(self) -> None: self._states: dict[str, StateSnapshot] = {} self._lock: Any = None # asyncio.Lock in async context def get(self, agent_id: str) -> StateSnapshot | None: """Retrieve current state for an agent. Returns None if not found.""" return self._states.get(agent_id) def set(self, agent_id: str, data: dict[str, Any], version: int = 0) -> StateSnapshot: """Set state for an agent with optimistic concurrency control. Args: agent_id: Unique identifier of the agent owning this state. data: Serializable state data dictionary. version: Expected current version for conflict detection. Pass -1 to force-set. Returns: Updated StateSnapshot instance. Raises: ValueError: If version mismatch detected (optimistic lock conflict). """ existing = self._states.get(agent_id) if existing is not None and version >= 0 and existing.version != version: raise ValueError( f"State version conflict for '{agent_id}': " f"expected {version}, current {existing.version}" ) snapshot = StateSnapshot( agent_id=agent_id, data=data, version=(existing.version + 1) if existing else 0, ) snapshot.compute_checksum() self._states[agent_id] = snapshot return snapshot def merge( self, source_agent: str, target_agent: str, fields: list[str] | None = None ) -> StateSnapshot: """Merge specific fields from one agent's state into another. Args: source_agent: Agent whose state is the source of truth. target_agent: Agent receiving merged state fields. fields: Specific field keys to merge. If None, merges all fields. Returns: Updated target StateSnapshot after merge. """ source = self._states.get(source_agent) if not source: raise KeyError(f"No state found for source agent '{source_agent}'") target = self.set( target_agent, {**(self._states.get(target_agent).data if self._states.get(target_agent) else {}), **source.data}, ) if fields: # Merge only specified fields current_target = self._states.get(target_agent) if current_target: merged = {k: v for k, v in source.data.items() if k in fields} base = {k: v for k, v in current_target.data.items() if k not in fields} target = self.set(target_agent, {**base, **merged}, version=current_target.version) return target def export_all(self) -> dict[str, StateSnapshot]: """Export all states as serializable dict.""" return {aid: s for aid, s in self._states.items()} def import_snapshot(self, snapshot: StateSnapshot) -> None: """Import a state snapshot, overwriting existing if present.""" if not snapshot.is_intact(): raise ValueError(f"Corrupted snapshot from agent '{snapshot.agent_id}'") self._states[snapshot.agent_id] = snapshot def delete(self, agent_id: str) -> bool: """Delete state for an agent. Returns True if deleted.""" return self._states.pop(agent_id, None) is not None ```