Introduction
Load testing LLM streaming responses is different from traditional API load testing. With a standard REST endpoint, you usually care about request rate, latency, and error percentage. With streaming LLM APIs, you also need to understand time to first token, stream duration, token delivery consistency, connection stability, and overall user-perceived responsiveness.
For AI applications, these metrics matter because users do not experience a streaming response as a single completed request. They experience it as a sequence: request accepted, first token appears, tokens continue at a readable pace, and the stream finishes without interruption. A model that returns a complete response in 12 seconds may feel slower than one that starts streaming in 800 ms and finishes in 15 seconds. That is why performance testing and stress testing LLM streaming responses requires a slightly different approach.
In this guide, you will learn how to load test LLM streaming responses using Locust on LoadForge. We will cover realistic streaming patterns, authenticated requests, SSE-style responses, concurrent chat workloads, and how to measure metrics that matter for AI & LLM systems. We will also show how LoadForge helps with distributed testing, real-time reporting, cloud-based infrastructure, global test locations, and CI/CD integration when validating streaming APIs at scale.
Prerequisites
Before you start load testing LLM streaming responses, make sure you have:
- A streaming LLM API endpoint to test
- An API key, bearer token, or session-based authentication method
- A clear understanding of your endpoint’s streaming protocol:
- Server-Sent Events (SSE)
- chunked HTTP responses
- newline-delimited JSON
- Sample prompts that reflect real user behavior
- Expected performance targets such as:
- time to first token under 1.5 seconds
- complete stream duration under 20 seconds
- error rate below 1%
- no stream truncation under peak load
You should also know whether your API supports payloads similar to:
/v1/chat/completions/api/v1/generate/stream/inference/chat/stream
For this guide, we will use realistic examples modeled after common LLM streaming APIs. These examples assume a bearer token in the Authorization header and a JSON request body with stream: true.
Understanding LLM Streaming Responses Under Load
LLM streaming endpoints behave differently from typical web APIs because each request often stays open for several seconds while tokens are emitted incrementally. This creates unique load testing challenges.
Key behaviors to measure
When load testing streaming APIs, focus on these metrics:
- Time to first token (TTFT): how quickly the first token arrives after the request is sent
- Stream completion time: total time from request start to final token
- Token cadence: whether tokens arrive smoothly or in bursts
- Stream integrity: whether the stream completes successfully without dropped chunks
- Concurrent connection handling: whether the service can maintain many open streams
- Error behavior under load: 429, 500, timeouts, broken pipe, incomplete chunks
Common bottlenecks in LLM streaming systems
Under load, LLM streaming APIs often fail in one of these areas:
- Model queueing delays causing poor TTFT
- GPU or inference worker saturation
- Reverse proxy timeout settings cutting off long streams
- SSE buffering or chunk flushing issues
- Authentication middleware slowing down request setup
- Rate limiting that triggers too aggressively under concurrency
- Network egress saturation affecting token delivery pace
Why traditional latency metrics are not enough
A single “response time” metric can hide the real user experience. For example:
- Request A: first token in 700 ms, completes in 18 s
- Request B: first token in 8 s, completes in 10 s
Traditional reporting may make Request B look faster overall, but users will usually prefer Request A because it feels more responsive. This is why load testing, performance testing, and stress testing LLM streaming responses should explicitly track TTFT and stream stability.
Writing Your First Load Test
Let’s start with a basic Locust script that sends a streaming chat completion request and measures:
- HTTP status
- time to first token
- total stream duration
- whether the stream completed cleanly
This example assumes an SSE-like endpoint at /v1/chat/completions.
from locust import HttpUser, task, between, events
import json
import time
import random
PROMPTS = [
"Explain the CAP theorem in simple terms.",
"Write a short Python function to validate an email address.",
"Summarize the benefits of using Redis for caching.",
"What are the tradeoffs between REST and GraphQL APIs?"
]
class LLMStreamingUser(HttpUser):
wait_time = between(1, 3)
host = "https://api.example-llm.com"
def on_start(self):
self.headers = {
"Authorization": "Bearer YOUR_API_TOKEN",
"Content-Type": "application/json",
"Accept": "text/event-stream"
}
@task
def stream_chat_completion(self):
payload = {
"model": "gpt-4o-mini",
"stream": True,
"temperature": 0.7,
"messages": [
{"role": "system", "content": "You are a helpful assistant for software developers."},
{"role": "user", "content": random.choice(PROMPTS)}
],
"max_tokens": 300
}
start_time = time.time()
first_token_time = None
chunk_count = 0
received_content = []
with self.client.post(
"/v1/chat/completions",
json=payload,
headers=self.headers,
stream=True,
catch_response=True,
timeout=60
) as response:
if response.status_code != 200:
response.failure(f"Unexpected status code: {response.status_code}")
return
try:
for line in response.iter_lines(decode_unicode=True):
if not line:
continue
if line.startswith("data: "):
data = line[6:].strip()
if data == "[DONE]":
break
if first_token_time is None:
first_token_time = time.time()
chunk_count += 1
try:
event = json.loads(data)
delta = event["choices"][0].get("delta", {})
content = delta.get("content")
if content:
received_content.append(content)
except json.JSONDecodeError:
response.failure("Invalid JSON chunk in stream")
return
total_duration = time.time() - start_time
if first_token_time is None:
response.failure("No streamed tokens received")
return
ttft_ms = int((first_token_time - start_time) * 1000)
full_text = "".join(received_content)
response.success()
events.request.fire(
request_type="STREAM",
name="ttft_/v1/chat/completions",
response_time=ttft_ms,
response_length=len(full_text),
exception=None,
context={}
)
events.request.fire(
request_type="STREAM",
name="duration_/v1/chat/completions",
response_time=int(total_duration * 1000),
response_length=chunk_count,
exception=None,
context={}
)
except Exception as e:
response.failure(f"Streaming error: {str(e)}")What this script does
This first test simulates a user making a streaming request to an LLM chat API. It reads the response incrementally and records:
ttft_/v1/chat/completions: time until the first chunk arrivesduration_/v1/chat/completions: total stream durationchunk_count: approximate measure of stream activityresponse_length: generated text size
This is a strong starting point for load testing LLM streaming responses because it focuses on user-perceived performance rather than only final request completion.
Running this on LoadForge
In LoadForge, you can upload this Locust script and run it from cloud-based infrastructure across multiple global test locations. This is especially useful if your AI users are spread across regions and you want to compare first-token latency by geography.
Advanced Load Testing Scenarios
Basic streaming tests are useful, but real AI applications often include authentication, multi-turn conversations, and different prompt sizes. Let’s move into more realistic performance testing scenarios.
Scenario 1: Authenticated streaming with session setup
Many AI platforms require an authentication step before the actual streaming request. The following example logs in, retrieves a JWT, and then uses it for a streaming inference call.
from locust import HttpUser, task, between, events
import json
import time
import random
USERS = [
{"email": "qa1@example.com", "password": "StrongPassword123!"},
{"email": "qa2@example.com", "password": "StrongPassword123!"},
{"email": "qa3@example.com", "password": "StrongPassword123!"}
]
CHAT_INPUTS = [
"Generate a customer support reply for a delayed shipment.",
"Draft a concise release note for a bug fix in our API.",
"Create a polite follow-up email after a sales demo."
]
class AuthenticatedLLMStreamingUser(HttpUser):
wait_time = between(2, 5)
host = "https://app.example-ai.com"
def on_start(self):
creds = random.choice(USERS)
login_payload = {
"email": creds["email"],
"password": creds["password"]
}
with self.client.post(
"/api/auth/login",
json=login_payload,
catch_response=True
) as response:
if response.status_code != 200:
response.failure(f"Login failed: {response.status_code}")
return
data = response.json()
self.token = data["access_token"]
self.headers = {
"Authorization": f"Bearer {self.token}",
"Content-Type": "application/json",
"Accept": "text/event-stream"
}
@task
def stream_reply_generation(self):
payload = {
"model": "llama-3.1-70b-instruct",
"stream": True,
"conversation_id": f"conv-{random.randint(1000, 9999)}",
"messages": [
{"role": "system", "content": "You are an assistant for business communications."},
{"role": "user", "content": random.choice(CHAT_INPUTS)}
],
"max_tokens": 250,
"temperature": 0.4
}
start = time.time()
first_token = None
bytes_received = 0
got_done = False
with self.client.post(
"/api/v1/chat/stream",
json=payload,
headers=self.headers,
stream=True,
catch_response=True,
timeout=90
) as response:
if response.status_code != 200:
response.failure(f"Stream request failed: {response.status_code}")
return
try:
for chunk in response.iter_lines(decode_unicode=True):
if not chunk:
continue
bytes_received += len(chunk)
if chunk.startswith("data: "):
data = chunk[6:].strip()
if first_token is None and data != "[DONE]":
first_token = time.time()
if data == "[DONE]":
got_done = True
break
if first_token is None:
response.failure("No first token received")
return
if not got_done:
response.failure("Stream ended without [DONE] marker")
return
ttft_ms = int((first_token - start) * 1000)
total_ms = int((time.time() - start) * 1000)
response.success()
events.request.fire(
request_type="STREAM",
name="ttft_authenticated_chat_stream",
response_time=ttft_ms,
response_length=bytes_received,
exception=None,
context={}
)
events.request.fire(
request_type="STREAM",
name="total_authenticated_chat_stream",
response_time=total_ms,
response_length=bytes_received,
exception=None,
context={}
)
except Exception as e:
response.failure(f"Exception while reading stream: {e}")This test is valuable because it includes the full user flow, not just the inference endpoint. In real-world load testing, authentication overhead can materially affect perceived performance.
Scenario 2: Multi-turn conversation with context growth
LLM performance often degrades as context windows grow. A single-turn prompt may stream quickly, while a 10-message conversation can produce much worse TTFT and completion time. This scenario simulates multi-turn chat sessions.
from locust import HttpUser, task, between, events
import json
import time
import random
class MultiTurnStreamingUser(HttpUser):
wait_time = between(3, 6)
host = "https://chat.example-llm.com"
def on_start(self):
self.headers = {
"Authorization": "Bearer YOUR_API_TOKEN",
"Content-Type": "application/json",
"Accept": "text/event-stream"
}
self.conversation = [
{"role": "system", "content": "You are a technical support assistant for a SaaS platform."}
]
@task
def continue_conversation(self):
user_message = random.choice([
"My webhook deliveries are failing with HTTP 401. What should I check?",
"How do I rotate API keys without downtime?",
"Why would request latency spike during peak traffic hours?",
"Can you explain how to configure retry logic for failed jobs?"
])
self.conversation.append({"role": "user", "content": user_message})
payload = {
"model": "gpt-4.1-mini",
"stream": True,
"messages": self.conversation,
"temperature": 0.3,
"max_tokens": 220
}
start = time.time()
first_token = None
assistant_reply = []
chunk_counter = 0
with self.client.post(
"/v1/chat/completions",
json=payload,
headers=self.headers,
stream=True,
catch_response=True,
timeout=120
) as response:
if response.status_code != 200:
response.failure(f"Bad status: {response.status_code}")
return
try:
for line in response.iter_lines(decode_unicode=True):
if not line or not line.startswith("data: "):
continue
data = line[6:].strip()
if data == "[DONE]":
break
if first_token is None:
first_token = time.time()
chunk_counter += 1
try:
event = json.loads(data)
delta = event["choices"][0].get("delta", {})
content = delta.get("content")
if content:
assistant_reply.append(content)
except Exception as e:
response.failure(f"Chunk parse error: {e}")
return
if first_token is None:
response.failure("No streamed content")
return
reply_text = "".join(assistant_reply)
self.conversation.append({"role": "assistant", "content": reply_text})
if len(self.conversation) > 9:
self.conversation = [self.conversation[0]] + self.conversation[-8:]
response.success()
events.request.fire(
request_type="STREAM",
name="ttft_multiturn_chat",
response_time=int((first_token - start) * 1000),
response_length=len(reply_text),
exception=None,
context={}
)
events.request.fire(
request_type="STREAM",
name="duration_multiturn_chat",
response_time=int((time.time() - start) * 1000),
response_length=chunk_counter,
exception=None,
context={}
)
except Exception as e:
response.failure(f"Streaming failed: {e}")This script is useful for stress testing context-heavy workloads, which are common in support bots, copilots, and enterprise assistants.
Scenario 3: Mixed workload with short and long prompts
Most production AI systems do not receive identical prompts. Some users ask simple questions, while others submit large documents or complex instructions. A mixed workload gives you more realistic performance testing data.
from locust import HttpUser, task, between, events
import json
import time
import random
SHORT_PROMPTS = [
"Define vector embeddings.",
"What is prompt injection?",
"Explain tokenization."
]
LONG_PROMPTS = [
"""Analyze the following incident summary and produce a root cause analysis:
At 14:03 UTC, API latency increased from 180 ms p95 to 3.2 s p95.
Background job queues grew by 8x. Database CPU hit 92%.
The deployment at 13:55 UTC introduced a new query path for account-level analytics.
Recommend immediate mitigation steps and long-term fixes.""",
"""Review this product requirement and identify implementation risks:
We want to add real-time collaborative editing, audit logs, role-based access control,
autosave every 2 seconds, and export to PDF for documents up to 200 pages.
Our current backend uses PostgreSQL, Redis, and a monolithic API service."""
]
class MixedLLMStreamingUser(HttpUser):
wait_time = between(1, 2)
host = "https://inference.example-ai.net"
def on_start(self):
self.headers = {
"Authorization": "Bearer YOUR_API_TOKEN",
"Content-Type": "application/json",
"Accept": "text/event-stream"
}
@task(3)
def short_prompt_stream(self):
self.run_stream_test(
prompt=random.choice(SHORT_PROMPTS),
workload_name="short_prompt_stream",
max_tokens=120
)
@task(1)
def long_prompt_stream(self):
self.run_stream_test(
prompt=random.choice(LONG_PROMPTS),
workload_name="long_prompt_stream",
max_tokens=500
)
def run_stream_test(self, prompt, workload_name, max_tokens):
payload = {
"model": "claude-3-5-sonnet",
"stream": True,
"input": prompt,
"max_tokens": max_tokens,
"metadata": {
"tenant_id": "acme-prod",
"client_app": "web-chat"
}
}
start = time.time()
first_token = None
event_count = 0
with self.client.post(
"/api/v1/generate/stream",
json=payload,
headers=self.headers,
stream=True,
catch_response=True,
timeout=120
) as response:
if response.status_code != 200:
response.failure(f"{workload_name} failed with {response.status_code}")
return
try:
for line in response.iter_lines(decode_unicode=True):
if not line:
continue
event_count += 1
if first_token is None:
first_token = time.time()
if first_token is None:
response.failure(f"{workload_name} returned no stream data")
return
response.success()
events.request.fire(
request_type="STREAM",
name=f"ttft_{workload_name}",
response_time=int((first_token - start) * 1000),
response_length=event_count,
exception=None,
context={}
)
events.request.fire(
request_type="STREAM",
name=f"duration_{workload_name}",
response_time=int((time.time() - start) * 1000),
response_length=event_count,
exception=None,
context={}
)
except Exception as e:
response.failure(f"{workload_name} stream exception: {e}")This pattern is especially useful when you want to compare how your LLM infrastructure behaves for lightweight versus context-heavy requests. It also helps identify whether queueing or token generation throughput degrades disproportionately for larger prompts.
Analyzing Your Results
When your test finishes, do not stop at average response time. For LLM streaming responses, the most important analysis should focus on user experience and stream reliability.
Metrics to prioritize
Time to first token
This is often the most important metric for streaming APIs. Watch:
- median TTFT for normal conditions
- p95 and p99 TTFT under load
- TTFT variation by region if using LoadForge global test locations
If TTFT spikes sharply as user count increases, you may have:
- inference queue saturation
- overloaded authentication or routing layers
- insufficient model replicas
- cold starts in serverless GPU infrastructure
Stream duration
Long stream duration is not always bad, especially for longer outputs. What matters is whether duration scales predictably with prompt size and token count.
Look for:
- sudden jumps in duration at specific concurrency levels
- high variance between similar requests
- duration increases without corresponding output length increases
Error rate and stream termination quality
Track:
- 429 Too Many Requests
- 500/502/503 backend failures
- incomplete streams
- missing end-of-stream markers like
[DONE] - client read timeouts
A stream that begins successfully but terminates halfway through is often worse than a fast failure because it damages user trust.
Connection behavior
Streaming APIs keep connections open longer than normal APIs. Under stress testing, this can reveal:
- exhausted connection pools
- reverse proxy limits
- idle timeout misconfiguration
- load balancer behavior under long-lived requests
Using LoadForge reporting effectively
LoadForge’s real-time reporting helps you observe how TTFT and stream duration change while the test is still running. This is particularly helpful for AI & LLM systems where saturation effects may appear gradually rather than instantly. You can also use distributed testing to simulate users from multiple regions and CI/CD integration to catch regressions in streaming performance before deployment.
Performance Optimization Tips
After load testing LLM streaming responses, these are the most common ways to improve results.
Reduce time to first token
- Keep authentication lightweight and cache validation where appropriate
- Warm model instances before peak traffic
- Reduce prompt preprocessing overhead
- Route requests intelligently to available inference workers
- Minimize synchronous logging and middleware in the request path
Improve stream stability
- Ensure proxies and gateways are configured for long-lived streaming connections
- Flush chunks immediately instead of buffering
- Tune keep-alive and idle timeout settings
- Monitor connection pool usage on app servers and gateways
Handle concurrency better
- Scale inference workers horizontally
- Separate short and long prompt workloads into different queues
- Apply fair scheduling to prevent large requests from starving small ones
- Use rate limiting policies that protect the system without breaking normal usage
Optimize prompt and output size
- Trim unnecessary conversation history
- Summarize prior context for long-running chats
- Set realistic
max_tokens - Avoid over-large system prompts unless necessary
Common Pitfalls to Avoid
Load testing streaming APIs can go wrong if your test does not reflect real application behavior.
Treating streams like normal HTTP responses
If you only measure final completion time, you will miss TTFT and token cadence, which are essential for user-perceived performance.
Ignoring authentication and session setup
A test that hits only the model endpoint may underestimate real latency. Include login, token refresh, or tenant routing if those happen in production.
Using unrealistic prompts
Short synthetic prompts may make your system look faster than it really is. Use representative prompt sizes, context lengths, and output expectations.
Not validating stream completion
A 200 status code does not guarantee a healthy stream. Always verify that tokens actually arrive and that the stream finishes correctly.
Overlooking regional effects
LLM streaming performance can vary significantly by geography. Network latency affects how quickly the first token is perceived. LoadForge’s global test locations can help surface these differences.
Running only one concurrency profile
Streaming systems often behave well at low concurrency and degrade suddenly at a threshold. Test ramp-up, sustained load, and stress conditions separately.
Conclusion
Load testing LLM streaming responses requires more than measuring raw API latency. To understand real user experience, you need to track time to first token, stream duration, chunk stability, and failure behavior under concurrency. With realistic Locust scripts, you can simulate how users actually interact with AI & LLM streaming APIs and uncover bottlenecks in inference, routing, authentication, and network delivery.
LoadForge makes this process much easier with cloud-based infrastructure, distributed testing, real-time reporting, global test locations, and CI/CD integration for repeatable performance testing and stress testing. If you are building or scaling streaming AI applications, now is the perfect time to validate how your system performs under real load. Try LoadForge and start load testing your LLM streaming responses with confidence.
LoadForge Team
LoadForge is a load and performance testing platform built on Locust. Our team has been shipping load tests against production systems since 2018, and we write these guides from real customer engagements.
Related guides
Keep going with more guides from the same category.
Load Testing LLM APIs with Streaming and Non-Streaming Requests
Compare streaming and non-streaming LLM API performance under load to find bottlenecks in latency, throughput, and UX.
How to Load Test an AI Gateway
Learn how to load test an AI gateway to validate routing, caching, rate limiting, and multi-model reliability at scale.
How to Load Test Azure OpenAI
Load test Azure OpenAI deployments to validate throughput, response times, quotas, and reliability for enterprise AI applications.