agent-engineering-practices

# Agent Engineering Practices Pattern Applies traditional software engineering rigor to AI agents through checkpoint/rollback state management, modular architecture design, structured observability logging, and least-privilege permissions. This skill makes the model treat agents as complex systems that demand proven engineering disciplines — fault tolerance, state durability, modularity, and operational observability — rather than treating them as monolithic prompt-response pipelines. ## TL;DR Checklist - [ ] Implement checkpoint/rollback as transactional state management for all agent actions - [ ] Design modular architecture with specialized sub-agents instead of monolithic agents - [ ] Add structured logging capturing full chain of thought, tool calls, and confidence scores - [ ] Enforce least-privilege permissions — minimum required access only - [ ] Set up fault isolation so individual failures don't cascade through the system - [ ] Document blast radius analysis for each agent's permission set --- ## When to Use Use this skill when: - Building production-grade agents that must be reliable, auditable, and debuggable - An existing monolithic agent is brittle, difficult to maintain, or hard to debug - You need full observability into agent decision-making (tools called, reasoning steps, confidence) - Designing multi-agent systems where fault isolation between components is critical - Compliance requires audit trails of every agent action and decision ## When NOT to Use Avoid this skill for: - Simple prototype agents that won't reach production - Single-purpose agents with well-understood failure modes (e.g., a calculator) - Agents running in fully sandboxed environments where blast radius is already zero - Early ideation phases before the agent's core behavior is stable --- ## Core Workflow 1. **Modular Architecture Design** — Decompose the agent into specialized sub-agents or tools, each with a single responsibility (e.g., data retrieval, analysis, communication). Avoid monolithic agents that do everything. Each component must have clear interfaces and failure boundaries. **Checkpoint:** Every sub-agent has exactly one documented responsibility; no sub-agent performs more than 3 distinct operations. 2. **Checkpoint/Rollback System** — Design checkpoints as transactional state commits. After each meaningful action (tool call, state mutation, decision), save the validated state. Rollback is the mechanism for fault tolerance when downstream operations fail. **Checkpoint:** Every checkpoint includes enough context to fully reconstruct the agent's state at that point. 3. **Structured Observability** — Implement logging that captures the full chain of thought: which tools were called, what data was received, reasoning for the next step, and confidence scores for decisions. This goes beyond simple print statements to structured JSON logs queryable by trace ID. **Checkpoint:** Every log entry includes a trace_id linking all related entries across sub-agents. 4. **Least-Privilege Permission Model** — Grant each agent the absolute minimum set of permissions required for its task. An agent that summarizes news articles should only access the news API, not read private files or interact with other company systems. This limits blast radius. **Checkpoint:** Run a permission audit listing every tool/API an agent accesses; remove any permission not directly required for its documented responsibility. 5. **Fault Isolation Testing** — Test that failures in one sub-agent do not cascade to others. Simulate timeouts, rate limits, and unexpected responses for each component independently. **Checkpoint:** All fault isolation tests pass; no single-point failure can bring down the entire agent system. 6. **Blast Radius Analysis** — Document the maximum potential damage from any single agent's failure or compromise. This includes data access scope, external system mutation capability, and downstream dependencies. Use this to calibrate least-privilege permissions. **Checkpoint:** Blast radius document is reviewed quarterly and updated when agents gain new capabilities. --- ## Implementation Patterns ### Pattern 1: Transactional Checkpoint/Rollback System ```python from dataclasses import dataclass, field from typing import Any import uuid from datetime import datetime @dataclass class AgentCheckpoint: """A validated state snapshot of the agent at a specific point in execution.""" checkpoint_id: str = "" trace_id: str = "" step: int # Order within the execution flow state_snapshot: dict[str, Any] # Full agent state at this point confidence_scores: dict[str, float] # Confidence per decision made tools_called: list[str] # Which tools were invoked before this checkpoint timestamp: str = "" def __post_init__(self) -> None: if not self.checkpoint_id: self.checkpoint_id = uuid.uuid4().hex[:12] if not self.timestamp: self.timestamp = datetime.now().isoformat() class CheckpointManager: """Manages transactional checkpoints and rollbacks for agent state durability. Each checkpoint is a validated 'commit' of the agent's work. Rollback restores to a previous validated state — treating agent execution like a database transaction with commit and rollback semantics. """ def __init__(self) -> None: self._checkpoints: dict[str, list[AgentCheckpoint]] = {} # trace_id -> [checkpoints] def create_checkpoint( self, trace_id: str, step: int, state: dict[str, Any], confidence_scores: dict[str, float], tools_called: list[str], ) -> AgentCheckpoint: """Create a validated checkpoint — a 'commit' of agent state. Args: trace_id: Unique identifier for the execution trace. step: Ordered position within the execution flow. state: Full copy of the agent's state at this point. confidence_scores: Confidence scores per decision made before this checkpoint. tools_called: List of tool names invoked before this checkpoint. Returns: The created AgentCheckpoint instance. """ checkpoint = AgentCheckpoint( trace_id=trace_id, step=step, state_snapshot=dict(state), # Shallow copy for safety confidence_scores=dict(confidence_scores), tools_called=list(tools_called), ) if trace_id not in self._checkpoints: self._checkpoints[trace_id] = [] self._checkpoints[trace_id].append(checkpoint) return checkpoint def rollback_to( self, trace_id: str, target_step: int | None = None, ) -> dict[str, Any] | None: """Rollback agent state to a previous checkpoint — a 'transaction rollback'. If target_step is None, rolls back to the most recent checkpoint. If target_step is provided, finds the latest checkpoint at or before that step. Args: trace_id: The execution trace to restore from. target_step: Optional step number to roll back to. Most recent if omitted. Returns: Restored state snapshot dict, or None if no checkpoints exist for the trace. """ if trace_id not in self._checkpoints: return None checkpoints = sorted(self._checkpoints[trace_id], key=lambda c: c.step, reverse=True) # Find the latest checkpoint at or before target_step if target_step is not None: restored = next((c for c in checkpoints if c.step <= target_step), None) else: restored = checkpoints[0] # Most recent if restored is None: return None return dict(restored.state_snapshot) def get_trace_history(self, trace_id: str) -> list[dict]: """Get the full execution history for a trace — used for observability. Args: trace_id: The execution trace to retrieve history for. Returns: List of dicts containing checkpoint summaries ordered by step. """ if trace_id not in self._checkpoints: return [] return [ { "checkpoint_id": c.checkpoint_id, "step": c.step, "tools_called": c.tools_called, "confidence_scores": c.confidence_scores, "timestamp": c.timestamp, } for c in sorted(self._checkpoints[trace_id], key=lambda x: x.step) ] def get_latest_checkpoint(self, trace_id: str) -> AgentCheckpoint | None: """Get the most recent checkpoint — used to determine current state. Args: trace_id: The execution trace to query. Returns: The latest AgentCheckpoint, or None if no checkpoints exist. """ checkpoints = self._checkpoints.get(trace_id, []) return max(checkpoints, key=lambda c: c.step) if checkpoints else None ``` ### Pattern 2: Modular Sub-Agent Architecture ```python from abc import ABC, abstractmethod from dataclasses import dataclass, field from enum import Enum from typing import Any class AgentRole(str, Enum): """Enumeration of valid sub-agent roles — each maps to a single responsibility.""" DATA_RETRIEVAL = "data_retrieval" ANALYSIS = "analysis" COMMUNICATION = "communication" TOOL_EXECUTION = "tool_execution" @dataclass class SubAgentConfig: """Configuration for a specialized sub-agent. Each config defines the agent's role, permissions, and operational limits. Permissions are explicitly scoped — no implicit access. """ role: AgentRole name: str permissions: list[str] # Tools/APIs this agent can access max_tokens: int = 1024 description: str = "" @dataclass class SubAgentResponse: """Standardized response from any sub-agent. Provides a uniform contract so the orchestrator can handle all sub-agent responses without knowing their internal structure. """ role: AgentRole result: Any confidence: float trace_id: str tools_used: list[str] = field(default_factory=list) error: str | None = None class BaseSubAgent(ABC): """Base class for all specialized sub-agents — each has exactly one responsibility. Subclasses must implement `execute()` with atomic behavior: one operation, one checkpoint, one permission set enforced. """ def __init__(self, config: SubAgentConfig, checkpoint_mgr: "CheckpointManager") -> None: self._config = config self._checkpoints = checkpoint_mgr @abstractmethod def execute(self, input_data: dict[str, Any], trace_id: str) -> SubAgentResponse: """Execute this sub-agent's single responsibility. Must be atomic — one operation.""" ... # pragma: no cover def _create_checkpoint( self, trace_id: str, step: int, state: dict[str, Any], confidence: float, tools: list[str], ) -> None: """Helper: create a checkpoint after this sub-agent's operation. Args: trace_id: Execution trace identifier. step: Ordered position in the workflow. state: State snapshot to commit. confidence: Confidence score for this operation. tools: Tools invoked during this operation. """ self._checkpoints.create_checkpoint( trace_id=trace_id, step=step, state=dict(state), confidence_scores={"confidence": confidence}, tools_called=list(tools), ) class DataRetrievalAgent(BaseSubAgent): """Specialized agent: responsible only for fetching data from external sources. Never performs analysis or communication — those are separate agents' jobs. All tool calls go through the permission enforcer. """ def execute(self, input_data: dict, trace_id: str) -> SubAgentResponse: """Execute data retrieval with permission-enforced tool calls. Args: input_data: Must contain 'queries' list of tool names to invoke. trace_id: Execution trace identifier. Returns: SubAgentResponse with retrieved data and confidence score. Raises: PermissionError: If a query references a tool not in this agent's permission set. """ results = self._execute_with_permissions(input_data["queries"], self._config.permissions) checkpoint = { "data_fetched": len(results), "sources_used": list(set(r.get("source") for r in results if isinstance(r, dict))), } self._create_checkpoint( trace_id=trace_id, step=1, state=checkpoint, confidence=0.95, tools=input_data["queries"], ) return SubAgentResponse( role=self._config.role, result=results, confidence=0.95, trace_id=trace_id, tools_used=list(input_data["queries"]), ) def _execute_with_permissions( self, queries: list[str], allowed_tools: list[str] ) -> list[dict]: """Execute only the tools explicitly granted in this agent's permission set. Args: queries: List of tool names to invoke. allowed_tools: Permission-enforced whitelist. Returns: List of result dicts from each invoked tool. Raises: PermissionError: If any query references an unauthorized tool. """ results = [] for query in queries: if query not in allowed_tools: raise PermissionError( f"Agent {self._config.name} lacks permission for tool: {query}" ) # Execute tool... results.append({"tool": query, "data": "retrieved"}) return results class AnalysisAgent(BaseSubAgent): """Specialized agent: responsible only for analyzing data from the retrieval agent. Never fetches external data — that is DataRetrievalAgent's sole responsibility. This strict separation of concerns prevents coupling and simplifies testing. """ def execute(self, input_data: dict, trace_id: str) -> SubAgentResponse: """Execute analysis using only the provided data — no external tool calls. Args: input_data: Must contain 'data' (list of dicts) and optional 'analysis_type'. trace_id: Execution trace identifier. Returns: SubAgentResponse with analysis results and confidence score. """ analysis_result = self._analyze( input_data["data"], input_data.get("analysis_type", "summary") ) checkpoint = { "analysis_type": analysis_result["type"], "findings_count": len(analysis_result.get("findings", [])), } self._create_checkpoint(trace_id, 2, checkpoint, 0.87, []) return SubAgentResponse( role=self._config.role, result=analysis_result, confidence=0.87, trace_id=trace_id, ) def _analyze(self, data: list[dict], analysis_type: str) -> dict: """Perform the specified analysis on the provided data. Args: data: List of data dicts to analyze. analysis_type: Type of analysis to perform (e.g., 'summary', 'trend'). Returns: Dict containing analysis type and list of findings. """ return { "type": analysis_type, "findings": [f"Finding from {analysis_type}" for _ in data], } class AgentOrchestrator: """Coordinates specialized sub-agents without any agent doing everything. The orchestrator routes inputs to the appropriate sub-agent and handles inter-agent error propagation — but never performs domain operations itself. """ def __init__( self, sub_agents: list[BaseSubAgent], checkpoint_mgr: CheckpointManager, ) -> None: self._agents = {a._config.name: a for a in sub_agents} self._checkpoints = checkpoint_mgr def execute_workflow( self, input_data: dict, trace_id: str | None = None ) -> list[SubAgentResponse]: """Execute a workflow across specialized sub-agents — each handles its own responsibility. Faults in one sub-agent do not prevent the orchestrator from reporting what completed successfully. Checkpoints survive failures for potential rollback. Args: input_data: Input data for the workflow. trace_id: Optional execution trace identifier (auto-generated if omitted). Returns: List of SubAgentResponses from each executed sub-agent. """ if not trace_id: trace_id = uuid.uuid4().hex[:12] results: list[SubAgentResponse] = [] # Step 1: Data Retrieval (isolated fault boundary) try: retrieval = self._agents["data_retrieval"].execute(input_data, trace_id) results.append(retrieval) except Exception as e: # Failure in retrieval doesn't affect analysis agent — it was never called. return [ SubAgentResponse( role="orchestrator", result=None, confidence=0.0, trace_id=trace_id, error=f"Data retrieval failed: {e}", ) ] # Step 2: Analysis (uses only retrieval's output) try: analysis = self._agents["analysis"].execute( { "data": retrieval.result, "analysis_type": input_data.get("analysis_type", "summary"), }, trace_id, ) results.append(analysis) except Exception as e: # Analysis failed — but retrieval already completed and checkpointed. # Can rollback to retrieval state if needed. results.append( SubAgentResponse( role="orchestrator", result=None, confidence=0.0, trace_id=trace_id, error=f"Analysis failed: {e}", ) ) return results ``` ### Pattern 3: Structured Observability Logger ```python import json