Content Module

Deprecated

This module is deprecated. Use KnowledgeBaseService instead.

The Content module provides low-level functionality for interacting with content stored in the knowledge base.

Overview

The unique_toolkit.content module encompasses all content-related functionality. Content can be any type of textual data that is stored in the Knowledgebase on the Unique platform. During ingestion, content is parsed, split into chunks, indexed, and stored in the database.

Note: This module is deprecated. Please use KnowledgeBaseService from unique_toolkit.services.knowledge_base for all knowledge base operations.

Components

Service

unique_toolkit.content.service.ContentService

Provides methods for searching, downloading and uploading content in the knowledge base.

Source code in unique_toolkit/unique_toolkit/content/service.py

@deprecated("Use KnowledgeBaseService instead")
class ContentService:
    """
    Provides methods for searching, downloading and uploading content in the knowledge base.
    """

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

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

    @overload
    def __init__(
        self,
        *,
        company_id: str,
        user_id: str,
        chat_id: str | None = None,
        metadata_filter: dict | None = None,
    ): ...

    """
        Initialize the ContentService with a company_id, user_id and chat_id and metadata_filter.
    """

    def __init__(
        self,
        event: Event | BaseEvent | None = None,
        company_id: str | None = None,
        user_id: str | None = None,
        chat_id: str | None = None,
        metadata_filter: dict | None = None,
    ):
        """
        Initialize the ContentService with a company_id, user_id and chat_id.
        """

        self._event = event  # Changed to protected attribute
        self._metadata_filter = None
        if event:
            self._company_id: str = event.company_id
            self._user_id: str = event.user_id
            if isinstance(event, (ChatEvent, Event)):
                self._metadata_filter = event.payload.metadata_filter
                self._chat_id: str | None = event.payload.chat_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
            self._chat_id: str | None = chat_id
            self._metadata_filter = metadata_filter

    @classmethod
    def from_event(cls, event: Event | ChatEvent | BaseEvent):
        """
        Initialize the ContentService with an event.
        """
        chat_id = None
        metadata_filter = None

        if isinstance(event, (ChatEvent | Event)):
            chat_id = event.payload.chat_id
            metadata_filter = event.payload.metadata_filter

        return cls(
            company_id=event.company_id,
            user_id=event.user_id,
            chat_id=chat_id,
            metadata_filter=metadata_filter,
        )

    @classmethod
    def from_settings(
        cls,
        settings: UniqueSettings | str | None = None,
        metadata_filter: dict | None = None,
    ):
        """
        Initialize the ContentService with a settings object and metadata filter.
        """

        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(),
            metadata_filter=metadata_filter,
        )

    @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

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

        Returns:
            str | None: The chat identifier.
        """
        return self._chat_id

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

        Args:
            value (str | None): The chat identifier.
        """
        self._chat_id = value

    @property
    @deprecated(
        "The metadata_filter property is deprecated and will be removed in a future version."
    )
    def metadata_filter(self) -> dict | None:
        """
        Get the metadata filter (deprecated).

        Returns:
            dict | None: The metadata filter.
        """
        return self._metadata_filter

    @metadata_filter.setter
    @deprecated(
        "The metadata_filter setter is deprecated and will be removed in a future version."
    )
    def metadata_filter(self, value: dict | None) -> None:
        """
        Set the metadata filter (deprecated).

        Args:
            value (dict | None): The metadata filter.
        """
        self._metadata_filter = value

    def search_content_chunks(
        self,
        search_string: str,
        search_type: ContentSearchType,
        limit: int,
        search_language: str = DEFAULT_SEARCH_LANGUAGE,
        chat_id: str = "",
        reranker_config: ContentRerankerConfig | None = None,
        scope_ids: list[str] | None = None,
        chat_only: bool | None = None,
        metadata_filter: dict | None = None,
        content_ids: list[str] | None = None,
        score_threshold: float | None = None,
    ) -> list[ContentChunk]:
        """
        Performs a synchronous search for content chunks in the knowledge base.

        Args:
            search_string (str): The search string.
            search_type (ContentSearchType): The type of search to perform.
            limit (int): The maximum number of results to return.
            search_language (str, optional): The language for the full-text search. Defaults to "english".
            chat_id (str, optional): The chat ID for context. Defaults to empty string.
            reranker_config (ContentRerankerConfig | None, optional): The reranker configuration. Defaults to None.
            scope_ids (list[str] | None, optional): The scope IDs to filter by. Defaults to None.
            chat_only (bool | None, optional): Whether to search only in the current chat. Defaults to None.
            metadata_filter (dict | None, optional): UniqueQL metadata filter. If unspecified/None, it tries to use the metadata filter from the event. Defaults to None.
            content_ids (list[str] | None, optional): The content IDs to search within. Defaults to None.
            score_threshold (float | None, optional): Sets the minimum similarity score for search results to be considered. Defaults to 0.

        Returns:
            list[ContentChunk]: The search results.

        Raises:
            Exception: If there's an error during the search operation.
        """

        if metadata_filter is None:
            metadata_filter = self._metadata_filter

        chat_id = chat_id or self._chat_id  # type: ignore

        if chat_only and not chat_id:
            raise ValueError("Please provide chat_id when limiting with chat_only")

        try:
            searches = search_content_chunks(
                user_id=self._user_id,
                company_id=self._company_id,
                chat_id=chat_id,
                search_string=search_string,
                search_type=search_type,
                limit=limit,
                search_language=search_language,
                reranker_config=reranker_config,
                scope_ids=scope_ids,
                chat_only=chat_only,
                metadata_filter=metadata_filter,
                content_ids=content_ids,
                score_threshold=score_threshold,
            )
            return searches
        except Exception as e:
            logger.error(f"Error while searching content chunks: {e}")
            raise e

    @deprecated("Use search_chunks_async instead")
    async def search_content_chunks_async(
        self,
        search_string: str,
        search_type: ContentSearchType,
        limit: int,
        search_language: str = DEFAULT_SEARCH_LANGUAGE,
        chat_id: str = "",
        reranker_config: ContentRerankerConfig | None = None,
        scope_ids: list[str] | None = None,
        chat_only: bool | None = None,
        metadata_filter: dict | None = None,
        content_ids: list[str] | None = None,
        score_threshold: float | None = None,
    ):
        """
        Performs an asynchronous search for content chunks in the knowledge base.

        Args:
            search_string (str): The search string.
            search_type (ContentSearchType): The type of search to perform.
            limit (int): The maximum number of results to return.
            search_language (str, optional): The language for the full-text search. Defaults to "english".
            chat_id (str, optional): The chat ID for context. Defaults to empty string.
            reranker_config (ContentRerankerConfig | None, optional): The reranker configuration. Defaults to None.
            scope_ids (list[str] | None, optional): The scope IDs to filter by. Defaults to None.
            chat_only (bool | None, optional): Whether to search only in the current chat. Defaults to None.
            metadata_filter (dict | None, optional): UniqueQL metadata filter. If unspecified/None, it tries to use the metadata filter from the event. Defaults to None.
            content_ids (list[str] | None, optional): The content IDs to search within. Defaults to None.
            score_threshold (float | None, optional): Sets the minimum similarity score for search results to be considered. Defaults to 0.

        Returns:
            list[ContentChunk]: The search results.

        Raises:
            Exception: If there's an error during the search operation.
        """
        if metadata_filter is None:
            metadata_filter = self._metadata_filter

        chat_id = chat_id or self._chat_id  # type: ignore

        if chat_only and not chat_id:
            raise ValueError("Please provide chat_id when limiting with chat_only.")

        try:
            searches = await search_content_chunks_async(
                user_id=self._user_id,
                company_id=self._company_id,
                chat_id=chat_id,
                search_string=search_string,
                search_type=search_type,
                limit=limit,
                search_language=search_language,
                reranker_config=reranker_config,
                scope_ids=scope_ids,
                chat_only=chat_only,
                metadata_filter=metadata_filter,
                content_ids=content_ids,
                score_threshold=score_threshold,
            )
            return searches
        except Exception as e:
            logger.error(f"Error while searching content chunks: {e}")
            raise e

    def search_contents(
        self,
        where: dict,
        chat_id: str = "",
    ) -> list[Content]:
        """
        Performs a search in the knowledge base by filter (and not a smilarity search)
        This function loads complete content of the files from the knowledge base in contrast to search_content_chunks.

        Args:
            where (dict): The search criteria.

        Returns:
            list[Content]: The search results.
        """
        chat_id = chat_id or self._chat_id  # type: ignore

        return search_contents(
            user_id=self._user_id,
            company_id=self._company_id,
            chat_id=chat_id,
            where=where,
        )

    async def search_contents_async(
        self,
        where: dict,
        chat_id: str = "",
    ) -> list[Content]:
        """
        Performs an asynchronous search for content files in the knowledge base by filter.

        Args:
            where (dict): The search criteria.

        Returns:
            list[Content]: The search results.
        """
        chat_id = chat_id or self._chat_id  # type: ignore

        return await search_contents_async(
            user_id=self._user_id,
            company_id=self._company_id,
            chat_id=chat_id,
            where=where,
        )

    def search_content_on_chat(self, chat_id: str) -> list[Content]:
        where = {"ownerId": {"equals": chat_id}}

        return self.search_contents(where, chat_id=chat_id)

    def upload_content_from_bytes(
        self,
        content: bytes,
        content_name: str,
        mime_type: str,
        scope_id: str | None = None,
        chat_id: str | None = None,
        skip_ingestion: bool = False,
        skip_excel_ingestion: bool = False,
        ingestion_config: unique_sdk.Content.IngestionConfig | None = None,
        metadata: dict | None = None,
    ) -> Content:
        """
        Uploads content to the knowledge base.

        Args:
            content (bytes): The content to upload.
            content_name (str): The name of the content.
            mime_type (str): The MIME type of the content.
            scope_id (str | None): The scope ID. Defaults to None.
            chat_id (str | None): The chat ID. Defaults to None.
            skip_ingestion (bool): Whether to skip ingestion. Defaults to False.
            skip_excel_ingestion (bool): Whether to skip excel ingestion. Defaults to False.
            ingestion_config (unique_sdk.Content.IngestionConfig | None): The ingestion configuration. Defaults to None.
            metadata (dict | None): The metadata to associate with the content. Defaults to None.

        Returns:
            Content: The uploaded content.
        """

        return upload_content_from_bytes(
            user_id=self._user_id,
            company_id=self._company_id,
            content=content,
            content_name=content_name,
            mime_type=mime_type,
            scope_id=scope_id,
            chat_id=chat_id,
            skip_ingestion=skip_ingestion,
            ingestion_config=ingestion_config,
            metadata=metadata,
        )

    async def upload_content_from_bytes_async(
        self,
        content: bytes,
        content_name: str,
        mime_type: str,
        scope_id: str | None = None,
        chat_id: str | None = None,
        skip_ingestion: bool = False,
        ingestion_config: unique_sdk.Content.IngestionConfig | None = None,
        metadata: dict | None = None,
    ) -> Content:
        """
        Uploads content to the knowledge base.

        Args:
            content (bytes): The content to upload.
            content_name (str): The name of the content.
            mime_type (str): The MIME type of the content.
            scope_id (str | None): The scope ID. Defaults to None.
            skip_ingestion (bool): Whether to skip ingestion. Defaults to False.
            skip_excel_ingestion (bool): Whether to skip excel ingestion. Defaults to False.
            ingestion_config (unique_sdk.Content.IngestionConfig | None): The ingestion configuration. Defaults to None.
            metadata (dict | None): The metadata to associate with the content. Defaults to None.

        Returns:
            Content: The uploaded content.
        """

        return await upload_content_from_bytes_async(
            user_id=self._user_id,
            company_id=self._company_id,
            content=content,
            content_name=content_name,
            mime_type=mime_type,
            scope_id=scope_id,
            chat_id=chat_id,
            skip_ingestion=skip_ingestion,
            ingestion_config=ingestion_config,
            metadata=metadata,
        )

    def upload_content(
        self,
        path_to_content: str,
        content_name: str,
        mime_type: str,
        scope_id: str | None = None,
        chat_id: str | None = None,
        skip_ingestion: bool = False,
        skip_excel_ingestion: bool = False,
        ingestion_config: unique_sdk.Content.IngestionConfig | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> Content:
        """
        Uploads content to the knowledge base.

        Args:
            path_to_content (str): The path to the content to upload.
            content_name (str): The name of the content.
            mime_type (str): The MIME type of the content.
            scope_id (str | None): The scope ID. Defaults to None.
            chat_id (str | None): The chat ID. Defaults to None.
            skip_ingestion (bool): Whether to skip ingestion. Defaults to False.
            skip_excel_ingestion (bool): Whether to skip excel ingestion. Defaults to False.
            ingestion_config (unique_sdk.Content.IngestionConfig | None): The ingestion configuration. Defaults to None.
            metadata (dict[str, Any] | None): The metadata to associate with the content. Defaults to None.

        Returns:
            Content: The uploaded content.
        """

        return upload_content(
            user_id=self._user_id,
            company_id=self._company_id,
            path_to_content=path_to_content,
            content_name=content_name,
            mime_type=mime_type,
            scope_id=scope_id,
            chat_id=chat_id,
            skip_ingestion=skip_ingestion,
            skip_excel_ingestion=skip_excel_ingestion,
            ingestion_config=ingestion_config,
            metadata=metadata,
        )

    def request_content_by_id(
        self,
        content_id: str,
        chat_id: str | None = None,
    ) -> Response:
        """
        Sends a request to download content from a chat.

        Args:
            content_id (str): The ID of the content to download.
            chat_id (str): The ID of the chat from which to download the content. Defaults to None to download from knowledge base.

        Returns:
            requests.Response: The response object containing the downloaded content.

        """
        chat_id = chat_id or self._chat_id  # type: ignore

        return request_content_by_id(
            user_id=self._user_id,
            company_id=self._company_id,
            content_id=content_id,
            chat_id=chat_id,
        )

    def download_content_to_file_by_id(
        self,
        content_id: str,
        chat_id: str | None = None,
        filename: str | None = None,
        tmp_dir_path: str | Path | None = "/tmp",
    ):
        """
        Downloads content from a chat and saves it to a file.

        Args:
            content_id (str): The ID of the content to download.
            chat_id (str | None): The ID of the chat to download from. Defaults to None and the file is downloaded from the knowledge base.
            filename (str | None): The name of the file to save the content as. If not provided, the original filename will be used. Defaults to None.
            tmp_dir_path (str | Path | None): The path to the temporary directory where the content will be saved. Defaults to "/tmp".

        Returns:
            Path: The path to the downloaded file.

        Raises:
            Exception: If the download fails or the filename cannot be determined.
        """

        chat_id = chat_id or self._chat_id  # type: ignore

        return download_content_to_file_by_id(
            user_id=self._user_id,
            company_id=self._company_id,
            content_id=content_id,
            chat_id=chat_id,
            filename=filename,
            tmp_dir_path=tmp_dir_path,
        )

    # TODO: Discuss if we should deprecate this method due to unclear use by content_name
    def download_content(
        self,
        content_id: str,
        content_name: str,
        chat_id: str | None = None,
        dir_path: str | Path | None = "/tmp",
    ) -> Path:
        """
        Downloads content to temporary directory

        Args:
            content_id (str): The id of the uploaded content.
            content_name (str): The name of the uploaded content.
            chat_id (Optional[str]): The chat_id, defaults to None.
            dir_path (Optional[Union[str, Path]]): The directory path to download the content to, defaults to "/tmp". If not provided, the content will be downloaded to a random directory inside /tmp. Be aware that this directory won't be cleaned up automatically.

        Returns:
            content_path: The path to the downloaded content in the temporary directory.

        Raises:
            Exception: If the download fails.
        """

        chat_id = chat_id or self._chat_id  # type: ignore

        return download_content(
            user_id=self._user_id,
            company_id=self._company_id,
            content_id=content_id,
            content_name=content_name,
            chat_id=chat_id,
            dir_path=dir_path,
        )

    def download_content_to_bytes(
        self,
        content_id: str,
        chat_id: str | None = None,
    ) -> bytes:
        """
        Downloads content to memory

        Args:
            content_id (str): The id of the uploaded content.
            chat_id (Optional[str]): The chat_id, defaults to None.

        Returns:
            bytes: The downloaded content.

        Raises:
            Exception: If the download fails.
        """
        chat_id = chat_id or self._chat_id  # type: ignore
        return download_content_to_bytes(
            user_id=self._user_id,
            company_id=self._company_id,
            content_id=content_id,
            chat_id=chat_id,
        )

    def get_documents_uploaded_to_chat(self) -> list[Content]:
        chat_contents = self.search_contents(
            where={
                "ownerId": {
                    "equals": self._chat_id,
                },
            },
        )

        content: list[Content] = []
        for c in chat_contents:
            if self.is_file_content(c.key):
                content.append(c)

        return content

    def get_images_uploaded_to_chat(self) -> list[Content]:
        chat_contents = self.search_contents(
            where={
                "ownerId": {
                    "equals": self._chat_id,
                },
            },
        )

        content: list[Content] = []
        for c in chat_contents:
            if self.is_image_content(c.key):
                content.append(c)

        return content

    def is_file_content(self, filename: str) -> bool:
        return is_file_content(filename=filename)

    def is_image_content(self, filename: str) -> bool:
        return is_image_content(filename=filename)

metadata_filter property writable

Get the metadata filter (deprecated).

Returns:

Type	Description
dict | None	dict | None: The metadata filter.

__init__(event=None, company_id=None, user_id=None, chat_id=None, metadata_filter=None)

__init__(event: Event | ChatEvent | BaseEvent)

__init__(
    *,
    company_id: str,
    user_id: str,
    chat_id: str | None = None,
    metadata_filter: dict | None = None,
)

Initialize the ContentService with a company_id, user_id and chat_id.

Source code in unique_toolkit/unique_toolkit/content/service.py

def __init__(
    self,
    event: Event | BaseEvent | None = None,
    company_id: str | None = None,
    user_id: str | None = None,
    chat_id: str | None = None,
    metadata_filter: dict | None = None,
):
    """
    Initialize the ContentService with a company_id, user_id and chat_id.
    """

    self._event = event  # Changed to protected attribute
    self._metadata_filter = None
    if event:
        self._company_id: str = event.company_id
        self._user_id: str = event.user_id
        if isinstance(event, (ChatEvent, Event)):
            self._metadata_filter = event.payload.metadata_filter
            self._chat_id: str | None = event.payload.chat_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
        self._chat_id: str | None = chat_id
        self._metadata_filter = metadata_filter

download_content(content_id, content_name, chat_id=None, dir_path='/tmp')

Downloads content to temporary directory

Parameters:

Name	Type	Description	Default
content_id	str	The id of the uploaded content.	required
content_name	str	The name of the uploaded content.	required
chat_id	Optional[str]	The chat_id, defaults to None.	None
dir_path	Optional[Union[str, Path]]	The directory path to download the content to, defaults to "/tmp". If not provided, the content will be downloaded to a random directory inside /tmp. Be aware that this directory won't be cleaned up automatically.	'/tmp'

Returns:

Name	Type	Description
content_path	Path	The path to the downloaded content in the temporary directory.

Raises:

Type	Description
Exception	If the download fails.

Source code in unique_toolkit/unique_toolkit/content/service.py

def download_content(
    self,
    content_id: str,
    content_name: str,
    chat_id: str | None = None,
    dir_path: str | Path | None = "/tmp",
) -> Path:
    """
    Downloads content to temporary directory

    Args:
        content_id (str): The id of the uploaded content.
        content_name (str): The name of the uploaded content.
        chat_id (Optional[str]): The chat_id, defaults to None.
        dir_path (Optional[Union[str, Path]]): The directory path to download the content to, defaults to "/tmp". If not provided, the content will be downloaded to a random directory inside /tmp. Be aware that this directory won't be cleaned up automatically.

    Returns:
        content_path: The path to the downloaded content in the temporary directory.

    Raises:
        Exception: If the download fails.
    """

    chat_id = chat_id or self._chat_id  # type: ignore

    return download_content(
        user_id=self._user_id,
        company_id=self._company_id,
        content_id=content_id,
        content_name=content_name,
        chat_id=chat_id,
        dir_path=dir_path,
    )

download_content_to_bytes(content_id, chat_id=None)

Downloads content to memory

Parameters:

Name	Type	Description	Default
content_id	str	The id of the uploaded content.	required
chat_id	Optional[str]	The chat_id, defaults to None.	None

Returns:

Name	Type	Description
bytes	bytes	The downloaded content.

Raises:

Type	Description
Exception	If the download fails.

Source code in unique_toolkit/unique_toolkit/content/service.py

def download_content_to_bytes(
    self,
    content_id: str,
    chat_id: str | None = None,
) -> bytes:
    """
    Downloads content to memory

    Args:
        content_id (str): The id of the uploaded content.
        chat_id (Optional[str]): The chat_id, defaults to None.

    Returns:
        bytes: The downloaded content.

    Raises:
        Exception: If the download fails.
    """
    chat_id = chat_id or self._chat_id  # type: ignore
    return download_content_to_bytes(
        user_id=self._user_id,
        company_id=self._company_id,
        content_id=content_id,
        chat_id=chat_id,
    )

from_settings(settings=None, metadata_filter=None) classmethod

Initialize the ContentService with a settings object and metadata filter.

Source code in unique_toolkit/unique_toolkit/content/service.py

@classmethod
def from_settings(
    cls,
    settings: UniqueSettings | str | None = None,
    metadata_filter: dict | None = None,
):
    """
    Initialize the ContentService with a settings object and metadata filter.
    """

    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(),
        metadata_filter=metadata_filter,
    )

