Source code for langchain_community.embeddings.infinity
"""written under MIT Licence, Michael Feil 2023."""
import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import Any, Callable, Dict, List, Optional, Tuple
import aiohttp
import numpy as np
import requests
from langchain_core.embeddings import Embeddings
from langchain_core.pydantic_v1 import BaseModel, Extra, root_validator
from langchain_core.utils import get_from_dict_or_env
__all__ = ["InfinityEmbeddings"]
[docs]class InfinityEmbeddings(BaseModel, Embeddings):
    """Embedding models for self-hosted https://github.com/michaelfeil/infinity
    This should also work for text-embeddings-inference and other
    self-hosted openai-compatible servers.
    Infinity is a class to interact with Embedding Models on https://github.com/michaelfeil/infinity
    Example:
        .. code-block:: python
            from langchain_community.embeddings import InfinityEmbeddings
            InfinityEmbeddings(
                model="BAAI/bge-small",
                infinity_api_url="http://localhost:7797/v1",
            )
    """
    model: str
    "Underlying Infinity model id."
    infinity_api_url: str = "http://localhost:7797/v1"
    """Endpoint URL to use."""
    client: Any = None  #: :meta private:
    """Infinity client."""
    # LLM call kwargs
    class Config:
        """Configuration for this pydantic object."""
        extra = Extra.forbid
    @root_validator(allow_reuse=True)
    def validate_environment(cls, values: Dict) -> Dict:
        """Validate that api key and python package exists in environment."""
        values["infinity_api_url"] = get_from_dict_or_env(