From 1fe0c19953df4e0a8bc15de3dc1df1da979da149 Mon Sep 17 00:00:00 2001
From: NolanTrem <34580718+NolanTrem@users.noreply.github.com>
Date: Fri, 6 Dec 2024 10:40:34 -0800
Subject: [PATCH 1/7] Fix agent responses for conversation

---
 py/core/main/api/templates/log_viewer.html | 30 ++++++------
 py/core/main/services/retrieval_service.py | 53 ++++++++++------------
 py/core/providers/logger/r2r_logger.py     | 14 ++++--
 3 files changed, 50 insertions(+), 47 deletions(-)
diff --git a/py/core/main/api/templates/log_viewer.html b/py/core/main/api/templates/log_viewer.html
index a3f9a6d5b..55ce6f68f 100644
--- a/py/core/main/api/templates/log_viewer.html
+++ b/py/core/main/api/templates/log_viewer.html
@@ -3,13 +3,13 @@
     <head>
         <title>R2R Log Viewer</title>
         <style>
-            body { 
-                margin: 20px; 
+            body {
+                margin: 20px;
                 font-family: monospace;
                 background: #f8f9fa;
             }
-            #logs { 
-                white-space: pre-wrap; 
+            #logs {
+                white-space: pre-wrap;
                 background: white;
                 padding: 20px;
                 border-radius: 4px;
@@ -31,38 +31,38 @@
     <body>
         <h2>R2R Log Viewer</h2>
         <div id="logs"><span class="status">Connecting to log stream...</span></div>
-        
+
         <!-- Include ansi_up via a CDN -->
         <script src="https://cdn.jsdelivr.net/npm/ansi_up@5.0.0/ansi_up.min.js"></script>
         <script>
             let ws = null;
             let ansi_up = new AnsiUp();
-            
+
             function connect() {
                 if (ws) {
                     ws.close();
                 }
-                
+
                 ws = new WebSocket(`ws://${window.location.host}/v3/logs/stream`);
-                
+
                 ws.onmessage = function(event) {
                     const logsDiv = document.getElementById("logs");
                     const newEntry = document.createElement('div');
                     newEntry.className = 'log-entry';
-                    
+
                     // Convert ANSI to HTML
                     const htmlContent = ansi_up.ansi_to_html(event.data);
                     newEntry.innerHTML = htmlContent;
                     logsDiv.appendChild(newEntry);
-                    
+
                     // Keep only the last 1000 entries
                     while (logsDiv.children.length > 1000) {
                         logsDiv.removeChild(logsDiv.firstChild);
                     }
-                    
+
                     logsDiv.scrollTop = logsDiv.scrollHeight;
                 };
-                
+
                 ws.onclose = function() {
                     const logsDiv = document.getElementById("logs");
                     const msg = document.createElement('div');
@@ -71,14 +71,14 @@ <h2>R2R Log Viewer</h2>
                     logsDiv.appendChild(msg);
                     setTimeout(connect, 1000);
                 };
-                
+
                 ws.onerror = function(err) {
                     console.error('WebSocket error:', err);
                 };
             }
-            
+
             connect();
-            
+
             window.onbeforeunload = function() {
                 if (ws) {
                     ws.close();
diff --git a/py/core/main/services/retrieval_service.py b/py/core/main/services/retrieval_service.py
index 3f9ef458d..4cd0afaad 100644
--- a/py/core/main/services/retrieval_service.py
+++ b/py/core/main/services/retrieval_service.py
@@ -302,32 +302,31 @@ async def agent(
                 ids = []
 
                 if conversation_id:
-                    # Fetch existing conversation
-                    conversation = (
-                        await self.logging_connection.get_conversation(
-                            conversation_id, branch_id
-                        )
-                    )
-                    if not conversation:
-                        logger.error(
-                            f"No conversation found for ID: {conversation_id}"
-                        )
-                        raise R2RException(
-                            status_code=404,
-                            message=f"Conversation not found: {conversation_id}",
+                    try:
+                        # Fetch existing conversation
+                        conversation = (
+                            await self.logging_connection.get_conversation(
+                                conversation_id=conversation_id,
+                                branch_id=branch_id,
+                            )
                         )
+                    except Exception as e:
+                        logger.error(f"Error logging conversation: {str(e)}")
                     # Assuming 'conversation' is a list of dicts with 'id' and 'message' keys
-                    messages_from_conversation = []
-                    for resp in conversation:
-                        if isinstance(resp, dict):
-                            msg = Message.from_dict(resp["message"])
-                            messages_from_conversation.append(msg)
-                            ids.append(resp["id"])
-                        else:
-                            logger.error(
-                                f"Unexpected type in conversation: {type(resp)}"
-                            )
-                    messages = messages_from_conversation + messages
+
+                    if conversation is not None:
+                        print("Gets into messages_from_conversation")
+                        messages_from_conversation: list[Message] = []
+                        for resp in conversation:
+                            if isinstance(resp, dict):
+                                msg = Message.from_dict(resp["message"])
+                                messages_from_conversation.append(msg)
+                                ids.append(resp["id"])
+                            else:
+                                logger.warning(
+                                    f"Unexpected type in conversation: {type(resp)}\n{resp}"
+                                )
+                        messages = messages_from_conversation + messages
                 else:
                     # Create new conversation
                     conversation_id = (
@@ -399,8 +398,6 @@ async def stream_response():
                     search_settings=search_settings,
                     rag_generation_config=rag_generation_config,
                     include_title_if_available=include_title_if_available,
-                    *args,
-                    **kwargs,
                 )
 
                 # Save the assistant's reply to the conversation
@@ -428,7 +425,7 @@ async def stream_response():
                     value=latency,
                 )
                 return {
-                    "messages": [msg.to_dict() for msg in results],
+                    "messages": results,
                     "conversation_id": str(
                         conversation_id
                     ),  # Ensure it's a string
@@ -443,7 +440,7 @@ async def stream_response():
                     )
                 raise HTTPException(
                     status_code=500,
-                    detail="Internal Server Error",
+                    detail=f"Internal Server Error - {str(e)}",
                 )
 
 
diff --git a/py/core/providers/logger/r2r_logger.py b/py/core/providers/logger/r2r_logger.py
index e46bac4ae..305eebb34 100644
--- a/py/core/providers/logger/r2r_logger.py
+++ b/py/core/providers/logger/r2r_logger.py
@@ -679,6 +679,7 @@ async def get_conversation(
                 )
             conversation_created_at = row[0]
 
+        print(f"Getting a branch_id: {branch_id}")
         if branch_id is None:
             # Get the most recent branch by created_at timestamp
             async with self.conn.execute(
@@ -691,14 +692,19 @@ async def get_conversation(
                 (conversation_id,),
             ) as cursor:
                 row = await cursor.fetchone()
+                print(f"Row: {row}")
                 branch_id = row[0] if row else None
 
         # If no branch exists, return empty results but with required fields
         if branch_id is None:
-            return {
-                "id": conversation_id,
-                "created_at": conversation_created_at,
-            }
+            logger.warning(
+                f"No branches found for conversation ID {conversation_id}"
+            )
+            return None
+            # return {
+            #     "id": conversation_id,
+            #     "created_at": conversation_created_at,
+            # }
 
         # Get all messages for this branch
         async with self.conn.execute(

From e85378e47af24143fda2e000612de803d1d7805d Mon Sep 17 00:00:00 2001
From: emrgnt-cmplxty <owen@algofi.org>
Date: Fri, 6 Dec 2024 10:48:55 -0800
Subject: [PATCH 2/7] fix retrieval

---
 py/cli/commands/retrieval.py     | 1 +
 py/pyproject.toml                | 2 +-
 py/shared/abstractions/search.py | 4 ++--
 3 files changed, 4 insertions(+), 3 deletions(-)

diff --git a/py/cli/commands/retrieval.py b/py/cli/commands/retrieval.py
index b472cd3d6..f3066c985 100644
--- a/py/cli/commands/retrieval.py
+++ b/py/cli/commands/retrieval.py
@@ -81,6 +81,7 @@ async def search(ctx, query, **kwargs):
     with timer():
         results = await client.retrieval.search(
             query,
+            "custom",
             search_settings,
         )
 
diff --git a/py/pyproject.toml b/py/pyproject.toml
index c352dd43c..6fcdf4d7c 100644
--- a/py/pyproject.toml
+++ b/py/pyproject.toml
@@ -5,7 +5,7 @@ build-backend = "poetry.core.masonry.api"
 [tool.poetry]
 name = "r2r"
 readme = "README.md"
-version = "3.3.4"
+version = "3.3.5"
 
 description = "SciPhi R2R"
 authors = ["Owen Colegrove <owen@sciphi.ai>"]
diff --git a/py/shared/abstractions/search.py b/py/shared/abstractions/search.py
index 74fbac3ea..e7789054d 100644
--- a/py/shared/abstractions/search.py
+++ b/py/shared/abstractions/search.py
@@ -217,10 +217,10 @@ def as_dict(self) -> dict:
             ),
             "graph_search_results": [
                 result.to_dict() for result in self.graph_search_results
-            ],
+            ] if self.graph_search_results else [],
             "web_search_results": [
                 result.to_dict() for result in self.web_search_results
-            ],
+            ] if self.web_search_results else [],
         }
 
 

From c4d5cf97c8cdaf927fe7a7f3fab413626c934a5b Mon Sep 17 00:00:00 2001
From: emrgnt-cmplxty <owen@algofi.org>
Date: Fri, 6 Dec 2024 15:07:36 -0800
Subject: [PATCH 3/7] fixup user_id xfer

---
 py/core/main/api/ingestion_router.py        | 12 ++--
 py/core/main/api/v3/chunks_router.py        |  6 +-
 py/core/main/api/v3/documents_router.py     | 79 ++++++++++++++++++---
 py/core/main/api/v3/retrieval_router.py     |  3 +-
 py/core/main/services/management_service.py |  6 +-
 py/core/main/services/retrieval_service.py  |  2 +-
 py/core/providers/database/graph.py         | 16 ++++-
 py/sdk/v3/documents.py                      |  7 +-
 py/sdk/v3/retrieval.py                      |  2 +-
 9 files changed, 104 insertions(+), 29 deletions(-)

diff --git a/py/core/main/api/ingestion_router.py b/py/core/main/api/ingestion_router.py
index 1fae038ac..425c6c100 100644
--- a/py/core/main/api/ingestion_router.py
+++ b/py/core/main/api/ingestion_router.py
@@ -166,16 +166,16 @@ async def ingest_files_app(
             # Check if the user is a superuser
             if not auth_user.is_superuser:
                 for metadata in metadatas or []:
-                    if "user_id" in metadata and (
+                    if "owner_id" in metadata and (
                         not auth_user.is_superuser
-                        and metadata["user_id"] != str(auth_user.id)
+                        and metadata["owner_id"] != str(auth_user.id)
                     ):
                         raise R2RException(
                             status_code=403,
                             message="Non-superusers cannot set user_id in metadata.",
                         )
                     # If user is not a superuser, set user_id in metadata
-                    metadata["user_id"] = str(auth_user.id)
+                    metadata["owner_id"] = str(auth_user.id)
 
             file_datas = await self._process_files(files)
 
@@ -293,14 +293,14 @@ async def update_files_app(
             """
             if not auth_user.is_superuser:
                 for metadata in metadatas or []:
-                    if "user_id" in metadata and metadata["user_id"] != str(
+                    if "owner_id" in metadata and metadata["owner_id"] != str(
                         auth_user.id
                     ):
                         raise R2RException(
                             status_code=403,
-                            message="Non-superusers cannot set user_id in metadata.",
+                            message="Non-superusers cannot set owner_id in metadata.",
                         )
-                    metadata["user_id"] = str(auth_user.id)
+                    metadata["owner_id"] = str(auth_user.id)
 
             file_datas = await self._process_files(files)
 
diff --git a/py/core/main/api/v3/chunks_router.py b/py/core/main/api/v3/chunks_router.py
index 6031e3c6c..82242b13c 100644
--- a/py/core/main/api/v3/chunks_router.py
+++ b/py/core/main/api/v3/chunks_router.py
@@ -73,7 +73,7 @@ def _select_filters(
             # for non-superusers, we filter by user_id and selected & allowed collections
             collection_filters = {
                 "$or": [
-                    {"user_id": {"$eq": auth_user.id}},
+                    {"owner_id": {"$eq": auth_user.id}},
                     {
                         "collection_ids": {
                             "$overlap": list(allowed_collections)
@@ -201,7 +201,7 @@ async def retrieve_chunk(
             # document = await self.services["management"].get_document(chunk.document_id)
             # TODO - Add collection ID check
             if not auth_user.is_superuser and str(auth_user.id) != str(
-                chunk["user_id"]
+                chunk["owner_id"]
             ):
                 raise R2RException("Not authorized to access this chunk", 403)
 
@@ -493,7 +493,7 @@ async def list_chunks(
 
             # Add user access control filter
             if not auth_user.is_superuser:
-                filters["user_id"] = {"$eq": str(auth_user.id)}
+                filters["owner_id"] = {"$eq": str(auth_user.id)}
 
             # Add metadata filters if provided
             if metadata_filter:
diff --git a/py/core/main/api/v3/documents_router.py b/py/core/main/api/v3/documents_router.py
index 8fb23513e..1135243c7 100644
--- a/py/core/main/api/v3/documents_router.py
+++ b/py/core/main/api/v3/documents_router.py
@@ -4,7 +4,7 @@
 import mimetypes
 import textwrap
 from io import BytesIO
-from typing import Optional
+from typing import Optional, Any
 from uuid import UUID
 
 from fastapi import Body, Depends, File, Form, Path, Query, UploadFile
@@ -18,6 +18,7 @@
     SearchSettings,
     RunType,
     UnprocessedChunk,
+    SearchMode,
     Workflow,
     generate_document_id,
     generate_id,
@@ -47,6 +48,22 @@
 MAX_CHUNKS_PER_REQUEST = 1024 * 100
 
 
+def merge_search_settings(
+    base: SearchSettings, overrides: SearchSettings
+) -> SearchSettings:
+    # Convert both to dict
+    base_dict = base.model_dump()
+    overrides_dict = overrides.model_dump(exclude_unset=True)
+
+    # Update base_dict with values from overrides_dict
+    # This ensures that any field set in overrides takes precedence
+    for k, v in overrides_dict.items():
+        base_dict[k] = v
+
+    # Construct a new SearchSettings from the merged dict
+    return SearchSettings(**base_dict)
+
+
 def merge_ingestion_config(
     base: IngestionConfig, overrides: IngestionConfig
 ) -> IngestionConfig:
@@ -72,6 +89,37 @@ def __init__(
         super().__init__(providers, services, orchestration_provider, run_type)
         self._register_workflows()
 
+
+    def _prepare_search_settings(
+        self,
+        auth_user: Any,
+        search_mode: SearchMode,
+        search_settings: Optional[SearchSettings],
+    ) -> SearchSettings:
+        """
+        Prepare the effective search settings based on the provided search_mode,
+        optional user-overrides in search_settings, and applied filters.
+        """
+
+        if search_mode != SearchMode.custom:
+            # Start from mode defaults
+            effective_settings = SearchSettings.get_default(search_mode.value)
+            if search_settings:
+                # Merge user-provided overrides
+                effective_settings = merge_search_settings(
+                    effective_settings, search_settings
+                )
+        else:
+            # Custom mode: use provided settings or defaults
+            effective_settings = search_settings or SearchSettings()
+
+        # Apply user-specific filters
+        effective_settings.filters = self._select_filters(
+            auth_user, effective_settings
+        )
+
+        return effective_settings
+
     # TODO - Remove this legacy method
     def _register_workflows(self):
         self.orchestration_provider.register_workflows(
@@ -294,16 +342,16 @@ async def create_document(
             # Check if the user is a superuser
             metadata = metadata or {}
             if not auth_user.is_superuser:
-                if "user_id" in metadata and (
+                if "owner_id" in metadata and (
                     not auth_user.is_superuser
-                    and metadata["user_id"] != str(auth_user.id)
+                    and metadata["owner_id"] != str(auth_user.id)
                 ):
                     raise R2RException(
                         status_code=403,
                         message="Non-superusers cannot set user_id in metadata.",
                     )
                 # If user is not a superuser, set user_id in metadata
-                metadata["user_id"] = str(auth_user.id)
+                metadata["owner_id"] = str(auth_user.id)
 
             if chunks:
                 if len(chunks) == 0:
@@ -987,7 +1035,7 @@ async def delete_document_by_filter(
             """
 
             filters_dict = {
-                "$and": [{"user_id": {"$eq": str(auth_user.id)}}, filters]
+                "$and": [{"owner_id": {"$eq": str(auth_user.id)}}, filters]
             }
             await self.services["management"].delete(filters=filters_dict)
 
@@ -1531,14 +1579,26 @@ async def search_documents(
                 ..., 
                 description="The search query to perform.",
             ),
-            settings: SearchSettings = Body(
+            search_mode: SearchMode = Body(
+                default=SearchMode.custom,
+                description=(
+                    "Default value of `custom` allows full control over search settings.\n\n"
+                    "Pre-configured search modes:\n"
+                    "`basic`: A simple semantic-based search.\n"
+                    "`advanced`: A more powerful hybrid search combining semantic and full-text.\n"
+                    "`custom`: Full control via `search_settings`.\n\n"
+                    "If `filters` or `limit` are provided alongside `basic` or `advanced`, "
+                    "they will override the default settings for that mode."
+                ),
+            ),
+            search_settings: SearchSettings = Body(
                 default_factory=SearchSettings,
                 description="Settings for document search",
             ),
             auth_user=Depends(self.providers.auth.auth_wrapper),
         ): # -> WrappedDocumentSearchResponse:  # type: ignore
             """
-            Perform a search query on the vector database and knowledge graph.
+            Perform a search query on the automatically generated document summaries in the system.
 
             This endpoint allows for complex filtering of search results using PostgreSQL-based queries.
             Filters can be applied to various fields such as document_id, and internal metadata values.
@@ -1546,9 +1606,12 @@ async def search_documents(
 
             Allowed operators include `eq`, `neq`, `gt`, `gte`, `lt`, `lte`, `like`, `ilike`, `in`, and `nin`.
             """
+            effective_settings = self._prepare_search_settings(
+                auth_user, search_mode, search_settings
+            )
 
             query_embedding = (
-                await self.service.providers.embedding.async_get_embedding(
+                await self.providers.embedding.async_get_embedding(
                     query
                 )
             )
diff --git a/py/core/main/api/v3/retrieval_router.py b/py/core/main/api/v3/retrieval_router.py
index b35829116..1685961b9 100644
--- a/py/core/main/api/v3/retrieval_router.py
+++ b/py/core/main/api/v3/retrieval_router.py
@@ -114,7 +114,7 @@ def _select_filters(
             # for non-superusers, we filter by user_id and selected & allowed collections
             collection_filters = {
                 "$or": [
-                    {"user_id": {"$eq": auth_user.id}},
+                    {"owner_id": {"$eq": auth_user.id}},
                     {
                         "collection_ids": {
                             "$overlap": list(allowed_collections)
@@ -854,6 +854,7 @@ async def completion(
             The messages list should contain alternating user and assistant messages, with an optional
             system message at the start. Each message should have a 'role' and 'content'.
             """
+            print('messages = ', messages)
 
             return await self.services["retrieval"].completion(
                 messages=messages,
diff --git a/py/core/main/services/management_service.py b/py/core/main/services/management_service.py
index 57fcaae01..f21c37deb 100644
--- a/py/core/main/services/management_service.py
+++ b/py/core/main/services/management_service.py
@@ -257,7 +257,7 @@ def validate_filters(filters: dict[str, Any]) -> None:
                         message=f"Invalid filter field: {field}",
                     )
 
-            for field in ["document_id", "user_id", "chunk_id"]:
+            for field in ["document_id", "owner_id", "chunk_id"]:
                 if field in filters:
                     op = next(iter(filters[field].keys()))
                     try:
@@ -307,10 +307,10 @@ def process_filter(filter_dict: dict[str, Any]):
                     relational_filters.setdefault(
                         "filter_document_ids", []
                     ).append(filter_dict["document_id"]["$eq"])
-                if "user_id" in filter_dict:
+                if "owner_id" in filter_dict:
                     relational_filters.setdefault(
                         "filter_user_ids", []
-                    ).append(filter_dict["user_id"]["$eq"])
+                    ).append(filter_dict["owner_id"]["$eq"])
                 if "collection_ids" in filter_dict:
                     relational_filters.setdefault(
                         "filter_collection_ids", []
diff --git a/py/core/main/services/retrieval_service.py b/py/core/main/services/retrieval_service.py
index 318e96e09..b45a0e750 100644
--- a/py/core/main/services/retrieval_service.py
+++ b/py/core/main/services/retrieval_service.py
@@ -120,7 +120,7 @@ async def search_documents(
         settings: SearchSettings,
         query_embedding: Optional[list[float]] = None,
     ) -> list[DocumentResponse]:
-
+        print('settings = ', settings)
         return await self.providers.database.search_documents(
             query_text=query,
             settings=settings,
diff --git a/py/core/providers/database/graph.py b/py/core/providers/database/graph.py
index 323364aaf..56e640483 100644
--- a/py/core/providers/database/graph.py
+++ b/py/core/providers/database/graph.py
@@ -2267,14 +2267,24 @@ def parse_condition(key: str, value: Any) -> str:  # type: ignore
                 logger.warning(
                     "Filtering by `document_id` is not supported with graph search, ignoring."
                 )
+                return ""  # Return an empty condition instead of None
+
             elif key == "chunk_id":
                 logger.warning(
                     "Filtering by `chunk_id` is not supported with graph search, ignoring."
                 )
+                return ""  # Return an empty condition instead of None
+
             elif key == "user_id":
                 logger.warning(
                     "Filtering by `user_id` is not supported with graph search, ignoring. Use `collection_ids` instead."
                 )
+                return ""  # Return an empty condition instead of None
+            elif key == "owner_id":
+                logger.warning(
+                    "Filtering by `owner_id` is not supported with graph search, ignoring. Use `collection_ids` instead."
+                )
+                return ""  # Return an empty condition instead of None
 
             else:
                 # Handle JSON-based filters
@@ -2395,8 +2405,9 @@ async def graph_search(
         where_clause = ""
         params: list[Union[str, int, bytes]] = [str(query_embedding), limit]
         if filters:
-            where_clause = self._build_filters(filters, params)
-            where_clause = f"WHERE {where_clause}"
+            conditions_list = self._build_filters(filters, params)
+            if conditions_list:
+                where_clause = "WHERE " + " AND ".join(conditions_list)
 
         # Modified query to include similarity score while keeping same structure
         QUERY = f"""
@@ -2407,6 +2418,7 @@ async def graph_search(
             ORDER BY {embedding_type} <=> $1
             LIMIT $2;
         """
+        print('QUERY = ', QUERY)
         results = await self.connection_manager.fetch_query(
             QUERY, tuple(params)
         )
diff --git a/py/sdk/v3/documents.py b/py/sdk/v3/documents.py
index c46d3da44..30926bbde 100644
--- a/py/sdk/v3/documents.py
+++ b/py/sdk/v3/documents.py
@@ -411,7 +411,7 @@ async def list(
     async def search(
         self,
         query: str,
-        # search_mode: Optional[str | SearchMode] = "custom",
+        search_mode: Optional[str | SearchMode] = "custom",
         search_settings: Optional[dict | SearchSettings] = None,
     ):
         """
@@ -429,13 +429,12 @@ async def search(
 
         if search_settings and not isinstance(search_settings, dict):
             search_settings = search_settings.model_dump()
-
         data = {
             "query": query,
             "search_settings": search_settings,
         }
-        # if search_mode:
-        # data["search_mode"] = search_mode
+        if search_mode:
+            data["search_mode"] = search_mode
 
         return await self.client._make_request(
             "POST",
diff --git a/py/sdk/v3/retrieval.py b/py/sdk/v3/retrieval.py
index bead7a729..8c522f342 100644
--- a/py/sdk/v3/retrieval.py
+++ b/py/sdk/v3/retrieval.py
@@ -72,7 +72,7 @@ async def completion(
             "messages": [msg.model_dump() for msg in cast_messages],
             "generation_config": generation_config,
         }
-
+        print('data = ', data)
         return await self.client._make_request(
             "POST",
             "retrieval/completion",

From c81c8a7261c02628059c2baedddea7b220665eb2 Mon Sep 17 00:00:00 2001
From: emrgnt-cmplxty <owen@algofi.org>
Date: Fri, 6 Dec 2024 15:43:10 -0800
Subject: [PATCH 4/7] adding back in document search

---
 py/core/__init__.py                     |   1 +
 py/core/base/__init__.py                |   1 +
 py/core/base/abstractions/__init__.py   |   2 +
 py/core/main/api/ingestion_router.py    | 735 ------------------------
 py/core/main/api/v3/chunks_router.py    |  78 +--
 py/core/main/api/v3/documents_router.py |  16 +-
 py/core/main/api/v3/retrieval_router.py |  42 +-
 py/sdk/models.py                        |   2 +
 py/sdk/v3/documents.py                  |   2 +-
 py/shared/abstractions/__init__.py      |   2 +
 py/shared/abstractions/search.py        |  41 ++
 py/shared/abstractions/vector.py        |   2 +-
 12 files changed, 58 insertions(+), 866 deletions(-)
 delete mode 100644 py/core/main/api/ingestion_router.py

diff --git a/py/core/__init__.py b/py/core/__init__.py
index 18edeb3da..bb1a879f9 100644
--- a/py/core/__init__.py
+++ b/py/core/__init__.py
@@ -79,6 +79,7 @@
     "GraphSearchSettings",
     "ChunkSearchResult",
     "SearchSettings",
+    "select_search_filters",
     "SearchMode",
     "HybridSearchSettings",
     # User abstractions
diff --git a/py/core/base/__init__.py b/py/core/base/__init__.py
index e9973bda4..b5594f4ab 100644
--- a/py/core/base/__init__.py
+++ b/py/core/base/__init__.py
@@ -52,6 +52,7 @@
     "ChunkSearchSettings",
     "ChunkSearchResult",
     "SearchSettings",
+    "select_search_filters",
     "SearchMode",
     "HybridSearchSettings",
     # User abstractions
diff --git a/py/core/base/abstractions/__init__.py b/py/core/base/abstractions/__init__.py
index d71c267db..d8d476013 100644
--- a/py/core/base/abstractions/__init__.py
+++ b/py/core/base/abstractions/__init__.py
@@ -64,6 +64,7 @@
     KGSearchResultType,
     SearchMode,
     SearchSettings,
+    select_search_filters,
     WebSearchResponse,
 )
 from shared.abstractions.user import Token, TokenData, User
@@ -134,6 +135,7 @@
     "ChunkSearchSettings",
     "ChunkSearchResult",
     "SearchSettings",
+    "select_search_filters",
     "SearchMode",
     "HybridSearchSettings",
     # KG abstractions
diff --git a/py/core/main/api/ingestion_router.py b/py/core/main/api/ingestion_router.py
deleted file mode 100644
index 425c6c100..000000000
--- a/py/core/main/api/ingestion_router.py
+++ /dev/null
@@ -1,735 +0,0 @@
-import base64
-import logging
-from io import BytesIO
-from pathlib import Path as pathlib_Path
-from typing import Optional, Union
-from uuid import UUID
-
-import yaml
-from fastapi import (
-    Body,
-    Depends,
-    File,
-    Form,
-    HTTPException,
-    Path,
-    Query,
-    UploadFile,
-)
-from pydantic import Json
-
-from core.base import R2RException, RawChunk, Workflow, generate_document_id
-from core.base.abstractions import (
-    IndexArgsHNSW,
-    IndexArgsIVFFlat,
-    IndexMeasure,
-    IndexMethod,
-    VectorTableName,
-)
-from core.base.api.models import (
-    WrappedCreateVectorIndexResponse,
-    WrappedDeleteVectorIndexResponse,
-    WrappedIngestionResponse,
-    WrappedListVectorIndicesResponse,
-    WrappedMetadataUpdateResponse,
-    WrappedUpdateResponse,
-)
-from core.providers import (
-    HatchetOrchestrationProvider,
-    SimpleOrchestrationProvider,
-)
-
-from ...base.logger.base import RunType
-from ..services.ingestion_service import IngestionService
-from .base_router import BaseRouter
-
-logger = logging.getLogger()
-
-
-class IngestionRouter(BaseRouter):
-    def __init__(
-        self,
-        service: IngestionService,
-        orchestration_provider: Union[
-            HatchetOrchestrationProvider, SimpleOrchestrationProvider
-        ],
-        run_type: RunType = RunType.INGESTION,
-    ):
-        super().__init__(service, orchestration_provider, run_type)
-        self.service: IngestionService = service
-
-    def _register_workflows(self):
-        self.orchestration_provider.register_workflows(
-            Workflow.INGESTION,
-            self.service,
-            {
-                "ingest-files": (
-                    "Ingest files task queued successfully."
-                    if self.orchestration_provider.config.provider != "simple"
-                    else "Ingestion task completed successfully."
-                ),
-                "ingest-chunks": (
-                    "Ingest chunks task queued successfully."
-                    if self.orchestration_provider.config.provider != "simple"
-                    else "Ingestion task completed successfully."
-                ),
-                "update-files": (
-                    "Update file task queued successfully."
-                    if self.orchestration_provider.config.provider != "simple"
-                    else "Update task queued successfully."
-                ),
-                "update-chunk": (
-                    "Update chunk task queued successfully."
-                    if self.orchestration_provider.config.provider != "simple"
-                    else "Chunk update completed successfully."
-                ),
-                "update-document-metadata": (
-                    "Update document metadata task queued successfully."
-                    if self.orchestration_provider.config.provider != "simple"
-                    else "Document metadata update completed successfully."
-                ),
-                "create-vector-index": (
-                    "Vector index creation task queued successfully."
-                    if self.orchestration_provider.config.provider != "simple"
-                    else "Vector index creation task completed successfully."
-                ),
-                "delete-vector-index": (
-                    "Vector index deletion task queued successfully."
-                    if self.orchestration_provider.config.provider != "simple"
-                    else "Vector index deletion task completed successfully."
-                ),
-                "select-vector-index": (
-                    "Vector index selection task queued successfully."
-                    if self.orchestration_provider.config.provider != "simple"
-                    else "Vector index selection task completed successfully."
-                ),
-            },
-        )
-
-    def _load_openapi_extras(self):
-        yaml_path = (
-            pathlib_Path(__file__).parent
-            / "data"
-            / "ingestion_router_openapi.yml"
-        )
-        with open(yaml_path, "r") as yaml_file:
-            yaml_content = yaml.safe_load(yaml_file)
-        return yaml_content
-
-    def _setup_routes(self):
-        # Note, we use the following verbose input parameters because FastAPI struggles to handle `File` input and `Body` inputs
-        # at the same time. Therefore, we must ues `Form` inputs for the metadata, document_ids
-        ingest_files_extras = self.openapi_extras.get("ingest_files", {})
-        ingest_files_descriptions = ingest_files_extras.get(
-            "input_descriptions", {}
-        )
-
-        @self.router.post(
-            "/ingest_files",
-            openapi_extra=ingest_files_extras.get("openapi_extra"),
-        )
-        @self.base_endpoint
-        async def ingest_files_app(
-            files: list[UploadFile] = File(
-                ..., description=ingest_files_descriptions.get("files")
-            ),
-            document_ids: Optional[Json[list[UUID]]] = Form(
-                None,
-                description=ingest_files_descriptions.get("document_ids"),
-            ),
-            collection_ids: Optional[Json[list[list[UUID]]]] = Form(
-                None,
-                description="Optional collection IDs for the documents, if provided the document will be assigned to them at ingestion.",
-            ),
-            metadatas: Optional[Json[list[dict]]] = Form(
-                None, description=ingest_files_descriptions.get("metadatas")
-            ),
-            ingestion_config: Optional[Json[dict]] = Form(
-                None,
-                description=ingest_files_descriptions.get("ingestion_config"),
-            ),
-            run_with_orchestration: Optional[bool] = Form(
-                True,
-                description=ingest_files_descriptions.get(
-                    "run_with_orchestration"
-                ),
-            ),
-            auth_user=Depends(self.service.providers.auth.auth_wrapper),
-        ) -> WrappedIngestionResponse:  # type: ignore
-            """
-            Ingests files into R2R, resulting in stored `Document` objects. Each document has corresponding `Chunk` objects which are used in vector indexing and search.
-
-            This endpoint supports multipart/form-data requests.
-
-            A valid user authentication token is required to access this endpoint, as regular users can only ingest files for their own access.
-            """
-            # Check if the user is a superuser
-            if not auth_user.is_superuser:
-                for metadata in metadatas or []:
-                    if "owner_id" in metadata and (
-                        not auth_user.is_superuser
-                        and metadata["owner_id"] != str(auth_user.id)
-                    ):
-                        raise R2RException(
-                            status_code=403,
-                            message="Non-superusers cannot set user_id in metadata.",
-                        )
-                    # If user is not a superuser, set user_id in metadata
-                    metadata["owner_id"] = str(auth_user.id)
-
-            file_datas = await self._process_files(files)
-
-            messages: list[dict[str, Union[str, None]]] = []
-            for it, file_data in enumerate(file_datas):
-                content_length = len(file_data["content"])
-                file_content = BytesIO(base64.b64decode(file_data["content"]))
-
-                file_data.pop("content", None)
-                document_id = (
-                    document_ids[it]
-                    if document_ids
-                    else generate_document_id(
-                        file_data["filename"], auth_user.id
-                    )
-                )
-
-                workflow_input = {
-                    "file_data": file_data,
-                    "document_id": str(document_id),
-                    "metadata": metadatas[it] if metadatas else None,
-                    "ingestion_config": ingestion_config,
-                    "user": auth_user.model_dump_json(),
-                    "size_in_bytes": content_length,
-                    "collection_ids": (
-                        [str(cid) for cid in collection_ids[it]]
-                        if collection_ids
-                        else None
-                    ),
-                    "is_update": False,
-                }
-
-                file_name = file_data["filename"]
-                await self.service.providers.database.store_file(
-                    document_id,
-                    file_name,
-                    file_content,
-                    file_data["content_type"],
-                )
-                if run_with_orchestration:
-                    raw_message: dict[str, Union[str, None]] = await self.orchestration_provider.run_workflow(  # type: ignore
-                        "ingest-files",
-                        {"request": workflow_input},
-                        options={
-                            "additional_metadata": {
-                                "document_id": str(document_id),
-                            }
-                        },
-                    )
-                    raw_message["document_id"] = str(document_id)
-                    messages.append(raw_message)
-                else:
-                    logger.info(
-                        f"Running ingestion without orchestration for file {file_name} and document_id {document_id}."
-                    )
-                    # TODO - Clean up implementation logic here to be more explicitly `synchronous`
-                    from core.main.orchestration import (
-                        simple_ingestion_factory,
-                    )
-
-                    simple_ingestor = simple_ingestion_factory(self.service)
-                    await simple_ingestor["ingest-files"](workflow_input)
-                    messages.append(
-                        {
-                            "message": "Ingestion task completed successfully.",
-                            "document_id": str(document_id),
-                            "task_id": None,
-                        }
-                    )
-
-            return messages  # type: ignore
-
-        update_files_extras = self.openapi_extras.get("update_files", {})
-        update_files_descriptions = update_files_extras.get(
-            "input_descriptions", {}
-        )
-
-        @self.router.post(
-            "/update_files",
-            openapi_extra=update_files_extras.get("openapi_extra"),
-        )
-        @self.base_endpoint
-        async def update_files_app(
-            files: list[UploadFile] = File(
-                ..., description=update_files_descriptions.get("files")
-            ),
-            document_ids: Optional[Json[list[UUID]]] = Form(
-                None, description=ingest_files_descriptions.get("document_ids")
-            ),
-            collection_ids: Optional[Json[list[list[UUID]]]] = Form(
-                None,
-                description="Optional collection IDs for the documents, if provided the document will be assigned to them at ingestion.",
-            ),
-            metadatas: Optional[Json[list[dict]]] = Form(
-                None, description=ingest_files_descriptions.get("metadatas")
-            ),
-            ingestion_config: Optional[Json[dict]] = Form(
-                None,
-                description=ingest_files_descriptions.get("ingestion_config"),
-            ),
-            run_with_orchestration: Optional[bool] = Form(
-                True,
-                description=ingest_files_descriptions.get(
-                    "run_with_orchestration"
-                ),
-            ),
-            auth_user=Depends(self.service.providers.auth.auth_wrapper),
-        ) -> WrappedUpdateResponse:
-            """
-            Ingests updated files into R2R, updating the corresponding `Document` and `Chunk` objects from previous ingestion.
-
-            This endpoint supports multipart/form-data requests, enabling you to update files and their associated metadatas into R2R.
-
-            A valid user authentication token is required to access this endpoint, as regular users can only update their own files.
-            """
-            if not auth_user.is_superuser:
-                for metadata in metadatas or []:
-                    if "owner_id" in metadata and metadata["owner_id"] != str(
-                        auth_user.id
-                    ):
-                        raise R2RException(
-                            status_code=403,
-                            message="Non-superusers cannot set owner_id in metadata.",
-                        )
-                    metadata["owner_id"] = str(auth_user.id)
-
-            file_datas = await self._process_files(files)
-
-            processed_data = []
-            for it, file_data in enumerate(file_datas):
-                content = base64.b64decode(file_data.pop("content"))
-                document_id = (
-                    document_ids[it]
-                    if document_ids
-                    else generate_document_id(
-                        file_data["filename"], auth_user.id
-                    )
-                )
-
-                await self.service.providers.database.store_file(
-                    document_id,
-                    file_data["filename"],
-                    BytesIO(content),
-                    file_data["content_type"],
-                )
-
-                processed_data.append(
-                    {
-                        "file_data": file_data,
-                        "file_length": len(content),
-                        "document_id": str(document_id),
-                    }
-                )
-
-            workflow_input = {
-                "file_datas": [item["file_data"] for item in processed_data],
-                "file_sizes_in_bytes": [
-                    item["file_length"] for item in processed_data
-                ],
-                "document_ids": [
-                    item["document_id"] for item in processed_data
-                ],
-                "metadatas": metadatas,
-                "ingestion_config": ingestion_config,
-                "user": auth_user.model_dump_json(),
-                "is_update": True,
-                "collection_ids": (
-                    [str(cid) for cid in collection_ids[it]]
-                    if collection_ids
-                    else None
-                ),
-            }
-
-            if run_with_orchestration:
-                raw_message: dict[str, Union[str, None]] = await self.orchestration_provider.run_workflow(  # type: ignore
-                    "update-files", {"request": workflow_input}, {}
-                )
-                raw_message["message"] = "Update task queued successfully."
-                raw_message["document_ids"] = workflow_input["document_ids"]
-
-                return raw_message  # type: ignore
-            else:
-                logger.info("Running update without orchestration.")
-                # TODO - Clean up implementation logic here to be more explicitly `synchronous`
-                from core.main.orchestration import simple_ingestion_factory
-
-                simple_ingestor = simple_ingestion_factory(self.service)
-                await simple_ingestor["update-files"](workflow_input)
-                return {  # type: ignore
-                    "message": "Update task completed successfully.",
-                    "document_ids": workflow_input["document_ids"],
-                    "task_id": None,
-                }
-
-        ingest_chunks_extras = self.openapi_extras.get("ingest_chunks", {})
-        ingest_chunks_descriptions = ingest_chunks_extras.get(
-            "input_descriptions", {}
-        )
-
-        @self.router.post(
-            "/ingest_chunks",
-            openapi_extra=ingest_chunks_extras.get("openapi_extra"),
-        )
-        @self.base_endpoint
-        async def ingest_chunks_app(
-            chunks: list[RawChunk] = Body(
-                {}, description=ingest_chunks_descriptions.get("chunks")
-            ),
-            document_id: Optional[str] = Body(
-                None, description=ingest_chunks_descriptions.get("document_id")
-            ),
-            metadata: Optional[dict] = Body(
-                None, description=ingest_files_descriptions.get("metadata")
-            ),
-            collection_ids: Optional[Json[list[list[UUID]]]] = Body(
-                None,
-                description="Optional collection IDs for the documents, if provided the document will be assigned to them at ingestion.",
-            ),
-            run_with_orchestration: Optional[bool] = Body(
-                True,
-                description=ingest_files_descriptions.get(
-                    "run_with_orchestration"
-                ),
-            ),
-            auth_user=Depends(self.service.providers.auth.auth_wrapper),
-        ) -> WrappedIngestionResponse:
-            """
-            Ingests `Chunk` objects into the system as raw text and associated metadata.
-
-            A valid user authentication token is required to access this endpoint, as regular users can only ingest chunks for their own access.
-            """
-            if document_id:
-                try:
-                    document_uuid = UUID(document_id)
-                except ValueError:
-                    raise R2RException(
-                        status_code=422, message="Invalid document ID format."
-                    )
-
-            if not document_id:
-                document_uuid = generate_document_id(
-                    chunks[0].text[:20], auth_user.id
-                )
-
-            workflow_input = {
-                "document_id": str(document_uuid),
-                "chunks": [chunk.model_dump() for chunk in chunks],
-                "metadata": metadata or {},
-                "user": auth_user.model_dump_json(),
-                "collection_ids": (
-                    [str(cid) for cid in collection_ids]
-                    if collection_ids
-                    else None
-                ),
-            }
-            if run_with_orchestration:
-                raw_message = await self.orchestration_provider.run_workflow(
-                    "ingest-chunks",
-                    {"request": workflow_input},
-                    options={
-                        "additional_metadata": {
-                            "document_id": str(document_uuid),
-                        }
-                    },
-                )
-                raw_message["document_id"] = str(document_uuid)
-
-                return [raw_message]  # type: ignore
-            else:
-                logger.info("Running ingest chunks without orchestration.")
-                # TODO - Clean up implementation logic here to be more explicitly `synchronous`
-                from core.main.orchestration import simple_ingestion_factory
-
-                simple_ingestor = simple_ingestion_factory(self.service)
-                await simple_ingestor["ingest-chunks"](workflow_input)
-                return [
-                    {  # type: ignore
-                        "message": "Ingestion task completed successfully.",
-                        "document_id": str(document_uuid),
-                        "task_id": None,
-                    }
-                ]
-
-        @self.router.post(
-            "/update_document_metadata/{document_id}",
-        )
-        @self.base_endpoint
-        async def update_document_metadata_app(
-            document_id: UUID = Path(
-                ..., description="The document ID of the document to update"
-            ),
-            metadata: dict = Body(
-                ...,
-                description="The new metadata to merge with existing metadata",
-            ),
-            auth_user=Depends(self.service.providers.auth.auth_wrapper),
-        ) -> WrappedMetadataUpdateResponse:
-            """
-            Updates the metadata of a previously ingested document and its associated chunks.
-
-            A valid user authentication token is required to access this endpoint, as regular users can only update their own documents.
-            """
-
-            try:
-                workflow_input = {
-                    "document_id": str(document_id),
-                    "metadata": metadata,
-                    "user": auth_user.model_dump_json(),
-                }
-
-                logger.info(
-                    "Running document metadata update without orchestration."
-                )
-                from core.main.orchestration import simple_ingestion_factory
-
-                simple_ingestor = simple_ingestion_factory(self.service)
-                await simple_ingestor["update-document-metadata"](
-                    workflow_input
-                )
-
-                return {  # type: ignore
-                    "message": "Update metadata task completed successfully.",
-                    "document_id": str(document_id),
-                    "task_id": None,
-                }
-
-            except Exception as e:
-                raise HTTPException(
-                    status_code=500,
-                    detail=f"Error updating document metadata: {str(e)}",
-                )
-
-        @self.router.put(
-            "/update_chunk/{document_id}/{extraction_id}",
-        )
-        @self.base_endpoint
-        async def update_chunk_app(
-            document_id: UUID = Path(
-                ..., description="The document ID of the chunk to update"
-            ),
-            extraction_id: UUID = Path(
-                ..., description="The extraction ID of the chunk to update"
-            ),
-            text: str = Body(
-                ..., description="The new text content for the chunk"
-            ),
-            metadata: Optional[dict] = Body(
-                None, description="Optional updated metadata"
-            ),
-            run_with_orchestration: Optional[bool] = Body(True),
-            auth_user=Depends(self.service.providers.auth.auth_wrapper),
-        ) -> WrappedUpdateResponse:
-            """
-            Updates a previously ingested `Chunk` object into the system as raw text and associated metadata.
-
-            A valid user authentication token is required to access this endpoint, as regular users can only ingest chunks for their own access.
-            """
-
-            try:
-                workflow_input = {
-                    "document_id": str(document_id),
-                    "extraction_id": str(extraction_id),
-                    "text": text,
-                    "metadata": metadata,
-                    "user": auth_user.model_dump_json(),
-                }
-
-                if run_with_orchestration:
-                    raw_message: dict[str, Union[str, None]] = await self.orchestration_provider.run_workflow(  # type: ignore
-                        "update-chunk", {"request": workflow_input}, {}
-                    )
-                    raw_message["message"] = "Update task queued successfully."
-                    raw_message["document_ids"] = [document_id]  # type: ignore
-
-                    return raw_message  # type: ignore
-                else:
-                    logger.info("Running chunk update without orchestration.")
-                    from core.main.orchestration import (
-                        simple_ingestion_factory,
-                    )
-
-                    simple_ingestor = simple_ingestion_factory(self.service)
-                    await simple_ingestor["update-chunk"](workflow_input)
-
-                    return {  # type: ignore
-                        "message": "Update task completed successfully.",
-                        "document_ids": workflow_input["document_ids"],
-                        "task_id": None,
-                    }
-
-            except Exception as e:
-                raise HTTPException(
-                    status_code=500, detail=f"Error updating chunk: {str(e)}"
-                )
-
-        create_vector_index_extras = self.openapi_extras.get(
-            "create_vector_index", {}
-        )
-        create_vector_descriptions = create_vector_index_extras.get(
-            "input_descriptions", {}
-        )
-
-        @self.router.post(
-            "/create_vector_index",
-            openapi_extra=create_vector_index_extras.get("openapi_extra"),
-        )
-        @self.base_endpoint
-        async def create_vector_index_app(
-            table_name: Optional[VectorTableName] = Body(
-                default=VectorTableName.VECTORS,
-                description=create_vector_descriptions.get("table_name"),
-            ),
-            index_method: IndexMethod = Body(
-                default=IndexMethod.hnsw,
-                description=create_vector_descriptions.get("index_method"),
-            ),
-            index_measure: IndexMeasure = Body(
-                default=IndexMeasure.cosine_distance,
-                description=create_vector_descriptions.get("index_measure"),
-            ),
-            index_arguments: Optional[
-                Union[IndexArgsIVFFlat, IndexArgsHNSW]
-            ] = Body(
-                None,
-                description=create_vector_descriptions.get("index_arguments"),
-            ),
-            index_name: Optional[str] = Body(
-                None,
-                description=create_vector_descriptions.get("index_name"),
-            ),
-            index_column: Optional[str] = Body(
-                None,
-                description=create_vector_descriptions.get("index_column"),
-            ),
-            concurrently: bool = Body(
-                default=True,
-                description=create_vector_descriptions.get("concurrently"),
-            ),
-            auth_user=Depends(self.service.providers.auth.auth_wrapper),
-        ) -> WrappedCreateVectorIndexResponse:
-            """
-            Create a vector index for a given table.
-
-            """
-
-            logger.info(
-                f"Creating vector index for {table_name} with method {index_method}, measure {index_measure}, concurrently {concurrently}"
-            )
-
-            raw_message = await self.orchestration_provider.run_workflow(
-                "create-vector-index",
-                {
-                    "request": {
-                        "table_name": table_name,
-                        "index_method": index_method,
-                        "index_measure": index_measure,
-                        "index_name": index_name,
-                        "index_column": index_column,
-                        "index_arguments": index_arguments,
-                        "concurrently": concurrently,
-                    },
-                },
-                options={
-                    "additional_metadata": {},
-                },
-            )
-
-            return raw_message  # type: ignore
-
-        list_vector_indices_extras = self.openapi_extras.get(
-            "create_vector_index", {}
-        )
-        list_vector_indices_descriptions = list_vector_indices_extras.get(
-            "input_descriptions", {}
-        )
-
-        @self.router.get(
-            "/list_vector_indices",
-            openapi_extra=list_vector_indices_extras.get("openapi_extra"),
-        )
-        @self.base_endpoint
-        async def list_vector_indices_app(
-            table_name: Optional[VectorTableName] = Query(
-                default=VectorTableName.VECTORS,
-                description=list_vector_indices_descriptions.get("table_name"),
-            ),
-            auth_user=Depends(self.service.providers.auth.auth_wrapper),
-        ) -> WrappedListVectorIndicesResponse:
-            indices = await self.service.providers.database.list_indices(
-                table_name=table_name
-            )
-            return {"indices": indices}  # type: ignore
-
-        delete_vector_index_extras = self.openapi_extras.get(
-            "delete_vector_index", {}
-        )
-        delete_vector_index_descriptions = delete_vector_index_extras.get(
-            "input_descriptions", {}
-        )
-
-        @self.router.delete(
-            "/delete_vector_index",
-            openapi_extra=delete_vector_index_extras.get("openapi_extra"),
-        )
-        @self.base_endpoint
-        async def delete_vector_index_app(
-            index_name: str = Body(
-                ...,
-                description=delete_vector_index_descriptions.get("index_name"),
-            ),
-            table_name: Optional[VectorTableName] = Body(
-                default=VectorTableName.VECTORS,
-                description=delete_vector_index_descriptions.get("table_name"),
-            ),
-            concurrently: bool = Body(
-                default=True,
-                description=delete_vector_index_descriptions.get(
-                    "concurrently"
-                ),
-            ),
-            auth_user=Depends(self.service.providers.auth.auth_wrapper),
-        ) -> WrappedDeleteVectorIndexResponse:
-            logger.info(
-                f"Deleting vector index {index_name} from table {table_name}"
-            )
-
-            raw_message = await self.orchestration_provider.run_workflow(
-                "delete-vector-index",
-                {
-                    "request": {
-                        "index_name": index_name,
-                        "table_name": table_name,
-                        "concurrently": concurrently,
-                    },
-                },
-                options={
-                    "additional_metadata": {},
-                },
-            )
-
-            return raw_message  # type: ignore
-
-    @staticmethod
-    async def _process_files(files):
-        import base64
-
-        file_datas = []
-        for file in files:
-            content = await file.read()
-            file_datas.append(
-                {
-                    "filename": file.filename,
-                    "content": base64.b64encode(content).decode("utf-8"),
-                    "content_type": file.content_type,
-                }
-            )
-        return file_datas
diff --git a/py/core/main/api/v3/chunks_router.py b/py/core/main/api/v3/chunks_router.py
index 82242b13c..3fd9bc310 100644
--- a/py/core/main/api/v3/chunks_router.py
+++ b/py/core/main/api/v3/chunks_router.py
@@ -16,6 +16,7 @@
     SearchSettings,
     UnprocessedChunk,
     UpdateChunk,
+    select_search_filters
 )
 from core.base.api.models import (
     GenericBooleanResponse,
@@ -49,44 +50,6 @@ def __init__(
     ):
         super().__init__(providers, services, orchestration_provider, run_type)
 
-    def _select_filters(
-        self,
-        auth_user: Any,
-        search_settings: SearchSettings,
-    ) -> dict[str, Any]:
-
-        filters = copy(search_settings.filters)
-        selected_collections = None
-        if not auth_user.is_superuser:
-            user_collections = set(auth_user.collection_ids)
-            for key in filters.keys():
-                if "collection_ids" in key:
-                    selected_collections = set(filters[key]["$overlap"])
-                    break
-
-            if selected_collections:
-                allowed_collections = user_collections.intersection(
-                    selected_collections
-                )
-            else:
-                allowed_collections = user_collections
-            # for non-superusers, we filter by user_id and selected & allowed collections
-            collection_filters = {
-                "$or": [
-                    {"owner_id": {"$eq": auth_user.id}},
-                    {
-                        "collection_ids": {
-                            "$overlap": list(allowed_collections)
-                        }
-                    },
-                ]  # type: ignore
-            }
-
-            filters.pop("collection_ids", None)
-
-            filters = {"$and": [collection_filters, filters]}  # type: ignore
-
-        return filters
 
     def _setup_routes(self):
         @self.router.post(
@@ -131,7 +94,7 @@ async def search_chunks(
             Allowed operators include `eq`, `neq`, `gt`, `gte`, `lt`, `lte`, `like`, `ilike`, `in`, and `nin`.
             """
 
-            search_settings.filters = self._select_filters(
+            search_settings.filters = select_search_filters(
                 auth_user, search_settings
             )
 
@@ -309,43 +272,6 @@ async def update_chunk(
                 # vector = existing_chunk.get('vector')
             )
 
-        #         @self.router.post(
-        #             "/chunks/{id}/enrich",
-        #             summary="Enrich Chunk",
-        #             openapi_extra={
-        #                 "x-codeSamples": [
-        #                     {
-        #                         "lang": "Python",
-        #                         "source": """
-        # from r2r import R2RClient
-
-        # client = R2RClient("http://localhost:7272")
-        # result = client.chunks.enrich(
-        #     id="b4ac4dd6-5f27-596e-a55b-7cf242ca30aa",
-        #     enrichment_config={"key": "value"}
-        # )
-        # """,
-        #                     }
-        #                 ]
-        #             },
-        #         )
-        #         @self.base_endpoint
-        #         async def enrich_chunk(
-        #             id: UUID = Path(...),
-        #             enrichment_config: dict = Body(...),
-        #             auth_user=Depends(self.providers.auth.auth_wrapper),
-        #         ) -> ResultsWrapper[ChunkResponse]:
-        #             """
-        #             Enrich a chunk with additional processing and metadata.
-
-        #             This endpoint allows adding additional enrichments to an existing chunk,
-        #             such as entity extraction, classification, or custom processing defined
-        #             in the enrichment_config.
-
-        #             Users can only enrich chunks they own unless they are superusers.
-        #             """
-        #             pass
-
         @self.router.delete(
             "/chunks/{id}",
             summary="Delete Chunk",
diff --git a/py/core/main/api/v3/documents_router.py b/py/core/main/api/v3/documents_router.py
index 1135243c7..c711f32b8 100644
--- a/py/core/main/api/v3/documents_router.py
+++ b/py/core/main/api/v3/documents_router.py
@@ -22,6 +22,7 @@
     Workflow,
     generate_document_id,
     generate_id,
+    select_search_filters
 )
 from core.base.abstractions import KGCreationSettings, KGRunType
 from core.base.api.models import (
@@ -114,7 +115,7 @@ def _prepare_search_settings(
             effective_settings = search_settings or SearchSettings()
 
         # Apply user-specific filters
-        effective_settings.filters = self._select_filters(
+        effective_settings.filters = select_search_filters(
             auth_user, effective_settings
         )
 
@@ -341,17 +342,6 @@ async def create_document(
                 )
             # Check if the user is a superuser
             metadata = metadata or {}
-            if not auth_user.is_superuser:
-                if "owner_id" in metadata and (
-                    not auth_user.is_superuser
-                    and metadata["owner_id"] != str(auth_user.id)
-                ):
-                    raise R2RException(
-                        status_code=403,
-                        message="Non-superusers cannot set user_id in metadata.",
-                    )
-                # If user is not a superuser, set user_id in metadata
-                metadata["owner_id"] = str(auth_user.id)
 
             if chunks:
                 if len(chunks) == 0:
@@ -1618,7 +1608,7 @@ async def search_documents(
             results = await self.services["retrieval"].search_documents(
                 query=query,
                 query_embedding=query_embedding,
-                settings=settings,
+                settings=effective_settings,
             )
             return results
 
diff --git a/py/core/main/api/v3/retrieval_router.py b/py/core/main/api/v3/retrieval_router.py
index 1685961b9..5f6e5bc9a 100644
--- a/py/core/main/api/v3/retrieval_router.py
+++ b/py/core/main/api/v3/retrieval_router.py
@@ -13,6 +13,7 @@
     R2RException,
     SearchMode,
     SearchSettings,
+    select_search_filters
 )
 from core.base.api.models import (
     WrappedAgentResponse,
@@ -84,50 +85,12 @@ def _prepare_search_settings(
             effective_settings = search_settings or SearchSettings()
 
         # Apply user-specific filters
-        effective_settings.filters = self._select_filters(
+        effective_settings.filters = select_search_filters(
             auth_user, effective_settings
         )
 
         return effective_settings
 
-    def _select_filters(
-        self,
-        auth_user: Any,
-        search_settings: SearchSettings,
-    ) -> dict[str, Any]:
-
-        filters = copy(search_settings.filters)
-        selected_collections = None
-        if not auth_user.is_superuser:
-            user_collections = set(auth_user.collection_ids)
-            for key in filters.keys():
-                if "collection_ids" in key:
-                    selected_collections = set(filters[key]["$overlap"])
-                    break
-
-            if selected_collections:
-                allowed_collections = user_collections.intersection(
-                    selected_collections
-                )
-            else:
-                allowed_collections = user_collections
-            # for non-superusers, we filter by user_id and selected & allowed collections
-            collection_filters = {
-                "$or": [
-                    {"owner_id": {"$eq": auth_user.id}},
-                    {
-                        "collection_ids": {
-                            "$overlap": list(allowed_collections)
-                        }
-                    },
-                ]  # type: ignore
-            }
-
-            filters.pop("collection_ids", None)
-
-            filters = {"$and": [collection_filters, filters]}  # type: ignore
-
-        return filters
 
     def _setup_routes(self):
 
@@ -854,7 +817,6 @@ async def completion(
             The messages list should contain alternating user and assistant messages, with an optional
             system message at the start. Each message should have a 'role' and 'content'.
             """
-            print('messages = ', messages)
 
             return await self.services["retrieval"].completion(
                 messages=messages,
diff --git a/py/sdk/models.py b/py/sdk/models.py
index 9ad98d6e7..a2f5b8f67 100644
--- a/py/sdk/models.py
+++ b/py/sdk/models.py
@@ -20,6 +20,7 @@
     R2RSerializable,
     SearchMode,
     SearchSettings,
+    select_search_filters,
     Token,
     User,
 )
@@ -45,6 +46,7 @@
     "Token",
     "ChunkSearchResult",
     "SearchSettings",
+    "select_search_filters",
     "SearchMode",
     "KGEntityDeduplicationSettings",
     "RAGResponse",
diff --git a/py/sdk/v3/documents.py b/py/sdk/v3/documents.py
index 30926bbde..3298f842c 100644
--- a/py/sdk/v3/documents.py
+++ b/py/sdk/v3/documents.py
@@ -12,7 +12,7 @@
     WrappedDocumentsResponse,
 )
 
-from ..models import IngestionMode, SearchSettings
+from ..models import IngestionMode, SearchSettings, SearchMode
 
 
 class DocumentsSDK:
diff --git a/py/shared/abstractions/__init__.py b/py/shared/abstractions/__init__.py
index d6f4c9a34..4e3ad8c6f 100644
--- a/py/shared/abstractions/__init__.py
+++ b/py/shared/abstractions/__init__.py
@@ -46,6 +46,7 @@
     KGRelationshipResult,
     KGSearchResultType,
     SearchMode,
+    select_search_filters,
     SearchSettings,
 )
 from .user import Token, TokenData, User
@@ -111,6 +112,7 @@
     "ChunkSearchSettings",
     "ChunkSearchResult",
     "SearchSettings",
+    "select_search_filters",
     "HybridSearchSettings",
     "SearchMode",
     # KG abstractions
diff --git a/py/shared/abstractions/search.py b/py/shared/abstractions/search.py
index e7789054d..d0f375606 100644
--- a/py/shared/abstractions/search.py
+++ b/py/shared/abstractions/search.py
@@ -1,5 +1,6 @@
 """Abstractions for search functionality."""
 
+from copy import copy
 from enum import Enum
 from typing import Any, Optional
 from uuid import UUID
@@ -457,3 +458,43 @@ class SearchMode(str, Enum):
     basic = "basic"
     advanced = "advanced"
     custom = "custom"
+
+
+
+def select_search_filters(
+    auth_user: Any,
+    search_settings: SearchSettings,
+) -> dict[str, Any]:
+
+    filters = copy(search_settings.filters)
+    selected_collections = None
+    if not auth_user.is_superuser:
+        user_collections = set(auth_user.collection_ids)
+        for key in filters.keys():
+            if "collection_ids" in key:
+                selected_collections = set(filters[key]["$overlap"])
+                break
+
+        if selected_collections:
+            allowed_collections = user_collections.intersection(
+                selected_collections
+            )
+        else:
+            allowed_collections = user_collections
+        # for non-superusers, we filter by user_id and selected & allowed collections
+        collection_filters = {
+            "$or": [
+                {"owner_id": {"$eq": auth_user.id}},
+                {
+                    "collection_ids": {
+                        "$overlap": list(allowed_collections)
+                    }
+                },
+            ]  # type: ignore
+        }
+
+        filters.pop("collection_ids", None)
+
+        filters = {"$and": [collection_filters, filters]}  # type: ignore
+
+    return filters
\ No newline at end of file
diff --git a/py/shared/abstractions/vector.py b/py/shared/abstractions/vector.py
index c755c96c2..863b6a210 100644
--- a/py/shared/abstractions/vector.py
+++ b/py/shared/abstractions/vector.py
@@ -202,7 +202,7 @@ def __str__(self) -> str:
             f"VectorEntry("
             f"chunk_id={self.id}, "
             f"document_id={self.document_id}, "
-            f"user_id={self.owner_id}, "
+            f"owner_id={self.owner_id}, "
             f"collection_ids={self.collection_ids}, "
             f"vector={self.vector}, "
             f"text={self.text}, "

From 3e2352b29997f756485ab741a8ae7e48f0f934ee Mon Sep 17 00:00:00 2001
From: emrgnt-cmplxty <owen@algofi.org>
Date: Fri, 6 Dec 2024 15:43:21 -0800
Subject: [PATCH 5/7] up

---
 py/core/base/abstractions/__init__.py      |  2 +-
 py/core/base/providers/ingestion.py        | 51 +++++++++++++++-------
 py/core/main/api/v3/chunks_router.py       |  3 +-
 py/core/main/api/v3/documents_router.py    | 20 ++++-----
 py/core/main/api/v3/logs_router.py         | 14 +++---
 py/core/main/api/v3/retrieval_router.py    |  3 +-
 py/core/main/app.py                        |  1 +
 py/core/main/app_entry.py                  |  3 +-
 py/core/main/config.py                     |  5 +--
 py/core/main/services/retrieval_service.py |  3 +-
 py/core/providers/database/graph.py        |  2 +-
 py/core/providers/logger/r2r_logger.py     |  2 +-
 py/core/utils/logging_config.py            |  4 +-
 py/sdk/models.py                           |  2 +-
 py/sdk/v3/documents.py                     |  5 +--
 py/sdk/v3/retrieval.py                     |  2 +-
 py/shared/abstractions/__init__.py         |  2 +-
 py/shared/abstractions/search.py           | 25 +++++------
 18 files changed, 79 insertions(+), 70 deletions(-)

diff --git a/py/core/base/abstractions/__init__.py b/py/core/base/abstractions/__init__.py
index d8d476013..703a7c568 100644
--- a/py/core/base/abstractions/__init__.py
+++ b/py/core/base/abstractions/__init__.py
@@ -64,8 +64,8 @@
     KGSearchResultType,
     SearchMode,
     SearchSettings,
-    select_search_filters,
     WebSearchResponse,
+    select_search_filters,
 )
 from shared.abstractions.user import Token, TokenData, User
 from shared.abstractions.vector import (
diff --git a/py/core/base/providers/ingestion.py b/py/core/base/providers/ingestion.py
index f199a6ab5..b4c6ecf0b 100644
--- a/py/core/base/providers/ingestion.py
+++ b/py/core/base/providers/ingestion.py
@@ -1,18 +1,19 @@
 import logging
 from abc import ABC
 from enum import Enum
+from typing import ClassVar
+
+from pydantic import BaseModel, Field
 
 from core.base.abstractions import ChunkEnrichmentSettings
 
-from .base import Provider, ProviderConfig
+from .base import AppConfig, Provider, ProviderConfig
 from .database import DatabaseProvider
 from .llm import CompletionProvider
-from .base import AppConfig
-from typing import ClassVar
-from pydantic import BaseModel, Field
 
 logger = logging.getLogger()
 
+
 class IngestionConfig(ProviderConfig):
     _defaults: ClassVar[dict] = {
         "app": AppConfig(),
@@ -31,7 +32,7 @@ class IngestionConfig(ProviderConfig):
         "chunks_for_document_summary": 128,
         "document_summary_model": "openai/gpt-4o-mini",
         "parser_overrides": {},
-        "extra_fields": {}
+        "extra_fields": {},
     }
 
     provider: str = Field(
@@ -41,40 +42,58 @@ class IngestionConfig(ProviderConfig):
         default_factory=lambda: IngestionConfig._defaults["excluded_parsers"]
     )
     chunk_enrichment_settings: ChunkEnrichmentSettings = Field(
-        default_factory=lambda: IngestionConfig._defaults["chunk_enrichment_settings"]
+        default_factory=lambda: IngestionConfig._defaults[
+            "chunk_enrichment_settings"
+        ]
     )
     extra_parsers: dict[str, str] = Field(
         default_factory=lambda: IngestionConfig._defaults["extra_parsers"]
     )
     audio_transcription_model: str = Field(
-        default_factory=lambda: IngestionConfig._defaults["audio_transcription_model"]
+        default_factory=lambda: IngestionConfig._defaults[
+            "audio_transcription_model"
+        ]
     )
     vision_img_prompt_name: str = Field(
-        default_factory=lambda: IngestionConfig._defaults["vision_img_prompt_name"]
+        default_factory=lambda: IngestionConfig._defaults[
+            "vision_img_prompt_name"
+        ]
     )
     vision_img_model: str = Field(
         default_factory=lambda: IngestionConfig._defaults["vision_img_model"]
     )
     vision_pdf_prompt_name: str = Field(
-        default_factory=lambda: IngestionConfig._defaults["vision_pdf_prompt_name"]
+        default_factory=lambda: IngestionConfig._defaults[
+            "vision_pdf_prompt_name"
+        ]
     )
     vision_pdf_model: str = Field(
         default_factory=lambda: IngestionConfig._defaults["vision_pdf_model"]
     )
     skip_document_summary: bool = Field(
-        default_factory=lambda: IngestionConfig._defaults["skip_document_summary"]
+        default_factory=lambda: IngestionConfig._defaults[
+            "skip_document_summary"
+        ]
     )
     document_summary_system_prompt: str = Field(
-        default_factory=lambda: IngestionConfig._defaults["document_summary_system_prompt"]
+        default_factory=lambda: IngestionConfig._defaults[
+            "document_summary_system_prompt"
+        ]
     )
     document_summary_task_prompt: str = Field(
-        default_factory=lambda: IngestionConfig._defaults["document_summary_task_prompt"]
+        default_factory=lambda: IngestionConfig._defaults[
+            "document_summary_task_prompt"
+        ]
     )
     chunks_for_document_summary: int = Field(
-        default_factory=lambda: IngestionConfig._defaults["chunks_for_document_summary"]
+        default_factory=lambda: IngestionConfig._defaults[
+            "chunks_for_document_summary"
+        ]
     )
     document_summary_model: str = Field(
-        default_factory=lambda: IngestionConfig._defaults["document_summary_model"]
+        default_factory=lambda: IngestionConfig._defaults[
+            "document_summary_model"
+        ]
     )
     parser_overrides: dict[str, str] = Field(
         default_factory=lambda: IngestionConfig._defaults["parser_overrides"]
@@ -121,7 +140,6 @@ def get_default(cls, mode: str, app) -> "IngestionConfig":
             # For `custom` or any unrecognized mode, return a base config
             return cls(app=app)
 
-
     @classmethod
     def set_default(cls, **kwargs):
         for key, value in kwargs.items():
@@ -131,7 +149,7 @@ def set_default(cls, **kwargs):
                 raise AttributeError(
                     f"No default attribute '{key}' in GenerationConfig"
                 )
-            
+
     class Config:
         populate_by_name = True
         json_schema_extra = {
@@ -152,6 +170,7 @@ class Config:
             "parser_overrides": {},
         }
 
+
 class IngestionProvider(Provider, ABC):
 
     config: IngestionConfig
diff --git a/py/core/main/api/v3/chunks_router.py b/py/core/main/api/v3/chunks_router.py
index 3fd9bc310..0eb5026ec 100644
--- a/py/core/main/api/v3/chunks_router.py
+++ b/py/core/main/api/v3/chunks_router.py
@@ -16,7 +16,7 @@
     SearchSettings,
     UnprocessedChunk,
     UpdateChunk,
-    select_search_filters
+    select_search_filters,
 )
 from core.base.api.models import (
     GenericBooleanResponse,
@@ -50,7 +50,6 @@ def __init__(
     ):
         super().__init__(providers, services, orchestration_provider, run_type)
 
-
     def _setup_routes(self):
         @self.router.post(
             "/chunks/search",
diff --git a/py/core/main/api/v3/documents_router.py b/py/core/main/api/v3/documents_router.py
index c711f32b8..2eb7138b4 100644
--- a/py/core/main/api/v3/documents_router.py
+++ b/py/core/main/api/v3/documents_router.py
@@ -4,7 +4,7 @@
 import mimetypes
 import textwrap
 from io import BytesIO
-from typing import Optional, Any
+from typing import Any, Optional
 from uuid import UUID
 
 from fastapi import Body, Depends, File, Form, Path, Query, UploadFile
@@ -15,14 +15,14 @@
     IngestionConfig,
     IngestionMode,
     R2RException,
-    SearchSettings,
     RunType,
-    UnprocessedChunk,
     SearchMode,
+    SearchSettings,
+    UnprocessedChunk,
     Workflow,
     generate_document_id,
     generate_id,
-    select_search_filters
+    select_search_filters,
 )
 from core.base.abstractions import KGCreationSettings, KGRunType
 from core.base.api.models import (
@@ -90,7 +90,6 @@ def __init__(
         super().__init__(providers, services, orchestration_provider, run_type)
         self._register_workflows()
 
-
     def _prepare_search_settings(
         self,
         auth_user: Any,
@@ -325,7 +324,7 @@ async def create_document(
                 ingestion_mode=ingestion_mode,
                 ingestion_config=ingestion_config,
             )
-            print('effective_ingestion_config = ', effective_ingestion_config)
+            print("effective_ingestion_config = ", effective_ingestion_config)
             if not file and not raw_text and not chunks:
                 raise R2RException(
                     status_code=422,
@@ -1566,7 +1565,7 @@ async def get_relationships(
         @self.base_endpoint
         async def search_documents(
             query: str = Body(
-                ..., 
+                ...,
                 description="The search query to perform.",
             ),
             search_mode: SearchMode = Body(
@@ -1586,7 +1585,7 @@ async def search_documents(
                 description="Settings for document search",
             ),
             auth_user=Depends(self.providers.auth.auth_wrapper),
-        ): # -> WrappedDocumentSearchResponse:  # type: ignore
+        ):  # -> WrappedDocumentSearchResponse:  # type: ignore
             """
             Perform a search query on the automatically generated document summaries in the system.
 
@@ -1601,9 +1600,7 @@ async def search_documents(
             )
 
             query_embedding = (
-                await self.providers.embedding.async_get_embedding(
-                    query
-                )
+                await self.providers.embedding.async_get_embedding(query)
             )
             results = await self.services["retrieval"].search_documents(
                 query=query,
@@ -1612,7 +1609,6 @@ async def search_documents(
             )
             return results
 
-
     @staticmethod
     async def _process_file(file):
         import base64
diff --git a/py/core/main/api/v3/logs_router.py b/py/core/main/api/v3/logs_router.py
index 7a6f1aa74..523c7ab5a 100644
--- a/py/core/main/api/v3/logs_router.py
+++ b/py/core/main/api/v3/logs_router.py
@@ -1,17 +1,19 @@
-from fastapi import WebSocket
-import contextlib
-from fastapi.requests import Request
-from fastapi.templating import Jinja2Templates
-from pathlib import Path
 import asyncio
+import contextlib
 import logging
+from pathlib import Path
+
 import aiofiles
+from fastapi import WebSocket
+from fastapi.requests import Request
+from fastapi.templating import Jinja2Templates
 
+from core.base.logger.base import RunType
 from core.providers import (
     HatchetOrchestrationProvider,
     SimpleOrchestrationProvider,
 )
-from core.base.logger.base import RunType
+
 from .base_router import BaseRouterV3
 
 
diff --git a/py/core/main/api/v3/retrieval_router.py b/py/core/main/api/v3/retrieval_router.py
index 5f6e5bc9a..2dd4b5615 100644
--- a/py/core/main/api/v3/retrieval_router.py
+++ b/py/core/main/api/v3/retrieval_router.py
@@ -13,7 +13,7 @@
     R2RException,
     SearchMode,
     SearchSettings,
-    select_search_filters
+    select_search_filters,
 )
 from core.base.api.models import (
     WrappedAgentResponse,
@@ -91,7 +91,6 @@ def _prepare_search_settings(
 
         return effective_settings
 
-
     def _setup_routes(self):
 
         @self.router.post(
diff --git a/py/core/main/app.py b/py/core/main/app.py
index c58c43b1a..0f75f0107 100644
--- a/py/core/main/app.py
+++ b/py/core/main/app.py
@@ -108,6 +108,7 @@ def _apply_cors(self):
 
     async def serve(self, host: str = "0.0.0.0", port: int = 7272):
         import uvicorn
+
         from core.utils.logging_config import configure_logging
 
         configure_logging()
diff --git a/py/core/main/app_entry.py b/py/core/main/app_entry.py
index 380379407..930548436 100644
--- a/py/core/main/app_entry.py
+++ b/py/core/main/app_entry.py
@@ -1,5 +1,3 @@
-from core.utils.logging_config import configure_logging
-
 import os
 from contextlib import asynccontextmanager
 from typing import Optional
@@ -10,6 +8,7 @@
 from fastapi.responses import JSONResponse
 
 from core.base import R2RException
+from core.utils.logging_config import configure_logging
 
 from .assembly import R2RBuilder, R2RConfig
 
diff --git a/py/core/main/config.py b/py/core/main/config.py
index f2343bb26..9d5d691fb 100644
--- a/py/core/main/config.py
+++ b/py/core/main/config.py
@@ -123,10 +123,7 @@ def __init__(self, config_data: dict[str, Any]):
         self.agent = AgentConfig.create(**self.agent, app=self.app)  # type: ignore
         self.orchestration = OrchestrationConfig.create(**self.orchestration, app=self.app)  # type: ignore
 
-
-        IngestionConfig.set_default(
-            **self.ingestion.dict()
-        )
+        IngestionConfig.set_default(**self.ingestion.dict())
 
         # override GenerationConfig defaults
         GenerationConfig.set_default(
diff --git a/py/core/main/services/retrieval_service.py b/py/core/main/services/retrieval_service.py
index b45a0e750..0c151328a 100644
--- a/py/core/main/services/retrieval_service.py
+++ b/py/core/main/services/retrieval_service.py
@@ -24,7 +24,6 @@
 from core.base.logger.base import RunType
 from core.providers.logger.r2r_logger import SqlitePersistentLoggingProvider
 from core.telemetry.telemetry_decorator import telemetry_event
-
 from shared.api.models.management.responses import MessageResponse
 
 from ..abstractions import R2RAgents, R2RPipelines, R2RPipes, R2RProviders
@@ -120,7 +119,7 @@ async def search_documents(
         settings: SearchSettings,
         query_embedding: Optional[list[float]] = None,
     ) -> list[DocumentResponse]:
-        print('settings = ', settings)
+        print("settings = ", settings)
         return await self.providers.database.search_documents(
             query_text=query,
             settings=settings,
diff --git a/py/core/providers/database/graph.py b/py/core/providers/database/graph.py
index 56e640483..e33c139ca 100644
--- a/py/core/providers/database/graph.py
+++ b/py/core/providers/database/graph.py
@@ -2418,7 +2418,7 @@ async def graph_search(
             ORDER BY {embedding_type} <=> $1
             LIMIT $2;
         """
-        print('QUERY = ', QUERY)
+        print("QUERY = ", QUERY)
         results = await self.connection_manager.fetch_query(
             QUERY, tuple(params)
         )
diff --git a/py/core/providers/logger/r2r_logger.py b/py/core/providers/logger/r2r_logger.py
index b63636b76..95f469fae 100644
--- a/py/core/providers/logger/r2r_logger.py
+++ b/py/core/providers/logger/r2r_logger.py
@@ -19,8 +19,8 @@
     RunType,
 )
 from shared.api.models.management.responses import (
-    MessageResponse,
     ConversationResponse,
+    MessageResponse,
 )
 
 logger = logging.getLogger()
diff --git a/py/core/utils/logging_config.py b/py/core/utils/logging_config.py
index 681220f6d..48a361d29 100644
--- a/py/core/utils/logging_config.py
+++ b/py/core/utils/logging_config.py
@@ -1,8 +1,8 @@
 import logging
-import sys
-from pathlib import Path
 import logging.config
 import re
+import sys
+from pathlib import Path
 
 
 class HTTPStatusFilter(logging.Filter):
diff --git a/py/sdk/models.py b/py/sdk/models.py
index a2f5b8f67..7b64d38ce 100644
--- a/py/sdk/models.py
+++ b/py/sdk/models.py
@@ -20,9 +20,9 @@
     R2RSerializable,
     SearchMode,
     SearchSettings,
-    select_search_filters,
     Token,
     User,
+    select_search_filters,
 )
 from shared.api.models import CombinedSearchResponse, RAGResponse
 
diff --git a/py/sdk/v3/documents.py b/py/sdk/v3/documents.py
index 3298f842c..5dd48031f 100644
--- a/py/sdk/v3/documents.py
+++ b/py/sdk/v3/documents.py
@@ -12,7 +12,7 @@
     WrappedDocumentsResponse,
 )
 
-from ..models import IngestionMode, SearchSettings, SearchMode
+from ..models import IngestionMode, SearchMode, SearchSettings
 
 
 class DocumentsSDK:
@@ -407,7 +407,6 @@ async def list(
             version="v3",
         )
 
-
     async def search(
         self,
         query: str,
@@ -441,4 +440,4 @@ async def search(
             "documents/search",
             json=data,
             version="v3",
-        )
\ No newline at end of file
+        )
diff --git a/py/sdk/v3/retrieval.py b/py/sdk/v3/retrieval.py
index 8c522f342..5a1f56b66 100644
--- a/py/sdk/v3/retrieval.py
+++ b/py/sdk/v3/retrieval.py
@@ -72,7 +72,7 @@ async def completion(
             "messages": [msg.model_dump() for msg in cast_messages],
             "generation_config": generation_config,
         }
-        print('data = ', data)
+        print("data = ", data)
         return await self.client._make_request(
             "POST",
             "retrieval/completion",
diff --git a/py/shared/abstractions/__init__.py b/py/shared/abstractions/__init__.py
index 4e3ad8c6f..7179e8ebf 100644
--- a/py/shared/abstractions/__init__.py
+++ b/py/shared/abstractions/__init__.py
@@ -46,8 +46,8 @@
     KGRelationshipResult,
     KGSearchResultType,
     SearchMode,
-    select_search_filters,
     SearchSettings,
+    select_search_filters,
 )
 from .user import Token, TokenData, User
 from .vector import (
diff --git a/py/shared/abstractions/search.py b/py/shared/abstractions/search.py
index d0f375606..13a8cc4a0 100644
--- a/py/shared/abstractions/search.py
+++ b/py/shared/abstractions/search.py
@@ -216,12 +216,16 @@ def as_dict(self) -> dict:
                 if self.chunk_search_results
                 else []
             ),
-            "graph_search_results": [
-                result.to_dict() for result in self.graph_search_results
-            ] if self.graph_search_results else [],
-            "web_search_results": [
-                result.to_dict() for result in self.web_search_results
-            ] if self.web_search_results else [],
+            "graph_search_results": (
+                [result.to_dict() for result in self.graph_search_results]
+                if self.graph_search_results
+                else []
+            ),
+            "web_search_results": (
+                [result.to_dict() for result in self.web_search_results]
+                if self.web_search_results
+                else []
+            ),
         }
 
 
@@ -460,7 +464,6 @@ class SearchMode(str, Enum):
     custom = "custom"
 
 
-
 def select_search_filters(
     auth_user: Any,
     search_settings: SearchSettings,
@@ -485,11 +488,7 @@ def select_search_filters(
         collection_filters = {
             "$or": [
                 {"owner_id": {"$eq": auth_user.id}},
-                {
-                    "collection_ids": {
-                        "$overlap": list(allowed_collections)
-                    }
-                },
+                {"collection_ids": {"$overlap": list(allowed_collections)}},
             ]  # type: ignore
         }
 
@@ -497,4 +496,4 @@ def select_search_filters(
 
         filters = {"$and": [collection_filters, filters]}  # type: ignore
 
-    return filters
\ No newline at end of file
+    return filters

From 3ab58f9013006a9ffbebf45138eb9a6623a80554 Mon Sep 17 00:00:00 2001
From: emrgnt-cmplxty <owen@algofi.org>
Date: Fri, 6 Dec 2024 15:45:54 -0800
Subject: [PATCH 6/7] up

---
 py/core/main/api/v3/documents_router.py | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/py/core/main/api/v3/documents_router.py b/py/core/main/api/v3/documents_router.py
index 337777bbf..004040db5 100644
--- a/py/core/main/api/v3/documents_router.py
+++ b/py/core/main/api/v3/documents_router.py
@@ -324,10 +324,6 @@ async def create_document(
                 ingestion_mode=ingestion_mode,
                 ingestion_config=ingestion_config,
             )
-<<<<<<< HEAD
-            print("effective_ingestion_config = ", effective_ingestion_config)
-=======
->>>>>>> 66f7f26d7778f9cbda2f039661204db80feb68d5
             if not file and not raw_text and not chunks:
                 raise R2RException(
                     status_code=422,

From 7995f6e6fc95d143e996c782c03306121e3515b3 Mon Sep 17 00:00:00 2001
From: emrgnt-cmplxty <owen@algofi.org>
Date: Fri, 6 Dec 2024 15:46:49 -0800
Subject: [PATCH 7/7] up

---
 py/core/main/services/retrieval_service.py | 1 -
 py/core/providers/database/graph.py        | 1 -
 py/core/providers/logger/r2r_logger.py     | 1 -
 py/sdk/v3/retrieval.py                     | 1 -
 4 files changed, 4 deletions(-)

diff --git a/py/core/main/services/retrieval_service.py b/py/core/main/services/retrieval_service.py
index 0c151328a..e0b3927cf 100644
--- a/py/core/main/services/retrieval_service.py
+++ b/py/core/main/services/retrieval_service.py
@@ -119,7 +119,6 @@ async def search_documents(
         settings: SearchSettings,
         query_embedding: Optional[list[float]] = None,
     ) -> list[DocumentResponse]:
-        print("settings = ", settings)
         return await self.providers.database.search_documents(
             query_text=query,
             settings=settings,
diff --git a/py/core/providers/database/graph.py b/py/core/providers/database/graph.py
index e33c139ca..50030f9a1 100644
--- a/py/core/providers/database/graph.py
+++ b/py/core/providers/database/graph.py
@@ -2418,7 +2418,6 @@ async def graph_search(
             ORDER BY {embedding_type} <=> $1
             LIMIT $2;
         """
-        print("QUERY = ", QUERY)
         results = await self.connection_manager.fetch_query(
             QUERY, tuple(params)
         )
diff --git a/py/core/providers/logger/r2r_logger.py b/py/core/providers/logger/r2r_logger.py
index 95f469fae..86d754821 100644
--- a/py/core/providers/logger/r2r_logger.py
+++ b/py/core/providers/logger/r2r_logger.py
@@ -714,7 +714,6 @@ async def get_conversation(
                 )
             conversation_created_at = row[0]
 
-        print(f"Getting a branch_id: {branch_id}")
         if branch_id is None:
             # Get the most recent branch by created_at timestamp
             async with self.conn.execute(
diff --git a/py/sdk/v3/retrieval.py b/py/sdk/v3/retrieval.py
index 5a1f56b66..921cd334c 100644
--- a/py/sdk/v3/retrieval.py
+++ b/py/sdk/v3/retrieval.py
@@ -72,7 +72,6 @@ async def completion(
             "messages": [msg.model_dump() for msg in cast_messages],
             "generation_config": generation_config,
         }
-        print("data = ", data)
         return await self.client._make_request(
             "POST",
             "retrieval/completion",
