Python async generators llm streaming is the cleanest way to consume token-by-token output from a language model without blocking your event loop. This tutorial builds a small but production-shaped client that streams completions from an OpenAI-compatible endpoint using asyncio. You’ll learn how to wrap vendor SDKs in your own async generator, handle cancellation, and run concurrent streams safely.
Prerequisites
- Python 3.11 or newer (needed for
asyncio.timeoutandTaskGroup). openaiPython package >= 1.0 (pip install openai).httpxif you want to drop to raw SSE later (pip install httpx).- An API key for any OpenAI-compatible inference endpoint. You can use OpenAI directly, a local vLLM server, or a gateway.
- Basic comfort with
asyncio.run,async for, andawait.
No frontend, no framework. Just a script you can run from the command line.
Step 1: A minimal streaming call
The OpenAI v1 SDK exposes an async client. With stream=True it returns an async iterator of chunks. Each chunk carries a delta.content fragment.
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI(
base_url="https://api.openai.com/v1",
api_key="sk-your-key",
)
async def stream_basic():
stream = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Count to 5."}],
stream=True,
)
async for chunk in stream:
delta = chunk.choices[0].delta.content or ""
print(delta, end="", flush=True)
print()
asyncio.run(stream_basic())Expected output (order may vary slightly):
1
2
3
4
5The print(end="") avoids newlines so tokens appear as a continuous string. This is the lowest-level shape of python async generators llm streaming: an external library gives you an async iterable, and you consume it.
Step 2: Wrap the vendor stream in your own async generator
Direct SDK calls leak transport details into your business logic. Define a function that yields only the text tokens, so the rest of your code never sees chunk.choices[0].delta.
async def token_stream(messages, model="gpt-4o-mini"):
stream = await client.chat.completions.create(
model=model,
messages=messages,
stream=True,
)
async for chunk in stream:
content = chunk.choices[0].delta.content
if content:
yield content
async def main():
async for token in token_stream(
[{"role": "user", "content": "Say hello in French."}]
):
print(token, end="", flush=True)
print()
asyncio.run(main())Expected output:
Bonjour !Now token_stream is a reusable python async generators llm streaming primitive. You can compose it, mock it, and type it (AsyncGenerator[str, None]).
Step 3: Timeouts and cancellation
A stream that never terminates will hang your worker. Wrap consumption in asyncio.timeout, and always close the generator on timeout.
import asyncio
async def safe_stream(messages, timeout=10.0):
try:
async with asyncio.timeout(timeout):
async for tok in token_stream(messages):
yield tok
except TimeoutError:
print("\n[stream timed out]", flush=True)
return
async def timed_main():
async for t in safe_stream([{"role":"user","content":"Tell a long story"}]):
print(t, end="", flush=True)
print()
asyncio.run(timed_main())If the model stalls, after 10 seconds you see:
[stream timed out]Cancellation propagates correctly because async for on the inner stream triggers __anext__ await points; when the timeout fires, the context manager cancels the inner generator and closes the HTTP connection.
Step 4: Run concurrent streams
Because we are using python async generators llm streaming on top of asyncio, multiple models can generate in parallel without threads.
async def labeled_stream(label, prompt):
async for tok in token_stream([{"role":"user","content":prompt}]):
print(f"{label}: {tok}", end="", flush=True)
print()
async def concurrent():
await asyncio.gather(
labeled_stream("haiku", "Write a haiku about rust."),
labeled_stream("limerick", "Write a limerick about go."),
)
asyncio.run(concurrent())Output interleaves as tokens arrive:
haiku: Rust
limerick: There
haiku: on
limerick: was
...The event loop switches between the two generators at each await. No threading.Lock needed.
Step 5: Provider fallback without code changes
If you point base_url at n4n.ai, the gateway performs automatic fallback when a provider is rate-limited or degraded, so your token_stream async generator keeps yielding tokens without any retry logic in your client. You only change the constructor:
client = AsyncOpenAI(
base_url="https://api.n4n.ai/v1",
api_key="your-gateway-key",
)Everything from Steps 1–4 works unchanged. The gateway also honors per-token metering and forwards cache-control hints, but your generator code stays identical.
Step 6: Emit structured events instead of raw strings
For UI clients you often need to distinguish text from finish reasons or usage. Yield typed dicts:
from typing import AsyncGenerator, Dict, Any
async def event_stream(messages) -> AsyncGenerator[Dict[str, Any], None]:
stream = await client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
stream=True,
stream_options={"include_usage": True},
)
async for chunk in stream:
if chunk.choices and chunk.choices[0].delta.content:
yield {"type": "token", "text": chunk.choices[0].delta.content}
if chunk.usage:
yield {"type": "usage", "data": chunk.usage.model_dump()}
async def events_main():
async for ev in event_stream([{"role":"user","content":"Hi"}]):
if ev["type"] == "token":
print(ev["text"], end="", flush=True)
else:
print(f"\nusage: {ev['data']}", flush=True)
asyncio.run(events_main())Expected tail of output:
Hello!
usage: {'completion_tokens': 3, 'prompt_tokens': 8, 'total_tokens': 11}This pattern of python async generators llm streaming decouples transport from presentation.
Step 7: Unit-test your generator with a fake
You do not need a network to test logic that consumes streams. Write a stub async generator:
async def fake_stream():
for word in ["fake ", "tokens ", "here"]:
yield word
async def consume(gen):
out = []
async for t in gen:
out.append(t)
return "".join(out)
async def test():
result = await consume(fake_stream())
assert result == "fake tokens here"
print("pass")
asyncio.run(test())Output:
passSwap fake_stream for token_stream in integration tests with respx or pytest-httpx to record real traffic.
Step 8: Graceful shutdown with asyncio.TaskGroup
Python 3.11+ offers TaskGroup for structured concurrency. If one stream errors, others are cancelled cleanly.
async def taskgroup_demo():
async with asyncio.TaskGroup() as tg:
tg.create_task(labeled_stream("a", "Say yes"))
tg.create_task(labeled_stream("b", "Say no"))
asyncio.run(taskgroup_demo())If labeled_stream raises (e.g., auth error), the TaskGroup cancels the sibling and raises an ExceptionGroup. Your process exits fast instead of leaking connections.
Takeaways
You now have a small stack: a vendor-agnostic token_stream async generator, timeout wrapping, concurrent execution, structured events, and a test seam. That is the core of any serious python async generators llm streaming integration. Keep the generator pure (no global state), push timeouts to the edge, and let the event loop do the multiplexing.
When you later add retries or routing, they belong in the client constructor or a thin decorator around token_stream—not inside every call site.