async def run_pipeline( source: DataBatcher, index: VectorIndex, batch_size: int = 512 ) -> PipelineResult: embedder = Embedder("text-embedding-3-large") async for batch in source.stream(batch_size): vecs = await embedder.encode(batch) await index.upsert(vecs, namespace="prod") return PipelineResult(status="ok") class VectorIndex: def __init__(self, dim: int = 3072): self.dim = dim self.store = pgvector.PGStore( dsn=settings.DATABASE_URL, table="embeddings_prod", dimension=dim, ) async def upsert( self, vectors: list[EmbedResult], namespace: str = "default", ) -> UpsertResult: records = [ VectorRecord( id=v.id, embedding=v.vector, metadata=v.metadata, namespace=namespace, ) for v in vectors ] return await self.store.batch_insert(records) @router.post("/v2/query") async def query_endpoint( req: QueryRequest, index: VectorIndex = Depends(get_index), embedder: Embedder = Depends(get_embedder), ) -> QueryResponse: vec = await embedder.encode_single(req.query) results = await index.search( vector=vec, k=req.top_k, namespace=req.namespace, filter=req.filter, ) return QueryResponse( results=results, latency_ms=ctx.elapsed_ms(), )