search_content_chunks(search_string, search_type, limit, search_language=DEFAULT_SEARCH_LANGUAGE, chat_id='', reranker_config=None, scope_ids=None, chat_only=None, metadata_filter=None, content_ids=None, score_threshold=None)

Performs a synchronous search for content chunks in the knowledge base.

Parameters:

Name	Type	Description	Default
search_string	str	The search string.	required
search_type	ContentSearchType	The type of search to perform.	required
limit	int	The maximum number of results to return.	required
search_language	str	The language for the full-text search. Defaults to "english".	DEFAULT_SEARCH_LANGUAGE
chat_id	str	The chat ID for context. Defaults to empty string.	''
reranker_config	ContentRerankerConfig | None	The reranker configuration. Defaults to None.	None
scope_ids	list[str] | None	The scope IDs to filter by. Defaults to None.	None
chat_only	bool | None	Whether to search only in the current chat. Defaults to None.	None
metadata_filter	dict | None	UniqueQL metadata filter. If unspecified/None, it tries to use the metadata filter from the event. Defaults to None.	None
content_ids	list[str] | None	The content IDs to search within. Defaults to None.	None
score_threshold	float | None	Sets the minimum similarity score for search results to be considered. Defaults to 0.	None

Returns:

Type	Description
list[ContentChunk]	list[ContentChunk]: The search results.

