Fleet Management for 100+ AI Bots: Lessons Learned

The Wild West of Multi-Agent Systems

When I first started running automated agents at scale, I thought I had it figured out. Spin up a few bots, point them at some tasks, collect the output. Simple, right?

Then I hit 50 agents. Then 100. Then things got weird.

Suddenly, my carefully orchestrated system felt like herding cats. Bots were stepping on each other's toes, rate limits were triggering cascade failures, and I couldn't tell which agent was doing what without diving into a dozen log files. The infrastructure that worked for 10 agents was actively hostile to 100.

If you're building toward multi-agent systems at scale, here's what I learned the hard way — so you don't have to.

The Scaling Trap: What Breaks First

1. Session Management Becomes a Nightmare

With a handful of agents, you can track sessions manually. With 100+, you need a proper session registry. Each agent needs its own identity, authentication context, and state.

Here's the pattern that saved me:

class AgentSessionManager:
    def __init__(self):
        self.sessions = {}  # agent_id -> session_data
        self.lock = asyncio.Lock()

    async def register_agent(self, agent_id: str, credentials: dict):
        async with self.lock:
            self.sessions[agent_id] = {
                "status": "idle",
                "task_queue": [],
                "last_heartbeat": time.time(),
                "auth_token": await self._generate_token(credentials)
            }

    async def assign_task(self, agent_id: str, task: dict):
        async with self.lock:
            if self.sessions.get(agent_id, {}).get("status") == "idle":
                self.sessions[agent_id]["task_queue"].append(task)
                self.sessions[agent_id]["status"] = "busy"
                return True
            return False

The key insight? Atomic operations on agent state. Without that lock, you'll get race conditions where two tasks get assigned to the same agent, or worse, an agent gets marked as available when it's actually still processing.

2. Health Checks That Actually Scale

Ping-pong health checks don't cut it at scale. You need heartbeat monitoring with exponential backoff and automatic decommissioning.

async def health_monitor(agent_id, session_manager, interval=30):
    missed_beats = 0
    max_missed = 3

    while True:
        await asyncio.sleep(interval)
        last_beat = session_manager.sessions[agent_id]["last_heartbeat"]

        if time.time() - last_beat > interval * 2:
            missed_beats += 1
            if missed_beats >= max_missed:
                await decommission_agent(agent_id)
                break
        else:
            missed_beats = 0

This pattern catches the silent failures — the agent that's technically "running" but has been stuck on a task for 10 minutes without making progress.

3. Task Queuing: Never Use a Single Queue

A single task queue is a bottleneck and a single point of failure. Instead, use per-agent priority queues with a global dispatcher.

class PriorityTaskDispatcher:
    def __init__(self):
        self.agent_queues = defaultdict(asyncio.PriorityQueue)
        self.capacity_map = {}  # agent_id -> max_concurrent_tasks

    async def dispatch(self, task):
        # Find the best agent based on load and capability
        best_agent = await self._select_optimal_agent(task)
        priority = task.get("priority", 10)
        await self.agent_queues[best_agent].put((priority, task))

    async def _select_optimal_agent(self, task):
        available = [
            aid for aid, cap in self.capacity_map.items()
            if self.agent_queues[aid].qsize() < cap
        ]
        # Pick the agent with the shortest queue
        return min(available, key=lambda aid: self.agent_queues[aid].qsize())

This isn't just about performance. It's about predictability. When you have 100 agents, you need to know that high-priority tasks won't get stuck behind a backlog of low-priority work.

Real World: Where This Gets Applied

These patterns aren't theoretical. I've seen platforms that handle this at scale, and the architecture matters. For instance, roborent.cc runs a fleet management system that coordinates both AI agents and human workers across thousands of concurrent tasks. Their approach to task delegation — routing social media tasks to one agent type, research tasks to another, verification to a third — is exactly what you'd build if you were designing a multi-agent system from scratch.

The difference is they've already solved the hard problems: session persistence across agent restarts, automatic failover when an agent goes silent, and payment reconciliation when a task involves multiple agents in a chain.

The Infrastructure You Need

Database: Not SQLite

Please, for the love of everything, don't use SQLite for multi-agent state management. You need PostgreSQL or at minimum MySQL with proper connection pooling.

CREATE TABLE agent_tasks (
    id UUID PRIMARY KEY,
    agent_id VARCHAR(64) NOT NULL,
    task_type VARCHAR(32) NOT NULL,
    status VARCHAR(16) NOT NULL DEFAULT 'pending',
    priority INTEGER NOT NULL DEFAULT 10,
    created_at TIMESTAMP NOT NULL DEFAULT NOW(),
    started_at TIMESTAMP,
    completed_at TIMESTAMP,
    result JSONB,
    FOREIGN KEY (agent_id) REFERENCES agents(id)
);

CREATE INDEX idx_agent_status ON agent_tasks(agent_id, status);
CREATE INDEX idx_priority ON agent_tasks(priority DESC) WHERE status = 'pending';

That composite index on agent_id and status is critical. Without it, your task assignment queries will slow to a crawl as the table grows.

Communication: Redis Pub/Sub

For inter-agent communication, HTTP polling is a trap. Use Redis Pub/Sub or a message queue like NATS.

import aioredis

class AgentMessageBus:
    def __init__(self):
        self.redis = await aioredis.from_url("redis://localhost")

    async def broadcast(self, channel, message):
        await self.redis.publish(channel, json.dumps(message))

    async def subscribe(self, channel):
        pubsub = self.redis.pubsub()
        await pubsub.subscribe(channel)
        async for message in pubsub.listen():
            if message["type"] == "message":
                yield json.loads(message["data"])

This gives you sub-millisecond message delivery and built-in fan-out when you need to notify all agents of a configuration change.

The Hidden Costs Nobody Talks About

1. Token Consumption

Each agent making API calls burns tokens — and at 100 agents, that adds up fast. Track token usage per agent, per task type, and set hard limits.

class TokenBudget:
    def __init__(self, daily_budget=1000000):
        self.daily_budget = daily_budget
        self.usage = defaultdict(int)

    async def can_proceed(self, agent_id, estimated_tokens):
        current = self.usage[agent_id]
        if current + estimated_tokens > self.daily_budget:
            return False
        self.usage[agent_id] = current + estimated_tokens
        return True

2. Log Volume

100 agents generating logs means you'll hit 10GB+ per day easily. Set up log rotation, sampling, and aggregation from day one. Elasticsearch or Loki for storage, with aggressive retention policies.

3. Coordination Overhead

Every coordination pattern — task assignment, health checks, result collection — adds latency. At scale, you need to measure and optimize this overhead. If your agents spend more time coordinating than working, you've over-engineered the system.

What I'd Do Differently

If I were building a 100+ agent fleet again from scratch:

1. Start with a task broker, not direct agent communication. RabbitMQ or Redis Streams as the backbone, not custom sockets.

2. Idempotency from day one. Every task should be safe to execute twice. Network failures will happen, and retries are