Aller au contenu

Utils Module

The utils module provides utility functions and helper methods for the Forra API.

UtilsAPI

Source code in forrasdk/api/utils.py 
 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
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
class UtilsAPI:
    def __init__(self, base_url: str, headers: dict, retry_strategy: Callable) -> None:
        self._base_url = base_url
        self._headers = headers
        self._retry_strategy = retry_strategy

    def chunk_document(self, file_path: str) -> dict:
        """
        Chunk a document into smaller parts for embedding using default Scout chunk algorythm.

        Args:
            file_path (str): The local file path of the document to be chunked.

        Returns:
            dict: The response from the API after chunking the document. {"file": {"chunks": [{chunk_to_embed}]}}
        """
        with open(file_path, "rb") as f:
            files = {"file": f}
            local_headers = self._headers.copy()
            local_headers.pop("Content-Type")

            response, status_code = RequestUtils.post(
                url=f"{self._base_url}/api/utils/chunk-document",
                headers=local_headers,
                files=files,
                retry_strategy=self._retry_strategy,
            )
        return response

    def get_document_text(self, file_path: str, args: Optional[dict] = None) -> dict:
        """
        Extract text content from a file.

        Args:
            file_path (str): The local file path of the file to extract text from.
            args (Optional[dict], optional): Additional arguments for text extraction. Defaults to None.

        Returns:
            dict: The response from the API after extracting text from the file.
        """
        with open(file_path, "rb") as f:
            files = {"file": f}
            local_headers = self._headers.copy()
            local_headers.pop("Content-Type")

            response, status_code = RequestUtils.post(
                url=f"{self._base_url}/api/utils/get-file-text-content",
                headers=local_headers,
                files=files,
                json_payload=args,
                retry_strategy=self._retry_strategy,
            )
        return response

    def llm_filter_documents(
        self,
        query: str,
        context: str,
        documents: dict[str, str],
        batch_size: int = 10,
        model_id: Optional[str] = None,
    ) -> list:
        """
        Filter a set of documents using an LLM based on a query and context.

        Args:
            query (str): The query string to use for filtering documents. Ex: What is the capital of France?
            context (str): Additional context to provide to the LLM during filtering. Ex: You are an expert in selecting content to answer geographical questions.
            documents (dict[str, str]): A dictionary mapping document IDs to document texts. EX: {"my_id": "Capital of france is paris", "my_id_2": "Irrelevant content"}
            batch_size (int, optional): Number of documents to process in each batch. Defaults to 10.
            model_id (Optional[str], optional): The ID of the LLM model to use for filtering. When not provided, use default model.

        Returns:
            list: List of ids that are relevant result to answer the question
        """
        response, status_code = RequestUtils.post(
            url=f"{self._base_url}/api/utils/llm-filter-documents",
            headers=self._headers,
            json_payload={
                "query": query,
                "context": context,
                "documents": documents,
                "batch_size": batch_size,
                "model_id": model_id,
            },
            retry_strategy=self._retry_strategy,
        )
        return response

    def filter_assistant_datas(
        self,
        objects: List[T],
        query: str,
        filter_prompt: str,
        batch_size: int = 10,
        model_id: Optional[str] = None,
    ) -> List[T]:
        """
        Filter a list of assistant data objects with content and metadata using an LLM.

        This is a generic method that works with any objects that have 'content' and 'metadata' fields,
        including AssistantSearchDataResponse and AssistantDataResponseItem.

        Args:
            objects: List of objects that implement ContentMetadataProtocol (have content and metadata fields)
            query: The user's query string to use for filtering
            filter_prompt: Context/instructions for the LLM during filtering
            batch_size: Number of objects to process in each batch (default: 10)
            model_id: Optional LLM model ID to use for filtering

        Returns:
            List of filtered objects (same type as input)

        Example:
            >>> from forratypes.assistants import AssistantSearchDataResponse
            >>> search_results = scout.current_assistant.search_data("python tutorials")
            >>> filtered_results = scout.utils.filter_assistant_datas(
            ...     objects=search_results,
            ...     query="How to handle exceptions?",
            ...     filter_prompt="You are an expert at finding relevant Python documentation."
            ... )
        """
        if not objects:
            return []

        # Create a mapping of IDs to objects for filtering
        id_to_object = {}
        documents_dict = {}

        for idx, obj in enumerate(objects):
            # Check if object has required fields
            if not isinstance(obj, ContentMetadataProtocol):
                raise TypeError(
                    f"Object at index {idx} does not have required 'content' and 'metadata' fields"
                )

            # Create unique ID for this object
            obj_id = str(idx)
            id_to_object[obj_id] = obj

            # Prepare document for LLM filtering
            # Include both content and metadata in the document text
            doc_text = obj.content
            if hasattr(obj, "metadata") and obj.metadata:
                # Add metadata as JSON string for context
                if isinstance(obj.metadata, dict):
                    doc_text += f"\n\nMetadata: {json.dumps(obj.metadata)}"
                elif hasattr(obj.metadata, "model_dump"):
                    # Handle Pydantic models
                    doc_text += f"\n\nMetadata: {json.dumps(obj.metadata.model_dump())}"
                else:
                    doc_text += f"\n\nMetadata: {str(obj.metadata)}"

            documents_dict[obj_id] = doc_text

        # Use the existing llm_filter_documents method
        filtered_ids = self.llm_filter_documents(
            query=query,
            context=filter_prompt,
            documents=documents_dict,
            batch_size=batch_size,
            model_id=model_id,
        )

        # Return the filtered objects in their original type
        return [
            id_to_object[obj_id] for obj_id in filtered_ids if obj_id in id_to_object
        ]

    def create_embeddings(self, texts: list[str]) -> list[list[float]]:
        """
        Create embeddings for a list of text strings.

        Args:
            texts (list[str]): List of text strings to create embeddings for.

        Returns:
            list[list[float]]: List of embedding vectors, where each vector is a list of floats.
        """
        response, status_code = RequestUtils.post(
            url=f"{self._base_url}/api/embeddings/",
            headers=self._headers,
            json_payload={"texts": texts},
            retry_strategy=self._retry_strategy,
        )

        return response

    def download_protected_file(self, protected_url: str) -> bytes:
        """
        Download a file from a protected URL by first getting a signed URL and then downloading.

        Args:
            protected_url (str): The protected URL path (e.g., "/protected/conversations/123/audio.mp3").

        Returns:
            bytes: The file content as bytes.

        Raises:
            Exception: If there's an error getting the signed URL or downloading the file.
        """

        response, status_code = RequestUtils.get(
            url=f"{self._base_url}/api{protected_url}",
            headers=self._headers,
            retry_strategy=self._retry_strategy,
        )

        signed_url_response = SignedUrlResponse.model_validate(response)

        download_response = requests.get(signed_url_response.url)

        try:
            download_response.raise_for_status()
        except requests.exceptions.HTTPError as e:
            error_message = "HTTP Error occurred while downloading file"
            if e.response is not None:
                try:
                    error_details = e.response.json()
                    error_message = f"Download Error: {error_details}"
                except (ValueError, AttributeError):
                    error_message = f"Download Error: HTTP {e.response.status_code} - {e.response.text or 'No response text'}"
            else:
                error_message = f"Download Error: {str(e)}"
            raise Exception(error_message) from e

        return download_response.content

