Class ModalEmbeddings

document_qa/custom_embeddings.py:14–86

LangChain ``Embeddings`` backed by an OpenAI-compatible HTTP API. The service must expose a ``POST /embeddings`` endpoint that accepts ``{"model": "…", "input": ["…"]}`` and returns the standard OpenAI response shape. Args: url: Base URL of the embedding service (e.g. ``"ht

class ModalEmbeddings(Embeddings):
    """LangChain ``Embeddings`` backed by an OpenAI-compatible HTTP API.

    The service must expose a ``POST /embeddings`` endpoint that accepts
    ``{"model": "…", "input": ["…"]}`` and returns the standard OpenAI
    response shape.

    Args:
        url: Base URL of the embedding service (e.g. ``"http://localhost:1234/v1"``).
        model_name: Model identifier(e.g. ``"intfloat/multilingual-e5-large-instruct"``).
        api_key: Optional bearer token for authenticated endpoints.
    """

    def __init__(self, url: str, model_name: str, api_key: str = None):
        self.url = url
        self.model_name = model_name
        self.api_key = api_key

    def embed(self, text: List[str]) -> List[List[float]]:
        """Embed a list of texts via the configured API.

        Newlines are replaced with spaces before sending, since most
        embedding models treat them as noise.

        Args:
            text: Strings to embed.

        Returns:
            list[list[float]]: One embedding vector per input string.

        Raises:
            requests.HTTPError: If the API returns a non-2xx status.
        """
        # Newlines degrade embedding quality for most models
        cleaned_text = [t.replace("\n", " ") for t in text]

        payload = {"text": "\n".join(cleaned_text)}

        headers = {}
        if self.api_key:
            headers = {"x-api-key": self.api_key}

        response = requests.post(self.url, data=payload, files=[], headers=headers)
        response.raise_for_status()

        # print(response.text)
        return response.json()

    def embed_documents(self, text: List[str]) -> List[List[str]]:
        """Embed multiple documents (LangChain interface).

        Args:
            text: Document strings to embed.

        Returns:
            list[list[float]]: One embedding vector per document.
        """
        return self.embed(text)