Raises:

Type	Description
Exception	If there's an error during the search operation.

Source code in unique_toolkit/unique_toolkit/content/service.py

def search_content_chunks(
    self,
    search_string: str,
    search_type: ContentSearchType,
    limit: int,
    search_language: str = DEFAULT_SEARCH_LANGUAGE,
    chat_id: str = "",
    reranker_config: ContentRerankerConfig | None = None,
    scope_ids: list[str] | None = None,
    chat_only: bool | None = None,
    metadata_filter: dict | None = None,
    content_ids: list[str] | None = None,
    score_threshold: float | None = None,
) -> list[ContentChunk]:
    """
    Performs a synchronous search for content chunks in the knowledge base.

    Args:
        search_string (str): The search string.
        search_type (ContentSearchType): The type of search to perform.
        limit (int): The maximum number of results to return.
        search_language (str, optional): The language for the full-text search. Defaults to "english".
        chat_id (str, optional): The chat ID for context. Defaults to empty string.
        reranker_config (ContentRerankerConfig | None, optional): The reranker configuration. Defaults to None.
        scope_ids (list[str] | None, optional): The scope IDs to filter by. Defaults to None.
        chat_only (bool | None, optional): Whether to search only in the current chat. Defaults to None.
        metadata_filter (dict | None, optional): UniqueQL metadata filter. If unspecified/None, it tries to use the metadata filter from the event. Defaults to None.
        content_ids (list[str] | None, optional): The content IDs to search within. Defaults to None.
        score_threshold (float | None, optional): Sets the minimum similarity score for search results to be considered. Defaults to 0.

    Returns:
        list[ContentChunk]: The search results.

    Raises:
        Exception: If there's an error during the search operation.
    """

    if metadata_filter is None:
        metadata_filter = self._metadata_filter

    chat_id = chat_id or self._chat_id  # type: ignore

    if chat_only and not chat_id:
        raise ValueError("Please provide chat_id when limiting with chat_only")

    try:
        searches = search_content_chunks(
            user_id=self._user_id,
            company_id=self._company_id,
            chat_id=chat_id,
            search_string=search_string,
            search_type=search_type,
            limit=limit,
            search_language=search_language,
            reranker_config=reranker_config,
            scope_ids=scope_ids,
            chat_only=chat_only,
            metadata_filter=metadata_filter,
            content_ids=content_ids,
            score_threshold=score_threshold,
        )
        return searches
    except Exception as e:
        logger.error(f"Error while searching content chunks: {e}")
        raise e