chunk_document(file_path)

Chunk a document into smaller parts for embedding using default Scout chunk algorythm.

Parameters:

Name Type Description Default
file_path str

The local file path of the document to be chunked.

 required

Returns:

Name Type Description
dict dict

The response from the API after chunking the document. {"file": {"chunks": [{chunk_to_embed}]}}
Source code in forrasdk/api/utils.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def chunk_document(self, file_path: str) -> dict:
    """
    Chunk a document into smaller parts for embedding using default Scout chunk algorythm.

    Args:
        file_path (str): The local file path of the document to be chunked.

    Returns:
        dict: The response from the API after chunking the document. {"file": {"chunks": [{chunk_to_embed}]}}
    """
    with open(file_path, "rb") as f:
        files = {"file": f}
        local_headers = self._headers.copy()
        local_headers.pop("Content-Type")

        response, status_code = RequestUtils.post(
            url=f"{self._base_url}/api/utils/chunk-document",
            headers=local_headers,
            files=files,
            retry_strategy=self._retry_strategy,
        )
    return response

create_embeddings(texts)

Create embeddings for a list of text strings.

Parameters:

Name Type Description Default
texts list[str]

List of text strings to create embeddings for.

 required

Returns:

Type Description
list[list[float]]

list[list[float]]: List of embedding vectors, where each vector is a list of floats.
Source code in forrasdk/api/utils.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def create_embeddings(self, texts: list[str]) -> list[list[float]]:
    """
    Create embeddings for a list of text strings.

    Args:
        texts (list[str]): List of text strings to create embeddings for.

    Returns:
        list[list[float]]: List of embedding vectors, where each vector is a list of floats.
    """
    response, status_code = RequestUtils.post(
        url=f"{self._base_url}/api/embeddings/",
        headers=self._headers,
        json_payload={"texts": texts},
        retry_strategy=self._retry_strategy,
    )

    return response

download_protected_file(protected_url)

Download a file from a protected URL by first getting a signed URL and then downloading.

Parameters:

Name Type Description Default
protected_url str

The protected URL path (e.g., "/protected/conversations/123/audio.mp3").

 required

