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:

1. Python's asyncio for concurrent agent execution
2. Simple reinforcement learning (Q-learning) for agent adaptation
3. 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):