Async embeddings requests in Python with asyncio and aiohttp

Sending embeddings for thousands of documents serially wastes IO latency. This guide shows how to build async embeddings python asyncio pipelines with aiohttp that batch requests concurrently and degrade gracefully when a provider throttles you.

Step 1: Install dependencies and isolate the environment

Use a fresh virtualenv. Only two packages are required: aiohttp for async HTTP and tqdm if you want progress bars (optional). Avoid dumping this into a shared system Python.

python -m venv .venv
source .venv/bin/activate
pip install aiohttp

Avoid the official OpenAI SDK if you want full control over connection pooling and concurrency. The SDK is fine for quick scripts, but aiohttp gives you explicit TCPConnector limits, reusable sessions, and no hidden thread locks. When you fire hundreds of concurrent embedding calls, those knobs matter. If you later need to swap providers, the code below stays identical because it speaks the OpenAI-compatible REST shape.

Step 2: Write the core async request function

The OpenAI-compatible embeddings endpoint expects a JSON body with model and input. The input field accepts a list of strings, but most providers cap batch size per request (often 2048 items or a token limit). We send one list per request and treat the call as atomic.

import aiohttp
import asyncio
from typing import List, Dict, Any

async def embed_batch(
    session: aiohttp.ClientSession,
    base_url: str,
    api_key: str,
    model: str,
    texts: List[str],
) -> List[List[float]]:
    payload = {"model": model, "input": texts}
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }
    async with session.post(f"{base_url}/v1/embeddings", json=payload, headers=headers) as resp:
        if resp.status != 200:
            text = await resp.text()
            raise RuntimeError(f"Embedding failed {resp.status}: {text[:200]}")
        data: Dict[str, Any] = await resp.json()
        # Response shape: {"data": [{"embedding": [...], "index": 0}, ...]}
        return [item["embedding"] for item in sorted(data["data"], key=lambda x: x["index"])]

This function is the atomic unit. It returns embeddings in the same order as texts because we sort by index. Never assume providers return ordered results—some load-balance and reorder. The RuntimeError wraps non-200 responses so the retry layer can catch it.

Step 3: Configure a single endpoint and auth

Hardcode nothing. Read from environment. If you route through a gateway such as n4n.ai, you get one OpenAI-compatible endpoint covering 240+ models with automatic fallback on rate limits, so the base_url stays constant across model swaps and you avoid per-provider branching.

import os

BASE_URL = os.environ.get("EMBED_BASE_URL", "https://api.openai.com")
API_KEY = os.environ["EMBED_API_KEY"]
MODEL = os.environ.get("EMBED_MODEL", "text-embedding-3-small")

A .env file makes local runs reproducible:

export EMBED_BASE_URL="https://api.openai.com"
export EMBED_API_KEY="sk-..."
export EMBED_MODEL="text-embedding-3-small"

Using a single base URL simplifies the aiohttp session: one TCPConnector, one connection pool, no DNS churn. If you later add client routing directives or provider cache-control hints, they forward transparently through a compliant gateway.

Step 4: Chunk inputs and bound concurrency

Throwing 10k texts at a semaphore with no chunking will blow up request size or hit a 413. Split into chunks of 100, then limit in-flight requests to 20 with asyncio.Semaphore. The patterns for async embeddings python asyncio scale to other LLM calls, but embedding batching has stricter token ceilings.

def chunked(items: List[str], size: int):
    for i in range(0, len(items), size):
        yield items[i : i + size]

async def embed_all(texts: List[str], max_concurrency: int = 20, chunk_size: int = 100):
    connector = aiohttp.TCPConnector(limit=max_concurrency)
    timeout = aiohttp.ClientTimeout(total=60)
    sem = asyncio.Semaphore(max_concurrency)

    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        async def bounded_embed(batch):
            async with sem:
                return await embed_batch(session, BASE_URL, API_KEY, MODEL, batch)

        tasks = [bounded_embed(chunk) for chunk in chunked(texts, chunk_size)]
        results = await asyncio.gather(*tasks)
    # results is list of lists; flatten
    return [emb for batch in results for emb in batch]