Returns:

Name Type Description
bytes bytes

The file content as bytes.

Raises:

Type Description
Exception

If there's an error getting the signed URL or downloading the file.
Source code in forrasdk/api/utils.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
def download_protected_file(self, protected_url: str) -> bytes:
    """
    Download a file from a protected URL by first getting a signed URL and then downloading.

    Args:
        protected_url (str): The protected URL path (e.g., "/protected/conversations/123/audio.mp3").

    Returns:
        bytes: The file content as bytes.

    Raises:
        Exception: If there's an error getting the signed URL or downloading the file.
    """

    response, status_code = RequestUtils.get(
        url=f"{self._base_url}/api{protected_url}",
        headers=self._headers,
        retry_strategy=self._retry_strategy,
    )

    signed_url_response = SignedUrlResponse.model_validate(response)

    download_response = requests.get(signed_url_response.url)

    try:
        download_response.raise_for_status()
    except requests.exceptions.HTTPError as e:
        error_message = "HTTP Error occurred while downloading file"
        if e.response is not None:
            try:
                error_details = e.response.json()
                error_message = f"Download Error: {error_details}"
            except (ValueError, AttributeError):
                error_message = f"Download Error: HTTP {e.response.status_code} - {e.response.text or 'No response text'}"
        else:
            error_message = f"Download Error: {str(e)}"
        raise Exception(error_message) from e

    return download_response.content

filter_assistant_datas(objects, query, filter_prompt, batch_size=10, model_id=None)

Filter a list of assistant data objects with content and metadata using an LLM.

This is a generic method that works with any objects that have 'content' and 'metadata' fields, including AssistantSearchDataResponse and AssistantDataResponseItem.

Parameters:

Name Type Description Default
objects List[T]

List of objects that implement ContentMetadataProtocol (have content and metadata fields)

 required
query str

The user's query string to use for filtering

 required
filter_prompt str

Context/instructions for the LLM during filtering

 required
batch_size int

Number of objects to process in each batch (default: 10)

 10
model_id Optional[str]

Optional LLM model ID to use for filtering

 None

Returns:

Type Description
List[T]

List of filtered objects (same type as input)
Example

from forratypes.assistants import AssistantSearchDataResponse search_results = scout.current_assistant.search_data("python tutorials") filtered_results = scout.utils.filter_assistant_datas( ... objects=search_results, ... query="How to handle exceptions?", ... filter_prompt="You are an expert at finding relevant Python documentation." ... )

Source code in forrasdk/api/utils.py 
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
def filter_assistant_datas(
    self,
    objects: List[T],
    query: str,
    filter_prompt: str,
    batch_size: int = 10,
    model_id: Optional[str] = None,
) -> List[T]:
    """
    Filter a list of assistant data objects with content and metadata using an LLM.

    This is a generic method that works with any objects that have 'content' and 'metadata' fields,
    including AssistantSearchDataResponse and AssistantDataResponseItem.

    Args:
        objects: List of objects that implement ContentMetadataProtocol (have content and metadata fields)
        query: The user's query string to use for filtering
        filter_prompt: Context/instructions for the LLM during filtering
        batch_size: Number of objects to process in each batch (default: 10)
        model_id: Optional LLM model ID to use for filtering

    Returns:
        List of filtered objects (same type as input)

    Example:
        >>> from forratypes.assistants import AssistantSearchDataResponse
        >>> search_results = scout.current_assistant.search_data("python tutorials")
        >>> filtered_results = scout.utils.filter_assistant_datas(
        ...     objects=search_results,
        ...     query="How to handle exceptions?",
        ...     filter_prompt="You are an expert at finding relevant Python documentation."
        ... )
    """
    if not objects:
        return []

    # Create a mapping of IDs to objects for filtering
    id_to_object = {}
    documents_dict = {}

    for idx, obj in enumerate(objects):
        # Check if object has required fields
        if not isinstance(obj, ContentMetadataProtocol):
            raise TypeError(
                f"Object at index {idx} does not have required 'content' and 'metadata' fields"
            )

        # Create unique ID for this object
        obj_id = str(idx)
        id_to_object[obj_id] = obj

        # Prepare document for LLM filtering
        # Include both content and metadata in the document text
        doc_text = obj.content
        if hasattr(obj, "metadata") and obj.metadata:
            # Add metadata as JSON string for context
            if isinstance(obj.metadata, dict):
                doc_text += f"\n\nMetadata: {json.dumps(obj.metadata)}"
            elif hasattr(obj.metadata, "model_dump"):
                # Handle Pydantic models
                doc_text += f"\n\nMetadata: {json.dumps(obj.metadata.model_dump())}"
            else:
                doc_text += f"\n\nMetadata: {str(obj.metadata)}"

        documents_dict[obj_id] = doc_text

    # Use the existing llm_filter_documents method
    filtered_ids = self.llm_filter_documents(
        query=query,
        context=filter_prompt,
        documents=documents_dict,
        batch_size=batch_size,
        model_id=model_id,
    )

    # Return the filtered objects in their original type
    return [
        id_to_object[obj_id] for obj_id in filtered_ids if obj_id in id_to_object
    ]

