Production Best Practices

Overview

Deploying multi-agent systems in production requires careful attention to error handling, retries, logging, monitoring, and system design. This guide covers the essential best practices for running Swarms agents reliably at scale.

Error Handling

Agent Error Hierarchy

Swarms provides a comprehensive exception hierarchy for different failure modes:

from swarms.structs.agent import (
    AgentError,  # Base exception
    AgentInitializationError,  # Initialization failures
    AgentRunError,  # Runtime failures
    AgentLLMError,  # LLM-related errors
    AgentToolError,  # Tool execution errors
    AgentMemoryError,  # Memory errors
    AgentToolExecutionError,  # Tool execution failures
)

try:
    agent = Agent(
        agent_name="Production-Agent",
        model_name="claude-sonnet-4-6",
        max_loops=3,
    )
    result = agent.run("Process this task")
except AgentInitializationError as e:
    logger.error(f"Failed to initialize agent: {e}")
    # Handle initialization failure (e.g., retry with different config)
except AgentLLMError as e:
    logger.error(f"LLM error: {e}")
    # Handle LLM failures (e.g., switch to fallback model)
except AgentToolError as e:
    logger.error(f"Tool error: {e}")
    # Handle tool failures (e.g., disable problematic tool)
except AgentRunError as e:
    logger.error(f"Runtime error: {e}")
    # Handle general runtime errors
except Exception as e:
    logger.error(f"Unexpected error: {e}")
    # Catch-all for unexpected errors

Graceful Error Recovery

import traceback
from loguru import logger

def run_agent_with_recovery(agent, task, max_retries=3):
    """
    Run agent with automatic recovery from transient errors.
    """
    for attempt in range(max_retries):
        try:
            result = agent.run(task)
            return result
        except (AgentLLMError, AgentRunError) as e:
            logger.warning(f"Attempt {attempt + 1} failed: {e}")
            if attempt < max_retries - 1:
                logger.info(f"Retrying in {2 ** attempt} seconds...")
                time.sleep(2 ** attempt)  # Exponential backoff
            else:
                logger.error(f"All {max_retries} attempts failed")
                raise
        except Exception as e:
            logger.error(f"Unexpected error: {e}")
            logger.error(traceback.format_exc())
            raise

Retry Strategies

Built-in Retry Configuration

from swarms import Agent

agent = Agent(
    agent_name="Resilient-Agent",
    model_name="claude-sonnet-4-6",
    retry_attempts=5,  # Number of retry attempts
    retry_interval=2,  # Initial retry interval (seconds)
    max_loops=3,
)

Exponential Backoff

import time
import random

def exponential_backoff_retry(
    func, 
    max_retries=5, 
    base_delay=1, 
    max_delay=60,
    jitter=True
):
    """
    Execute function with exponential backoff retry.
    """
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            
            delay = min(base_delay * (2 ** attempt), max_delay)
            if jitter:
                delay *= (0.5 + random.random())  # Add jitter
            
            logger.warning(
                f"Attempt {attempt + 1} failed: {e}. "
                f"Retrying in {delay:.2f}s..."
            )
            time.sleep(delay)

# Usage
result = exponential_backoff_retry(
    lambda: agent.run("Complex task"),
    max_retries=5,
    base_delay=1,
    max_delay=60,
)

Fallback Models

Use fallback models for resilience:

agent = Agent(
    agent_name="Resilient-Agent",
    fallback_models=[
        "claude-sonnet-4-6",           # Primary model
        "gpt-5.4",      # First fallback
        "gpt-3.5-turbo",    # Second fallback
    ],
    max_loops=3,
)

# Agent automatically tries fallback models if primary fails
result = agent.run("Generate a report")

Logging

Structured Logging with Loguru

Swarms uses Loguru for powerful, structured logging:

from loguru import logger
import sys

# Configure logging for production
logger.remove()  # Remove default handler
logger.add(
    sys.stderr,
    format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
    level="INFO",
    colorize=True,
)

# Add file logging with rotation
logger.add(
    "logs/agent_{time}.log",
    rotation="500 MB",  # Rotate when file reaches 500MB
    retention="10 days",  # Keep logs for 10 days
    compression="zip",  # Compress rotated logs
    level="DEBUG",
    format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {name}:{function}:{line} - {message}",
)