The TCPConnector(limit=...) and Semaphore are redundant but defensive: the connector caps TCP connections, the semaphore caps pending tasks. Tune chunk_size to your provider’s max batch tokens. If you embed long documents, add a tiktoken estimate and shrink chunks dynamically—don’t guess.

Step 5: Add retries with backoff for 429/5xx

Providers rate-limit embedding endpoints aggressively. Wrap the call with a simple retry that respects Retry-After if present. Catch aiohttp.ClientError separately from RuntimeError so network resets trigger retries too.

import random
from aiohttp import ClientError

async def embed_batch_retry(session, base_url, api_key, model, texts, tries=4):
    for attempt in range(tries):
        try:
            return await embed_batch(session, base_url, api_key, model, texts)
        except (RuntimeError, ClientError) as e:
            if attempt == tries - 1:
                raise
            # crude backoff: 1s, 2s, 4s + jitter
            await asyncio.sleep((2 ** attempt) + random.random())

Swap embed_batch for embed_batch_retry inside bounded_embed. For production, use aiohttp-retry or parse the Retry-After header from the response instead of guessing. Embedding requests are idempotent as long as you don’t change model mid-retry, so blind retries are safe.

Step 6: Run the loop and inspect output

A minimal main that embeds sample sentences and validates dimensions:

async def main():
    sample = [f"Document number {i} about async embeddings python asyncio" for i in range(500)]
    embeddings = await embed_all(sample, max_concurrency=20, chunk_size=50)
    assert len(embeddings) == len(sample), "Count mismatch"
    dim = len(embeddings[0])
    print(f"Got {len(embeddings)} vectors of dim {dim}")
    # Spot-check: first vector should be non-zero
    assert any(x != 0.0 for x in embeddings[0]), "Zero vector returned"

if __name__ == "__main__":
    asyncio.run(main())

Run with python embed.py. Known dimensions: 1536 for text-embedding-3-small, 3072 for text-embedding-3-large, 768 for many open-source models. If your dim is wrong, you called a different model than configured.

Step 7: Verify success and measure qualitatively

Success criteria are concrete:

len(embeddings) == len(texts).
Every vector has the expected dimension.
No exceptions surfaced after retries.

To verify concurrency actually helps, time the run:

time python embed.py

Then compare against a synchronous loop with requests posting the same chunks serially. The async version should finish in a fraction of the wall-clock time because it overlaps network waits. Do not trust micro-benchmarks with tiny inputs; test with at least 1k items. If you use a gateway that provides per-token usage metering, check the response usage field to confirm billed tokens match len(texts) * approx_tokens. The OpenAI-compatible response includes "usage": {"prompt_tokens": N, "total_tokens": N}.

Step 8: Harden with logging and graceful shutdown

Print statements are not observability. Use logging with a JSON formatter, and log the model, chunk size, and status on failures. Wrap asyncio.run in a try/except that cancels pending tasks on KeyboardInterrupt so you don’t leave half-open TCP connections.

import logging
logging.basicConfig(level=logging.INFO)

async def main():
    try:
        # ... embed_all call ...
        logging.info("embed complete", extra={"count": len(embeddings)})
    except asyncio.CancelledError:
        logging.warning("embed interrupted")

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

Common pitfalls we hit in production: missing await on resp.json() causes silent None returns; reusing a session across threads breaks because aiohttp sessions are single-event-loop bound; ignoring index ordering leads to misaligned vectors; too high concurrency triggers 429s that retry storms worsen—start at 10 and increase.

If you embed fewer than 50 items per day, synchronous code is fine. Async embeddings python asyncio pays off only when you batch continuously or serve online requests with mixed IO. For offline ETL, consider a managed batch API instead of hammering the synchronous endpoint. That’s the full pipeline—copy the functions, set env vars, run.