--- title: "Load Testing AI Agent Systems: Simulating 10,000 Concurrent Conversations" description: "Learn how to load test AI agent systems by simulating thousands of concurrent conversations, collecting meaningful metrics, identifying bottlenecks, and building capacity planning models that predict scaling needs." canonical: https://callsphere.ai/blog/load-testing-ai-agent-systems-10000-conversations category: "Learn Agentic AI" tags: ["Load Testing", "AI Agents", "Performance Testing", "Capacity Planning", "Locust", "Metrics"] author: "CallSphere Team" published: 2026-03-17T00:00:00.000Z updated: 2026-05-06T05:15:21.443Z --- # Load Testing AI Agent Systems: Simulating 10,000 Concurrent Conversations > Learn how to load test AI agent systems by simulating thousands of concurrent conversations, collecting meaningful metrics, identifying bottlenecks, and building capacity planning models that predict scaling needs. ## 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: ```mermaid flowchart LR PR(["PR opened"]) UNIT["Unit tests"] EVAL["Eval harness PromptFoo or Braintrust"] GOLD[("Golden set 200 tagged cases")] JUDGE["LLM as judge plus regex graders"] SCORE["Aggregate score and per slice"] GATE{"Score regress 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 ``` ```python 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: ```yaml # 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: ```python 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: ```python # 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: ```python 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 --- Source: https://callsphere.ai/blog/load-testing-ai-agent-systems-10000-conversations