get_document_text(file_path, args=None)

Extract text content from a file.

Parameters:

Name Type Description Default
file_path str

The local file path of the file to extract text from.

 required
args Optional[dict]

Additional arguments for text extraction. Defaults to None.

 None

Returns:

Name Type Description
dict dict

The response from the API after extracting text from the file.
Source code in forrasdk/api/utils.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def get_document_text(self, file_path: str, args: Optional[dict] = None) -> dict:
    """
    Extract text content from a file.

    Args:
        file_path (str): The local file path of the file to extract text from.
        args (Optional[dict], optional): Additional arguments for text extraction. Defaults to None.

    Returns:
        dict: The response from the API after extracting text from the file.
    """
    with open(file_path, "rb") as f:
        files = {"file": f}
        local_headers = self._headers.copy()
        local_headers.pop("Content-Type")

        response, status_code = RequestUtils.post(
            url=f"{self._base_url}/api/utils/get-file-text-content",
            headers=local_headers,
            files=files,
            json_payload=args,
            retry_strategy=self._retry_strategy,
        )
    return response

llm_filter_documents(query, context, documents, batch_size=10, model_id=None)

Filter a set of documents using an LLM based on a query and context.

Parameters:

Name Type Description Default
query str

The query string to use for filtering documents. Ex: What is the capital of France?

 required
context str

Additional context to provide to the LLM during filtering. Ex: You are an expert in selecting content to answer geographical questions.

 required
documents dict[str, str]

A dictionary mapping document IDs to document texts. EX: {"my_id": "Capital of france is paris", "my_id_2": "Irrelevant content"}

 required
batch_size int

Number of documents to process in each batch. Defaults to 10.

 10
model_id Optional[str]

The ID of the LLM model to use for filtering. When not provided, use default model.

 None

Returns:

Name Type Description
list list

List of ids that are relevant result to answer the question
Source code in forrasdk/api/utils.py 
 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
def llm_filter_documents(
    self,
    query: str,
    context: str,
    documents: dict[str, str],
    batch_size: int = 10,
    model_id: Optional[str] = None,
) -> list:
    """
    Filter a set of documents using an LLM based on a query and context.

    Args:
        query (str): The query string to use for filtering documents. Ex: What is the capital of France?
        context (str): Additional context to provide to the LLM during filtering. Ex: You are an expert in selecting content to answer geographical questions.
        documents (dict[str, str]): A dictionary mapping document IDs to document texts. EX: {"my_id": "Capital of france is paris", "my_id_2": "Irrelevant content"}
        batch_size (int, optional): Number of documents to process in each batch. Defaults to 10.
        model_id (Optional[str], optional): The ID of the LLM model to use for filtering. When not provided, use default model.

    Returns:
        list: List of ids that are relevant result to answer the question
    """
    response, status_code = RequestUtils.post(
        url=f"{self._base_url}/api/utils/llm-filter-documents",
        headers=self._headers,
        json_payload={
            "query": query,
            "context": context,
            "documents": documents,
            "batch_size": batch_size,
            "model_id": model_id,
        },
        retry_strategy=self._retry_strategy,
    )
    return response

upload_file_to_signed_url(signed_url_response, file_path, file_key=None)

Upload a file to a signed URL provided by the Scout API.

Parameters:

Name Type Description Default
signed_url_response SignedUploadUrlResponse

The response object containing the signed URL and upload details.

 required
file_path str

The local path to the file to be uploaded.

 required
file_key str | None

The key/name to use for the file in the upload. If None, uses the filename from file_path.

 None

Returns:

Name Type Description
int int

The HTTP status code of the upload response.
Source code in forrasdk/api/utils.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def upload_file_to_signed_url(
    signed_url_response: SignedUploadUrlResponse,
    file_path: str,
    file_key: str | None = None,
) -> int:
    """
    Upload a file to a signed URL provided by the Scout API.

    Args:
        signed_url_response (SignedUploadUrlResponse): The response object containing the signed URL and upload details.
        file_path (str): The local path to the file to be uploaded.
        file_key (str | None, optional): The key/name to use for the file in the upload. If None, uses the filename from file_path.

    Returns:
        int: The HTTP status code of the upload response.
    """
    if file_key is None:
        file_key = file_path.split("/")[-1]

    return upload_file(signed_url_response, file_path, file_key)