U
    hH                     @   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 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)	BaseModel	SecretStr)convert_to_secret_strget_from_dict_or_envpre_init)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   I/tmp/pip-unpacked-wheel-9gdii04g/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< G d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   timeoutc                   @   s   e Zd ZdZdS )zEmbaasEmbeddings.ConfigZforbidN)r   r   r   extrar   r   r   r   Config;   s   r&   )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_paramsG   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_payloadL   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_factorZallowed_methodsraise_on_statuszhttp://)r"   zhttps://)headersjsonr$   c                 S   s   g | ]}|d  qS )	embeddingr   ).0itemr   r   r   
<listcomp>l   s     z4EmbaasEmbeddings._handle_request.<locals>.<listcomp>data)r    Zget_secret_valuerequestsSessionr   r"   mountr   postr   r$   r7   )r-   r/   r6   sessionretriesresponseparsed_response
embeddingsr   r   r   _handle_requestS   s*    z EmbaasEmbeddings._handle_requestc              
   C   s   |  |}z| |W S  tjjk
r } zJ|jdks>|jjsLtd| |j }d|krptd|d   W 5 d}~X Y nX dS )z)Generate embeddings using the Embaas API.Nz'Error raised by embaas embeddings API: messagez1Validation Error raised by embaas embeddings API:)	r0   rF   r=   
exceptionsRequestExceptionrC   text
ValueErrorr7   )r-   r   r/   erD   r   r   r   _generate_embeddingsp   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)r9   i)r   r   r   r;      s    z4EmbaasEmbeddings.embed_documents.<locals>.<listcomp>r   c                    s   g | ]}  |qS r   )rM   )r9   batchr,   r   r   r;      s     c                 S   s   g | ]}|D ]}|qqS r   r   )r9   rP   r8   r   r   r   r;      s       )rangelenrN   )r-   r   ZbatchesrE   r   )r-   r   r   embed_documents   s
    	
z EmbaasEmbeddings.embed_documents)rJ   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   )rS   )r-   rJ   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&   r   r   r*   propertyr   r   r.   r   r   r0   floatrF   rM   rS   rT   r   r   r   r   r      s"   
r   )typingr   r   r   r   r   r=   Zlangchain_core.embeddingsr   Zlangchain_core.pydantic_v1r   r	   Zlangchain_core.utilsr
   r   r   Zrequests.adaptersr   r   Ztyping_extensionsr   r   rN   rU   r   r   r   r   r   r   <module>   s   