Notes

omparing your implementation against the five recommendations in the table:

Recommendation Status Comments
✅ 1. Use asyncio + AsyncOpenAI for embeddings ❌ Not implemented You're still using the synchronous OpenAI client. get_embeddings() blocks until the API returns. No async requests are being made.
✅ 2. Make thread pool configurable (max_workers=min(32, os.cpu_count()+4)) ✅ Implemented You have MAX_WORKERS = int(os.getenv("MAX_WORKERS", min(32, os.cpu_count() + 4))). This matches the recommendation.
✅ 3. Producer/consumer pipeline between embedding and upsert ❌ Not implemented Your code first embeds all documents, stores all embeddings, builds all points, and only then starts upserting. There is no streaming pipeline or queue.
✅ 4. Stream batches instead of building everything first ❌ Not implemented Same issue as above. Memory usage is still O(n) because you keep every embedding and every point in memory before upserting.
✅ 5. Cache duplicate embeddings ❌ Not implemented Every document is sent to OpenAI, even if duplicates exist. There is no LRU cache or dictionary lookup.


"""
OpenAI Connection
"""

import os
from typing import List

from dotenv import load_dotenv
from openai import AsyncOpenAI

load_dotenv()

openai_client = AsyncOpenAI(
api_key=os.getenv("OPENAI_API_KEY")
)

EMBEDDING_MODEL = "text-embedding-3-large"


async def get_embeddings(texts: List[str]) -> List[List[float]]:
"""Batch embeddings."""
response = await openai_client.embeddings.create(
model=EMBEDDING_MODEL,
input=texts,
)
return [item.embedding for item in response.data]

2. Add an embedding cache

Near the top of your ingestion file:

from functools import lru_cache

_embedding_cache = {}
async def cached_embeddings(texts):
"""
Only embed unseen texts.
"""

missing = [t for t in texts if t not in _embedding_cache]

if missing:
vectors = await get_embeddings(missing)

for text, vec in zip(missing, vectors):
_embedding_cache[text] = vec

return [_embedding_cache[t] for t in texts]

Now duplicate documents never call OpenAI twice.

3. Producer
import asyncio


async def producer(
queue: asyncio.Queue,
documents,
batch_size,
):
"""
Generate embedding batches.
"""

for i in range(0, len(documents), batch_size):

docs = documents[i:i + batch_size]

embeddings = await cached_embeddings(docs)

points = [
PointStruct(
id=str(uuid.uuid4()),
vector=emb,
payload={"text": doc},
)
for doc, emb in zip(docs, embeddings)
]

await queue.put(points)

await queue.put(None)

Notice:

only one batch exists
embeddings are immediately converted to PointStruct
nothing accumulates in RAM
4. Consumer
async def consumer(
queue,
collection_name,
):
while True:

batch = await queue.get()

if batch is None:
break

await asyncio.to_thread(
safe_upsert,
collection_name,
batch,
)

queue.task_done()

The upsert runs in a background thread while the producer is already embedding the next batch.

5. Main

Replace the entire bottom half of main() with

queue = asyncio.Queue(maxsize=5)

await asyncio.gather(

producer(
queue,
documents,
args.batch_size,
),

consumer(
queue,
args.collection,
),
)
6. Entry point

Replace

main()

with

import asyncio

asyncio.run(main())

and

def main():

becomes

async def main():
7. Remove

These become unnecessary:

embeddings = get_embeddings(documents)

points = ...

batches = ...

and

ThreadPoolExecutor

because asyncio.Queue already pipelines the work.

8. Optional: Multiple consumers

For larger datasets, start several consumers:

workers = [
asyncio.create_task(
consumer(queue, args.collection)
)
for _ in range(5)
]

await producer(
queue,
documents,
args.batch_size,
)

await queue.join()

for _ in workers:
await queue.put(None)

await asyncio.gather(*workers)

This allows several Qdrant upserts to proceed concurrently while embeddings for later batches are still being generated.