Orchestrating a python asyncio stream multiple models fan-out lets you compare responses from different LLMs in real time without blocking on the slowest one. This tutorial builds a small concurrent streaming client against an OpenAI-compatible endpoint, showing how to fire off several model requests and merge their token streams as they arrive. You will end up with a pattern that scales to any number of models behind a single API.
Prerequisites
- Python 3.10 or newer (uses
asyncioprimitives;TaskGroupoptional but we stick togatherfor broad compatibility). openaiPython package >= 1.0 (pip install openai).- An API key for an OpenAI-compatible inference gateway. We’ll use
n4n.aibecause it exposes 240+ models behind one OpenAI-compatible URL and handles provider fallback automatically, but the code works against any compliant server. - Basic comfort with
async/awaitand event loops.
Set the key in your environment:
export N4N_API_KEY="sk-..."Client setup
Instantiate the async client once. Reusing a single client matters: it keeps a connection pool and avoids TLS handshake overhead per request.
import os
from openai import AsyncOpenAI
client = AsyncOpenAI(
base_url="https://api.n4n.ai/v1",
api_key=os.environ["N4N_API_KEY"],
)If you target a different provider, change base_url and the env var name. The rest of the code is identical.
Streaming one model
A single-model stream is a coroutine that pulls deltas and pushes them into a shared asyncio.Queue. Using a queue decouples production (network I/O) from consumption (printing or aggregating), which is the cleanest way to interleave multiple streams.
import asyncio
async def stream_model(model: str, prompt: str, queue: asyncio.Queue):
try:
stream = await client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
stream=True,
temperature=0.7,
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
await queue.put((model, delta))
except Exception as e:
await queue.put((model, f"\n[error {model}: {e}]\n"))
finally:
await queue.put((model, None)) # sentinel marks end of this modelThe sentinel (None payload) lets the consumer know when a model finished, without relying on task ordering.
Fan-out concurrency
The core of python asyncio stream multiple models is running stream_model for each model as a separate task and consuming the queue concurrently. asyncio.gather runs them together; the event loop multiplexes the network reads.
async def main():
models = ["gpt-4o-mini", "llama-3.1-8b-instruct", "mistral-7b-instruct"]
prompt = "Explain asyncio in one sentence."
queue: asyncio.Queue = asyncio.Queue()
producers = [asyncio.create_task(stream_model(m, prompt, queue)) for m in models]
async def consume():
pending = len(models)
while pending:
model, data = await queue.get()
if data is None:
pending -= 1
print(f"\n--- {model} stream closed ---")
else:
print(f"[{model}] {data}", end="", flush=True)
await asyncio.gather(*producers, consume())Run it with asyncio.run(main()). The output interleaves tokens from all three models as they arrive, like a multi-column terminal feed.
Expected output at checkpoint
A typical first run prints something like:
[gpt-4o-mini] Asyncio is a Python library for writing concurrent code using the async/await syntax.
[llama-3.1-8b-instruct] asyncio lets you handle many I/O-bound tasks concurrently without threads.
[mistral-7b-instruct] It provides an event loop to schedule coroutines.
[gpt-4o-mini] It uses an event loop to manage tasks.
--- gpt-4o-mini stream closed ---
--- llama-3.1-8b-instruct stream closed ---
--- mistral-7b-instruct stream closed ---Order varies per run. The key point: no model waits for another to finish before emitting tokens.
Adding timeouts
A stalled provider shouldn’t hang the whole fan-out. Wrap each stream in asyncio.wait_for with a per-model budget.
async def stream_model(model: str, prompt: str, queue: asyncio.Queue, timeout=30.0):
try:
await asyncio.wait_for(_stream_inner(model, prompt, queue), timeout=timeout)
except asyncio.TimeoutError:
await queue.put((model, f"\n[timeout {model}]\n"))
finally:
await queue.put((model, None))
async def _stream_inner(model, prompt, queue):
stream = await client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
stream=True,
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
await queue.put((model, delta))If a model exceeds 30s to first/any chunk, its task is cancelled and the consumer notes the timeout.
Graceful fallback
Even with a gateway that performs automatic fallback when a provider is rate-limited, you should own failure modes client-side. A simple retry with a secondary model keeps the demo robust:
async def stream_with_fallback(primary, fallback, prompt, queue):
try:
await _stream_inner(primary, prompt, queue)
except Exception:
await queue.put((primary, f"\n[fallback {primary} -> {fallback}]\n"))
await _stream_inner(fallback, prompt, queue)
finally:
await queue.put((primary, None))When pointed at n4n.ai, the endpoint may already have routed around a dead provider; the client fallback is a second layer for hard API errors.
Full runnable script
import os
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI(
base_url="https://api.n4n.ai/v1",
api_key=os.environ["N4N_API_KEY"],
)
async def _stream_inner(model, prompt, queue):
stream = await client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
stream=True,
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
await queue.put((model, delta))
async def stream_model(model, prompt, queue, timeout=30.0):
try:
await asyncio.wait_for(_stream_inner(model, prompt, queue), timeout=timeout)
except Exception as e:
await queue.put((model, f"\n[error {model}: {e}]\n"))
finally:
await queue.put((model, None))
async def main():
models = ["gpt-4o-mini", "llama-3.1-8b-instruct", "mistral-7b-instruct"]
prompt = "Explain asyncio in one sentence."
queue: asyncio.Queue = asyncio.Queue()
producers = [asyncio.create_task(stream_model(m, prompt, queue)) for m in models]
async def consume():
pending = len(models)
while pending:
model, data = await queue.get()
if data is None:
pending -= 1
print(f"\n--- {model} done ---")
else:
print(f"[{model}] {data}", end="", flush=True)
await asyncio.gather(*producers, consume())
if __name__ == "__main__":
asyncio.run(main())Operational notes
- Connection limits: The openai client uses a default connection pool. Streaming many models simultaneously is I/O bound, but if you fan out to 50+ models, set
max_connectionsviahttpxlimits passed toAsyncOpenAI. - Backpressure:
asyncio.Queueis unbounded by default. For long-running services, bound it (asyncio.Queue(maxsize=1000)) andawait queue.putwill exert backpressure on slow consumers. - Ordering: Tokens from a single model arrive in order; cross-model order is arbitrary by design. If you need aligned columns, buffer per model and flush on newline.
- Cost metering: Gateway per-token usage metering (as provided by n4n.ai) still applies to streams; you can read
chunk.usageon the final chunk if the provider sends it, but withstream=Trueusage is often emitted on the last delta.
The python asyncio stream multiple models pattern is not just for side-by-side demos. It powers parallel candidate generation, multi-model voting, and latency hedging in production inference routes. Build the queue-first design once and you can swap models without touching the concurrency logic.