DEV Community: biao lin

Monitoring and Observability for Autonomous AI Systems

biao lin — Thu, 18 Jun 2026 01:38:17 +0000

Monitoring and Observability for Autonomous AI Systems

Autonomous AI systems—from self-driving cars to algorithmic trading bots and robotic process automation (RPA) agents—operate with minimal human intervention. When they fail, the consequences can be catastrophic. Traditional monitoring (checking if a service is up) is insufficient. We need observability: the ability to ask arbitrary questions about a system's internal state based on its external outputs.

In this post, we'll explore how to implement monitoring and observability for autonomous AI systems, covering metrics, logging, and dashboards with concrete code examples.

Why Observability Matters for Autonomous AI

Autonomous systems exhibit non-deterministic behavior. Unlike a web server that either returns 200 or 500, an AI agent might make a series of "correct" decisions that collectively lead to a suboptimal outcome. Key challenges include:

Hidden state drift: Model performance degrades over time due to data drift.
Decision lineage: Understanding why an agent took a particular action.
Anomaly detection: Identifying when the system enters an unknown state.

Observability provides three pillars: metrics (quantitative measurements), logs (structured events), and traces (request flow across components).

1. Instrumenting Metrics

Metrics are numeric aggregations. For autonomous systems, we need both technical metrics (CPU, memory) and business metrics (success rate, action latency).

Prometheus-Compatible Metrics

Here's a Python example using the prometheus_client library to instrument an AI decision engine:

# metrics.py
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import random
import time

# Define metrics
decisions_total = Counter('ai_decisions_total', 'Total decisions made', ['model_version', 'action_type'])
decision_latency = Histogram('ai_decision_latency_seconds', 'Decision latency in seconds', 
                             buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 5.0])
model_confidence = Gauge('ai_model_confidence', 'Current model confidence score', ['model_id'])
error_rate = Gauge('ai_error_rate', 'Rolling error rate (last 100 decisions)')

# Rolling window for error rate
errors = []
MAX_WINDOW = 100

def make_decision(model_version, action_type):
    start = time.time()

    # Simulate decision making
    is_error = random.random() < 0.05  # 5% error rate
    confidence = random.uniform(0.7, 0.99)

    # Record metrics
    decisions_total.labels(model_version=model_version, action_type=action_type).inc()
    decision_latency.observe(time.time() - start)
    model_confidence.labels(model_id='ensemble-v3').set(confidence)

    # Update error rate
    errors.append(1 if is_error else 0)
    if len(errors) > MAX_WINDOW:
        errors.pop(0)
    error_rate.set(sum(errors) / len(errors))

    return is_error

if __name__ == '__main__':
    start_http_server(8000)  # Expose metrics at /metrics
    while True:
        make_decision('v3.1', random.choice(['navigate', 'grasp', 'inspect']))
        time.sleep(0.1)

Key metrics to track:

ai_decisions_total: Volume of decisions broken down by type.
ai_decision_latency_seconds: Performance degradation detection.
ai_model_confidence: Track when models become uncertain.
ai_error_rate: Rolling window alerts on anomaly spikes.

2. Structured Logging

Logs must be machine-parseable and correlated. Avoid free-text messages; use structured JSON with consistent fields.

Python Logger with JSON Output

# logging_config.py
import logging
import json
import sys
from datetime import datetime

class StructuredLogger:
    def __init__(self, name, log_level=logging.INFO):
        self.logger = logging.getLogger(name)
        self.logger.setLevel(log_level)

        handler = logging.StreamHandler(sys.stdout)
        formatter = logging.Formatter('%(message)s')
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)

    def _log(self, level, message, **kwargs):
        record = {
            'timestamp': datetime.utcnow().isoformat() + 'Z',
            'level': level,
            'logger': self.logger.name,
            'message': message,
            **kwargs
        }
        self.logger.log(getattr(logging, level), json.dumps(record))

    def info(self, message, **kwargs):
        self._log('INFO', message, **kwargs)

    def error(self, message, **kwargs):
        self._log('ERROR', message, **kwargs)

    def warn(self, message, **kwargs):
        self._log('WARNING', message, **kwargs)

# Usage in autonomous agent
logger = StructuredLogger('autonomous_agent')

def process_observation(observation):
    logger.info('Processing observation', 
                observation_id=observation['id'],
                sensor_type=observation['sensor'],
                timestamp=observation['timestamp'])

    if observation['confidence'] < 0.5:
        logger.warn('Low confidence observation', 
                   confidence=observation['confidence'],
                   observation_id=observation['id'])

    try:
        result = ai_model.predict(observation)
        logger.info('Prediction made', 
                   prediction=result['action'],
                   probability=result['probability'])
        return result
    except Exception as e:
        logger.error('Prediction failed',
                    error=str(e),
                    observation_id=observation['id'],
                    exception_type=type(e).__name__)
        raise

Important fields for AI systems:

decision_id: Traceability across components.
model_version: Which model made the decision.
input_hash: Reproduce inputs later.
confidence: Model's certainty.
context: Environment state (temperature, traffic, etc.).

3. Building the Dashboard

A Grafana dashboard provides real-time visibility. Here's a JSON model for a dashboard focused on autonomous agent health:

{
  "dashboard": {
    "title": "Autonomous AI System Overview",
    "panels": [
      {
        "title": "Decision Rate",
        "type": "graph",
        "targets": [
          {
            "expr": "rate(ai_decisions_total[5m])",
            "legendFormat": "{{action_type}}"
          }
        ],
        "gridPos": {"h": 8, "w": 12, "x": 0, "y": 0}
      },
      {
        "title": "Decision Latency (p99)",
        "type": "heatmap",
        "targets": [
          {
            "expr": "histogram_quantile(0.99, sum(rate(ai_decision_latency_seconds_bucket[5m])) by (le))",
            "legendFormat": "p99"
          }
        ],
        "gridPos": {"h": 8, "w": 12, "x": 12, "y": 0}
      },
      {
        "title": "Model Confidence Distribution",
        "type": "stat",
        "targets": [
          {
            "expr": "avg(ai_model_confidence) by (model_id)",
            "legendFormat": "{{model_id}}"
          }
        ],
        "gridPos": {"h": 6, "w": 8, "x": 0, "y": 8}
      },
      {
        "title": "Error Rate (Rolling 100)",
        "type": "graph",
        "targets": [
          {
            "expr": "ai_error_rate",
            "legendFormat": "Error Rate"
          }
        ],
        "gridPos": {"h": 6, "w": 8, "x": 8, "y": 8}
      },
      {
        "title": "Recent Errors",
        "type": "logs",
        "targets": [
          {
            "expr": "{logger=\"autonomous_agent\", level=\"ERROR\"} | json",
            "refId": "A"
          }
        ],
        "gridPos": {"h": 6, "w": 8, "x": 16, "y": 8}
      }
    ]
  }
}

Alerting Rules (Prometheus)


