Load Testing Enterprise Agents: Simulating Concurrent User Load

An agent that works perfectly with one user may collapse under a hundred. LLM rate limits, database connections, tool API quotas, and memory constraints all surface under load. Load testing enterprise agents before production deployment prevents embarrassing failures and unexpected cost overruns.

Why Load Test Agents

Agents are different from traditional web apps:

• LLM calls are slow — 1-30 seconds per call, not milliseconds
• Token limits are hard — Rate limits per minute, per day, per key
• State is complex — Checkpoints, context windows, conversation history
• Tools have quotas — API rate limits, database connection pools
• Costs scale linearly — Every token costs money

Load Testing Architecture

Load Generator
    ├── Virtual User 1 → Agent API
    ├── Virtual User 2 → Agent API
    ├── ...
    └── Virtual User N → Agent API
         ↓
    Agent Fleet
    ├── Worker 1 (LangGraph)
    ├── Worker 2 (LangGraph)
    └── Worker N (LangGraph)
         ↓
    Infrastructure
    ├── LLM API (OpenAI/Anthropic)
    ├── Vector Database
    ├── Tool APIs
    └── Message Queue

Implementation with Locust

Define Agent Workloads

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

class AgentUser(HttpUser):
    wait_time = between(5, 15)  # Wait between tasks
    
    def on_start(self):
        """Initialize user session."""
        self.session_id = f"load-test-{random.randint(1000, 9999)}"
        self.tasks_completed = 0
    
    @task(3)
    def simple_query(self):
        """Simple agent query (70% of traffic)."""
        queries = [
            "What are your business hours?",
            "How do I reset my password?",
            "Can you explain your pricing?",
            "What's your refund policy?",
            "How do I contact support?",
        ]
        
        self._send_agent_request(random.choice(queries))
    
    @task(2)
    def complex_task(self):
        """Complex multi-step task (20% of traffic)."""
        tasks = [
            "Research competitor pricing and create a comparison table",
            "Analyze this CSV data and generate a report with insights",
            "Write a Python script to scrape and summarize product reviews",
            "Review this code for security vulnerabilities and suggest fixes",
            "Create a marketing email campaign for our new product launch",
        ]
        
        self._send_agent_request(random.choice(tasks))
    
    @task(1)
    def long_running(self):
        """Long-running task (10% of traffic)."""
        self._send_agent_request(
            "Research the latest developments in quantum computing, "
            "write a comprehensive report, and create a presentation outline"
        )
    
    def _send_agent_request(self, task: str):
        payload = {
            "task": task,
            "session_id": self.session_id,
            "task_type": "auto",
        }
        
        with self.client.post(
            "/api/agent/execute",
            json=payload,
            catch_response=True,
            name="/api/agent/execute"
        ) as response:
            if response.status_code == 200:
                result = response.json()
                if result.get("status") == "completed":
                    response.success()
                else:
                    response.failure(f"Agent returned status: {result.get('status')}")
            elif response.status_code == 429:
                response.failure("Rate limited")
            else:
                response.failure(f"HTTP {response.status_code}")

Custom Metrics Collection

from locust import events
import time

class AgentMetrics:
    def __init__(self):
        self.metrics = {
            "ttft": [],  # Time to first token
            "total_time": [],
            "tokens_per_second": [],
            "error_rate": 0,
            "rate_limit_hits": 0,
        }
    
    def record_request(self, start_time: float, result: dict):
        total_time = time.time() - start_time
        self.metrics["total_time"].append(total_time)
        
        if result.get("ttft"):
            self.metrics["ttft"].append(result["ttft"])
        
        if result.get("tokens_per_second"):
            self.metrics["tokens_per_second"].append(result["tokens_per_second"])

@events.request.add_listener
def on_request(request_type, name, response_time, response_length, exception, **kwargs):
    if exception:
        metrics.metrics["error_rate"] += 1
    
    if response_time:
        metrics.metrics["total_time"].append(response_time / 1000)

Analyze Results

import statistics

class LoadTestAnalyzer:
    def analyze(self, metrics: dict) -> dict:
        total_times = metrics["total_time"]
        
        return {
            "total_requests": len(total_times),
            "avg_response_time": statistics.mean(total_times),
            "p50_response_time": statistics.median(total_times),
            "p95_response_time": self._percentile(total_times, 95),
            "p99_response_time": self._percentile(total_times, 99),
            "throughput_rps": len(total_times) / sum(total_times),
            "error_rate": metrics["error_rate"] / max(len(total_times), 1),
            "avg_ttft": statistics.mean(metrics["ttft"]) if metrics["ttft"] else 0,
        }
    
    def _percentile(self, data: list, percentile: int) -> float:
        sorted_data = sorted(data)
        index = int(len(sorted_data) * percentile / 100)
        return sorted_data[min(index, len(sorted_data) - 1)]

Bottleneck Identification

