Load Testing AI Agent Systems: Simulating 10,000 Concurrent Conversations

Why Standard Load Testing Fails for AI Agents

Traditional load testing tools like Apache Bench or wrk fire thousands of HTTP requests per second and measure response times. But AI agent conversations are fundamentally different: each conversation is a multi-turn stateful session where each request depends on the previous response. A single conversation might take 30 seconds to 5 minutes. The bottleneck is not requests per second — it is concurrent long-lived sessions.

You need load testing that simulates realistic conversation patterns: connection establishment, multi-turn exchanges with think time between turns, tool call chains, and graceful session termination.

Building a Conversation Simulator

Use Locust, which supports custom user behavior classes that model stateful interactions:

flowchart LR
    PR(["PR opened"])
    UNIT["Unit tests"]
    EVAL["Eval harness<br/>PromptFoo or Braintrust"]
    GOLD[("Golden set<br/>200 tagged cases")]
    JUDGE["LLM as judge<br/>plus regex graders"]
    SCORE["Aggregate score<br/>and per slice"]
    GATE{"Score regress<br/>more than 2 percent?"}
    BLOCK(["Block merge"])
    MERGE(["Merge to main"])
    PR --> UNIT --> EVAL --> GOLD --> JUDGE --> SCORE --> GATE
    GATE -->|Yes| BLOCK
    GATE -->|No| MERGE
    style EVAL fill:#4f46e5,stroke:#4338ca,color:#fff
    style GATE fill:#f59e0b,stroke:#d97706,color:#1f2937
    style BLOCK fill:#dc2626,stroke:#b91c1c,color:#fff
    style MERGE fill:#059669,stroke:#047857,color:#fff

from locust import HttpUser, task, between, events
import json
import time
import random

# Sample conversation scripts for realistic testing
CONVERSATION_SCRIPTS = [
    [
        "What are your business hours?",
        "Do you offer weekend support?",
        "How do I contact sales?",
    ],
    [
        "I need help with my order #12345",
        "It was supposed to arrive yesterday",
        "Can you expedite the replacement?",
        "Thank you for your help",
    ],
    [
        "Tell me about your enterprise plan",
        "What integrations do you support?",
        "Can you connect with Salesforce?",
        "What is the pricing for 500 seats?",
        "I would like to schedule a demo",
    ],
]

class AgentConversationUser(HttpUser):
    wait_time = between(2, 8)  # Think time between turns
    host = "https://agents-staging.example.com"

    def on_start(self):
        # Create a new session
        resp = self.client.post(
            "/api/sessions",
            json={"user_id": f"loadtest-{self.greenlet.minimal_ident}"},
        )
        self.session_id = resp.json()["session_id"]
        self.script = random.choice(CONVERSATION_SCRIPTS)
        self.turn_index = 0

    @task
    def send_message(self):
        if self.turn_index >= len(self.script):
            # Conversation finished, start a new one
            self.on_start()
            return

        message = self.script[self.turn_index]
        start = time.time()

        with self.client.post(
            f"/api/sessions/{self.session_id}/messages",
            json={"message": message},
            catch_response=True,
            name="/api/sessions/[id]/messages",
        ) as resp:
            latency = time.time() - start

            if resp.status_code == 200:
                body = resp.json()
                if "response" in body and len(body["response"]) > 0:
                    resp.success()
                else:
                    resp.failure("Empty response from agent")
            elif resp.status_code == 429:
                resp.failure("Rate limited")
            else:
                resp.failure(f"Status {resp.status_code}")

        self.turn_index += 1

Run the load test with a ramp-up pattern that gradually increases concurrent users:

# locustfile configuration
# Run: locust -f loadtest.py --headless
#   --users 10000 --spawn-rate 100
#   --run-time 30m --csv results

# This ramps up 100 new users per second until 10,000
# concurrent conversation sessions are active

Metrics That Matter

Standard latency percentiles (p50, p95, p99) are necessary but not sufficient. For AI agents, track these additional metrics:

import time
from prometheus_client import Histogram, Counter, Gauge

