From 0d12a1339b6219a3c71ab34245aff555349ffa89 Mon Sep 17 00:00:00 2001 From: Rudy Celekli <47457359+rudycelekli@users.noreply.github.com> Date: Fri, 2 Oct 2026 14:21:53 -0400 Subject: [PATCH] Match embedding responses to their original chunk indices Signed-off-by: Rudy Celekli <47457359+rudycelekli@users.noreply.github.com> --- engineering/engineering-rag-pipeline-engineer.md | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/engineering/engineering-rag-pipeline-engineer.md b/engineering/engineering-rag-pipeline-engineer.md index 5b85882b..694c848f 100644 --- a/engineering/engineering-rag-pipeline-engineer.md +++ b/engineering/engineering-rag-pipeline-engineer.md @@ -160,7 +160,11 @@ async def embed_batch(texts: list[str], batch_size: int = 100) -> list[list[floa input=batch, model="text-embedding-3-small" ) - all_embeddings.extend([r.embedding for r in response.data]) + # The API supplies each embedding's input index; do not rely on wire order. + indexed = sorted(response.data, key=lambda item: item.index) + if [item.index for item in indexed] != list(range(len(batch))): + raise ValueError('Incomplete or duplicate embedding indices') + all_embeddings.extend(item.embedding for item in indexed) return all_embeddings async def ingest_document(document_id: str, chunks: list[dict], pool: asyncpg.Pool): @@ -170,6 +174,8 @@ async def ingest_document(document_id: str, chunks: list[dict], pool: asyncpg.Po """ texts = [c["content"] for c in chunks] embeddings = await embed_batch(texts) + if len(embeddings) != len(chunks): + raise ValueError('Embedding count must match chunk count before inserting') async with pool.acquire() as conn: await register_vector(conn)