When you build LLM features, the difference between a snappy UI and a janky one is how you handle the firehose of text coming back from the model. Using a python asyncio queue streaming tokens approach lets you decouple network I/O from rendering, batching, or downstream processing without blocking the event loop.
Below is an ordered path from a minimal producer/consumer pair to patterns that survive real workloads.
Why buffer streamed tokens at all
LLM APIs emit tokens in chunks at variable rates. Your consumer—whether it writes to a WebSocket, feeds a text-to-speech engine, or persists to a database—rarely runs at the exact same speed as the network delivers bytes.
Without a buffer, you either block the receiving coroutine on slow writes or drop tokens. An asyncio.Queue sits between the two and absorbs bursts. It also gives you a clean seam for cancellation: kill the consumer task and the producer can detect the drain, or vice versa.
The queue is not free. It adds a small per-item overhead and introduces a place where state can get stuck if you ignore shutdown. But for any service that faces unpredictable clients, that trade is worth it.
Basic pattern: one producer, one consumer
The minimal setup is a coroutine that reads the stream and puts each token, and a second coroutine that gets and processes.
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def produce(queue: asyncio.Queue, prompt: str):
stream = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
stream=True,
)
async for chunk in stream:
token = chunk.choices[0].delta.content
if token:
await queue.put(token)
async def consume(queue: asyncio.Queue):
while True:
token = await queue.get()
# simulate slow consumer (websocket send, TTS, etc.)
print(token, end="", flush=True)
queue.task_done()
async def main():
q: asyncio.Queue = asyncio.Queue()
prod = asyncio.create_task(produce(q, "Explain asyncio queues."))
cons = asyncio.create_task(consume(q))
await prod
await q.join() # wait for all tokens processed
cons.cancel()
asyncio.run(main())This works for a script, but the consumer loop never exits on its own. In a long-running service you need a shutdown signal, which we cover later.
The simplest python asyncio queue streaming tokens setup uses a single coroutine producing into the queue and another consuming; it already shows the core decoupling.
Handling backpressure with maxsize
The default asyncio.Queue is unbounded. If your consumer stalls for 10 seconds, the producer keeps filling memory. Set maxsize to apply backpressure: await queue.put() suspends the producer until the consumer catches up.
q: asyncio.Queue = asyncio.Queue(maxsize=200)Tradeoff: a slow or dead consumer now blocks the producer. If the stream connection has its own timeout, the put may raise or the request may fail. Wrap critical puts with a timeout:
try:
await asyncio.wait_for(queue.put(token), timeout=5.0)
except asyncio.TimeoutError:
# consumer is stuck; break the stream
breakDo not use put_nowait unless you are prepared to handle asyncio.QueueFull and implement your own drop/latest-value policy. For token streaming, dropping early tokens corrupts output, so blocking is usually safer than dropping.
Merging multiple streams with a single queue
Sometimes you run parallel requests—multiple model candidates, or the same prompt to different providers for fallback. A single queue can merge them, but you must tag each token with its source if attribution matters.
class TaggedToken:
def __init__(self, source: str, text: str):
self.source = source
self.text = text
async def produce_tagged(queue: asyncio.Queue, source: str, prompt: str):
stream = await client.chat.completions.create(
model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], stream=True
)
async for chunk in stream:
if chunk.choices[0].delta.content:
await queue.put(TaggedToken(source, chunk.choices[0].delta.content))
async def merge_consumer(queue: asyncio.Queue):
while True:
item = await queue.get()
# interleaved sources; handle per-source buffering if needed
print(f"[{item.source}] {item.text}", end="", flush=True)
queue.task_done()If you query several models behind a single OpenAI-compatible endpoint such as n4n.ai—which fronts 240+ models and auto-falls back on provider errors—you can still treat each response stream as a producer feeding the same queue. The queue logic does not care which backend served the token.
Batching tokens without losing latency
A raw token-by-token consumer is simple but can hammer downstream systems that prefer larger chunks (e.g., a sentence segmenter or a TTS engine). Wrap the consumer with a timed batch collector.
async def batch_consumer(queue, max_batch=5, max_wait=0.2):
batch = []
while True:
try:
token = await asyncio.wait_for(queue.get(), timeout=max_wait)
except asyncio.TimeoutError:
if batch:
emit(batch)
batch.clear()
continue
batch.append(token)
if len(batch) >= max_batch:
emit(batch)
batch.clear()The timeout ensures partial batches flush even if the stream trickles. You trade a little latency for fewer writes—usually a good deal for network egress.
Graceful shutdown and cancellation
Production code needs a defined end. The cleanest pattern is a sentinel value or an asyncio.Event.
SENTINEL = None
async def produce_with_stop(queue, prompt, stop_event):
stream = await client.chat.completions.create(
model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], stream=True
)
async for chunk in stream:
if stop_event.is_set():
break
token = chunk.choices[0].delta.content
if token:
await queue.put(token)
await queue.put(SENTINEL)
async def consume_until_sentinel(queue):
while True:
item = await queue.get()
if item is SENTINEL:
queue.task_done()
break
process(item)
queue.task_done()When the API stream ends naturally, the producer puts the sentinel. On external shutdown, set stop_event and cancel the producer task; the consumer still drains remaining items then exits on sentinel or on queue.get() cancellation.
Pitfall: if you task.cancel() the consumer without a sentinel, you may leave tokens in the queue and queue.join() will hang forever because task_done() was never called for them. Always use try/finally around get to call task_done(), or rely on sentinel draining.
Common pitfalls and tradeoffs
Unbounded queues. Easy to forget maxsize. Under load, a stalled WebSocket client will silently eat RAM until OOM. Set a limit and decide explicitly what happens when full.
Blocking calls in the consumer. Calling time.sleep, synchronous DB drivers, or heavy CPU work inside consume blocks the entire event loop. Offload to asyncio.to_thread or use async libraries.
Missing task_done. Every get must pair with task_done (or use join carefully). A mismatch hangs queue.join() and masks shutdown bugs.
Cross-stream ordering. A single queue preserves FIFO per producer, but when merging multiple producers, tokens from different streams interleave non-deterministically. If you need per-stream ordering at the output, keep separate queues or buffer per source in the consumer.
Cancellation leaks. asyncio.CancelledError inherits from BaseException. If you catch it broadly and swallow, tasks never terminate. Use try: ... except asyncio.CancelledError: raise or finally for cleanup only.
Sentinel vs. event. A sentinel works for single-consumer loops. For multiple consumers, use a broadcast asyncio.Event or inject one sentinel per consumer count.
When not to use a queue
If your consumer is purely a fast async def that writes to an async WebSocket and never blocks, you can iterate the stream directly without a queue. The python asyncio queue streaming tokens pattern earns its complexity only when there is a real speed mismatch, a need to merge/fan-out, or a requirement for clean cancellation boundaries.
For most LLM gateway integrations, the queue sits at the edge between the network stream and your application logic. Keep it bounded, tag merged sources, and always define how it empties. That is the difference between a demo and a service that survives a slow client.