Source code for langchain_community.embeddings.javelin_ai_gateway

from __future__ import annotations

from typing import Any, Iterator, List, Optional

from langchain_core.embeddings import Embeddings
from pydantic import BaseModel


def _chunk(texts: List[str], size: int) -> Iterator[List[str]]:
    for i in range(0, len(texts), size):
        yield texts[i : i + size]



[docs]
class JavelinAIGatewayEmbeddings(Embeddings, BaseModel):
    """Javelin AI Gateway embeddings.

    To use, you should have the ``javelin_sdk`` python package installed.
    For more information, see https://docs.getjavelin.io

    Example:
        .. code-block:: python

            from langchain_community.embeddings import JavelinAIGatewayEmbeddings

            embeddings = JavelinAIGatewayEmbeddings(
                gateway_uri="<javelin-ai-gateway-uri>",
                route="<your-javelin-gateway-embeddings-route>"
            )
    """

    client: Any
    """javelin client."""

    route: str
    """The route to use for the Javelin AI Gateway API."""

    gateway_uri: Optional[str] = None
    """The URI for the Javelin AI Gateway API."""

    javelin_api_key: Optional[str] = None
    """The API key for the Javelin AI Gateway API."""

    def __init__(self, **kwargs: Any):
        try:
            from javelin_sdk import (
                JavelinClient,
                UnauthorizedError,
            )
        except ImportError:
            raise ImportError(
                "Could not import javelin_sdk python package. "
                "Please install it with `pip install javelin_sdk`."
            )

        super().__init__(**kwargs)
        if self.gateway_uri:
            try:
                self.client = JavelinClient(
                    base_url=self.gateway_uri, api_key=self.javelin_api_key
                )
            except UnauthorizedError as e:
                raise ValueError("Javelin: Incorrect API Key.") from e

    def _query(self, texts: List[str]) -> List[List[float]]:
        embeddings = []
        for txt in _chunk(texts, 20):
            try:
                resp = self.client.query_route(self.route, query_body={"input": txt})
                resp_dict = resp.dict()

                embeddings_chunk = resp_dict.get("llm_response", {}).get("data", [])
                for item in embeddings_chunk:
                    if "embedding" in item:
                        embeddings.append(item["embedding"])
            except ValueError as e:
                print("Failed to query route: " + str(e))  # noqa: T201

        return embeddings

    async def _aquery(self, texts: List[str]) -> List[List[float]]:
        embeddings = []
        for txt in _chunk(texts, 20):
            try:
                resp = await self.client.aquery_route(
                    self.route, query_body={"input": txt}
                )
                resp_dict = resp.dict()

                embeddings_chunk = resp_dict.get("llm_response", {}).get("data", [])
                for item in embeddings_chunk:
                    if "embedding" in item:
                        embeddings.append(item["embedding"])
            except ValueError as e:
                print("Failed to query route: " + str(e))  # noqa: T201

        return embeddings


[docs]
    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        return self._query(texts)



[docs]
    def embed_query(self, text: str) -> List[float]:
        return self._query([text])[0]



[docs]
    async def aembed_documents(self, texts: List[str]) -> List[List[float]]:
        return await self._aquery(texts)



[docs]
    async def aembed_query(self, text: str) -> List[float]:
        result = await self._aquery([text])
        return result[0]