Merge pull request #65 from DefangLabs/jordan/uwsgi-lazy-apps

refactor RAGSystem initialization

Author: commit111

app/Dockerfile

@@ -46,4 +46,4 @@ RUN python test_intercom.py
 ENV FLASK_APP=app.py
 
 # Run the application using uWSGI
-CMD ["uwsgi", "--http", "0.0.0.0:5050", "--wsgi-file", "app.py", "--callable", "app", "--processes", "4", "--threads", "2"]
+CMD ["uwsgi", "--lazy-apps", "--http", "0.0.0.0:5050", "--wsgi-file", "app.py", "--callable", "app", "--processes", "4", "--threads", "2"]

app/app.py

@@ -1,6 +1,6 @@
 from flask import Flask, request, jsonify, render_template, Response, stream_with_context, session, send_from_directory
 from flask_wtf.csrf import CSRFProtect
-from rag_system import rag_system
+from rag_system import RAGSystem
 import hashlib
 import subprocess
 import os
@@ -26,6 +26,8 @@
 app.config['SESSION_COOKIE_HTTPONLY'] = True
 app.config['SESSION_COOKIE_SECURE'] = bool(os.getenv('SESSION_COOKIE_SECURE'))
 
+app.rag_system = RAGSystem()
+
 csrf = CSRFProtect(app)
 
 # Initialize Redis connection
@@ -52,13 +54,13 @@ def handle_ask_request(request, session):
     if 'anonymous_id' not in session:
         session['anonymous_id'] = str(uuid.uuid4())
     anonymous_id = session['anonymous_id']
-    
+
     # Determine the source based on the user agent
     user_agent = request.headers.get('User-Agent', '')
     source = 'Ask Defang Discord Bot' if 'Discord Bot' in user_agent else 'Ask Defang Website'
 
     # Use the shared generate function directly
-    return Response(stream_with_context(generate(query, source, anonymous_id)), content_type='text/markdown')
+    return Response(stream_with_context(generate(app.rag_system, query, source, anonymous_id)), content_type='text/markdown')
 
 @app.route('/', methods=['GET', 'POST'])
 def index():
@@ -110,7 +112,7 @@ def trigger_rebuild():
 
         print("Rebuilding embeddings...")
         try:
-            rag_system.rebuild_embeddings()
+            app.rag_system.rebuild_embeddings()
         except Exception as e:
             print(f"Error rebuilding embeddings: {str(e)}")
             return jsonify({"error": "Error rebuilding embeddings", "details": str(e)}), 500
@@ -136,7 +138,7 @@ def debug_context():
         query = data.get('query', '')
         if not query:
             return jsonify({"error": "Query is required"}), 400
-        context = rag_system.get_context(query)
+        context = app.rag_system.get_context(query)
         return jsonify({"context": context})
 
 
@@ -180,7 +182,7 @@ def handle_webhook():
             return 'OK'
         # Fetch the conversation and generate an LLM answer for the user
         logger.info(f"Detected a user reply in conversation {conversation_id}; fetching an answer from LLM...")
-        answer_intercom_conversation(conversation_id)
+        answer_intercom_conversation(app.rag_system, conversation_id)
     else:
         logger.info(f"Received webhook for unsupported topic: {topic}; no action taken.")
     return 'OK'

app/intercom.py

@@ -43,7 +43,7 @@ def fetch_intercom_conversation(conversation_id):
     if response.status_code != 200:
         logger.error(f"Failed to fetch conversation {conversation_id} from Intercom; status code: {response.status_code}, response: {response.text}")
         return jsonify({"error": "Failed to fetch conversation from Intercom"}), response.status_code
-    
+
     return response, response.status_code
 
 # Determines the user query from the Intercom conversation response
@@ -123,7 +123,7 @@ def set_conversation_human_replied(conversation_id, redis_client):
         logger.info(f"Added conversation_id {conversation_id} to Redis set admin_replied_conversations")
     except Exception as e:
         logger.error(f"Error adding conversation_id to Redis: {e}")
-    
+
 # Check if a conversation is already marked as replied by a human admin
 def is_conversation_human_replied(conversation_id, redis_client):
     try:
@@ -158,7 +158,7 @@ def post_intercom_reply(conversation_id, response_text):
 
 
 # Returns a generated LLM answer to the Intercom conversation based on previous user message history
-def answer_intercom_conversation(conversation_id):
+def answer_intercom_conversation(rag, conversation_id):
     logger.info(f"Received request to get conversation {conversation_id}")
     # Retrieves the history of the conversation thread in Intercom
     conversation, status_code = fetch_intercom_conversation(conversation_id)
@@ -174,9 +174,9 @@ def answer_intercom_conversation(conversation_id):
 
     # Use a deterministic, non-reversible hash for anonymous_id for Intercom conversations
     anon_hash = hashlib.sha256(f"intercom-{conversation_id}".encode()).hexdigest()
-    
+
     # Generate the exact response using the RAG system
-    llm_response = "".join(generate(user_query, 'Intercom Conversation', anon_hash))
+    llm_response = "".join(generate(rag, user_query, 'Intercom Conversation', anon_hash))
     llm_response = llm_response + " 🤖" # Add a marker to indicate the end of the response
 
     logger.info(f"LLM response: {llm_response}")
@@ -200,5 +200,5 @@ def check_intercom_ip(request):
     if remote_ip not in INTERCOM_ALLOWED_IPS:
         # logger.info(f"Rejected webhook from unauthorized IP: {remote_ip}")
         return False
-    
+
     return True

app/rag_system.py

@@ -187,6 +187,3 @@ def get_context(self, retrieved_docs):
         for doc in retrieved_docs:
             retrieved_text.append(f"{doc['about']}. {doc['text']}")
         return "\n\n".join(retrieved_text)
-
-# Instantiate the RAGSystem
-rag_system = RAGSystem()

app/utils.py

@@ -1,14 +1,13 @@
-from rag_system import rag_system
 import sys
 import traceback
 import segment.analytics as analytics
 
 # Shared function to generate response stream from RAG system
-def generate(query, source, anonymous_id):
+def generate(rag, query, source, anonymous_id):
     full_response = ""
     print(f"Received query: {str(query)}", file=sys.stderr)
     try:
-        for token in rag_system.answer_query_stream(query):
+        for token in rag.answer_query_stream(query):
             yield token
             full_response += token
     except Exception as e:
@@ -26,5 +25,5 @@ def generate(query, source, anonymous_id):
             event='Chatbot Question submitted',
             properties={'query': query, 'response': full_response, 'source': source}
         )
-    
+
     return full_response

compose.yaml

@@ -26,7 +26,6 @@ services:
       INTERCOM_TOKEN:
       INTERCOM_ADMIN_ID:
       REDIS_URL: redis://redis:6379/0
-    command: uwsgi --http 0.0.0.0:5050 --wsgi-file app.py --callable app --processes 4 --threads 2
     deploy:
       resources:
         reservations: