o ưiC@sdZddlmZddlZddlmZmZmZmZm Z m Z ddl m Z ddl mZmZddlmZddlmZdd lmZerJdd lmZdd lmZGd d d eeZdS)aG Vertex AI-specific RAG Ingestion implementation. Vertex AI RAG Engine handles embedding and chunking internally when files are uploaded, so this implementation skips the embedding step and directly uploads files to RAG corpora. Based on: https://docs.cloud.google.com/vertex-ai/generative-ai/docs/model-reference/rag-api-v1 ) annotationsN) TYPE_CHECKINGAnyDictListOptionalTuple)verbose_logger)get_async_httpx_clienthttpxSpecialProvider)get_vertex_base_url) VertexBase)BaseRAGIngestion)Router)RAGIngestOptionsc@sdeZdZdZ d.d/ddZd0d dZd1ddZ d.d2ddZ  d3d4d%d&Zd5d)d*Z d6d,d-Z dS)7VertexAIRAGIngestiona Vertex AI RAG Engine ingestion implementation. Key differences from base: - Embedding is handled by Vertex AI RAG Engine when files are uploaded - Files are uploaded using the RAG API (import or upload) - Chunking is done by Vertex AI RAG Engine (supports custom chunking config) - Supports Google Cloud Storage (GCS) and Google Drive sources - Supports custom parsing configurations (layout parser, LLM parser) Ningest_options'RAGIngestOptions'routerOptional['Router']cCsPtj|||dt|t|j}|||_||pd|_| ||_ dS)N)rrz us-central1) r__init__r dictvector_store_configZsafe_get_vertex_ai_project project_idZget_vertex_ai_locationlocationZsafe_get_vertex_ai_credentialsvertex_credentials)selfrrZlitellm_paramsr`/home/app/Keep/.python/lib/python3.10/site-packages/litellm/rag/ingestion/vertex_ai_ingestion.pyr)s    zVertexAIRAGIngestion.__init__chunks List[str]returnOptional[List[List[float]]]csdS)z Vertex AI RAG Engine handles embedding internally - skip this step. Returns: None (Vertex AI embeds when files are uploaded to RAG corpus) Nr)rrrrrembed9s zVertexAIRAGIngestion.embed file_contentOptional[bytes]filename Optional[str] content_type embeddings#Tuple[Optional[str], Optional[str]]csr|jstd|jd}|s!|j|jpd|jddIdH}d}|r5|r5|r5|j||||dIdH}||fS)a  Store content in Vertex AI RAG corpus. Vertex AI workflow: 1. Create RAG corpus (if not provided) 2. Upload file using RAG API (Vertex AI handles chunking/embedding) Args: file_content: Raw file bytes filename: Name of the file content_type: MIME type chunks: Ignored - Vertex AI handles chunking embeddings: Ignored - Vertex AI handles embedding Returns: Tuple of (rag_corpus_id, file_id) zVvertex_project is required for Vertex AI RAG ingestion. Set it in vector_store config.Zvector_store_idzlitellm-rag-corpus description) display_namer+N) rag_corpus_idr&r$r()r ValueErrorrget_create_rag_corpusZ ingest_name_upload_file_to_corpus)rr$r&r(rr)r-Zresult_file_idrrrstoreFs(     zVertexAIRAGIngestion.storer,strr+cs|j|j|jdd\}}|js||_t|j}|d|jd|jd}d|i}|r/||d<|jd}|r;||d <|jd } | rUd |vrKi|d <d d | ii|d d <td|tdt j |ddt t j ddid} | j||d|dddIdH} | jdvrd| j} t| t| | } tdt j | dd| dr| didd }n| dd }td!||j||d"IdH}td#||S)$a Create a Vertex AI RAG corpus. Args: display_name: Display name for the corpus description: Optional description Returns: RAG corpus ID (format: projects/{project}/locations/{location}/ragCorpora/{corpus_id}) vertex_ai credentialsrZcustom_llm_providerz/v1beta1/projects/z /locations/z /ragCorporaZ displayNamer+vector_db_configZvectorDbConfigembedding_modelZvertexPredictionEndpointZendpointZragEmbeddingModelConfigzCreating RAG corpus: Request body: indenttimeoutN@Z llm_providerparamsBearer application/json Authorizationz Content-TypejsonheadersNzFailed to create RAG corpus: zCreate corpus response: doneresponsenamezPolling operation: )operation_name access_tokenzCreated RAG corpus: )_ensure_access_tokenrrr rrr/r debugrFdumpsr r RAGpost status_codetexterror Exception_poll_operation)rr,r+rPrbase_urlurl request_bodyr7r8clientrL error_msg response_data corpus_namerOrrrr0ysv           z'VertexAIRAGIngestion._create_rag_corpus@rOrP max_retriesint retry_delayfloatcs ddl}t|j}|d|}ttjddid}t|D]i} |j|dd|id IdH} | jd krCd | j } t | t | | } | d rrd | vr[| d } t d| | didd}|rk|St d| t d| dd|||IdHqt d|d)a Poll a long-running operation until it completes. Args: operation_name: The operation name (e.g., "operations/123456") access_token: Access token for authentication max_retries: Maximum number of polling attempts retry_delay: Delay between polling attempts in seconds Returns: The corpus name from the completed operation Raises: Exception: If operation fails or times out rN /v1beta1/r=r>r?rDrA)rGrIzFailed to poll operation: rKrXzOperation failed: rLrMrNz&No corpus name in operation response: z Operation not done yet, attempt /zOperation timed out after z attempts)asyncior rr r rTranger/rVrWr rXrYrFrRsleep)rrOrPrdrfrkr[r\r^attemptrLr_Zoperation_datarXrarrrrZs<       z$VertexAIRAGIngestion._poll_operationr-bytesc s|j|j|jdd\}}t|j}|d|d}dd|ii} |jd} | r.| | dd<|j} | rnt| t rn| d } | d } | sF| rnd | vrNi| d <d d iii| d d<| d dd d }| rh| |d <| rn| |d <t d|t dt j | dddt | df|||pdfd}ttjddid}|j||d|dddIdH}|jdvrd|j}t |t|z!| }|d id!d"}|s|d!d"}t d#||WSty}zt d$|WYd}~d%Sd}~ww)&aA Upload a file to Vertex AI RAG corpus using multipart upload. Args: rag_corpus_id: RAG corpus resource name filename: Name of the file file_content: File content bytes content_type: MIME type Returns: File ID or resource name r4r5z/upload/v1beta1/z/ragFiles:uploadZrag_filer,Zfile_descriptionr+ chunk_size chunk_overlapZupload_rag_file_configZrag_file_chunking_configZfixed_length_chunkingZrag_file_transformation_configzUploading file to RAG corpus: z Metadata: r:r;NrBzapplication/octet-stream)metadatafiler=gr@r?rA multipart)rDzX-Goog-Upload-Protocol)filesrGrHzFailed to upload file: ZragFilerMrNzUpload complete. File ID: z!Could not parse upload response: Zuploaded)rQrrr rrr/chunking_strategy isinstancerr rRrFrSr r rTrUrVrWrXrYwarning)rr-r&r$r(rP_r[r\rrr+rvrprqZchunking_configrur^rLr_r`Zfile_iderrrr1s          z+VertexAIRAGIngestion._upload_file_to_corpusgcs_uriscsX|j|j|jdd\}}t|j}|d|d}ddd|iii}|j}|rGt|trG|d} |d } | s:| rG| p=d | p@d d |dd <|j d} | rU| |dd<t d|t dt j |ddttjddid} | j||d|dddIdH} | jdvrd| j}t |t|| }|dd}t d ||S)!a Import files from Google Cloud Storage into RAG corpus. Args: rag_corpus_id: RAG corpus resource name gcs_uris: List of GCS URIs (e.g., ["gs://bucket/file.pdf"]) Returns: Operation name for tracking import progress r4r5rhz/ragFiles:importZimportRagFilesConfigZ gcsSourceurisrprqirI)Z chunkSizeZ chunkOverlapZragFileChunkingConfigZmax_embedding_requests_per_minZmaxEmbeddingRequestsPerMinzImporting files from GCS: r9r:r;r=r>r?rArBrCrENrHzFailed to import files: rMrNzImport operation started: )rQrrr rrvrwrr/rr rRrFrSr r rTrUrVrWrXrY)rr-r{rPryr[r\r]rvrprqZmax_embedding_qpmr^rLr_r`rOrrr_import_files_from_gcssb         z+VertexAIRAGIngestion._import_files_from_gcs)N)rrrr)rr r!r") r$r%r&r'r(r'rr r)r"r!r*)r,r3r+r'r!r3)rbrc) rOr3rPr3rdrerfrgr!r3) r-r3r&r3r$ror(r'r!r3)r-r3r{r r!r3) __name__ __module__ __qualname____doc__rr#r2r0rZr1r}rrrrrs  6 f Cor)r __future__rrFtypingrrrrrrZlitellm._loggingr Z&litellm.llms.custom_httpx.http_handlerr r Z#litellm.llms.vertex_ai.common_utilsr Z&litellm.llms.vertex_ai.vertex_llm_baser Z$litellm.rag.ingestion.base_ingestionrZlitellmrZlitellm.types.ragrrrrrrs