search_contents(where, chat_id='')

Performs a search in the knowledge base by filter (and not a smilarity search) This function loads complete content of the files from the knowledge base in contrast to search_content_chunks.

Parameters:

Name	Type	Description	Default
where	dict	The search criteria.	required

Returns:

Type	Description
list[Content]	list[Content]: The search results.

Source code in unique_toolkit/unique_toolkit/content/service.py

def search_contents(
    self,
    where: dict,
    chat_id: str = "",
) -> list[Content]:
    """
    Performs a search in the knowledge base by filter (and not a smilarity search)
    This function loads complete content of the files from the knowledge base in contrast to search_content_chunks.

    Args:
        where (dict): The search criteria.

    Returns:
        list[Content]: The search results.
    """
    chat_id = chat_id or self._chat_id  # type: ignore

    return search_contents(
        user_id=self._user_id,
        company_id=self._company_id,
        chat_id=chat_id,
        where=where,
    )

upload_content(path_to_content, content_name, mime_type, scope_id=None, chat_id=None, skip_ingestion=False, skip_excel_ingestion=False, ingestion_config=None, metadata=None)

Uploads content to the knowledge base.

Parameters:

Name	Type	Description	Default
path_to_content	str	The path to the content to upload.	required
content_name	str	The name of the content.	required
mime_type	str	The MIME type of the content.	required
scope_id	str | None	The scope ID. Defaults to None.	None
chat_id	str | None	The chat ID. Defaults to None.	None
skip_ingestion	bool	Whether to skip ingestion. Defaults to False.	False
skip_excel_ingestion	bool	Whether to skip excel ingestion. Defaults to False.	False
ingestion_config	IngestionConfig | None	The ingestion configuration. Defaults to None.	None
metadata	dict[str, Any] | None	The metadata to associate with the content. Defaults to None.	None

Returns:

Name	Type	Description
Content	Content	The uploaded content.

Source code in unique_toolkit/unique_toolkit/content/service.py

def upload_content(
    self,
    path_to_content: str,
    content_name: str,
    mime_type: str,
    scope_id: str | None = None,
    chat_id: str | None = None,
    skip_ingestion: bool = False,
    skip_excel_ingestion: bool = False,
    ingestion_config: unique_sdk.Content.IngestionConfig | None = None,
    metadata: dict[str, Any] | None = None,
) -> Content:
    """
    Uploads content to the knowledge base.

    Args:
        path_to_content (str): The path to the content to upload.
        content_name (str): The name of the content.
        mime_type (str): The MIME type of the content.
        scope_id (str | None): The scope ID. Defaults to None.
        chat_id (str | None): The chat ID. Defaults to None.
        skip_ingestion (bool): Whether to skip ingestion. Defaults to False.
        skip_excel_ingestion (bool): Whether to skip excel ingestion. Defaults to False.
        ingestion_config (unique_sdk.Content.IngestionConfig | None): The ingestion configuration. Defaults to None.
        metadata (dict[str, Any] | None): The metadata to associate with the content. Defaults to None.

    Returns:
        Content: The uploaded content.
    """

    return upload_content(
        user_id=self._user_id,
        company_id=self._company_id,
        path_to_content=path_to_content,
        content_name=content_name,
        mime_type=mime_type,
        scope_id=scope_id,
        chat_id=chat_id,
        skip_ingestion=skip_ingestion,
        skip_excel_ingestion=skip_excel_ingestion,
        ingestion_config=ingestion_config,
        metadata=metadata,
    )

upload_content_from_bytes(content, content_name, mime_type, scope_id=None, chat_id=None, skip_ingestion=False, skip_excel_ingestion=False, ingestion_config=None, metadata=None)

Uploads content to the knowledge base.

Parameters:

Name	Type	Description	Default
content	bytes	The content to upload.	required
content_name	str	The name of the content.	required
mime_type	str	The MIME type of the content.	required
scope_id	str | None	The scope ID. Defaults to None.	None
chat_id	str | None	The chat ID. Defaults to None.	None
skip_ingestion	bool	Whether to skip ingestion. Defaults to False.	False
skip_excel_ingestion	bool	Whether to skip excel ingestion. Defaults to False.	False
ingestion_config	IngestionConfig | None	The ingestion configuration. Defaults to None.	None
metadata	dict | None	The metadata to associate with the content. Defaults to None.	None

Returns:

Name	Type	Description
Content	Content	The uploaded content.

Source code in unique_toolkit/unique_toolkit/content/service.py

