Add docstrings and minor formatting improvements for code clarity

Author: sahilds1

server/api/services/conversions_services.py

@@ -2,6 +2,23 @@
 
 
 def convert_uuids(data):
+    """
+    Recursively convert UUID objects to strings in nested data structures.
+
+    Traverses dictionaries, lists, and other data structures to find UUID objects
+    and converts them to their string representation for serialization.
+
+    Parameters
+    ----------
+    data : any
+        The data structure to process (dict, list, UUID, or any other type)
+
+    Returns
+    -------
+    any
+        The data structure with all UUID objects converted to strings.
+        Structure and types are preserved except for UUID -> str conversion.
+    """
     if isinstance(data, dict):
         return {key: convert_uuids(value) for key, value in data.items()}
     elif isinstance(data, list):

server/api/services/embedding_services.py

@@ -1,29 +1,63 @@
 # services/embedding_services.py
+
+from pgvector.django import L2Distance
+
 from .sentencetTransformer_model import TransformerModel
+
 # Adjust import path as needed
 from ..models.model_embeddings import Embeddings
-from pgvector.django import L2Distance
 
 
-def get_closest_embeddings(user, message_data, document_name=None, guid=None, num_results=10):
+def get_closest_embeddings(
+    user, message_data, document_name=None, guid=None, num_results=10
+):
+    """
+    Find the closest embeddings to a given message for a specific user.
+
+    Parameters
+    ----------
+    user : User
+        The user whose uploaded documents will be searched
+    message_data : str
+        The input message to find similar embeddings for
+    document_name : str, optional
+        Filter results to a specific document name
+    guid : str, optional
+        Filter results to a specific document GUID (takes precedence over document_name)
+    num_results : int, default 10
+        Maximum number of results to return
+
+    Returns
+    -------
+    list[dict]
+        List of dictionaries containing embedding results with keys:
+        - name: document name
+        - text: embedded text content
+        - page_number: page number in source document
+        - chunk_number: chunk number within the document
+        - distance: L2 distance from query embedding
+        - file_id: GUID of the source file
+    """
+
     #
     transformerModel = TransformerModel.get_instance().model
     embedding_message = transformerModel.encode(message_data)
     # Start building the query based on the message's embedding
-    closest_embeddings_query = Embeddings.objects.filter(
-        upload_file__uploaded_by=user
-    ).annotate(
-        distance=L2Distance(
-            'embedding_sentence_transformers', embedding_message)
-    ).order_by('distance')
+    closest_embeddings_query = (
+        Embeddings.objects.filter(upload_file__uploaded_by=user)
+        .annotate(
+            distance=L2Distance("embedding_sentence_transformers", embedding_message)
+        )
+        .order_by("distance")
+    )
 
     # Filter by GUID if provided, otherwise filter by document name if provided
     if guid:
         closest_embeddings_query = closest_embeddings_query.filter(
-            upload_file__guid=guid)
+            upload_file__guid=guid
+        )
     elif document_name:
-        closest_embeddings_query = closest_embeddings_query.filter(
-            name=document_name)
+        closest_embeddings_query = closest_embeddings_query.filter(name=document_name)
 
     # Slice the results to limit to num_results
     closest_embeddings_query = closest_embeddings_query[:num_results]

server/api/services/tools/tools.py

@@ -1,6 +1,8 @@
-from django.db import connection
 from typing import Dict, Any, Callable, List
 from dataclasses import dataclass
+
+from django.db import connection
+
 from .database import ask_database, get_database_info
 
 database_schema_dict = get_database_info(connection)
@@ -11,13 +13,15 @@
     ]
 )
 
+
 @dataclass
 class ToolFunction:
     name: str
     func: Callable
     description: str
     parameters: Dict[str, Any]
 
+
 def create_tool_dict(tool: ToolFunction) -> Dict[str, Any]:
     return {
         "type": "function",
@@ -28,10 +32,11 @@ def create_tool_dict(tool: ToolFunction) -> Dict[str, Any]:
                 "type": "object",
                 "properties": tool.parameters,
                 "required": list(tool.parameters.keys()),
-            }
-        }
+            },
+        },
     }
 
+
 TOOL_FUNCTIONS = [
     ToolFunction(
         name="ask_database",
@@ -56,60 +61,58 @@ def create_tool_dict(tool: ToolFunction) -> Dict[str, Any]:
                 SQL should be written using this database schema:
                 {database_schema_string}
                 The query should be returned in plain text, not in JSON.
-                """
+                """,
             }
-        }
+        },
     ),
 ]
 
 # Automatically generate the tool_functions dictionary and tools list
-tool_functions: Dict[str, Callable] = {
-    tool.name: tool.func for tool in TOOL_FUNCTIONS
-}
+tool_functions: Dict[str, Callable] = {tool.name: tool.func for tool in TOOL_FUNCTIONS}
+
+tools: List[Dict[str, Any]] = [create_tool_dict(tool) for tool in TOOL_FUNCTIONS]
 
-tools: List[Dict[str, Any]] = [
-    create_tool_dict(tool) for tool in TOOL_FUNCTIONS
-]
 
 def validate_tool_inputs(tool_function_name, tool_arguments):
     """Validate the inputs for the execute_tool function."""
     if not isinstance(tool_function_name, str) or not tool_function_name:
         raise ValueError("Invalid tool function name")
-    
+
     if not isinstance(tool_arguments, dict):
         raise ValueError("Tool arguments must be a dictionary")
-    
+
     # Check if the tool_function_name exists in the tools
     tool = next((t for t in tools if t["function"]["name"] == tool_function_name), None)
     if not tool:
         raise ValueError(f"Tool function '{tool_function_name}' does not exist")
-    
+
     # Validate the tool arguments based on the tool's parameters
     parameters = tool["function"].get("parameters", {})
     required_params = parameters.get("required", [])
     for param in required_params:
         if param not in tool_arguments:
             raise ValueError(f"Missing required parameter: {param}")
-    
+
     # Check if the parameter types match the expected types
     properties = parameters.get("properties", {})
     for param, prop in properties.items():
-        expected_type = prop.get('type')
+        expected_type = prop.get("type")
         if param in tool_arguments:
-            if expected_type == 'string' and not isinstance(tool_arguments[param], str):
+            if expected_type == "string" and not isinstance(tool_arguments[param], str):
                 raise ValueError(f"Parameter '{param}' must be of type string")
-            
+
+
 def execute_tool(function_name: str, arguments: Dict[str, Any]) -> str:
     """
     Execute the appropriate function based on the function name.
-    
+
     :param function_name: The name of the function to execute
     :param arguments: A dictionary of arguments to pass to the function
     :return: The result of the function execution
     """
     # Validate tool inputs
     validate_tool_inputs(function_name, arguments)
-    
+
     try:
         return tool_functions[function_name](**arguments)
     except Exception as e:

server/api/views/conversations/urls.py

@@ -1,13 +1,12 @@
 from django.urls import path, include
-from api.views.conversations import views
 from rest_framework.routers import DefaultRouter
-# from views import ConversationViewSet
+
+from api.views.conversations import views
 
 router = DefaultRouter()
-router.register(r'conversations', views.ConversationViewSet,
-                basename='conversation')
+router.register(r"conversations", views.ConversationViewSet, basename="conversation")
 
 urlpatterns = [
     path("chatgpt/extract_text/", views.extract_text, name="post_web_text"),
-    path("chatgpt/", include(router.urls))
+    path("chatgpt/", include(router.urls)),
 ]