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 defines an exception hierarchy under AgentError. Every class is exported from the top-level swarms package:

Exception	When `Agent` raises it
`AgentInitializationError`	The constructor rejects the configuration, for example `max_loops` that is neither a positive integer nor `"auto"`
`AgentLLMError`	Every one of the `retry_attempts` LLM calls failed, on the primary model and on every fallback model
`AgentToolExecutionError`	A tool batch failed `tool_retry_attempts` times. The agent catches this itself and shows the failure to the model, so it does not escape `run()`
`AgentRunError`, `AgentLLMInitializationError`	Part of the hierarchy for your own code; the `Agent` run path does not raise them

run() raises ValueError before any LLM call when the task is empty and interactive is False.

from loguru import logger
from swarms import Agent, AgentInitializationError, AgentLLMError

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"Invalid agent configuration: {e}")
except AgentLLMError as e:
    # The original provider error is chained as e.__cause__
    logger.error(f"Every model and retry failed: {e}")

In 15.0.3 and earlier, an agent whose retries were all exhausted logged the failure and returned an empty string, and fallback_models were never tried. run() now raises AgentLLMError, after first handing the task to each fallback model. The agent also removes the turns the failed run added to its memory, so a fallback model does not see the task twice.

Graceful Error Recovery

import time
import traceback

from loguru import logger
from swarms import AgentLLMError


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 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,        # LLM call attempts per loop (default 3)
    tool_retry_attempts=3,   # attempts per tool batch (default 3)
    max_loops=3,
)

Retries run back to back, with no delay between attempts. Wrap run() in your own backoff, as below, when you need to wait out rate limits.

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-5.4-mini",    # Second fallback
    ],
    max_loops=3,
)

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

Each model gets the full retry_attempts before the agent moves to the next one. When the last model fails, run() raises AgentLLMError.

Fallback swarms

SwarmRouter can fall back across swarm architectures. When the primary swarm_type raises, during construction or during run(), the router builds the next type in fallback_swarms from the same agents and configuration and gives it the same task. The first one to complete wins:

from swarms import Agent, SwarmRouter
from swarms.structs.swarm_router import SwarmRouterRunError

agents = [
    Agent(agent_name="Researcher", model_name="gpt-5.4", max_loops=1),
    Agent(agent_name="Analyst", model_name="gpt-5.4", max_loops=1),
    Agent(agent_name="Writer", model_name="gpt-5.4", max_loops=1),
]

router = SwarmRouter(
    agents=agents,
    swarm_type="HierarchicalSwarm",
    fallback_swarms=["SequentialWorkflow", "ConcurrentWorkflow"],
)

try:
    result = router.run("Assess whether we should adopt usage-based pricing.")
    print(f"Served by: {router.active_swarm_type}")
    for attempt in router.fallback_attempts:
        print(f"Failed first: {attempt['swarm_type']} -> {attempt['error']}")
except SwarmRouterRunError as e:
    print(f"Every swarm type failed: {e}")

Without fallback_swarms, the original exception propagates unchanged. Including "AgentRearrange" in the list requires rearrange_flow. See SwarmRouter for details.

Logging

Structured Logging with Loguru

Swarms uses Loguru. Out of the box it logs to the console and to {WORKSPACE_DIR}/logs/: a combined swarms_YYYY-MM-DD.log rotated daily and kept for 10 days, plus one <module>.log per Swarms module that logs. Calling logger.remove() drops those handlers too, so add back any file sinks you still want:

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
)

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

Tracing

Swarms emits OpenTelemetry spans for every agent and swarm run by default. See Telemetry for what is captured, where it is sent, and how to opt out with SWARMS_TELEMETRY_ON=false.

Token usage

Every Agent keeps a running total of the token counts its provider reports, and SwarmRouter sums them across its agents:

result = agent.run("Summarize the quarterly report.")
print(agent.usage)
# {'input_tokens': ..., 'output_tokens': ..., 'cached_tokens': ..., 'reasoning_tokens': ..., 'total_tokens': ...}

router.run("Assess the market.")
print(router.usage)

Totals are lifetime totals for the instance, so record the difference between two readings to cost a single request. See Token usage for the structures that report usage and how streaming calls are counted.

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 a lightweight agent configured with max_loops=1.
    """
    try:
        agent.run("Reply with OK.")
        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")),
    "verbose": os.getenv("VERBOSE", "false").lower() == "true",
}

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

Configuration Validation

from pydantic import BaseModel, Field, 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=1, le=10)
    
    @field_validator('model_name')
    @classmethod
    def validate_model(cls, v):
        allowed_models = ["claude-sonnet-4-6", "gpt-5.4", "gpt-5.4-mini"]
        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.model_dump())

Security Best Practices

API Key Management

import os

from swarms import Agent

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

# Pass a key explicitly when it does not come from the provider's standard env var
agent = Agent(
    model_name="claude-sonnet-4-6",
    llm_api_key=ANTHROPIC_API_KEY,
)

When the provider’s standard variable (ANTHROPIC_API_KEY, OPENAI_API_KEY, …) is set, you can omit llm_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

# Bound the context window and let the agent auto-compress near the limit
agent = Agent(
    agent_name="Memory-Managed-Agent",
    model_name="claude-sonnet-4-6",
    context_length=8000,       # Context window size
    context_compression=True,  # Auto-summarize when nearing the limit (default)
)

# Manual compaction: collapse history into a single summary message.
# `agent.short_memory` is the agent's underlying Conversation object.
agent.short_memory.compact(summary="Summary of the conversation so far.")

Concurrent Requests

An Agent instance keeps one conversation. If several threads call run() on the same instance, their turns land in the same history. For independent requests, give each request its own agent:

from concurrent.futures import ThreadPoolExecutor

from swarms import Agent


def make_agent() -> Agent:
    return Agent(
        agent_name="Request-Worker",
        model_name="claude-sonnet-4-6",
        max_loops=1,
        print_on=False,
    )


executor = ThreadPoolExecutor(max_workers=10)

def process_tasks(tasks):
    futures = [executor.submit(make_agent().run, task) for task in tasks]
    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="stateful_agent.json",
)

A relative saved_state_path is written inside the agent’s own workspace directory, {WORKSPACE_DIR}/agents/<agent-name>-<id>/. An absolute path is used as given.

Manual State Management

from swarms import Agent

# Save state to an explicit path
agent.save("/var/lib/myapp/agent_states/agent_state.json")

# Restore it into a new agent
restored = Agent(
    agent_name="Restored-Agent",
    model_name="claude-sonnet-4-6",
)
restored.load("/var/lib/myapp/agent_states/agent_state.json")

load() opens the path exactly as given, so use absolute paths for state you plan to restore. Passing load_state_path to the constructor only sets the default path for a later load() call; it does not load anything by itself.

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_empty_task(self):
        with self.assertRaises(ValueError):
            self.agent.run("")  # Rejected before any LLM call

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

​Fallback swarms

​Logging

​Structured Logging with Loguru

​Agent-Specific Logging

​Contextual Logging

​Monitoring

​Tracing

​Token usage

​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

​Concurrent Requests

​State Management

​Autosave

​Manual State Management

​Testing

​Unit Testing

​Integration Testing

​Related Resources

Overview

Error Handling

Agent Error Hierarchy

Graceful Error Recovery

Retry Strategies

Built-in Retry Configuration

Exponential Backoff

Fallback Models

Fallback swarms

Logging

Structured Logging with Loguru

Agent-Specific Logging

Contextual Logging

Monitoring

Tracing

Token usage

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

Concurrent Requests

State Management

Autosave

Manual State Management

Testing

Unit Testing

Integration Testing

Related Resources