def upload_content_from_bytes(
    self,
    content: bytes,
    content_name: str,
    mime_type: str,
    scope_id: str | None = None,
    chat_id: str | None = None,
    skip_ingestion: bool = False,
    skip_excel_ingestion: bool = False,
    ingestion_config: unique_sdk.Content.IngestionConfig | None = None,
    metadata: dict | None = None,
) -> Content:
    """
    Uploads content to the knowledge base.

    Args:
        content (bytes): The content to upload.
        content_name (str): The name of the content.
        mime_type (str): The MIME type of the content.
        scope_id (str | None): The scope ID. Defaults to None.
        chat_id (str | None): The chat ID. Defaults to None.
        skip_ingestion (bool): Whether to skip ingestion. Defaults to False.
        skip_excel_ingestion (bool): Whether to skip excel ingestion. Defaults to False.
        ingestion_config (unique_sdk.Content.IngestionConfig | None): The ingestion configuration. Defaults to None.
        metadata (dict | None): The metadata to associate with the content. Defaults to None.

    Returns:
        Content: The uploaded content.
    """

    return upload_content_from_bytes(
        user_id=self._user_id,
        company_id=self._company_id,
        content=content,
        content_name=content_name,
        mime_type=mime_type,
        scope_id=scope_id,
        chat_id=chat_id,
        skip_ingestion=skip_ingestion,
        ingestion_config=ingestion_config,
        metadata=metadata,
    )

Schemas

unique_toolkit.content.schemas.Content

Bases: BaseModel

Source code in unique_toolkit/unique_toolkit/content/schemas.py

class Content(BaseModel):
    model_config = model_config
    id: str = Field(
        default="",
        description="The id of the content. The id starts with 'cont_' followed by an alphanumeric string of length 24.",
        examples=["cont_abcdefgehijklmnopqrstuvwx"],
    )
    key: str = Field(
        default="",
        description="The key of the content. For documents this is the the filename",
    )
    title: str | None = Field(
        default=None,
        description="The title of the content. For documents this is the title of the document.",
    )
    url: str | None = None
    chunks: list[ContentChunk] = []
    write_url: str | None = None
    read_url: str | None = None
    created_at: datetime | None = None
    updated_at: datetime | None = None
    expired_at: datetime | None = None
    metadata: dict[str, Any] | None = None
    ingestion_config: dict | None = None
    ingestion_state: str | None = None

unique_toolkit.content.schemas.ContentChunk

Bases: BaseModel

Source code in unique_toolkit/unique_toolkit/content/schemas.py

class ContentChunk(BaseModel):
    model_config = model_config
    id: str = Field(
        default="",
        description="The id of the content this chunk belongs to. The id starts with 'cont_' followed by an alphanumeric string of length 24.",
        examples=["cont_abcdefgehijklmnopqrstuvwx"],
    )
    text: str = Field(default="", description="The text content of the chunk.")
    order: int = Field(
        default=0,
        description="The order of the chunk in the original content. Concatenating the chunks in order will give the original content.",
    )
    key: str | None = Field(
        default=None,
        description="The key of the chunk. For document chunks this is the the filename",
    )
    chunk_id: str | None = Field(
        default=None,
        description="The id of the chunk. The id starts with 'chunk_' followed by an alphanumeric string of length 24.",
        examples=["chunk_abcdefgehijklmnopqrstuv"],
    )
    url: str | None = Field(
        default=None,
        description="For chunk retrieved from the web this is the url of the chunk.",
    )
    title: str | None = Field(
        default=None,
        description="The title of the chunk. For document chunks this is the title of the document.",
    )
    start_page: int | None = Field(
        default=None,
        description="The start page of the chunk. For document chunks this is the start page of the document.",
    )
    end_page: int | None = Field(
        default=None,
        description="The end page of the chunk. For document chunks this is the end page of the document.",
    )

    object: str | None = None
    metadata: ContentMetadata | None = None
    internally_stored_at: datetime | None = None
    created_at: datetime | None = None
    updated_at: datetime | None = None

unique_toolkit.content.schemas.ContentSearchType

Bases: StrEnum

Source code in unique_toolkit/unique_toolkit/content/schemas.py

class ContentSearchType(StrEnum):
    COMBINED = "COMBINED"
    VECTOR = "VECTOR"

unique_toolkit.content.schemas.ContentRerankerConfig

Bases: BaseModel

Source code in unique_toolkit/unique_toolkit/content/schemas.py

class ContentRerankerConfig(BaseModel):
    model_config = model_config
    deployment_name: str = Field(serialization_alias="deploymentName")
    options: dict | None = None

Functions

unique_toolkit.content.functions.search_content_chunks(user_id, company_id, chat_id, search_string, search_type, limit, search_language=DEFAULT_SEARCH_LANGUAGE, reranker_config=None, scope_ids=None, chat_only=None, metadata_filter=None, content_ids=None, score_threshold=None)

Performs a synchronous search for content chunks in the knowledge base.

Parameters:

Name	Type	Description	Default
search_string	str	The search string.	required
search_type	ContentSearchType	The type of search to perform.	required
limit	int	The maximum number of results to return.	required
search_language	str	The language for the full-text search. Defaults to "english".	DEFAULT_SEARCH_LANGUAGE
reranker_config	ContentRerankerConfig | None	The reranker configuration. Defaults to None.	None
scope_ids	list[str] | None	The scope IDs. Defaults to None.	None
chat_only	bool | None	Whether to search only in the current chat. Defaults to None.	None
metadata_filter	dict | None	UniqueQL metadata filter. If unspecified/None, it tries to use the metadata filter from the event. Defaults to None.	None
content_ids	list[str] | None	The content IDs to search. Defaults to None.	None
score_threshold	float | None	The minimum score threshold for results. Defaults to 0.	None

Source code in unique_toolkit/unique_toolkit/content/functions.py