# Add error-only log file
logger.add(
    "logs/errors_{time}.log",
    rotation="100 MB",
    retention="30 days",
    compression="zip",
    level="ERROR",
    filter=lambda record: record["level"].name == "ERROR",
)

Agent-Specific Logging

# Enable verbose logging for specific agents
agent = Agent(
    agent_name="Debug-Agent",
    model_name="claude-sonnet-4-6",
    verbose=True,  # Enable verbose logging
    print_on=True,  # Print outputs to console
)

# Control logging in AOP
from swarms.structs.aop import AOP

deployer = AOP(
    server_name="ProductionCluster",
    verbose=True,
    traceback_enabled=True,  # Enable detailed tracebacks
    log_level="INFO",  # Set log level
)

Contextual Logging

from loguru import logger
import contextvars

# Create context var for request ID
request_id_var = contextvars.ContextVar('request_id', default='unknown')

# Add request ID to all logs
logger.configure(
    patcher=lambda record: record.update(
        request_id=request_id_var.get()
    )
)

def process_request(request_id, task):
    request_id_var.set(request_id)
    logger.info(f"Processing task: {task}")
    
    try:
        result = agent.run(task)
        logger.info(f"Task completed successfully")
        return result
    except Exception as e:
        logger.error(f"Task failed: {e}")
        raise

Monitoring

Performance Metrics

import time
from loguru import logger

class AgentMetrics:
    def __init__(self):
        self.total_requests = 0
        self.successful_requests = 0
        self.failed_requests = 0
        self.total_duration = 0.0
    
    def record_request(self, success: bool, duration: float):
        self.total_requests += 1
        if success:
            self.successful_requests += 1
        else:
            self.failed_requests += 1
        self.total_duration += duration
    
    def get_metrics(self):
        avg_duration = (
            self.total_duration / self.total_requests 
            if self.total_requests > 0 else 0
        )
        success_rate = (
            self.successful_requests / self.total_requests 
            if self.total_requests > 0 else 0
        )
        
        return {
            "total_requests": self.total_requests,
            "successful_requests": self.successful_requests,
            "failed_requests": self.failed_requests,
            "success_rate": success_rate,
            "avg_duration": avg_duration,
        }

metrics = AgentMetrics()

def monitored_agent_run(agent, task):
    start_time = time.time()
    success = False
    
    try:
        result = agent.run(task)
        success = True
        return result
    except Exception as e:
        logger.error(f"Agent run failed: {e}")
        raise
    finally:
        duration = time.time() - start_time
        metrics.record_request(success, duration)
        logger.info(f"Request completed in {duration:.2f}s")

Health Checks

def health_check(agent):
    """
    Perform health check on agent.
    """
    try:
        # Quick test run
        result = agent.run("Hello", max_loops=1)
        return {
            "status": "healthy",
            "agent_name": agent.agent_name,
            "model": agent.model_name,
            "timestamp": time.time(),
        }
    except Exception as e:
        return {
            "status": "unhealthy",
            "agent_name": agent.agent_name,
            "error": str(e),
            "timestamp": time.time(),
        }

# Periodic health checks
import threading

def periodic_health_check(agent, interval=60):
    def check():
        while True:
            status = health_check(agent)
            logger.info(f"Health check: {status}")
            time.sleep(interval)
    
    thread = threading.Thread(target=check, daemon=True)
    thread.start()

Configuration Management

Environment-Based Configuration

import os
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Production configuration
PROD_CONFIG = {
    "model_name": os.getenv("MODEL_NAME", "claude-sonnet-4-6"),
    "max_loops": int(os.getenv("MAX_LOOPS", "3")),
    "retry_attempts": int(os.getenv("RETRY_ATTEMPTS", "5")),
    "timeout": int(os.getenv("TIMEOUT", "120")),
    "verbose": os.getenv("VERBOSE", "false").lower() == "true",
}

agent = Agent(
    agent_name="Production-Agent",
    **PROD_CONFIG,
)

Configuration Validation

from pydantic import BaseModel, Field, validator