# Core latency metrics
agent_turn_latency = Histogram(
    "agent_turn_latency_seconds",
    "Time to complete one conversation turn",
    buckets=[0.5, 1, 2, 3, 5, 8, 13, 21, 30],
)

llm_call_latency = Histogram(
    "llm_call_latency_seconds",
    "Time for each LLM API call",
    labelnames=["model", "endpoint"],
    buckets=[0.2, 0.5, 1, 2, 5, 10, 20],
)

# Throughput metrics
active_sessions = Gauge(
    "active_agent_sessions",
    "Number of currently active conversation sessions",
)

tool_calls_total = Counter(
    "agent_tool_calls_total",
    "Total tool calls made by agents",
    labelnames=["tool_name", "status"],
)

# Error metrics
agent_errors = Counter(
    "agent_errors_total",
    "Agent errors by type",
    labelnames=["error_type"],
)

# Cost metrics
tokens_consumed = Counter(
    "llm_tokens_consumed_total",
    "Total tokens consumed",
    labelnames=["model", "direction"],  # input/output
)

Identifying Bottlenecks

During load testing, watch for these common bottleneck patterns:

Database connection exhaustion appears as a sudden spike in turn latency across all sessions simultaneously. Monitor the pg_stat_activity view to see how many connections are active versus waiting.

LLM API rate limits show as increasing 429 responses or growing queue depth. Most providers return rate limit headers you can track.

Memory growth from accumulating conversation context is a slow-burn problem. Monitor per-pod memory usage over a 30-minute test to see if it grows linearly with session count:

# Add to your health endpoint for monitoring during tests
import psutil
import os

@app.get("/debug/resources")
async def resource_usage():
    process = psutil.Process(os.getpid())
    return {
        "memory_mb": process.memory_info().rss / 1024 / 1024,
        "cpu_percent": process.cpu_percent(),
        "open_files": len(process.open_files()),
        "connections": len(process.connections()),
        "threads": process.num_threads(),
    }

Capacity Planning Model

After collecting load test data, build a simple capacity model:

def estimate_capacity(
    target_concurrent_sessions: int,
    avg_turn_latency_seconds: float,
    avg_turns_per_session: int,
    avg_session_duration_minutes: float,
    sessions_per_pod: int,
    memory_per_session_mb: float,
    pod_memory_limit_mb: float,
) -> dict:
    pods_by_sessions = (
        target_concurrent_sessions / sessions_per_pod
    )
    pods_by_memory = (
        target_concurrent_sessions
        * memory_per_session_mb
        / pod_memory_limit_mb
    )
    pods_needed = max(pods_by_sessions, pods_by_memory)

    # Add 30% headroom for spikes
    pods_recommended = int(pods_needed * 1.3) + 1

    return {
        "target_sessions": target_concurrent_sessions,
        "pods_by_session_limit": int(pods_by_sessions) + 1,
        "pods_by_memory_limit": int(pods_by_memory) + 1,
        "pods_recommended": pods_recommended,
        "estimated_monthly_cost": pods_recommended * 73.0,
    }

FAQ

How do I simulate realistic conversation patterns in load tests?

Record real conversation transcripts (anonymized) from production and replay them as test scripts. Include realistic think times between turns (2 to 15 seconds) and vary conversation lengths (2 to 10 turns). Use a mix of simple and complex conversations that matches your production distribution.

What is a good target for agent turn latency under load?

For real-time conversational agents, target p50 under 3 seconds and p95 under 8 seconds. For async task agents, p50 under 10 seconds and p95 under 30 seconds. These targets include LLM API latency, which is typically the dominant factor.

How often should I run load tests?

Run a baseline load test after any infrastructure change (new region, database migration, scaling configuration change). Run weekly regression tests at 50 percent of peak capacity to catch gradual degradation. Run full-scale tests monthly or before major releases.

---

#LoadTesting #AIAgents #PerformanceTesting #CapacityPlanning #Locust #Metrics #AgenticAI #LearnAI #AIEngineering