Streaming and batch

How you send requests matters as much as what you send. A chat interface should show words as soon as they are generated; an evaluation job over ten thousand inputs should not wait for each request in turn; and a nightly classification job does not need answers immediately at all. This lesson covers the three patterns — streaming, concurrency, and batch processing — and when to use each.

Last updated: 01 Nov, 2026

Why streaming?

Definition · Streaming

A response mode where the API sends the reply in small pieces (chunks of tokens) as the model generates them, instead of one complete response at the end.

Definition · Time to first token (TTFT)

How long the user waits before the first piece of the reply appears. Streaming does not make the full answer faster, but it cuts perceived waiting from seconds to a fraction of a second.

A model generating 400 tokens might take eight seconds. Without streaming, the user stares at a spinner for eight seconds. With streaming, text starts appearing almost immediately and they read while the rest is generated.

Python

stream = client.chat.completions.create(
    model="gpt-4.1-mini",
    messages=[{"role": "user", "content": "List three risks of AI agents."}],
    stream=True,
)
parts = []
for chunk in stream:
    delta = chunk.choices[0].delta.content or ""     # some chunks carry no text
    parts.append(delta)
    print(delta, end="", flush=True)
print()
full_reply = "".join(parts)

1. Runaway loops that burn tokens. 2. Wrong or harmful actions taken through tools. 3. Leaking private data through prompts or tool results.

Simulating a stream

A stream is just an iterator that yields text pieces over time, so a generator reproduces it exactly. This lets you build and test streaming UIs offline.

Python

import time


def fake_stream(text: str, delay: float = 0.1):
    for word in text.split(" "):
        time.sleep(delay)              # the model "thinking" about the next token
        yield word + " "


start = time.perf_counter()
first_token_at = None
parts = []
for piece in fake_stream("Agents need limits, logs, and approvals."):
    if first_token_at is None:
        first_token_at = time.perf_counter() - start
    parts.append(piece)
    print(piece, end="", flush=True)
total = time.perf_counter() - start
print()
print(f"first token after {first_token_at:.1f}s, full reply after {total:.1f}s")

Agents need limits, logs, and approvals. first token after 0.1s, full reply after 0.6s

The user saw text after 0.1 s instead of waiting 0.6 s for the whole reply. With real models the gap is much larger.

Watch out

With streaming you cannot validate the full output before showing it. If you must check the answer (for example, JSON that must parse, or a safety filter), either buffer the stream and check at the end, or stream to the user and retract or flag the message if the final check fails.

Concurrency for many independent requests

Definition · Concurrency

Running several requests at overlapping times. LLM calls spend almost all their time waiting on the network, so sending many at once (up to your rate limit) finishes far sooner than sending them one after another.

Definition · Semaphore

A counter that limits how many tasks may run a section of code at the same time. Use one to cap concurrent API calls below your rate limit.

Python

import asyncio
import time


async def fake_classify(ticket: str) -> str:
    await asyncio.sleep(0.3)                       # simulated API latency
    return "billing" if "refund" in ticket or "charge" in ticket else "technical"


async def classify_all(tickets: list[str], max_concurrent: int = 2) -> list[str]:
    limit = asyncio.Semaphore(max_concurrent)

    async def one(ticket: str) -> str:
        async with limit:                          # at most 2 calls in flight
            return await fake_classify(ticket)

    return await asyncio.gather(*(one(t) for t in tickets))   # keeps input order


tickets = ["I want a refund", "Backup fails with E-504", "Double charge on my card", "App crashes on start"]
start = time.perf_counter()
labels = asyncio.run(classify_all(tickets))
print(labels)
print(f"4 calls took {time.perf_counter() - start:.1f}s (sequential would take 1.2s)")

['billing', 'technical', 'billing', 'technical'] 4 calls took 0.6s (sequential would take 1.2s)

asyncio.gather returns results in the same order as the inputs, even though the calls finish in any order. With the real SDK, use AsyncOpenAI and await client.chat.completions.create(...) inside one().

The Batch API for offline jobs

Definition · Batch API

A provider endpoint where you upload a file of many requests and collect the results later (within 24 hours). It costs about half the price of normal requests and has separate, higher rate limits — ideal for evaluations, backfills, and nightly classification.

Each line of the input file is one complete request with your own custom_id, so you can match results back to inputs.

Python

import json

tickets = {"t-1": "I want a refund", "t-2": "Backup fails with E-504"}

with open("batch_input.jsonl", "w", encoding="utf-8") as f:
    for ticket_id, text in tickets.items():
        request = {
            "custom_id": ticket_id,
            "method": "POST",
            "url": "/v1/chat/completions",
            "body": {
                "model": "gpt-4.1-mini",
                "messages": [
                    {"role": "system", "content": "Classify the ticket as billing or technical. Reply with one word."},
                    {"role": "user", "content": text},
                ],
                "temperature": 0,
            },
        }
        f.write(json.dumps(request) + "\n")

with open("batch_input.jsonl", encoding="utf-8") as f:
    lines = f.readlines()
print(len(lines), "requests")
print(json.loads(lines[0])["custom_id"], "->", json.loads(lines[0])["body"]["messages"][1]["content"])

2 requests t-1 -> I want a refund

Python

batch_file = client.files.create(file=open("batch_input.jsonl", "rb"), purpose="batch")
batch = client.batches.create(
    input_file_id=batch_file.id, endpoint="/v1/chat/completions", completion_window="24h"
)
print(batch.id, batch.status)

# later - poll until finished, then download the results file
batch = client.batches.retrieve(batch.id)
if batch.status == "completed":
    for line in client.files.content(batch.output_file_id).text.splitlines():
        result = json.loads(line)
        print(result["custom_id"], result["response"]["body"]["choices"][0]["message"]["content"])

batch_abc123 validating ... t-1 billing t-2 technical

Choosing a pattern

Pattern	Use when	Latency	Cost
Single call	One request, answer needed now, no UI	Full generation time	Standard
Streaming	A person is watching the answer appear	Very low time to first token	Standard
Concurrent calls	Many independent requests needed within minutes	About the slowest single call per wave	Standard
Batch API	Large jobs that can wait hours	Up to 24 hours	About 50% cheaper

Try it

Change max_concurrent to 1 and then 4 and predict the run time before you run it.
Add a 10% random failure to fake_classify and retry failed tickets with backoff.
Write a function that reads a Batch API output file and returns {custom_id: answer}, recording any errored lines separately.

Key takeaways

Stream whenever a person is waiting for the answer; it cuts perceived latency dramatically.
Streaming means you cannot validate the full answer before it is shown — plan for that.
Run independent calls concurrently with asyncio and cap them with a semaphore.
Use the Batch API for large jobs that can wait: half the price and separate rate limits.
Always attach your own ids to batched requests so results can be matched to inputs.