Most LLM integrations block the event loop or swallow streaming errors. This tutorial builds a production-shaped python asyncopenai async llm client that streams tokens, handles concurrency, and reports usage without contorting your asyncio code. We’ll use the official OpenAI SDK’s async interface against an OpenAI-compatible endpoint.
Prerequisites
- Python 3.10 or newer (asyncio timeout and task primitives are stable).
pip install "openai>=1.30.0"for theAsyncOpenAIclient.- An API key for an OpenAI-compatible inference gateway. For example, n4n.ai exposes one OpenAI-compatible endpoint that addresses 240+ models, applies automatic fallback when a provider is rate-limited or degraded, and returns per-token usage metering.
- Comfort with
async def,await, andasyncio.run.
Configure the async client
The AsyncOpenAI class mirrors the sync client but every network call is awaitable. Instantiate it once at module scope; the client manages its own connection pool.
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI(
api_key="sk-your-key",
base_url="https://api.n4n.ai/v1", # swap for any compatible base
)
async def simple_call():
resp = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Return the word 'pong'."}],
)
print(resp.choices[0].message.content)
if __name__ == "__main__":
asyncio.run(simple_call())Expected output:
pongStream tokens with async iteration
Non-streaming calls waste round-trips for long generations. Set stream=True and iterate asynchronously. Each chunk carries a delta with incremental content.
async def stream_tokens():
stream = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Spell 'async' letter by letter."}],
stream=True,
)
async for chunk in stream:
delta = chunk.choices[0].delta.content or ""
print(delta, end="", flush=True)
print()
asyncio.run(stream_tokens())Expected streaming output
a
s
y
n
cTokens arrive incrementally; your terminal shows them concatenated without newlines. The chunk.choices[0].finish_reason is None until the final packet.
Fire concurrent requests
asyncio shines when you batch independent prompts. Use asyncio.gather to run many completions in parallel without threads. Bound parallelism with a Semaphore to avoid overwhelming the gateway.
async def batch(prompts, limit=5):
sem = asyncio.Semaphore(limit)
async def one(p):
async with sem:
r = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": p}],
)
return r.choices[0].message.content
return await asyncio.gather(*(one(p) for p in prompts))
prompts = ["Say A", "Say B", "Say C"]
results = asyncio.run(batch(prompts))
print(results)Expected output (order preserved):
['A', 'B', 'C']Enforce timeouts
A hung connection shouldn’t stall your service. Wrap calls in asyncio.wait_for to cancel the underlying request when the deadline passes.
async def bounded_call():
try:
return await asyncio.wait_for(simple_call(), timeout=5.0)
except asyncio.TimeoutError:
print("provider too slow")
asyncio.run(bounded_call())Cancellation propagates: the AsyncOpenAI client aborts the HTTP request on CancelledError.
Meter token usage
The response object carries usage on non-streaming calls. For streaming, request usage in the final chunk via stream_options.
async def with_usage():
resp = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Explain asyncio in one line."}],
)
print(resp.usage.model_dump())
async def stream_with_usage():
stream = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Count to 3."}],
stream=True,
stream_options={"include_usage": True},
)
async for chunk in stream:
if chunk.usage:
print("USAGE:", chunk.usage.model_dump())
asyncio.run(with_usage())
asyncio.run(stream_with_usage())Expected output shape for with_usage:
{"prompt_tokens": 12, "completion_tokens": 18, "total_tokens": 30}Exact counts vary by model and prompt.
Wrap it in a client class
A small wrapper hides setup and centralizes error policy. Use an async context manager if you want explicit lifecycle control.
class LLMClient:
def __init__(self, api_key, base_url, model="gpt-4o-mini"):
self.client = AsyncOpenAI(api_key=api_key, base_url=base_url)
self.model = model
async def complete(self, prompt, stream=False):
return await self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
stream=stream,
)
async def stream_text(self, prompt):
stream = await self.complete(prompt, stream=True)
async for chunk in stream:
yield chunk.choices[0].delta.content or ""
async def demo():
llm = LLMClient("sk-your-key", "https://api.n4n.ai/v1")
async for tok in llm.stream_text("List three Python async primitives."):
print(tok, end="", flush=True)
asyncio.run(demo())Routing and cache hints
Some gateways let you pin a provider or leverage cache-control per request. n4n.ai honors client routing directives and forwards provider cache-control hints, so you can pass extra headers or model suffixes without custom middleware.
resp = await client.chat.completions.create(
model="gpt-4o-mini@provider-x",
messages=[{"role": "user", "content": "Hi"}],
extra_headers={"cache-control": "max-age=300"},
)Error handling and fallback
Even with a gateway that provides automatic fallback, your client should catch APIError and retry with backoff. The SDK raises APIConnectionError, RateLimitError, and APIStatusError as subclasses of APIError.
from openai import APIError
async def resilient_call(prompt, retries=3):
for i in range(retries):
try:
return await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
)
except APIError as e:
if i == retries - 1:
raise
await asyncio.sleep(0.5 * (2 ** i))
asyncio.run(resilient_call("Test"))Cancellation and cleanup
When your process shuts down, pending streams should be cancelled. asyncio.TaskGroup (3.11+) or explicit task.cancel() prevents orphaned connections.
async def supervised():
async with asyncio.TaskGroup() as tg:
tg.create_task(stream_tokens())
tg.create_task(stream_tokens())
# Cancelling the outer task cancels children.Testing the client
Use pytest-asyncio and mock the transport. The AsyncOpenAI client accepts a http_client argument where you can inject httpx.AsyncMockTransport to return canned responses.
import httpx
from openai import AsyncOpenAI
def mock_client():
transport = httpx.AsyncMockTransport([
httpx.MockResponse(200, json={"choices": [{"message": {"content": "ok"}}]})
])
return AsyncOpenAI(api_key="x", base_url="http://test", http_client=httpx.AsyncClient(transport=transport))Where to take it next
You now have a python asyncopenai async llm client that streams, batches, times out, and meters. Extend the LLMClient with your own retry and logging policies; the async core stays unchanged. Add structured output parsing, prompt templating, or a queue-backed worker pool as your system grows.