def search_content_chunks(
    user_id: str,
    company_id: str,
    chat_id: str,
    search_string: str,
    search_type: ContentSearchType,
    limit: int,
    search_language: str = DEFAULT_SEARCH_LANGUAGE,
    reranker_config: ContentRerankerConfig | None = None,
    scope_ids: list[str] | None = None,
    chat_only: bool | None = None,
    metadata_filter: dict | None = None,
    content_ids: list[str] | None = None,
    score_threshold: float | None = None,
) -> list[ContentChunk]:
    """
    Performs a synchronous search for content chunks in the knowledge base.

    Args:
        search_string (str): The search string.
        search_type (ContentSearchType): The type of search to perform.
        limit (int): The maximum number of results to return.
        search_language (str): The language for the full-text search. Defaults to "english".
        reranker_config (ContentRerankerConfig | None): The reranker configuration. Defaults to None.
        scope_ids (list[str] | None): The scope IDs. Defaults to None.
        chat_only (bool | None): Whether to search only in the current chat. Defaults to None.
        metadata_filter (dict | None): UniqueQL metadata filter. If unspecified/None, it tries to use the metadata filter from the event. Defaults to None.
        content_ids (list[str] | None): The content IDs to search. Defaults to None.
        score_threshold (float | None): The minimum score threshold for results. Defaults to 0.
    Returns:
        list[ContentChunk]: The search results.
    """
    if not scope_ids:
        logger.warning("No scope IDs provided for search.")

    if content_ids:
        logger.info(f"Searching for content chunks with content_ids: {content_ids}")

    try:
        searches = unique_sdk.Search.create(
            user_id=user_id,
            company_id=company_id,
            chatId=chat_id,
            searchString=search_string,
            searchType=search_type.name,
            scopeIds=scope_ids,
            limit=limit,
            reranker=(
                reranker_config.model_dump(by_alias=True) if reranker_config else None
            ),
            language=search_language,
            chatOnly=chat_only,
            metaDataFilter=metadata_filter,
            contentIds=content_ids,
            scoreThreshold=score_threshold,
        )
        return map_to_content_chunks(searches)
    except Exception as e:
        logger.error(f"Error while searching content chunks: {e}")
        raise e

unique_toolkit.content.functions.search_contents(user_id, company_id, chat_id, where, include_failed_content=False)

Performs an asynchronous search for content files in the knowledge base by filter.

Parameters:

Name	Type	Description	Default
user_id	str	The user ID.	required
company_id	str	The company ID.	required
chat_id	str	The chat ID.	required
where	dict	The search criteria.	required

Returns:

Type	Description
list[Content]	list[Content]: The search results.

Source code in unique_toolkit/unique_toolkit/content/functions.py

def search_contents(
    user_id: str,
    company_id: str,
    chat_id: str,
    where: dict,
    include_failed_content: bool = False,
) -> list[Content]:
    """
    Performs an asynchronous search for content files in the knowledge base by filter.

    Args:
        user_id (str): The user ID.
        company_id (str): The company ID.
        chat_id (str): The chat ID.
        where (dict): The search criteria.

    Returns:
        list[Content]: The search results.
    """
    if where.get("contentId"):
        logger.info(f"Searching for content with content_id: {where['contentId']}")

    try:
        contents = unique_sdk.Content.search(
            user_id=user_id,
            company_id=company_id,
            chatId=chat_id,
            # TODO add type parameter in SDK
            where=where,  # type: ignore
            includeFailedContent=include_failed_content,
        )
        return map_contents(contents)
    except Exception as e:
        logger.error(f"Error while searching contents: {e}")
        raise e

unique_toolkit.content.functions.upload_content(user_id, company_id, path_to_content, content_name, mime_type, scope_id=None, chat_id=None, skip_ingestion=False, skip_excel_ingestion=False, ingestion_config=None, metadata=None)

Uploads content to the knowledge base.

Parameters:

Name	Type	Description	Default
user_id	str	The user ID.	required
company_id	str	The company ID.	required
path_to_content	str	The path to the content to upload.	required
content_name	str	The name of the content.	required
mime_type	str	The MIME type of the content.	required
scope_id	str | None	The scope ID. Defaults to None.	None
chat_id	str | None	The chat ID. Defaults to None.	None
skip_ingestion	bool	Whether to skip ingestion. Defaults to False.	False
skip_excel_ingestion	bool	Whether to skip excel ingestion. Defaults to False.	False
ingestion_config	IngestionConfig | None	The ingestion configuration. Defaults to None.	None
metadata	dict[str, Any] | None	The metadata for the content. Defaults to None.	None

Returns:

Name	Type	Description
Content	Content	The uploaded content.

Source code in unique_toolkit/unique_toolkit/content/functions.py

def upload_content(
    user_id: str,
    company_id: str,
    path_to_content: str,
    content_name: str,
    mime_type: str,
    scope_id: str | None = None,
    chat_id: str | None = None,
    skip_ingestion: bool = False,
    skip_excel_ingestion: bool = False,
    ingestion_config: unique_sdk.Content.IngestionConfig | None = None,
    metadata: dict[str, Any] | None = None,
) -> Content:
    """
    Uploads content to the knowledge base.

    Args:
        user_id (str): The user ID.
        company_id (str): The company ID.
        path_to_content (str): The path to the content to upload.
        content_name (str): The name of the content.
        mime_type (str): The MIME type of the content.
        scope_id (str | None): The scope ID. Defaults to None.
        chat_id (str | None): The chat ID. Defaults to None.
        skip_ingestion (bool): Whether to skip ingestion. Defaults to False.
        skip_excel_ingestion (bool): Whether to skip excel ingestion. Defaults to False.
        ingestion_config (unique_sdk.Content.IngestionConfig | None): The ingestion configuration. Defaults to None.
        metadata ( dict[str, Any] | None): The metadata for the content. Defaults to None.

    Returns:
        Content: The uploaded content.
    """

    try:
        return _trigger_upload_content(
            user_id=user_id,
            company_id=company_id,
            content=path_to_content,
            content_name=content_name,
            mime_type=mime_type,
            scope_id=scope_id,
            chat_id=chat_id,
            skip_ingestion=skip_ingestion,
            skip_excel_ingestion=skip_excel_ingestion,
            ingestion_config=ingestion_config,
            metadata=metadata,
        )
    except Exception as e:
        logger.error(f"Error while uploading content: {e}")
        raise e

unique_toolkit.content.functions.download_content(user_id, company_id, content_id, content_name, chat_id=None, dir_path='/tmp')

Downloads content to temporary directory

Parameters:

Name	Type	Description	Default
user_id	str	The user ID.	required
company_id	str	The company ID.	required
content_id	str	The id of the uploaded content.	required
content_name	str	The name of the uploaded content.	required
chat_id	str | None	The chat_id, defaults to None.	None
dir_path	str | Path	The directory path to download the content to, defaults to "/tmp". If not provided, the content will be downloaded to a random directory inside /tmp. Be aware that this directory won't be cleaned up automatically.	'/tmp'

