Skip to content

cohere.embedders

A module for calling OpenAI's Embeddings models.

CohereEmbedder

Bases: BaseEmbedder[CohereEmbeddingResponse]

Cohere Embedder

model max_dimensions embed-english-v3.0 1024 embed-multilingual-v3.0 1024 embed-english-light-v3.0 384 embed-multilingual-light-v3.0 384 embed-english-v2.0 4096 embed-english-light-v2.0 1024 embed-multilingual-v2.0 768

Example:

import os
from mirascope.cohere import CohereEmbedder

os.environ["CO_API_KEY"] = "YOUR_COHERE_API_KEY"

cohere_embedder = CohereEmbedder()
response = cohere_embedder.embed(["your text to embed"])
print(response)
Source code in mirascope/cohere/embedders.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class CohereEmbedder(BaseEmbedder[CohereEmbeddingResponse]):
    """Cohere Embedder

    model                           max_dimensions
    embed-english-v3.0              1024
    embed-multilingual-v3.0         1024
    embed-english-light-v3.0        384
    embed-multilingual-light-v3.0   384
    embed-english-v2.0              4096
    embed-english-light-v2.0        1024
    embed-multilingual-v2.0         768

    Example:

    ```python
    import os
    from mirascope.cohere import CohereEmbedder

    os.environ["CO_API_KEY"] = "YOUR_COHERE_API_KEY"

    cohere_embedder = CohereEmbedder()
    response = cohere_embedder.embed(["your text to embed"])
    print(response)
    ```
    """

    dimensions: Optional[int] = 1024
    embedding_params: ClassVar[CohereEmbeddingParams] = CohereEmbeddingParams(
        model="embed-english-v3.0"
    )

    def embed(self, inputs: list[str]) -> CohereEmbeddingResponse:
        """Call the embedder with multiple inputs"""
        co = Client(api_key=self.api_key, base_url=self.base_url)
        embedding_type = (
            self.embedding_params.embedding_types[0]
            if self.embedding_params.embedding_types
            else None
        )
        start_time = datetime.datetime.now().timestamp() * 1000
        embed = co.embed
        if self.embedding_params.logfire:
            embed = self.embedding_params.logfire(
                co.embed, "cohere", response_type=CohereEmbeddingResponse
            )  # pragma: no cover
        response = embed(texts=inputs, **self.embedding_params.kwargs())
        return CohereEmbeddingResponse(
            response=response,
            start_time=start_time,
            end_time=datetime.datetime.now().timestamp() * 1000,
            embedding_type=embedding_type,
        )

    async def embed_async(self, inputs: list[str]) -> CohereEmbeddingResponse:
        """Asynchronously call the embedder with multiple inputs"""
        co = AsyncClient(api_key=self.api_key, base_url=self.base_url)
        embedding_type = (
            self.embedding_params.embedding_types[0]
            if self.embedding_params.embedding_types
            else None
        )
        start_time = datetime.datetime.now().timestamp() * 1000
        embed = co.embed
        if self.embedding_params.logfire_async:
            embed = self.embedding_params.logfire_async(
                co.embed,
                "cohere",
                response_type=None,  # note: no content to extract, so we set to `None`
            )  # pragma: no cover
        response = await embed(texts=inputs, **self.embedding_params.kwargs())
        return CohereEmbeddingResponse(
            response=response,
            start_time=start_time,
            end_time=datetime.datetime.now().timestamp() * 1000,
            embedding_type=embedding_type,
        )

    def __call__(
        self, input: list[str]
    ) -> Optional[Union[list[list[float]], list[list[int]]]]:
        """Call the embedder with a input

        Chroma expects parameter to be `input`.
        """
        response = self.embed(input)
        embeddings = response.embeddings
        return embeddings

__call__(input)

Call the embedder with a input

Chroma expects parameter to be input.

Source code in mirascope/cohere/embedders.py 
88
89
90
91
92
93
94
95
96
97
def __call__(
    self, input: list[str]
) -> Optional[Union[list[list[float]], list[list[int]]]]:
    """Call the embedder with a input

    Chroma expects parameter to be `input`.
    """
    response = self.embed(input)
    embeddings = response.embeddings
    return embeddings

embed(inputs)

Call the embedder with multiple inputs

Source code in mirascope/cohere/embedders.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def embed(self, inputs: list[str]) -> CohereEmbeddingResponse:
    """Call the embedder with multiple inputs"""
    co = Client(api_key=self.api_key, base_url=self.base_url)
    embedding_type = (
        self.embedding_params.embedding_types[0]
        if self.embedding_params.embedding_types
        else None
    )
    start_time = datetime.datetime.now().timestamp() * 1000
    embed = co.embed
    if self.embedding_params.logfire:
        embed = self.embedding_params.logfire(
            co.embed, "cohere", response_type=CohereEmbeddingResponse
        )  # pragma: no cover
    response = embed(texts=inputs, **self.embedding_params.kwargs())
    return CohereEmbeddingResponse(
        response=response,
        start_time=start_time,
        end_time=datetime.datetime.now().timestamp() * 1000,
        embedding_type=embedding_type,
    )

embed_async(inputs) async

Asynchronously call the embedder with multiple inputs

Source code in mirascope/cohere/embedders.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
async def embed_async(self, inputs: list[str]) -> CohereEmbeddingResponse:
    """Asynchronously call the embedder with multiple inputs"""
    co = AsyncClient(api_key=self.api_key, base_url=self.base_url)
    embedding_type = (
        self.embedding_params.embedding_types[0]
        if self.embedding_params.embedding_types
        else None
    )
    start_time = datetime.datetime.now().timestamp() * 1000
    embed = co.embed
    if self.embedding_params.logfire_async:
        embed = self.embedding_params.logfire_async(
            co.embed,
            "cohere",
            response_type=None,  # note: no content to extract, so we set to `None`
        )  # pragma: no cover
    response = await embed(texts=inputs, **self.embedding_params.kwargs())
    return CohereEmbeddingResponse(
        response=response,
        start_time=start_time,
        end_time=datetime.datetime.now().timestamp() * 1000,
        embedding_type=embedding_type,
    )