class BottleneckDetector:
    def __init__(self):
        self.thresholds = {
            "llm_latency": 5.0,  # seconds
            "tool_latency": 2.0,
            "db_latency": 0.5,
            "error_rate": 0.05,
            "rate_limit_percentage": 0.8,
        }
    
    def detect(self, metrics: dict, infrastructure_metrics: dict) -> list:
        bottlenecks = []
        
        # Check LLM latency
        if infrastructure_metrics.get("llm_avg_latency", 0) > self.thresholds["llm_latency"]:
            bottlenecks.append({
                "component": "LLM API",
                "issue": "High latency",
                "current": infrastructure_metrics["llm_avg_latency"],
                "threshold": self.thresholds["llm_latency"],
                "recommendation": "Consider using a faster model or increasing concurrency limit"
            })
        
        # Check rate limits
        rate_limit_usage = infrastructure_metrics.get("rate_limit_usage", 0)
        if rate_limit_usage > self.thresholds["rate_limit_percentage"]:
            bottlenecks.append({
                "component": "LLM API Rate Limit",
                "issue": "Approaching rate limit",
                "current": f"{rate_limit_usage:.1%}",
                "threshold": f"{self.thresholds['rate_limit_percentage']:.1%}",
                "recommendation": "Increase rate limit or add request queuing"
            })
        
        # Check error rate
        error_rate = metrics.get("error_rate", 0)
        if error_rate > self.thresholds["error_rate"]:
            bottlenecks.append({
                "component": "Agent Fleet",
                "issue": "High error rate",
                "current": f"{error_rate:.1%}",
                "threshold": f"{self.thresholds['error_rate']:.1%}",
                "recommendation": "Check logs for errors, add retry logic"
            })
        
        return bottlenecks

Scaling Strategies

class AutoScaler:
    def __init__(self):
        self.metrics_history = []
    
    def should_scale(self, current_metrics: dict) -> dict:
        self.metrics_history.append(current_metrics)
        
        if len(self.metrics_history) < 5:
            return {"action": "none"}
        
        recent = self.metrics_history[-5:]
        
        # Scale up if latency is increasing
        avg_latency = statistics.mean([m.get("avg_response_time", 0) for m in recent])
        if avg_latency > 10:
            return {"action": "scale_up", "reason": "High latency"}
        
        # Scale down if load is low
        avg_throughput = statistics.mean([m.get("throughput_rps", 0) for m in recent])
        if avg_throughput < 1:
            return {"action": "scale_down", "reason": "Low throughput"}
        
        return {"action": "none"}

# Kubernetes HPA
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: agent-fleet
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: agent-workers
  minReplicas: 2
  maxReplicas: 20
  metrics:
  - type: Pods
    pods:
      metric:
        name: agent_latency_p95
      target:
        type: AverageValue
        averageValue: "10000"  # 10 seconds

Cost Analysis

class CostAnalyzer:
    def __init__(self, pricing: dict):
        self.pricing = pricing  # cost per 1M tokens
    
    def estimate_load_test_cost(
        self,
        concurrent_users: int,
        duration_minutes: int,
        avg_tokens_per_request: int
    ) -> dict:
        total_requests = concurrent_users * duration_minutes * 60 / 10  # avg 10s per request
        total_tokens = total_requests * avg_tokens_per_request
        
        cost = (total_tokens / 1_000_000) * self.pricing["per_1m_tokens"]
        
        return {
            "total_requests": int(total_requests),
            "total_tokens": int(total_tokens),
            "estimated_cost": f"${cost:.2f}",
            "cost_per_request": f"${cost / total_requests:.4f}",
        }

# Example
analyzer = CostAnalyzer({"per_1m_tokens": 10})
cost = analyzer.estimate_load_test_cost(
    concurrent_users=50,
    duration_minutes=10,
    avg_tokens_per_request=2000
)
print(cost)
# {'total_requests': 3000, 'total_tokens': 6000000, 'estimated_cost': '$60.00'}

Running the Load Test

# Basic load test
locust -f agent_load_test.py --host=http://localhost:8080 --users=50 --spawn-rate=5

# Distributed load test
locust -f agent_load_test.py --host=http://localhost:8080 --master --users=200 --spawn-rate=10
locust -f agent_load_test.py --host=http://localhost:8080 --worker --master-host=192.168.1.100

# Headless mode for CI/CD
locust -f agent_load_test.py --host=http://localhost:8080 \
    --users=100 --spawn-rate=10 --run-time=10m \
    --headless --csv=results

Best Practices

• Test at 2x expected peak — Ensure headroom for growth
• Monitor costs — Track token usage during tests
• Test failure scenarios — What happens when LLM API is slow?
• Test rate limits — Verify graceful degradation
• Baseline before changes — Compare performance after each change
• Automate in CI/CD — Run load tests on every deployment

Conclusion

Load testing enterprise agents reveals bottlenecks before users do. Test with realistic workloads, monitor LLM costs, identify rate limit thresholds, and verify auto-scaling works. Start with small concurrent loads and increase gradually, always tracking both performance metrics and cost.