Returns:

Name	Type	Description
content_path	Path	The path to the downloaded content in the temporary directory.

Raises:

Type	Description
Exception	If the download fails.

Source code in unique_toolkit/unique_toolkit/content/functions.py

def download_content(
    user_id: str,
    company_id: str,
    content_id: str,
    content_name: str,
    chat_id: str | None = None,
    dir_path: str | Path | None = "/tmp",
) -> Path:
    """
    Downloads content to temporary directory

    Args:
        user_id (str): The user ID.
        company_id (str): The company ID.
        content_id (str): The id of the uploaded content.
        content_name (str): The name of the uploaded content.
        chat_id (str | None): The chat_id, defaults to None.
        dir_path (str | Path): The directory path to download the content to, defaults to "/tmp". If not provided, the content will be downloaded to a random directory inside /tmp. Be aware that this directory won't be cleaned up automatically.

    Returns:
        content_path: The path to the downloaded content in the temporary directory.

    Raises:
        Exception: If the download fails.
    """

    logger.info(f"Downloading content with content_id: {content_id}")
    response = request_content_by_id(user_id, company_id, content_id, chat_id)

    random_dir = tempfile.mkdtemp(dir=dir_path)
    content_path = Path(random_dir) / content_name

    if response.status_code == 200:
        with open(content_path, "wb") as file:
            file.write(response.content)
    else:
        error_msg = f"Error downloading file: Status code {response.status_code}"
        logger.error(error_msg)
        raise Exception(error_msg)

    return content_path

Utilities

unique_toolkit.content.utils.sort_content_chunks(content_chunks)

Sorts the content chunks based on their 'order' in the original content. This function sorts the search results based on their 'order' in ascending order. It also performs text modifications by replacing the string within the tags <|/content|> with 'text part {order}' and removing any <|info|> tags (Which is useful in referencing the chunk). Parameters: - content_chunks (list): A list of ContentChunkt objects. Returns: - list: A list of ContentChunk objects sorted according to their order.

Source code in unique_toolkit/unique_toolkit/content/utils.py

def sort_content_chunks(content_chunks: list[ContentChunk]):
    """
    Sorts the content chunks based on their 'order' in the original content.
    This function sorts the search results based on their 'order' in ascending order.
    It also performs text modifications by replacing the string within the tags <|/content|>
    with 'text part {order}' and removing any <|info|> tags (Which is useful in referencing the chunk).
    Parameters:
    - content_chunks (list): A list of ContentChunkt objects.
    Returns:
    - list: A list of ContentChunk objects sorted according to their order.
    """
    doc_id_to_chunks = _map_content_id_to_chunks(content_chunks)
    sorted_chunks: list[ContentChunk] = []
    for chunks in doc_id_to_chunks.values():
        chunks.sort(key=lambda x: x.order)
        for i, s in enumerate(chunks):
            s.text = re.sub(
                r"<\|/content\|>", f" text part {s.order}<|/content|>", s.text
            )
            s.text = re.sub(r"<\|info\|>(.*?)<\|\/info\|>", "", s.text)
            pages_postfix = _generate_pages_postfix([s])
            s.key = s.key + pages_postfix if s.key else s.key
            s.title = s.title + pages_postfix if s.title else s.title
        sorted_chunks.extend(chunks)
    return sorted_chunks

unique_toolkit.content.utils.merge_content_chunks(content_chunks)

Merges multiple search results based on their 'id', removing redundant content and info markers.

This function groups search results by their 'id' and then concatenates their texts, cleaning up any content or info markers in subsequent chunks beyond the first one.

Parameters: - content_chunks (list): A list of objects, each representing a search result with 'id' and 'text' keys.

Returns: - list: A list of objects with merged texts for each unique 'id'.

Source code in unique_toolkit/unique_toolkit/content/utils.py

def merge_content_chunks(content_chunks: list[ContentChunk]):
    """
    Merges multiple search results based on their 'id', removing redundant content and info markers.

    This function groups search results by their 'id' and then concatenates their texts,
    cleaning up any content or info markers in subsequent chunks beyond the first one.

    Parameters:
    - content_chunks (list): A list of objects, each representing a search result with 'id' and 'text' keys.

    Returns:
    - list: A list of objects with merged texts for each unique 'id'.
    """

    doc_id_to_chunks = _map_content_id_to_chunks(content_chunks)
    merged_chunks: list[ContentChunk] = []
    for chunks in doc_id_to_chunks.values():
        chunks.sort(key=lambda x: x.order)
        for i, s in enumerate(chunks):
            ## skip first element
            if i > 0:
                ## replace the string within the tags <|content|>...<|/content|> and <|info|> and <|/info|>
                s.text = re.sub(r"<\|content\|>(.*?)<\|\/content\|>", "", s.text)
                s.text = re.sub(r"<\|info\|>(.*?)<\|\/info\|>", "", s.text)

        pages_postfix = _generate_pages_postfix(chunks)
        chunks[0].text = "\n".join(str(s.text) for s in chunks)
        chunks[0].key = (
            chunks[0].key + pages_postfix if chunks[0].key else chunks[0].key
        )
        chunks[0].title = (
            chunks[0].title + pages_postfix if chunks[0].title else chunks[0].title
        )
        chunks[0].end_page = chunks[-1].end_page
        merged_chunks.append(chunks[0])

    return merged_chunks

unique_toolkit.content.utils.count_tokens(text, encoding_model='cl100k_base')

Counts the number of tokens in the provided text.

This function encodes the input text using a predefined encoding scheme and returns the number of tokens in the encoded text.

Parameters: - text (str): The text to count tokens for.

Returns: - int: The number of tokens in the text.

Source code in unique_toolkit/unique_toolkit/content/utils.py

def count_tokens(text: str, encoding_model="cl100k_base") -> int:
    """
    Counts the number of tokens in the provided text.

    This function encodes the input text using a predefined encoding scheme
    and returns the number of tokens in the encoded text.

    Parameters:
    - text (str): The text to count tokens for.

    Returns:
    - int: The number of tokens in the text.
    """
    encoding = tiktoken.get_encoding(encoding_model)
    return len(encoding.encode(text))