Skip to content

Embedding Module

unique_toolkit.embedding.service

EmbeddingService

Bases: BaseService

Provides methods to interact with the Embedding service.

Source code in unique_toolkit/unique_toolkit/embedding/service.py 
 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
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
class EmbeddingService(BaseService):
    """
    Provides methods to interact with the Embedding service.
    """

    @deprecated(
        "Use __init__ with company_id and user_id instead or use the classmethod `from_event`"
    )
    @overload
    def __init__(self, event: Event | BaseEvent): ...

    """
        Initialize the EmbeddingService with an event (deprecated)
    """

    @overload
    def __init__(self, *, company_id: str, user_id: str): ...

    """
        Initialize the EmbeddingService with a company_id and user_id.
    """

    def __init__(
        self,
        event: Event | BaseEvent | None = None,
        company_id: str | None = None,
        user_id: str | None = None,
    ):
        self._event = event
        if event:
            self._company_id: str = event.company_id
            self._user_id: str = event.user_id
        else:
            [company_id, user_id] = validate_required_values([company_id, user_id])
            self._company_id: str = company_id
            self._user_id: str = user_id

    @classmethod
    def from_event(cls, event: Event | BaseEvent):
        """
        Initialize the EmbeddingService with an event.
        """
        return cls(company_id=event.company_id, user_id=event.user_id)

    @classmethod
    def from_settings(cls, settings: UniqueSettings | str | None = None):
        """
        Initialize the EmbeddingService with a settings object.
        """

        if settings is None:
            settings = UniqueSettings.from_env_auto_with_sdk_init()
        elif isinstance(settings, str):
            settings = UniqueSettings.from_env_auto_with_sdk_init(filename=settings)

        return cls(
            company_id=settings.auth.company_id.get_secret_value(),
            user_id=settings.auth.user_id.get_secret_value(),
        )

    @property
    @deprecated(
        "The event property is deprecated and will be removed in a future version."
    )
    def event(self) -> Event | BaseEvent | None:
        """
        Get the event object (deprecated).

        Returns:
            Event | BaseEvent | None: The event object.
        """
        return self._event

    @property
    @deprecated(
        "The company_id property is deprecated and will be removed in a future version."
    )
    def company_id(self) -> str | None:
        """
        Get the company identifier (deprecated).

        Returns:
            str | None: The company identifier.
        """
        return self._company_id

    @company_id.setter
    @deprecated(
        "The company_id setter is deprecated and will be removed in a future version."
    )
    def company_id(self, value: str) -> None:
        """
        Set the company identifier (deprecated).

        Args:
            value (str | None): The company identifier.
        """
        self._company_id = value

    @property
    @deprecated(
        "The user_id property is deprecated and will be removed in a future version."
    )
    def user_id(self) -> str | None:
        """
        Get the user identifier (deprecated).

        Returns:
            str | None: The user identifier.
        """
        return self._user_id

    @user_id.setter
    @deprecated(
        "The user_id setter is deprecated and will be removed in a future version."
    )
    def user_id(self, value: str) -> None:
        """
        Set the user identifier (deprecated).

        Args:
            value (str | None): The user identifier.
        """
        self._user_id = value

    def embed_texts(
        self,
        texts: list[str],
        timeout: int = DEFAULT_TIMEOUT,
    ) -> Embeddings:
        """
        Embed text.

        Args:
            texts (list[str]): The texts to embed.
            timeout (int): The timeout in milliseconds. Defaults to 600000.

        Returns:
            Embeddings: The Embedding object.

        Raises:
            Exception: If an error occurs.
        """
        return embed_texts(
            user_id=self._user_id,
            company_id=self._company_id,
            texts=texts,
            timeout=timeout,
        )

    async def embed_texts_async(
        self,
        texts: list[str],
        timeout: int = DEFAULT_TIMEOUT,
    ) -> Embeddings:
        """
        Embed text asynchronously.

        Args:
            text (str): The text to embed.
            timeout (int): The timeout in milliseconds. Defaults to 600000.

        Returns:
            Embeddings: The Embedding object.

        Raises:
            Exception: If an error occurs.
        """
        return await embed_texts_async(
            user_id=self._user_id,
            company_id=self._company_id,
            texts=texts,
            timeout=timeout,
        )

company_id property writable

Get the company identifier (deprecated).

Returns:

Type Description
str | None

str | None: The company identifier.

event property

Get the event object (deprecated).

Returns:

Type Description
Event | BaseEvent | None

Event | BaseEvent | None: The event object.

user_id property writable

Get the user identifier (deprecated).

Returns:

Type Description
str | None

str | None: The user identifier.

embed_texts(texts, timeout=DEFAULT_TIMEOUT)

Embed text.

Parameters:

Name Type Description Default
texts list[str]

The texts to embed.

 required
timeout int

The timeout in milliseconds. Defaults to 600000.

 DEFAULT_TIMEOUT

Returns:

Name Type Description
Embeddings Embeddings

The Embedding object.

Raises:

Type Description
Exception

If an error occurs.
Source code in unique_toolkit/unique_toolkit/embedding/service.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def embed_texts(
    self,
    texts: list[str],
    timeout: int = DEFAULT_TIMEOUT,
) -> Embeddings:
    """
    Embed text.

    Args:
        texts (list[str]): The texts to embed.
        timeout (int): The timeout in milliseconds. Defaults to 600000.

    Returns:
        Embeddings: The Embedding object.

    Raises:
        Exception: If an error occurs.
    """
    return embed_texts(
        user_id=self._user_id,
        company_id=self._company_id,
        texts=texts,
        timeout=timeout,
    )

embed_texts_async(texts, timeout=DEFAULT_TIMEOUT) async

Embed text asynchronously.

Parameters:

Name Type Description Default
text str

The text to embed.

 required
timeout int

The timeout in milliseconds. Defaults to 600000.

 DEFAULT_TIMEOUT

Returns:

Name Type Description
Embeddings Embeddings

The Embedding object.

Raises:

Type Description
Exception

If an error occurs.
Source code in unique_toolkit/unique_toolkit/embedding/service.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
async def embed_texts_async(
    self,
    texts: list[str],
    timeout: int = DEFAULT_TIMEOUT,
) -> Embeddings:
    """
    Embed text asynchronously.

    Args:
        text (str): The text to embed.
        timeout (int): The timeout in milliseconds. Defaults to 600000.

    Returns:
        Embeddings: The Embedding object.

    Raises:
        Exception: If an error occurs.
    """
    return await embed_texts_async(
        user_id=self._user_id,
        company_id=self._company_id,
        texts=texts,
        timeout=timeout,
    )

from_event(event) classmethod

Initialize the EmbeddingService with an event.

Source code in unique_toolkit/unique_toolkit/embedding/service.py 
51
52
53
54
55
56
@classmethod
def from_event(cls, event: Event | BaseEvent):
    """
    Initialize the EmbeddingService with an event.
    """
    return cls(company_id=event.company_id, user_id=event.user_id)

from_settings(settings=None) classmethod

Initialize the EmbeddingService with a settings object.

Source code in unique_toolkit/unique_toolkit/embedding/service.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@classmethod
def from_settings(cls, settings: UniqueSettings | str | None = None):
    """
    Initialize the EmbeddingService with a settings object.
    """

    if settings is None:
        settings = UniqueSettings.from_env_auto_with_sdk_init()
    elif isinstance(settings, str):
        settings = UniqueSettings.from_env_auto_with_sdk_init(filename=settings)

    return cls(
        company_id=settings.auth.company_id.get_secret_value(),
        user_id=settings.auth.user_id.get_secret_value(),
    )