yaml
# alerts.yml
groups:
  - name: autonomous_ai
    rules:
      - alert: HighErrorRate
        expr: ai_error_rate > 0.2
        for: 5m
        annotations:
          summary: "Error rate exceeding 20% for 5 minutes"
          description: "Current error rate: {{ $value | humanizePercentage }}"

      - alert: ModelConfidenceDrop
        expr: avg(ai_model_confidence) < 0.6
        annotations:
          summary: "Average model confidence dropped below 60%"

      - alert: LatencySpike
        expr: histogram_quantile(0.99, rate(ai_

Building a Plug-and-Play Agent Architecture with Python

biao lin — Thu, 18 Jun 2026 01:36:24 +0000

Building a Plug-and-Play Agent Architecture with Python

Modern software systems increasingly rely on intelligent agents that can perceive their environment, make decisions, and execute actions. However, building a monolithic agent system often leads to tight coupling, poor maintainability, and limited extensibility. In this post, we'll design a plug-and-play agent architecture using Python that allows you to hot-swap components, add new capabilities without modifying core code, and maintain clean separation of concerns.

Why Plug-and-Play?

Traditional agent architectures suffer from several pain points:

Rigid component relationships: Changing one part often requires changes throughout the system
Difficult testing: Hard to isolate and test individual components
Poor reusability: Components are tightly coupled to specific applications
Limited extensibility: Adding new capabilities requires modifying existing code

A plug-and-play architecture solves these by defining clear interfaces, using dependency injection, and implementing a registry pattern for dynamic component discovery.

Core Architecture Overview

Our architecture consists of three main layers:

Agent Registry: Central catalog of available components
Plugin System: Mechanism for loading and integrating components
Agent Core: Runtime engine that orchestrates components

Let's implement each layer with practical code examples.

1. The Agent Registry

The registry serves as a service locator for all agent components. It provides lookup, registration, and lifecycle management.

# agent_registry.py
from typing import Dict, Type, Any, Optional, List
import inspect
import logging

logger = logging.getLogger(__name__)

class AgentRegistry:
    """Central registry for agent components."""

    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._components: Dict[str, Dict[str, Any]] = {}
            cls._instance._plugins: Dict[str, Type] = {}
        return cls._instance

    def register_component(self, name: str, component: Any, 
                          component_type: str, metadata: Optional[Dict] = None):
        """Register a component with the registry."""
        self._components[name] = {
            'instance': component,
            'type': component_type,
            'metadata': metadata or {}
        }
        logger.info(f"Registered component '{name}' of type '{component_type}'")

    def get_component(self, name: str) -> Optional[Any]:
        """Retrieve a registered component by name."""
        if name in self._components:
            return self._components[name]['instance']
        return None

    def get_components_by_type(self, component_type: str) -> List[Dict]:
        """Get all components of a specific type."""
        return [
            {name: info} for name, info in self._components.items()
            if info['type'] == component_type
        ]

    def unregister_component(self, name: str):
        """Remove a component from the registry."""
        if name in self._components:
            del self._components[name]
            logger.info(f"Unregistered component '{name}'")

    def list_components(self) -> List[str]:
        """List all registered component names."""
        return list(self._components.keys())

    def register_plugin(self, plugin_class: Type):
        """Register a plugin class for dynamic loading."""
        if hasattr(plugin_class, 'plugin_name'):
            name = plugin_class.plugin_name
        else:
            name = plugin_class.__name__
        self._plugins[name] = plugin_class
        logger.info(f"Registered plugin '{name}'")

    def get_plugin(self, name: str) -> Optional[Type]:
        """Get a registered plugin class."""
        return self._plugins.get(name)

2. The Plugin System

The plugin system handles dynamic loading of components from external modules, enabling runtime extensibility.

# plugin_system.py
import importlib
import pkgutil
import inspect
import logging
from typing import List, Type, Optional
from pathlib import Path

logger = logging.getLogger(__name__)

class PluginSystem:
    """Manages dynamic plugin loading and lifecycle."""

    def __init__(self, registry: 'AgentRegistry'):
        self.registry = registry
        self.loaded_plugins = {}

    def discover_plugins(self, package_path: str) -> List[str]:
        """Discover available plugins in a package."""
        discovered = []
        try:
            package = importlib.import_module(package_path)
            for importer, modname, ispkg in pkgutil.iter_modules(package.__path__):
                if ispkg:  # Only load packages (not individual modules)
                    discovered.append(f"{package_path}.{modname}")
            logger.info(f"Discovered {len(discovered)} potential plugins")
        except ImportError as e:
            logger.error(f"Failed to discover plugins: {e}")
        return discovered

    def load_plugin(self, plugin_module_path: str) -> Optional[Type]:
        """Load a plugin from a module path."""
        try:
            module = importlib.import_module(plugin_module_path)

            # Find plugin classes (classes that inherit from BasePlugin)
            for name, obj in inspect.getmembers(module):
                if (inspect.isclass(obj) and 
                    issubclass(obj, BasePlugin) and 
                    obj != BasePlugin):
                    self.registry.register_plugin(obj)
                    self.loaded_plugins[plugin_module_path] = obj
                    logger.info(f"Loaded plugin '{name}' from {plugin_module_path}")
                    return obj
            logger.warning(f"No plugin class found in {plugin_module_path}")
        except ImportError as e:
            logger.error(f"Failed to load plugin {plugin_module_path}: {e}")
        return None

    def load_all_plugins(self, package_path: str) -> int:
        """Discover and load all plugins in a package."""
        count = 0
        plugin_paths = self.discover_plugins(package_path)
        for path in plugin_paths:
            if self.load_plugin(path):
                count += 1
        logger.info(f"Loaded {count} plugins from {package_path}")
        return count

    def unload_plugin(self, plugin_module_path: str):
        """Unload a previously loaded plugin."""
        if plugin_module_path in self.loaded_plugins:
            plugin_class = self.loaded_plugins[plugin_module_path]
            # Clean up any registered components from this plugin
            for name, info in list(self.registry._components.items()):
                if hasattr(info['instance'], '__class__'):
                    if info['instance'].__class__ == plugin_class:
                        self.registry.unregister_component(name)
            del self.loaded_plugins[plugin_module_path]
            logger.info(f"Unloaded plugin from {plugin_module_path}")

3. Base Plugin Interface

All plugins must inherit from a base class that defines the contract.

# base_plugin.py
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional

class BasePlugin(ABC):
    """Abstract base class for all plugins."""

    plugin_name: str = None
    plugin_version: str = "1.0.0"
    plugin_description: str = ""

    def __init__(self, config: Optional[Dict[str, Any]] = None):
        self.config = config or {}
        self.initialized = False

    @abstractmethod
    def initialize(self) -> bool:
        """Initialize the plugin. Return True if successful."""
        pass

    @abstractmethod
    def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        """Execute the plugin's main functionality."""
        pass

    def shutdown(self):
        """Cleanup resources when plugin is unloaded."""
        self.initialized = False

    def get_info(self) -> Dict[str, Any]:
        """Return plugin metadata."""
        return {
            'name': self.plugin_name or self.__class__.__name__,
            'version': self.plugin_version,
            'description': self.plugin_description,
            'initialized': self.initialized
        }

4. Example Plugins

Let's create some practical plugins to demonstrate the system.

# plugins/text_processor/__init__.py
from base_plugin import BasePlugin
from typing import Dict, Any

class TextProcessorPlugin(BasePlugin):
    """Plugin for text processing operations."""

    plugin_name = "Text Processor"
    plugin_version = "1.0.0"
    plugin_description = "Provides text analysis and transformation capabilities"

    def initialize(self) -> bool:
        self.initialized = True
        return True

    def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        text = context.get('text', '')
        operation = context.get('operation', 'word_count')

        if operation == 'word_count':
            result = len(text.split())
        elif operation == 'char_count':
            result = len(text)
        elif operation == 'reverse':
            result = text[::-1]
        elif operation == 'uppercase':
            result = text.upper()
        else:
            result = f"Unknown operation: {operation}"

        return {'result': result, 'operation': operation, 'plugin': self.plugin_name}


python
# plugins/data_validator/__init__.py
from base_plugin import

Python Data Validation Patterns for Production Systems

biao lin — Wed, 17 Jun 2026 16:05:54 +0000

Python Data Validation Patterns for Production Systems

Data validation is a critical component of any production system. Invalid data can cause silent failures, corrupt databases, create security vulnerabilities, and lead to poor user experiences. In this post, we'll explore battle-tested patterns for validating data in Python production systems, focusing on practical implementations that scale.

Why Data Validation Matters

Consider a simple e-commerce checkout system. A user submits an order with a negative quantity, or a product ID that doesn't exist. Without proper validation, these inputs could:

Crash your application
Create negative inventory records
Allow price manipulation attacks
Cause downstream data processing failures

Let's build a robust validation framework that prevents these issues.

Pattern 1: Functional Validators

The simplest pattern uses pure functions for validation. Each validator takes a value and returns either the validated value or raises an exception.

from typing import Any, Callable, TypeVar

T = TypeVar('T')

def validate_email(email: str) -> str:
    """Validate email format."""
    import re
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    if not re.match(pattern, email):
        raise ValueError(f"Invalid email format: {email}")
    return email

def validate_positive_int(value: Any) -> int:
    """Validate positive integer."""
    if not isinstance(value, int):
        raise TypeError(f"Expected int, got {type(value).__name__}")
    if value <= 0:
        raise ValueError(f"Value must be positive, got {value}")
    return value

def validate_non_empty_string(value: Any) -> str:
    """Validate non-empty string."""
    if not isinstance(value, str):
        raise TypeError(f"Expected string, got {type(value).__name__}")
    if not value.strip():
        raise ValueError("String cannot be empty")
    return value

Pattern 2: Schema-Based Validation

For complex data structures, schema-based validation is more maintainable. Here's a lightweight schema system:

from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field

@dataclass
class Field:
    type: type
    required: bool = True
    validators: List[Callable] = field(default_factory=list)
    default: Any = None

class Schema:
    def __init__(self, fields: Dict[str, Field]):
        self.fields = fields

    def validate(self, data: Dict[str, Any]) -> Dict[str, Any]:
        validated = {}
        errors = {}

        for field_name, field_spec in self.fields.items():
            value = data.get(field_name)

            # Check required fields
            if field_spec.required and value is None:
                errors[field_name] = "Field is required"
                continue

            # Set default if not provided
            if value is None and field_spec.default is not None:
                value = field_spec.default

            # Type validation
            if value is not None and not isinstance(value, field_spec.type):
                errors[field_name] = f"Expected {field_spec.type.__name__}, got {type(value).__name__}"
                continue

            # Custom validators
            for validator in field_spec.validators:
                try:
                    value = validator(value)
                except (ValueError, TypeError) as e:
                    errors[field_name] = str(e)
                    break
            else:
                validated[field_name] = value

        if errors:
            raise ValidationError(errors)

        return validated

class ValidationError(Exception):
    def __init__(self, errors: Dict[str, str]):
        self.errors = errors
        super().__init__(str(errors))

Usage Example

# Define validators
def validate_phone(phone: str) -> str:
    if not phone.startswith('+') or len(phone) < 10:
        raise ValueError("Invalid phone number format")
    return phone

# Create schema
user_schema = Schema({
    'name': Field(str, validators=[validate_non_empty_string]),
    'email': Field(str, validators=[validate_email]),
    'age': Field(int, validators=[validate_positive_int]),
    'phone': Field(str, required=False, validators=[validate_phone]),
    'role': Field(str, default='user')
})

# Validate data
user_data = {
    'name': 'Alice',
    'email': 'alice@example.com',
    'age': 30,
    'phone': '+1234567890'
}

try:
    validated_user = user_schema.validate(user_data)
    print(f"Validated user: {validated_user}")
except ValidationError as e:
    print(f"Validation failed: {e.errors}")

Pattern 3: Pydantic Integration

For production systems, I recommend using Pydantic for complex validation. It provides powerful built-in validators and serialization:

from pydantic import BaseModel, Field, validator
from typing import Optional
from datetime import datetime
import re

class ProductCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    price: float = Field(..., gt=0)
    sku: str = Field(..., regex=r'^[A-Z]{2}-\d{4}$')
    tags: list[str] = Field(default_factory=list)
    created_at: Optional[datetime] = None

    @validator('name')
    def name_must_be_meaningful(cls, v):
        if not re.search(r'[a-zA-Z]', v):
            raise ValueError('Name must contain at least one letter')
        return v.strip()

    @validator('price')
    def price_must_have_two_decimals(cls, v):
        if round(v, 2) != v:
            raise ValueError('Price must have at most 2 decimal places')
        return v

# Usage
try:
    product = ProductCreate(
        name='  Wireless Mouse  ',
        price=29.99,
        sku='WM-001',
        tags=['electronics', 'accessories']
    )
    print(f"Validated product: {product.dict()}")
except Exception as e:
    print(f"Validation error: {e}")

Pattern 4: Validation Pipeline

For complex workflows, chain validators together in a pipeline:

from typing import List, Tuple, Any

class ValidationPipeline:
    def __init__(self):
        self.steps: List[Tuple[str, Callable]] = []

    def add_step(self, name: str, validator: Callable):
        self.steps.append((name, validator))

    def process(self, data: Any) -> Any:
        for step_name, validator in self.steps:
            try:
                data = validator(data)
                print(f"Step '{step_name}' passed")
            except (ValueError, TypeError) as e:
                raise RuntimeError(f"Validation failed at step '{step_name}': {e}")
        return data

# Example pipeline
pipeline = ValidationPipeline()
pipeline.add_step("sanitize", lambda x: x.strip())
pipeline.add_step("email_format", validate_email)
pipeline.add_step("check_domain", lambda email: email if 'company.com' in email else (_ for _ in ()).throw(ValueError("Not company email")))

# Usage
try:
    result = pipeline.process("  user@company.com  ")
    print(f"Pipeline result: {result}")
except RuntimeError as e:
    print(f"Pipeline failed: {e}")

Pattern 5: Async Validation

For I/O-bound validation (database lookups, API calls), use async validators:

import asyncio
from typing import Any, Awaitable

class AsyncValidator:
    def __init__(self):
        self.validators: list[Awaitable] = []

    async def validate(self, data: Any) -> Any:
        results = await asyncio.gather(
            *[v(data) for v in self.validators],
            return_exceptions=True
        )

        errors = []
        for result in results:
            if isinstance(result, Exception):
                errors.append(str(result))

        if errors:
            raise ValidationError({"async": errors})
        return data

# Example async validators
async def check_user_exists(user_id: int) -> int:
    # Simulate DB lookup
    await asyncio.sleep(0.1)
    if user_id == 999:
        raise ValueError(f"User {user_id} not found")
    return user_id

async def check_rate_limit(user_id: int) -> int:
    # Simulate rate limit check
    await asyncio.sleep(0.05)
    if user_id == 500:
        raise ValueError("Rate limit exceeded")
    return user_id

# Usage
async def main():
    validator = AsyncValidator()
    validator.validators = [check_user_exists, check_rate_limit]

    try:
        await validator.validate(123)
        print("Async validation passed")
    except ValidationError as e:
        print(f"Async validation failed: {e}")

# asyncio.run(main())

Error Handling Best Practices

**Be explicit about errors

Implementing a Self-Healing Distributed System with Python

biao lin — Wed, 17 Jun 2026 16:03:34 +0000

Implementing a Self-Healing Distributed System with Python

Distributed systems are powerful, but they come with inherent complexities: network partitions, node failures, and resource exhaustion can bring down even the most carefully designed architectures. A self-healing system automatically detects failures and recovers without human intervention, ensuring high availability and resilience. In this post, we'll build a practical self-healing distributed system in Python, complete with health checks, auto-restart mechanisms, and fallback strategies.

The Core Principles

A self-healing system revolves around three key mechanisms:

Health Monitoring: Continuously check if components are alive and responsive.
Automated Recovery: Restart or replace failed components without manual intervention.
Graceful Degradation: When primary services fail, fall back to secondary services to maintain partial functionality.

We'll implement these using asyncio for concurrency, aiohttp for HTTP-based health checks, and Python's subprocess module for managing child processes.

System Architecture

Our system will consist of:

A Service Manager that orchestrates multiple worker processes.
Worker Services that perform actual work (simulated with HTTP servers).
A Health Checker that probes workers periodically.
A Fallback Service that kicks in when primary workers fail.

+------------------+       +------------------+
|  Service Manager |       |  Health Checker  |
|  (orchestrator)  |<----->|  (monitoring)    |
+--------+---------+       +--------+---------+
         |                           |
         v                           v
+------------------+       +------------------+
|  Worker Process 1|       |  Worker Process 2|
|  (primary)       |       |  (fallback)      |
+------------------+       +------------------+

Implementation

1. The Worker Service

First, let's create a simple HTTP-based worker that simulates occasional failures.

# worker.py
import asyncio
import random
from aiohttp import web

async def handle_health(request):
    """Health check endpoint"""
    # Simulate random failure (20% chance)
    if random.random() < 0.2:
        return web.Response(status=500, text="Internal Error")
    return web.json_response({"status": "healthy"})

async def handle_work(request):
    """Main work endpoint"""
    await asyncio.sleep(0.1)  # Simulate processing
    return web.json_response({"result": "work_done", "worker_id": request.app['worker_id']})

def create_app(worker_id, port):
    app = web.Application()
    app['worker_id'] = worker_id
    app.router.add_get('/health', handle_health)
    app.router.add_get('/work', handle_work)
    return app

if __name__ == '__main__':
    import sys
    worker_id = sys.argv[1] if len(sys.argv) > 1 else 'unknown'
    port = int(sys.argv[2]) if len(sys.argv) > 2 else 8080
    web.run_app(create_app(worker_id, port), port=port)

2. The Service Manager

The manager is the brain of our self-healing system. It spawns workers, monitors them, and takes corrective action.

# service_manager.py
import asyncio
import subprocess
import signal
import time
from typing import Dict, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta

@dataclass
class WorkerInstance:
    process: subprocess.Popen
    port: int
    worker_id: str
    last_healthy: datetime
    restart_count: int = 0

class ServiceManager:
    def __init__(self):
        self.workers: Dict[str, WorkerInstance] = {}
        self.next_worker_id = 0
        self.base_port = 9000
        self.max_restarts = 3
        self.restart_window = timedelta(minutes=5)

    async def spawn_worker(self) -> WorkerInstance:
        """Start a new worker process"""
        worker_id = f"worker-{self.next_worker_id}"
        self.next_worker_id += 1
        port = self.base_port + len(self.workers)

        print(f"[Manager] Spawning {worker_id} on port {port}")

        process = subprocess.Popen(
            ['python', 'worker.py', worker_id, str(port)],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )

        instance = WorkerInstance(
            process=process,
            port=port,
            worker_id=worker_id,
            last_healthy=datetime.now()
        )
        self.workers[worker_id] = instance

        # Give worker time to start
        await asyncio.sleep(1)
        return instance

    async def check_worker_health(self, worker_id: str) -> bool:
        """Check if a worker is healthy via HTTP"""
        import aiohttp

        worker = self.workers.get(worker_id)
        if not worker:
            return False

        try:
            async with aiohttp.ClientSession() as session:
                url = f"http://localhost:{worker.port}/health"
                async with session.get(url, timeout=2) as response:
                    if response.status == 200:
                        worker.last_healthy = datetime.now()
                        return True
                    else:
                        print(f"[Health] {worker_id} returned status {response.status}")
                        return False
        except Exception as e:
            print(f"[Health] {worker_id} check failed: {e}")
            return False

    def should_restart(self, worker_id: str) -> bool:
        """Check if worker hasn't exceeded restart limits"""
        worker = self.workers.get(worker_id)
        if not worker:
            return False

        # Reset count if outside window
        now = datetime.now()
        if now - worker.last_healthy > self.restart_window:
            worker.restart_count = 0

        return worker.restart_count < self.max_restarts

    async def restart_worker(self, worker_id: str):
        """Restart a failed worker"""
        worker = self.workers.get(worker_id)
        if not worker:
            return

        print(f"[Manager] Restarting {worker_id} (attempt {worker.restart_count + 1})")

        # Kill old process
        worker.process.terminate()
        try:
            worker.process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            worker.process.kill()
            worker.process.wait()

        # Spawn new one
        new_worker = await self.spawn_worker()
        new_worker.restart_count = worker.restart_count + 1
        self.workers[worker_id] = new_worker

    async def health_monitor_loop(self):
        """Continuously monitor all workers"""
        while True:
            for worker_id in list(self.workers.keys()):
                is_healthy = await self.check_worker_health(worker_id)

                if not is_healthy:
                    if self.should_restart(worker_id):
                        await self.restart_worker(worker_id)
                    else:
                        print(f"[Manager] {worker_id} exceeded restart limit. Escalating...")
                        # In production, this would trigger alerts or scale up fallback
                        await self.activate_fallback(worker_id)

            await asyncio.sleep(5)  # Check every 5 seconds

    async def activate_fallback(self, failed_worker_id: str):
        """Activate fallback mechanism"""
        print(f"[Fallback] Activating fallback for {failed_worker_id}")

        # Simulate fallback: create a new worker with different config
        fallback_id = f"fallback-{self.next_worker_id}"
        self.next_worker_id += 1
        port = self.base_port + len(self.workers) + 100  # Different port range

        process = subprocess.Popen(
            ['python', 'worker.py', fallback_id, str(port)],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )

        instance = WorkerInstance(
            process=process,
            port=port,
            worker_id=fallback_id,
            last_healthy=datetime.now(),
            restart_count=0
        )
        self.workers[fallback_id] = instance
        print(f"[Fallback] {fallback_id} started on port {port}")

    async def run(self):
        """Main entry point"""
        # Start initial workers
        initial_workers = 3
        print(f"[Manager] Starting {initial_workers} workers...")

        for _ in range(initial_workers):
            await self.spawn_worker()

        # Start health monitoring
        await self.health_monitor_loop()

if __name__ == '__main__':
    manager = ServiceManager()
    try:
        asyncio.run(manager.run())
    except KeyboardInterrupt:
        print("[Manager] Shutting down...")
        for worker in manager.workers.values():
            worker.process.terminate()

3. The Client with Automatic Fallback

No self-healing system

Building an Autonomous AI Agent Colony with Python and FastAPI

biao lin — Wed, 17 Jun 2026 15:59:13 +0000

Building an Autonomous AI Agent Colony with Python and FastAPI

Introduction

Imagine a digital ecosystem where specialized AI agents collaborate, communicate, and self-heal without human intervention. This isn't science fiction — it's a practical architecture you can build today using Python and FastAPI. In this post, I'll walk you through creating an autonomous AI agent colony where agents work together, monitor each other, and recover from failures automatically.

Core Architecture

Our agent colony consists of three main components:

Agent Nodes - Specialized workers that perform specific tasks
API Server - Central communication hub built with FastAPI
Self-Healing Watcher - A monitoring system that ensures colony health

The magic happens when these components work together autonomously, creating a resilient system that can adapt to changing conditions.

Setting Up the API Server

Let's start with the heart of our colony — the FastAPI server that manages agent registration, communication, and task distribution.

from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing import Dict, List, Optional
import uvicorn
import asyncio
from datetime import datetime
import uuid

app = FastAPI(title="AI Agent Colony API")

# Data models
class AgentRegistration(BaseModel):
    name: str
    agent_type: str
    capabilities: List[str]
    endpoint: str

class AgentStatus(BaseModel):
    agent_id: str
    status: str
    last_heartbeat: datetime
    current_task: Optional[str] = None

class TaskAssignment(BaseModel):
    task_id: str
    task_type: str
    parameters: Dict
    priority: int = 1

# In-memory storage (use Redis for production)
agents: Dict[str, AgentRegistration] = {}
tasks: Dict[str, TaskAssignment] = {}
agent_statuses: Dict[str, AgentStatus] = {}

@app.post("/register_agent")
async def register_agent(agent: AgentRegistration):
    agent_id = str(uuid.uuid4())
    agents[agent_id] = agent
    agent_statuses[agent_id] = AgentStatus(
        agent_id=agent_id,
        status="idle",
        last_heartbeat=datetime.now()
    )
    return {"agent_id": agent_id, "status": "registered"}

@app.post("/heartbeat/{agent_id}")
async def heartbeat(agent_id: str):
    if agent_id not in agents:
        raise HTTPException(status_code=404, detail="Agent not found")
    agent_statuses[agent_id].last_heartbeat = datetime.now()
    agent_statuses[agent_id].status = "active"
    return {"status": "acknowledged"}

@app.get("/get_task/{agent_id}")
async def get_task(agent_id: str):
    if agent_id not in agents:
        raise HTTPException(status_code=404, detail="Agent not found")

    # Find highest priority task that matches agent capabilities
    agent = agents[agent_id]
    for task_id, task in tasks.items():
        if task.task_type in agent.capabilities and task.priority > 0:
            tasks[task_id].priority -= 1  # Decrease priority to balance load
            agent_statuses[agent_id].current_task = task_id
            return task

    return {"task_id": None, "message": "No tasks available"}

@app.post("/submit_result")
async def submit_result(task_id: str, result: Dict):
    if task_id not in tasks:
        raise HTTPException(status_code=404, detail="Task not found")
    # Process and store result
    return {"status": "completed", "task_id": task_id}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Building the Dashboard

A real-time dashboard helps visualize colony health. Here's a lightweight implementation using FastAPI's WebSocket support with a simple HTML frontend.

from fastapi import WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
import json

# WebSocket connection manager
class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            try:
                await connection.send_text(message)
            except:
                self.disconnect(connection)

manager = ConnectionManager()

@app.get("/dashboard")
async def get_dashboard():
    html_content = """
    <!DOCTYPE html>
    <html>
    <head>
        <title>Agent Colony Dashboard</title>
        <style>
            body { font-family: Arial, sans-serif; background: #1a1a2e; color: #eee; }
            .container { max-width: 1200px; margin: 0 auto; padding: 20px; }
            .agent-card { background: #16213e; border-radius: 8px; padding: 15px; margin: 10px; }
            .status-active { color: #00ff88; }
            .status-idle { color: #ffd700; }
            .status-error { color: #ff4444; }
            #metrics { display: grid; grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); gap: 20px; }
        </style>
    </head>
    <body>
        <div class="container">
            <h1>🤖 Agent Colony Dashboard</h1>
            <div id="metrics"></div>
            <h2>Active Agents</h2>
            <div id="agents"></div>
        </div>
        <script>
            const ws = new WebSocket('ws://localhost:8000/ws');
            ws.onmessage = function(event) {
                const data = JSON.parse(event.data);
                document.getElementById('agents').innerHTML = 
                    data.agents.map(agent => `
                        <div class="agent-card">
                            <h3>${agent.name}</h3>
                            <p>Status: <span class="status-${agent.status}">${agent.status}</span></p>
                            <p>Type: ${agent.type}</p>
                            <p>Last Task: ${agent.lastTask || 'None'}</p>
                        </div>
                    `).join('');

                document.getElementById('metrics').innerHTML = `
                    <div class="agent-card">
                        <h3>Total Agents</h3>
                        <p>${data.metrics.totalAgents}</p>
                    </div>
                    <div class="agent-card">
                        <h3>Active Tasks</h3>
                        <p>${data.metrics.activeTasks}</p>
                    </div>
                    <div class="agent-card">
                        <h3>System Health</h3>
                        <p>${data.metrics.health}</p>
                    </div>
                `;
            };
        </script>
    </body>
    </html>
    """
    return HTMLResponse(content=html_content)

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            # Send updated metrics every 2 seconds
            dashboard_data = {
                "agents": [
                    {
                        "name": agents[aid].name,
                        "status": agent_statuses[aid].status,
                        "type": agents[aid].agent_type,
                        "lastTask": agent_statuses[aid].current_task
                    }
                    for aid in agents
                ],
                "metrics": {
                    "totalAgents": len(agents),
                    "activeTasks": len([t for t in tasks.values() if t.priority > 0]),
                    "health": "Good" if len(agents) > 0 else "Critical"
                }
            }
            await manager.broadcast(json.dumps(dashboard_data))
            await asyncio.sleep(2)
    except WebSocketDisconnect:
        manager.disconnect(websocket)

The Self-Healing Watcher

This is where the true autonomy lives. The watcher monitors agent health and automatically spawns replacements when needed.


python
import aiohttp
import asyncio
from datetime import datetime, timedelta
import subprocess
import sys

class SelfHealingWatcher:
    def __init__(self, api_url: str, check_interval: int = 10):
        self.api_url = api_url
        self.check_interval = check_interval
        self.failed_agents = set()
        self.agent_templates = {
            "data_processor": {
                "script": "data_agent.py",
                "capabilities": ["data_processing", "analysis"]
            },
            "api_connector": {
                "script": "api_agent.py",
                "capabilities": ["api_calls", "web_scraping"]
            },
            "reporter": {
                "script": "report_agent.py",
                "capabilities": ["reporting", "visualization"]
            }
        }

    async def check_agent_health

Secure Credential Management for Automated Deployment Pipelines

biao lin — Wed, 17 Jun 2026 15:58:14 +0000

Secure Credential Management for Automated Deployment Pipelines

In modern DevOps environments, automated deployment pipelines are the backbone of continuous delivery. These pipelines often require access to sensitive credentials—API keys, database passwords, SSH keys, cloud provider tokens, and more. Mishandling these secrets can lead to catastrophic security breaches. This post explores a pragmatic approach to credential management using symmetric encryption with Python's cryptography.fernet module, balancing security with operational simplicity.

The Credential Challenge

Hardcoding credentials in pipeline configuration files is an obvious anti-pattern. Equally problematic are:

Storing secrets in environment variables that persist across runs
Using unencrypted files in shared storage
Embedding credentials in source control (even in private repos)
Relying on pipeline platform secrets managers without proper rotation

A robust solution should provide:

Encryption at rest for stored credentials
Decryption on demand only when needed
Auditability of credential access
Rotation support without pipeline code changes
Platform independence across CI/CD tools

Enter Fernet Encryption

Fernet is a symmetric encryption algorithm included in the cryptography library. It provides:

AES-128-CBC encryption with PKCS7 padding
HMAC authentication using SHA256
Time-limited token validity (optional)
URL-safe base64 encoding

This makes Fernet ideal for credential files that need to be stored alongside pipeline code but remain unreadable without the correct key.

Implementation Architecture

Our solution consists of three components:

Key Management Service (KMS) - Stores the master encryption key
Credential Vault - Encrypted file containing all secrets
Pipeline Helper - Decrypts and injects credentials at runtime

Step 1: Key Generation and Storage

First, generate a Fernet key:

from cryptography.fernet import Fernet

def generate_key():
    """Generate a new Fernet key"""
    key = Fernet.generate_key()
    with open("master.key", "wb") as key_file:
        key_file.write(key)
    return key

# One-time setup
key = generate_key()
print(f"Key generated: {key.decode()}")

Critical: The key file (master.key) must be stored securely. Options include:

HashiCorp Vault
AWS Secrets Manager / Azure Key Vault
CI/CD platform's encrypted environment variables
Hardware Security Module (HSM) for enterprise

Never commit the key to version control. Add master.key to .gitignore.

Step 2: Credential Vault Creation

Create a JSON structure to hold credentials, then encrypt it:

import json
from cryptography.fernet import Fernet

def create_vault(credentials: dict, key: bytes) -> bytes:
    """Encrypt credentials dictionary into a vault file"""
    fernet = Fernet(key)
    json_data = json.dumps(credentials, indent=2).encode()
    encrypted_data = fernet.encrypt(json_data)

    with open("credentials.enc", "wb") as vault:
        vault.write(encrypted_data)

    return encrypted_data

# Example credentials
secrets = {
    "database": {
        "host": "db.example.com",
        "port": 5432,
        "username": "app_user",
        "password": "s3cret!P@ss"
    },
    "api_keys": {
        "stripe": "sk_live_xxxxx",
        "aws": "AKIAIOSFODNN7EXAMPLE"
    },
    "ssh": {
        "private_key": "-----BEGIN RSA PRIVATE KEY-----\nMIIEpA..."
    }
}

# Load key (from secure storage)
with open("master.key", "rb") as f:
    key = f.read()

create_vault(secrets, key)

The resulting credentials.enc file is safe to store in:

Pipeline artifact storage
Shared network drives
Version control (if you trust the repo's access controls)

Step 3: Decryption and Injection

Now create the helper that pipelines will call:

import json
import os
import sys
from cryptography.fernet import Fernet

class CredentialManager:
    def __init__(self, vault_path: str, key: bytes):
        self.fernet = Fernet(key)
        self.vault_path = vault_path
        self._credentials = None

    def load_vault(self) -> dict:
        """Decrypt and load the credential vault"""
        with open(self.vault_path, "rb") as f:
            encrypted_data = f.read()

        try:
            decrypted_data = self.fernet.decrypt(encrypted_data)
            self._credentials = json.loads(decrypted_data)
        except Exception as e:
            print(f"Failed to decrypt vault: {e}", file=sys.stderr)
            sys.exit(1)

        return self._credentials

    def get_credential(self, path: str) -> str:
        """
        Retrieve a credential using dot notation path.
        Example: get_credential("database.password")
        """
        if not self._credentials:
            self.load_vault()

        parts = path.split(".")
        current = self._credentials

        for part in parts:
            if isinstance(current, dict) and part in current:
                current = current[part]
            else:
                raise KeyError(f"Credential path '{path}' not found")

        return current

    def inject_env(self, prefix: str = "SECRET_"):
        """Inject all credentials as environment variables"""
        if not self._credentials:
            self.load_vault()

        def _flatten(obj, parent_key=""):
            items = []
            for k, v in obj.items():
                new_key = f"{parent_key}_{k}".upper() if parent_key else k.upper()
                if isinstance(v, dict):
                    items.extend(_flatten(v, new_key).items())
                else:
                    items.append((f"{prefix}{new_key}", str(v)))
            return dict(items)

        env_vars = _flatten(self._credentials)
        os.environ.update(env_vars)
        return env_vars

# Pipeline usage
if __name__ == "__main__":
    # Key should come from environment variable (injected by CI/CD platform)
    key = os.environ.get("MASTER_KEY")
    if not key:
        print("MASTER_KEY environment variable not set", file=sys.stderr)
        sys.exit(1)

    manager = CredentialManager("credentials.enc", key.encode())

    # Option 1: Get individual credentials
    db_password = manager.get_credential("database.password")
    os.environ["DB_PASSWORD"] = db_password

    # Option 2: Inject all secrets at once
    env_map = manager.inject_env("MYAPP_")
    # Now MYAPP_DATABASE_HOST, MYAPP_STRIPE_KEY etc. are available

Pipeline Integration

Here's how to use this in common CI/CD platforms:

GitHub Actions

name: Deploy
on: [push]
jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Setup credentials
        run: |
          pip install cryptography
          python -c "
          from credential_manager import CredentialManager
          manager = CredentialManager('credentials.enc', b'${{ secrets.MASTER_KEY }}')
          manager.inject_env('DEPLOY_')
          "
      - name: Deploy
        run: |
          echo "Using DB: $DEPLOY_DATABASE_HOST"
          # Actual deployment commands

GitLab CI

variables:
  MASTER_KEY: $MASTER_KEY  # Set in CI/CD Settings

deploy:
  script:
    - pip install cryptography
    - python -c "
      from credential_manager import CredentialManager
      import os
      manager = CredentialManager('credentials.enc', os.environ['MASTER_KEY'].encode())
      manager.inject_env('DEPLOY_')
      "
    - ./deploy.sh

Security Best Practices

Key Rotation: Implement a rotation script that re-encrypts the vault with a new key:

def rotate_key(old_key: bytes, new_key: bytes, vault_path: str):
    """Re-encrypt vault with new key"""
    old_fernet = Fernet(old_key)
    new_fernet = Fernet(new_key)

    with open(vault_path, "rb") as f:
        data = old_fernet.decrypt(f.read())

    new_encrypted = new_fernet.encrypt(data)

    with open(vault_path, "wb") as f:
        f.write(new_encrypted)

Audit Logging: Log every credential access with timestamp and pipeline ID:


python
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def get_credential(self, path):
    logger.info(f"Credential accessed: {path} (pipeline: {os.environ.get('CI_PIPELINE_ID', 'local')})")
    # ... rest

Rate Limiting and Circuit Breakers in Distributed AI Systems

biao lin — Wed, 17 Jun 2026 15:51:39 +0000

Rate Limiting and Circuit Breakers in Distributed AI Systems

Distributed AI systems are inherently complex, handling massive volumes of requests, variable latency from model inference, and dependencies on external services like GPU clusters, databases, or third-party APIs. Without proper safeguards, a single misbehaving component or a sudden traffic surge can cascade into system-wide failure. Two fundamental patterns—rate limiting and circuit breakers—provide essential protection. This post explores their roles, implementation strategies, and practical Python examples tailored for AI workloads.

Why Distributed AI Systems Need These Patterns

Consider a typical AI pipeline: a user sends a prompt, which hits a load balancer, then an API gateway, then an inference service (e.g., a large language model), which may call a vector database or a fine-tuning API. Each component has capacity limits:

GPU inference servers can handle limited concurrent requests.
External APIs (e.g., OpenAI, HuggingFace) impose rate limits.
Database connections are finite.

Without rate limiting, a single abusive client can exhaust resources. Without circuit breakers, a failing downstream service can cause cascading timeouts and resource exhaustion across the entire system.

Rate Limiting: Controlling Request Flow

Rate limiting restricts how many requests a client, user, or service can make in a given time window. It prevents resource starvation and ensures fair access.

Common Algorithms

Algorithm	Pros	Cons
Token Bucket	Smooth burst handling, easy to implement	Memory per bucket
Leaky Bucket	Constant outflow rate, simple	Less flexible for bursts
Fixed Window	Simple, low overhead	Boundary spikes (reset issues)
Sliding Window	Smoother than fixed, accurate	Slightly more complex

For AI systems, token bucket is often preferred because it allows short bursts (e.g., a user sending a batch of prompts) while maintaining a long-term average.

Python Implementation: Token Bucket Rate Limiter

import time
import threading
from collections import defaultdict

class TokenBucket:
    def __init__(self, rate: float, capacity: int):
        """
        rate: tokens per second
        capacity: maximum bucket size (burst)
        """
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last_refill = time.monotonic()
        self.lock = threading.Lock()

    def _refill(self):
        now = time.monotonic()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.last_refill = now

    def consume(self, tokens: int = 1) -> bool:
        with self.lock:
            self._refill()
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

# Per-user rate limiter for AI inference
class AIInferenceRateLimiter:
    def __init__(self, default_rate: float = 10, default_capacity: int = 20):
        self.buckets = defaultdict(lambda: TokenBucket(default_rate, default_capacity))

    def check_request(self, user_id: str) -> bool:
        return self.buckets[user_id].consume(1)

# Usage
limiter = AIInferenceRateLimiter(rate=5, capacity=10)  # 5 req/s, burst up to 10
if limiter.check_request("user_123"):
    print("Proceed with model inference")
else:
    print("429 Too Many Requests")

AI-Specific Considerations

Token-level throttling: For LLMs, rate-limit by input/output token count, not just requests. A request generating 4096 tokens is much heavier than one generating 100.
Priority queues: Premium users get higher rate limits or separate buckets.
Distributed state: Use Redis or etcd for shared rate limiter state across multiple service instances.

# Redis-based distributed rate limiter (simplified)
import redis
import time

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def check_rate_limit(user_id: str, max_requests: int, window_seconds: int) -> bool:
    key = f"ratelimit:{user_id}:{int(time.time()) // window_seconds}"
    current = r.get(key)
    if current and int(current) >= max_requests:
        return False
    r.incr(key)
    r.expire(key, window_seconds + 1)
    return True

Circuit Breakers: Fail Fast and Recover Gracefully

A circuit breaker monitors calls to a downstream service (e.g., a GPU inference server, a vector database). When failures exceed a threshold, the circuit opens, and subsequent calls fail immediately (or use a fallback). After a cooldown, it transitions to half-open to test if the service has recovered.

States

Closed (normal) --> Open (failing) --> Half-Open (testing) --> Closed or Open

Python Implementation: Circuit Breaker for AI Inference

import time
import logging
from enum import Enum

class CircuitState(Enum):
    CLOSED = 1
    OPEN = 2
    HALF_OPEN = 3

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 30.0,
                 half_open_max_requests: int = 3):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_requests = half_open_max_requests

        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
        self.half_open_requests = 0
        self.lock = threading.Lock()

    def call(self, func, fallback=None, *args, **kwargs):
        with self.lock:
            if self.state == CircuitState.OPEN:
                if time.monotonic() - self.last_failure_time >= self.recovery_timeout:
                    self.state = CircuitState.HALF_OPEN
                    self.half_open_requests = 0
                    logging.info("Circuit breaker entering half-open state")
                else:
                    logging.warning("Circuit breaker open, using fallback")
                    return self._execute_fallback(fallback, *args, **kwargs)

            if self.state == CircuitState.HALF_OPEN:
                if self.half_open_requests >= self.half_open_max_requests:
                    logging.warning("Half-open max requests reached, using fallback")
                    return self._execute_fallback(fallback, *args, **kwargs)
                self.half_open_requests += 1

        # Execute the actual call (outside lock to avoid blocking)
        try:
            result = func(*args, **kwargs)
            with self.lock:
                if self.state == CircuitState.HALF_OPEN:
                    self.state = CircuitState.CLOSED
                    self.failure_count = 0
                    logging.info("Circuit breaker closed after successful half-open call")
                elif self.state == CircuitState.CLOSED:
                    self.failure_count = 0
            return result
        except Exception as e:
            with self.lock:
                self.failure_count += 1
                self.last_failure_time = time.monotonic()
                if self.state == CircuitState.HALF_OPEN:
                    self.state = CircuitState.OPEN
                    logging.warning("Circuit breaker reopened after half-open failure")
                elif self.state == CircuitState.CLOSED and self.failure_count >= self.failure_threshold:
                    self.state = CircuitState.OPEN
                    logging.error(f"Circuit breaker opened after {self.failure_count} failures")
            return self._execute_fallback(fallback, *args, **kwargs)

    def _execute_fallback(self, fallback, *args, **kwargs):
        if fallback:
            return fallback(*args, **kwargs)
        raise Exception("Circuit breaker open and no fallback provided")

# Usage with AI inference call
def call_gpu_inference(prompt: str) -> str:
    # Simulated call to GPU inference server
    import random
    if random.random() < 0.3:  # 30% failure rate
        raise ConnectionError("GPU server timeout")
    return f"Response to: {prompt}"

def fallback_response(prompt: str) -> str:
    return f"Fallback: {prompt[:50]}... (service degraded)"

circuit_breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=10.0)

for i in range(20):
    result = circuit_breaker.call(call_gpu_inference, fallback_response, f"Prompt {i}")
    print(f"Request {i}: {result}")
    time.sleep(0.5)

AI-Specific Circuit Breaker Features

Failure types: Distinguish between transient errors (timeouts, 503s) and persistent errors (400 bad request). Only count transient errors toward the threshold.
Slow calls: Treat abnormally slow responses (e.g., >10s for LLM inference) as failures. This prevents resource hoarding.
**Bulk

Building a Self-Evolving Multi-Agent System with Python

biao lin — Wed, 17 Jun 2026 15:51:08 +0000

Building a Self-Evolving Multi-Agent System with Python

Multi-agent systems (MAS) have become a cornerstone of modern distributed AI, enabling complex problem-solving through the collaboration of autonomous agents. But what if these agents could not only execute tasks but also learn, adapt, and evolve their own strategies over time? In this post, we'll build a self-evolving multi-agent system in Python, where agents use reinforcement learning to improve their behavior dynamically. We'll explore the architecture, implement core components, and demonstrate how the system can optimize itself without manual intervention.

The Vision: Agents That Improve Themselves

A self-evolving MAS consists of agents that:

Operate in a shared environment
Learn from interactions (successes and failures)
Update their internal policies based on collective experience
Adapt to new tasks or changing conditions

To achieve this, we'll combine three key technologies:

Python's asyncio for concurrent agent execution
Simple reinforcement learning (Q-learning) for agent adaptation
A shared memory structure for experience replay and knowledge sharing

System Architecture

Our system has four main components:

Environment: A grid world where agents collect rewards
Agents: Independent entities with Q-tables
Coordinator: Manages agent lifecycle and experience sharing
Evolution Engine: Periodically mutates and selects optimal agents

Let's implement each piece.

Step 1: The Environment

We'll create a simple 5x5 grid world with rewards and obstacles.

import numpy as np
from typing import Tuple, List, Dict
import random

class GridWorld:
    """A simple 5x5 grid environment."""
    def __init__(self):
        self.size = 5
        self.reset()

    def reset(self):
        self.agent_pos = [0, 0]
        self.goal_pos = [4, 4]
        self.obstacles = [(1, 2), (2, 2), (3, 3)]
        self.rewards = {(4, 4): 10, (1, 1): -2, (3, 4): 5}
        return self._get_state()

    def _get_state(self) -> int:
        """Encode position as state index."""
        return self.agent_pos[0] * self.size + self.agent_pos[1]

    def step(self, action: int) -> Tuple[int, float, bool]:
        """Take action: 0=up, 1=down, 2=left, 3=right."""
        row, col = self.agent_pos
        if action == 0: row = max(0, row - 1)
        elif action == 1: row = min(self.size - 1, row + 1)
        elif action == 2: col = max(0, col - 1)
        elif action == 3: col = min(self.size - 1, col + 1)

        # Check obstacles
        if (row, col) not in self.obstacles:
            self.agent_pos = [row, col]

        state = self._get_state()
        reward = self.rewards.get(tuple(self.agent_pos), 0)
        done = tuple(self.agent_pos) == self.goal_pos
        return state, reward, done

Step 2: The Q-Learning Agent

Each agent maintains its own Q-table and learns from experience.

class QAgent:
    """Reinforcement learning agent with Q-learning."""
    def __init__(self, agent_id: int, state_size: int, action_size: int, 
                 learning_rate: float = 0.1, discount: float = 0.95, 
                 epsilon: float = 0.1):
        self.id = agent_id
        self.q_table = np.zeros((state_size, action_size))
        self.lr = learning_rate
        self.discount = discount
        self.epsilon = epsilon
        self.fitness = 0  # Track total reward for evolution

    def choose_action(self, state: int) -> int:
        """Epsilon-greedy action selection."""
        if random.random() < self.epsilon:
            return random.randint(0, 3)  # Explore
        return np.argmax(self.q_table[state])  # Exploit

    def learn(self, state: int, action: int, reward: float, 
              next_state: int, done: bool):
        """Update Q-values using Q-learning update rule."""
        target = reward
        if not done:
            target += self.discount * np.max(self.q_table[next_state])
        td_error = target - self.q_table[state][action]
        self.q_table[state][action] += self.lr * td_error

    def mutate(self, mutation_rate: float = 0.01):
        """Randomly mutate Q-table entries for evolution."""
        mutation_mask = np.random.random(self.q_table.shape) < mutation_rate
        self.q_table[mutation_mask] += np.random.normal(0, 0.1, 
                                                       size=self.q_table[mutation_mask].shape)
        # Clip to reasonable range
        self.q_table = np.clip(self.q_table, -10, 10)

Step 3: The Multi-Agent Coordinator

This component manages concurrent agent training and experience sharing.

import asyncio
from collections import deque

class Coordinator:
    """Manages multiple agents and coordinates learning."""
    def __init__(self, num_agents: int = 10, shared_memory_size: int = 1000):
        self.env = GridWorld()
        self.state_size = self.env.size * self.env.size
        self.action_size = 4
        self.agents = [QAgent(i, self.state_size, self.action_size) 
                      for i in range(num_agents)]
        self.shared_memory = deque(maxlen=shared_memory_size)
        self.best_agent = None
        self.best_fitness = float('-inf')

    async def train_agent(self, agent: QAgent, episodes: int = 100):
        """Train a single agent asynchronously."""
        for _ in range(episodes):
            state = self.env.reset()
            total_reward = 0
            done = False

            while not done:
                action = agent.choose_action(state)
                next_state, reward, done = self.env.step(action)
                agent.learn(state, action, reward, next_state, done)
                total_reward += reward

                # Store experience in shared memory
                self.shared_memory.append((state, action, reward, next_state, done))
                state = next_state

            agent.fitness = total_reward

    async def train_all(self, episodes_per_agent: int = 100):
        """Train all agents concurrently."""
        tasks = [self.train_agent(agent, episodes_per_agent) 
                for agent in self.agents]
        await asyncio.gather(*tasks)

        # Update best agent
        for agent in self.agents:
            if agent.fitness > self.best_fitness:
                self.best_fitness = agent.fitness
                self.best_agent = agent

Step 4: The Evolution Engine

This is where self-evolution happens. We periodically select top agents and create offspring.

class EvolutionEngine:
    """Handles agent evolution through selection and mutation."""
    def __init__(self, coordinator: Coordinator, 
                 selection_ratio: float = 0.3, 
                 mutation_rate: float = 0.05):
        self.coordinator = coordinator
        self.selection_ratio = selection_ratio
        self.mutation_rate = mutation_rate
        self.generation = 0

    def evolve(self):
        """Perform one generation of evolution."""
        self.generation += 1
        agents = self.coordinator.agents

        # Sort by fitness (descending)
        agents.sort(key=lambda a: a.fitness, reverse=True)

        # Select top performers
        num_selected = max(2, int(len(agents) * self.selection_ratio))
        selected = agents[:num_selected]

        # Create next generation
        new_generation = []

        # Keep the best agent unchanged (elitism)
        new_generation.append(copy.deepcopy(selected[0]))

        # Create offspring through mutation
        while len(new_generation) < len(agents):
            parent = random.choice(selected)
            offspring = copy.deepcopy(parent)
            offspring.mutate(self.mutation_rate)
            new_generation.append(offspring)

        self.coordinator.agents = new_generation
        print(f"Generation {self.generation}: Best fitness = {selected[0].fitness:.2f}")

Step 5: Putting It All Together

Now we'll create the main training loop that evolves the system.


python
import copy

async def main():
    # Initialize system
    coordinator = Coordinator(num_agents=20)
    evolution = EvolutionEngine(coordinator, selection_ratio=0.3)

    # Run evolution for multiple generations
    num_generations = 20
    episodes_per_agent = 50

    for gen in range(num_generations):