a
    bgT                     @   s   d dl mZmZmZmZmZ d dlZd dlmZ d dl	m
Z
mZmZ d dlmZmZmZ d dlmZmZ d dlmZmZ dZd	ZG d
d deZG dd deeZdS )    )AnyDictListMappingOptionalN)
Embeddings)convert_to_secret_strget_from_dict_or_envpre_init)	BaseModel
ConfigDict	SecretStr)HTTPAdapterRetry)NotRequired	TypedDict   z$https://api.embaas.io/v1/embeddings/c                   @   s2   e Zd ZU dZeed< ee ed< ee ed< dS )EmbaasEmbeddingsPayloadz&Payload for the Embaas embeddings API.modeltextsinstructionN)__name__
__module____qualname____doc__str__annotations__r   r    r   r   s/var/www/html/cobodadashboardai.evdpl.com/venv/lib/python3.9/site-packages/langchain_community/embeddings/embaas.pyr      s   
r   c                   @   s  e Zd ZU dZdZeed< dZee ed< e	Z
eed< dZee ed< dZee ed	< d
Zee ed< eddZeeedddZeeeef dddZee edddZeeee  dddZee eee  dddZee eee  dddZeee ddd Z dS )!EmbaasEmbeddingsa  Embaas's embedding service.

    To use, you should have the
    environment variable ``EMBAAS_API_KEY`` set with your API key, or pass
    it as a named parameter to the constructor.

    Example:
        .. code-block:: python

            # initialize with default model and instruction
            from langchain_community.embeddings import EmbaasEmbeddings
            emb = EmbaasEmbeddings()

            # initialize with custom model and instruction
            from langchain_community.embeddings import EmbaasEmbeddings
            emb_model = "instructor-large"
            emb_inst = "Represent the Wikipedia document for retrieval"
            emb = EmbaasEmbeddings(
                model=emb_model,
                instruction=emb_inst
            )
    ze5-large-v2r   Nr   api_urlembaas_api_key   max_retries   timeoutZforbid)extra)valuesreturnc                 C   s   t t|dd}||d< |S )z?Validate that api key and python package exists in environment.r!   ZEMBAAS_API_KEY)r   r	   )clsr'   r!   r   r   r   validate_environment?   s
    
z%EmbaasEmbeddings.validate_environment)r(   c                 C   s   | j | jdS )zGet the identifying params.r   r   r+   selfr   r   r   _identifying_paramsH   s    z$EmbaasEmbeddings._identifying_params)r   r(   c                 C   s"   t || jd}| jr| j|d< |S )z&Generates payload for the API request.)r   r   r   )r   r   r   )r-   r   payloadr   r   r   _generate_payloadM   s    
z"EmbaasEmbeddings._generate_payload)r/   r(   c                 C   s   d| j   dd}t }t| jddgdd}|dt|d	 |d
t|d	 |j| j	||| j
d}| }dd |d D }|S )z;Sends a request to the Embaas API and handles the response.zBearer zapplication/json)AuthorizationzContent-Typeg      ?POSTT)totalbackoff_factorallowed_methodsraise_on_statuszhttp://)r#   zhttps://)headersjsonr%   c                 S   s   g | ]}|d  qS )	embeddingr   ).0itemr   r   r   
<listcomp>m       z4EmbaasEmbeddings._handle_request.<locals>.<listcomp>data)r!   Zget_secret_valuerequestsSessionr   r#   mountr   postr    r%   r8   )r-   r/   r7   sessionretriesresponseparsed_response
embeddingsr   r   r   _handle_requestT   s*    z EmbaasEmbeddings._handle_requestc              
   C   s   |  |}z| |W S  tjjy } zR|jdu s<|jjsJtd| |j }d|v rntd|d   W Y d}~n
d}~0 0 dS )z)Generate embeddings using the Embaas API.Nz'Error raised by embaas embeddings API: messagez1Validation Error raised by embaas embeddings API:)	r0   rH   r?   
exceptionsRequestExceptionrE   text
ValueErrorr8   )r-   r   r/   erF   r   r   r   _generate_embeddingsq   s    

z%EmbaasEmbeddings._generate_embeddingsc                    s>   fddt dttD } fdd|D }dd |D S )zGet embeddings for a list of texts.

        Args:
            texts: The list of texts to get embeddings for.

        Returns:
            List of embeddings, one for each text.
        c                    s   g | ]} ||t   qS r   )MAX_BATCH_SIZE)r:   i)r   r   r   r<      s   z4EmbaasEmbeddings.embed_documents.<locals>.<listcomp>r   c                    s   g | ]}  |qS r   )rO   )r:   batchr,   r   r   r<      r=   c                 S   s   g | ]}|D ]}|qqS r   r   )r:   rR   r9   r   r   r   r<      r=   )rangelenrP   )r-   r   ZbatchesrG   r   )r-   r   r   embed_documents   s
    	
z EmbaasEmbeddings.embed_documents)rL   r(   c                 C   s   |  |gd S )zGet embeddings for a single text.

        Args:
            text: The text to get embeddings for.

        Returns:
            List of embeddings.
        r   )rU   )r-   rL   r   r   r   embed_query   s    	zEmbaasEmbeddings.embed_query)!r   r   r   r   r   r   r   r   r   EMBAAS_API_URLr    r!   r   r#   intr%   r   Zmodel_configr
   r   r*   propertyr   r   r.   r   r   r0   floatrH   rO   rU   rV   r   r   r   r   r      s&   
r   )typingr   r   r   r   r   r?   Zlangchain_core.embeddingsr   Zlangchain_core.utilsr   r	   r
   Zpydanticr   r   r   Zrequests.adaptersr   r   Ztyping_extensionsr   r   rP   rW   r   r   r   r   r   r   <module>   s   