class AgentConfig(BaseModel):
    agent_name: str = Field(..., min_length=1)
    model_name: str
    max_loops: int = Field(default=3, ge=1, le=100)
    retry_attempts: int = Field(default=3, ge=0, le=10)
    timeout: int = Field(default=120, ge=1)
    
    @validator('model_name')
    def validate_model(cls, v):
        allowed_models = ["claude-sonnet-4-6", "gpt-5.4", "claude-sonnet-3.5"]
        if v not in allowed_models:
            raise ValueError(f"Model must be one of {allowed_models}")
        return v

# Load and validate config
config = AgentConfig(
    agent_name="Production-Agent",
    model_name="claude-sonnet-4-6",
    max_loops=5,
)

agent = Agent(**config.dict())

Security Best Practices

API Key Management

import os
from cryptography.fernet import Fernet

# Never hardcode API keys
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")

# Use environment variables or secure vaults
agent = Agent(
    model_name="claude-sonnet-4-6",
    llm_api_key=OPENAI_API_KEY,
)

Input Validation

import re

def sanitize_input(text: str) -> str:
    """
    Sanitize user input to prevent injection attacks.
    """
    # Remove control characters
    text = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', text)
    
    # Limit length
    max_length = 10000
    if len(text) > max_length:
        text = text[:max_length]
    
    return text.strip()

# Use sanitized input
user_task = sanitize_input(request.get("task"))
result = agent.run(user_task)

Safety Prompts

agent = Agent(
    agent_name="Safe-Agent",
    model_name="claude-sonnet-4-6",
    safety_prompt_on=True,  # Enable safety guardrails
)

Resource Management

Memory Management

# Truncate conversation history to manage memory
agent = Agent(
    agent_name="Memory-Managed-Agent",
    model_name="claude-sonnet-4-6",
    context_length=8000,  # Context window size
)

# Manual truncation
agent.truncate_history(max_messages=50)

Connection Pooling

from concurrent.futures import ThreadPoolExecutor

# Use thread pool for concurrent requests
executor = ThreadPoolExecutor(max_workers=10)

def process_tasks(tasks):
    futures = []
    for task in tasks:
        future = executor.submit(agent.run, task)
        futures.append(future)
    
    return [f.result() for f in futures]

State Management

Autosave

agent = Agent(
    agent_name="Stateful-Agent",
    model_name="claude-sonnet-4-6",
    autosave=True,  # Auto-save state after each run
    saved_state_path="./agent_states/",
)

Manual State Management

# Save state
agent.save_state()

# Load state
from swarms import Agent

agent = Agent(
    agent_name="Restored-Agent",
    model_name="claude-sonnet-4-6",
    load_state_path="./agent_states/agent_state.json",
)

Testing

Unit Testing

import unittest
from swarms import Agent

class TestAgent(unittest.TestCase):
    def setUp(self):
        self.agent = Agent(
            agent_name="Test-Agent",
            model_name="gpt-5.4",
            max_loops=1,
        )
    
    def test_basic_run(self):
        result = self.agent.run("Say hello")
        self.assertIsNotNone(result)
        self.assertIn("hello", result.lower())
    
    def test_error_handling(self):
        with self.assertRaises(AgentRunError):
            self.agent.run("")  # Empty task should fail

Integration Testing

def test_agent_integration():
    """Test agent with real LLM."""
    agent = Agent(
        agent_name="Integration-Test-Agent",
        model_name="gpt-5.4",
        max_loops=2,
    )
    
    # Test with various inputs
    test_cases = [
        "Simple task",
        "Multi-step reasoning task",
        "Task requiring tool use",
    ]
    
    for task in test_cases:
        result = agent.run(task)
        assert result is not None
        assert len(result) > 0

​Overview

​Error Handling

​Agent Error Hierarchy

​Graceful Error Recovery

​Retry Strategies

​Built-in Retry Configuration

​Exponential Backoff

​Fallback Models

​Logging

​Structured Logging with Loguru

​Agent-Specific Logging

​Contextual Logging

​Monitoring

​Performance Metrics

​Health Checks

​Configuration Management

​Environment-Based Configuration

​Configuration Validation

​Security Best Practices

​API Key Management

​Input Validation

​Safety Prompts

​Resource Management

​Memory Management

​Connection Pooling

​State Management

​Autosave

​Manual State Management

​Testing

​Unit Testing

​Integration Testing

​Related Resources