async def _get_enriched_chunk_text(
        self,
        chunk_idx: int,
        chunk: dict,
        document_id: UUID,
        document_summary: str | None,
        chunk_enrichment_settings: ChunkEnrichmentSettings,
        list_document_chunks: list[dict],
    ) -> VectorEntry:
        """Helper for chunk_enrichment.

        Leverages an LLM to rewrite or expand chunk text, then re-embeds it.
        """
        preceding_chunks = [
            list_document_chunks[idx]["text"]
            for idx in range(
                max(0, chunk_idx - chunk_enrichment_settings.n_chunks),
                chunk_idx,
            )
        ]
        succeeding_chunks = [
            list_document_chunks[idx]["text"]
            for idx in range(
                chunk_idx + 1,
                min(
                    len(list_document_chunks),
                    chunk_idx + chunk_enrichment_settings.n_chunks + 1,
                ),
            )
        ]
        try:
            # Obtain the updated text from the LLM
            updated_chunk_text = (
                (
                    await self.providers.llm.aget_completion(
                        messages=await self.providers.database.prompts_handler.get_message_payload(
                            task_prompt_name=chunk_enrichment_settings.chunk_enrichment_prompt,
                            task_inputs={
                                "document_summary": document_summary or "None",
                                "chunk": chunk["text"],
                                "preceding_chunks": (
                                    "\n".join(preceding_chunks)
                                    if preceding_chunks
                                    else "None"
                                ),
                                "succeeding_chunks": (
                                    "\n".join(succeeding_chunks)
                                    if succeeding_chunks
                                    else "None"
                                ),
                                "chunk_size": self.config.ingestion.chunk_size
                                or 1024,
                            },
                        ),
                        generation_config=chunk_enrichment_settings.generation_config
                        or GenerationConfig(model=self.config.app.fast_llm),
                    )
                )
                .choices[0]
                .message.content
            )
        except Exception:
            updated_chunk_text = chunk["text"]
            chunk["metadata"]["chunk_enrichment_status"] = "failed"
        else:
            chunk["metadata"]["chunk_enrichment_status"] = (
                "success" if updated_chunk_text else "failed"
            )

        if not updated_chunk_text or not isinstance(updated_chunk_text, str):
            updated_chunk_text = str(chunk["text"])
            chunk["metadata"]["chunk_enrichment_status"] = "failed"

        # Re-embed
        data = await self.providers.embedding.async_get_embedding(
            updated_chunk_text
        )
        chunk["metadata"]["original_text"] = chunk["text"]

        return VectorEntry(
            id=generate_id(str(chunk["id"])),
            vector=Vector(data=data, type=VectorType.FIXED, length=len(data)),
            document_id=document_id,
            owner_id=chunk["owner_id"],
            collection_ids=chunk["collection_ids"],
            text=updated_chunk_text,
            metadata=chunk["metadata"],
        )
Helper for chunk_enrichment.

        Leverages an LLM to rewrite or expand chunk text, then re-embeds it.
        
py/core/main/services/ingestion_service.py