async def embed_document(
        self,
        chunked_documents: list[dict],
        embedding_batch_size: int = 8,
    ) -> AsyncGenerator[VectorEntry, None]:
        """Inline replacement for the old embedding_pipe.run(...).

        Batches the embedding calls and yields VectorEntry objects.
        """
        if not chunked_documents:
            return

        concurrency_limit = (
            self.providers.embedding.config.concurrent_request_limit or 5
        )
        extraction_batch: list[DocumentChunk] = []
        tasks: set[asyncio.Task] = set()

        async def process_batch(
            batch: list[DocumentChunk],
        ) -> list[VectorEntry]:
            # All text from the batch
            texts = [
                (
                    ex.data.decode("utf-8")
                    if isinstance(ex.data, bytes)
                    else ex.data
                )
                for ex in batch
            ]
            # Retrieve embeddings in bulk
            vectors = await self.providers.embedding.async_get_embeddings(
                texts,  # list of strings
            )
            # Zip them back together
            results = []
            for raw_vector, extraction in zip(vectors, batch, strict=False):
                results.append(
                    VectorEntry(
                        id=extraction.id,
                        document_id=extraction.document_id,
                        owner_id=extraction.owner_id,
                        collection_ids=extraction.collection_ids,
                        vector=Vector(data=raw_vector, type=VectorType.FIXED),
                        text=(
                            extraction.data.decode("utf-8")
                            if isinstance(extraction.data, bytes)
                            else str(extraction.data)
                        ),
                        metadata={**extraction.metadata},
                    )
                )
            return results

        async def run_process_batch(batch: list[DocumentChunk]):
            return await process_batch(batch)

        # Convert each chunk dict to a DocumentChunk
        for chunk_dict in chunked_documents:
            extraction = DocumentChunk.from_dict(chunk_dict)
            extraction_batch.append(extraction)

            # If we hit a batch threshold, spawn a task
            if len(extraction_batch) >= embedding_batch_size:
                tasks.add(
                    asyncio.create_task(run_process_batch(extraction_batch))
                )
                extraction_batch = []

            # If tasks are at concurrency limit, wait for the first to finish
            while len(tasks) >= concurrency_limit:
                done, tasks = await asyncio.wait(
                    tasks, return_when=asyncio.FIRST_COMPLETED
                )
                for t in done:
                    for vector_entry in await t:
                        yield vector_entry

        # Handle any leftover items
        if extraction_batch:
            tasks.add(asyncio.create_task(run_process_batch(extraction_batch)))

        # Gather remaining tasks
        for future_task in asyncio.as_completed(tasks):
            for vector_entry in await future_task:
                yield vector_entry
Inline replacement for the old embedding_pipe.run(...).

        Batches the embedding calls and yields VectorEntry objects.
        
py/core/main/services/ingestion_service.py