Voyageai multimodal

libs/core/langchain_core/embeddings/__init__.py

@@ -1,4 +1,5 @@
 from langchain_core.embeddings.embeddings import Embeddings
+from langchain_core.embeddings.multimodal_embeddings import MultimodalEmbeddings, MultimodalInput, Content
 from langchain_core.embeddings.fake import DeterministicFakeEmbedding, FakeEmbeddings
 
-__all__ = ["DeterministicFakeEmbedding", "Embeddings", "FakeEmbeddings"]
+__all__ = ["DeterministicFakeEmbedding", "Embeddings", "FakeEmbeddings", "MultimodalEmbeddings", "MultimodalInput", "Content"]

libs/partners/voyageai/langchain_voyageai/multimodal_embeddings.py

@@ -0,0 +1,120 @@
+import logging
+from typing import Any, List, Optional
+
+import voyageai  # type: ignore
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    PrivateAttr,
+    SecretStr,
+    model_validator,
+)
+from typing_extensions import Self
+
+from langchain_core.embeddings import MultimodalEmbeddings, MultimodalInput, Content
+from langchain_core.embeddings.multimodal_embeddings import ContentType
+from langchain_core.utils import secret_from_env
+
+logger = logging.getLogger(__name__)
+
+
+class VoyageAIMultimodalEmbeddings(BaseModel, MultimodalEmbeddings):
+    """VoyageAIMultimodalEmbeddings embedding model.
+
+    Example:
+        .. code-block:: python
+
+            from langchain_voyageai import VoyageAIMultimodalEmbeddings
+
+            model = VoyageAIMultimodalEmbeddings()
+    """
+
+    _client: voyageai.Client = PrivateAttr()
+    _aclient: voyageai.client_async.AsyncClient = PrivateAttr()
+    model: str
+    show_progress_bar: bool = False
+    truncation: Optional[bool] = None
+    voyage_api_key: SecretStr = Field(
+        alias="api_key",
+        default_factory=secret_from_env(
+            "VOYAGE_API_KEY",
+            error_message="Must set `VOYAGE_API_KEY` environment variable or "
+            "pass `api_key` to VoyageAIEmbeddings constructor.",
+        ),
+    )
+
+    model_config = ConfigDict(
+        extra="forbid",
+        populate_by_name=True,
+    )
+
+    @classmethod
+    def _langchain_content_to_content(cls, content: Content) -> dict:
+        result = {
+            "type": content.type.value
+        }
+        if content.type == ContentType.text:
+            result["text"] = content.data
+        elif content.type == ContentType.image_url:
+            result["image_url"] = content.data
+        elif content.type == ContentType.image_base64:
+            result["image_base64"] = content.data
+
+        return result
+
+    @classmethod
+    def _langchain_multimodal_input_to_input(cls, multimodal_input: MultimodalInput) -> dict:
+        return {
+            "content": [cls._langchain_content_to_content(x) for x in multimodal_input.input]
+        }
+
+    @model_validator(mode="before")
+    @classmethod
+    def default_values(cls, values: dict) -> Any:
+        return values
+
+    @model_validator(mode="after")
+    def validate_environment(self) -> Self:
+        """Validate that VoyageAI credentials exist in environment."""
+        api_key_str = self.voyage_api_key.get_secret_value()
+        self._client = voyageai.Client(api_key=api_key_str)
+        self._aclient = voyageai.client_async.AsyncClient(api_key=api_key_str)
+        return self
+
+    def embed_documents(self, multimodal_inputs: list[MultimodalInput]) -> List[List[float]]:
+        """Embed search docs."""
+
+        return self._client.multimodal_embed(
+            [self._langchain_multimodal_input_to_input(x) for x in multimodal_inputs],
+            model=self.model,
+            input_type="document",
+            truncation=self.truncation,
+        ).embeddings
+
+    def embed_query(self, multimodal_input: MultimodalInput) -> List[float]:
+        """Embed query text."""
+        return self._client.multimodal_embed(
+            [self._langchain_multimodal_input_to_input(multimodal_input)],
+            model=self.model,
+            input_type="query",
+            truncation=self.truncation
+        ).embeddings[0]
+
+    async def aembed_documents(self, multimodal_inputs: list[MultimodalInput]) -> List[List[float]]:
+        result = await self._aclient.multimodal_embed(
+            [self._langchain_multimodal_input_to_input(x) for x in multimodal_inputs],
+            model=self.model,
+            input_type="document",
+            truncation=self.truncation,
+        )
+        return result.embeddings
+
+    async def aembed_query(self, multimodal_input: MultimodalInput) -> List[float]:
+        result = await self._aclient.multimodal_embed(
+            [self._langchain_multimodal_input_to_input(multimodal_input)],
+            model=self.model,
+            input_type="document",
+            truncation=self.truncation,
+        )
+        return result.embeddings[0]

libs/partners/voyageai/pyproject.toml

@@ -21,7 +21,7 @@ disallow_untyped_defs = "True"
 [tool.poetry.dependencies]
 python = ">=3.9,<4.0"
 langchain-core = "^0.3.15"
-voyageai = ">=0.2.1,<1"
+voyageai = ">=0.3.2,<1"
 pydantic = ">=2,<3"
 
 [tool.ruff.lint]