From 307c70a63dbd0e61c5e0994a746edebd36fa80fc Mon Sep 17 00:00:00 2001 From: PromptEngineer <134474669+PromtEngineer@users.noreply.github.com> Date: Tue, 10 Mar 2026 10:12:24 -0700 Subject: [PATCH 1/2] Remove ~1,200 lines of dead code across backend, RAG system, and frontend MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Systematically identified and removed dead code in 5 phases ordered by risk: Phase 1 - Dead imports/comments: - Remove unused `import pymupdf`, duplicate `from PIL import Image`, commented `ChunkStore` import in retrieval_pipeline.py - Remove commented `BM25Generator` import in embedders.py - Remove unused `parse_qs` import in api_server.py - Remove commented-out overview refresh block in agent/loop.py - Remove dead `json as _json` import alias in backend/server.py - Remove commented-out remove button in empty-chat-state.tsx Phase 2 - Dead backend functions/variables: - Remove `ensure_overview_path()` (defined but never called) in server.py - Remove `pdf_processor` global variable (assigned but never read) in server.py - Remove deprecated `handle_pdf_upload()` (returns 410, never called) in server.py - Remove `get_simple_pdf_processor()` (never called) in simple_pdf_processor.py - Remove `main()` test function and `if __name__` block in ollama_client.py - Remove unimplemented `--stop` arg and handler in run_system.py Phase 3 - Dead RAG functions/config: - Remove `_get_bm25_retriever()` (references nonexistent BM25Retriever class) - Remove `show_graph()` (references nonexistent config key, would KeyError) - Remove dead `bm25` and `graph_rag` entries from PIPELINE_CONFIGS Phase 4 - Dead frontend files/exports: - Delete 6 unused files: IndexWizard.tsx, GlassSelect.tsx, badge.tsx, skeleton.tsx, sidebar.tsx (replaced by session-sidebar.tsx), chat-bubble-demo.tsx - Remove unused `LocalGPTChat` import in demo.tsx - Remove unused `layout` prop from ChatBubble component - Remove `hasExcessiveWhitespace()` (exported, never called) in textNormalization.ts - Remove `cleanupEmptySessions()` and `uploadPDFs()` (never called) in api.ts Phase 5 - Unused file: - Delete rag_system/api_server_with_progress.py (443 lines, never imported) Verified: all Python files compile, TypeScript type-check passes, RAG pipeline E2E test passes (index → query → streaming all functional). Co-Authored-By: Claude Opus 4.6 --- backend/ollama_client.py | 34 -- backend/server.py | 29 +- backend/simple_pdf_processor.py | 13 - rag_system/agent/loop.py | 12 - rag_system/api_server.py | 2 +- rag_system/api_server_with_progress.py | 443 --------------------- rag_system/indexing/embedders.py | 1 - rag_system/main.py | 41 +- rag_system/pipelines/retrieval_pipeline.py | 17 - run_system.py | 9 - src/components/IndexWizard.tsx | 72 ---- src/components/demo.tsx | 1 - src/components/ui/GlassSelect.tsx | 13 - src/components/ui/badge.tsx | 46 --- src/components/ui/chat-bubble-demo.tsx | 103 ----- src/components/ui/chat-bubble.tsx | 2 - src/components/ui/empty-chat-state.tsx | 7 - src/components/ui/sidebar.tsx | 260 ------------ src/components/ui/skeleton.tsx | 13 - src/lib/api.ts | 69 ---- src/utils/textNormalization.ts | 22 - 21 files changed, 3 insertions(+), 1206 deletions(-) delete mode 100644 rag_system/api_server_with_progress.py delete mode 100644 src/components/IndexWizard.tsx delete mode 100644 src/components/ui/GlassSelect.tsx delete mode 100644 src/components/ui/badge.tsx delete mode 100644 src/components/ui/chat-bubble-demo.tsx delete mode 100644 src/components/ui/sidebar.tsx delete mode 100644 src/components/ui/skeleton.tsx diff --git a/backend/ollama_client.py b/backend/ollama_client.py index 33ee103d..52e21770 100644 --- a/backend/ollama_client.py +++ b/backend/ollama_client.py @@ -166,37 +166,3 @@ def chat_stream(self, message: str, model: str = "llama3.2", conversation_histor except requests.exceptions.RequestException as e: yield f"Connection error: {e}" -def main(): - """Test the Ollama client""" - client = OllamaClient() - - # Check if Ollama is running - if not client.is_ollama_running(): - print("❌ Ollama is not running. Please start Ollama first.") - print("Install: https://ollama.ai") - print("Run: ollama serve") - return - - print("✅ Ollama is running!") - - # List available models - models = client.list_models() - print(f"Available models: {models}") - - # Try to use llama3.2, pull if needed - model_name = "llama3.2" - if model_name not in [m.split(":")[0] for m in models]: - print(f"Model {model_name} not found. Pulling...") - if client.pull_model(model_name): - print(f"✅ Model {model_name} pulled successfully!") - else: - print(f"❌ Failed to pull model {model_name}") - return - - # Test chat - print("\n🤖 Testing chat...") - response = client.chat("Hello! Can you tell me a short joke?", model_name) - print(f"AI: {response}") - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/backend/server.py b/backend/server.py index 859040ef..71dce5ae 100644 --- a/backend/server.py +++ b/backend/server.py @@ -730,19 +730,6 @@ def handle_index_documents(self, session_id: str): print(f"❌ Exception during indexing: {str(e)}") self.send_json_response({"error": f"An unexpected error occurred: {str(e)}"}, status_code=500) - def handle_pdf_upload(self, session_id: str): - """ - Processes PDF files: extracts text and stores it in the database. - DEPRECATED: This is the old method. Use handle_file_upload instead. - """ - # This function is now deprecated in favor of the new indexing workflow - # but is kept for potential legacy/compatibility reasons. - # For new functionality, it should not be used. - self.send_json_response({ - "warning": "This upload method is deprecated. Use the new file upload and indexing flow.", - "message": "No action taken." - }, status_code=410) # 410 Gone - def handle_get_models(self): """Get available models from both Ollama and HuggingFace, grouped by capability""" try: @@ -907,15 +894,9 @@ def handle_build_index(self, index_id: str): # Set per-index overview file path overview_path = f"index_store/overviews/{index_id}.jsonl" - # Ensure config_override includes overview_path - def ensure_overview_path(cfg: dict): - cfg["overview_path"] = overview_path - - # we'll inject later when we build config_override - # Delegate to advanced RAG API same as session indexing rag_api_url = "http://localhost:8001/index" - import requests, json as _json + import requests # Use the index's dedicated LanceDB table so retrieval matches table_name = index.get("vector_table_name") payload = { @@ -1098,14 +1079,6 @@ def main(): print(f"❌ Error initializing PDF processor: {e}") print("⚠️ PDF processing disabled - server will run without RAG functionality") - # Set a global reference to the initialized processor if needed elsewhere - global pdf_processor - pdf_processor = pdf_module.simple_pdf_processor - if pdf_processor: - print("✅ Global PDF processor initialized") - else: - print("⚠️ PDF processing disabled - server will run without RAG functionality") - # Cleanup empty sessions on startup print("🧹 Cleaning up empty sessions...") cleanup_count = db.cleanup_empty_sessions() diff --git a/backend/simple_pdf_processor.py b/backend/simple_pdf_processor.py index e1f8dfad..3b67db17 100644 --- a/backend/simple_pdf_processor.py +++ b/backend/simple_pdf_processor.py @@ -199,16 +199,3 @@ def initialize_simple_pdf_processor(): print(f"❌ Failed to initialize PDF processor: {str(e)}") simple_pdf_processor = None -def get_simple_pdf_processor(): - """Get the global PDF processor instance""" - global simple_pdf_processor - if simple_pdf_processor is None: - initialize_simple_pdf_processor() - return simple_pdf_processor - -if __name__ == "__main__": - # Test the simple PDF processor - print("🧪 Testing simple PDF processor...") - - processor = SimplePDFProcessor() - print("✅ Simple PDF processor test completed!") \ No newline at end of file diff --git a/rag_system/agent/loop.py b/rag_system/agent/loop.py index b7da704d..e76a4f87 100644 --- a/rag_system/agent/loop.py +++ b/rag_system/agent/loop.py @@ -267,18 +267,6 @@ async def _run_async(self, query: str, table_name: str = None, session_id: str = # 🚀 NEW: Get conversation history history = self.chat_histories.get(session_id, []) if session_id else [] - # 🔄 Refresh overviews for this session if available - # if session_id and session_id != getattr(self, "_current_overview_session", None): - # candidate_path = os.path.join("index_store", "overviews", f"{session_id}.jsonl") - # if os.path.exists(candidate_path): - # self._load_overviews(candidate_path) - # self._current_overview_session = session_id - # else: - # # Fall back to global overviews if per-session file not found - # if self._current_overview_session != "GLOBAL": - # self._load_overviews(self._global_overview_path) - # self._current_overview_session = "GLOBAL" - query_type = await self._triage_query_async(query, history) print(f"🎯 ROUTING DEBUG: Final triage decision: '{query_type}'") print(f"Agent Triage Decision: '{query_type}'") diff --git a/rag_system/api_server.py b/rag_system/api_server.py index de148361..af0f6c0d 100644 --- a/rag_system/api_server.py +++ b/rag_system/api_server.py @@ -1,7 +1,7 @@ import json import http.server import socketserver -from urllib.parse import urlparse, parse_qs +from urllib.parse import urlparse import os import requests import sys diff --git a/rag_system/api_server_with_progress.py b/rag_system/api_server_with_progress.py deleted file mode 100644 index 438dcb34..00000000 --- a/rag_system/api_server_with_progress.py +++ /dev/null @@ -1,443 +0,0 @@ -import json -import threading -import time -from typing import Dict, List, Any -import logging -from urllib.parse import urlparse, parse_qs -import http.server -import socketserver - -# Import the core logic and batch processing utilities -from rag_system.main import get_agent -from rag_system.utils.batch_processor import ProgressTracker, timer - -# Set up logging -logging.basicConfig(level=logging.INFO) -logger = logging.getLogger(__name__) - -# Global progress tracking storage -ACTIVE_PROGRESS_SESSIONS: Dict[str, Dict[str, Any]] = {} - -# --- Global Singleton for the RAG Agent --- -print("🧠 Initializing RAG Agent... (This may take a moment)") -RAG_AGENT = get_agent() -if RAG_AGENT is None: - print("❌ Critical error: RAG Agent could not be initialized. Exiting.") - exit(1) -print("✅ RAG Agent initialized successfully.") - -class ServerSentEventsHandler: - """Handler for Server-Sent Events (SSE) for real-time progress updates""" - - active_connections: Dict[str, Any] = {} - - @classmethod - def add_connection(cls, session_id: str, response_handler): - """Add a new SSE connection""" - cls.active_connections[session_id] = response_handler - logger.info(f"SSE connection added for session: {session_id}") - - @classmethod - def remove_connection(cls, session_id: str): - """Remove an SSE connection""" - if session_id in cls.active_connections: - del cls.active_connections[session_id] - logger.info(f"SSE connection removed for session: {session_id}") - - @classmethod - def send_event(cls, session_id: str, event_type: str, data: Dict[str, Any]): - """Send an SSE event to a specific session""" - if session_id not in cls.active_connections: - return - - try: - handler = cls.active_connections[session_id] - event_data = json.dumps(data) - message = f"event: {event_type}\ndata: {event_data}\n\n" - handler.wfile.write(message.encode('utf-8')) - handler.wfile.flush() - except Exception as e: - logger.error(f"Failed to send SSE event: {e}") - cls.remove_connection(session_id) - -class RealtimeProgressTracker(ProgressTracker): - """Enhanced ProgressTracker that sends updates via Server-Sent Events""" - - def __init__(self, total_items: int, operation_name: str, session_id: str): - super().__init__(total_items, operation_name) - self.session_id = session_id - self.last_update = 0 - self.update_interval = 1 # Update every 1 second - - # Initialize session progress - ACTIVE_PROGRESS_SESSIONS[session_id] = { - "operation_name": operation_name, - "total_items": total_items, - "processed_items": 0, - "errors_encountered": 0, - "start_time": self.start_time, - "status": "running", - "current_step": "", - "eta_seconds": 0, - "throughput": 0, - "progress_percentage": 0 - } - - # Send initial progress update - self._send_progress_update() - - def update(self, items_processed: int, errors: int = 0, current_step: str = ""): - """Update progress and send notification""" - super().update(items_processed, errors) - - # Update session data - session_data = ACTIVE_PROGRESS_SESSIONS.get(self.session_id) - if session_data: - session_data.update({ - "processed_items": self.processed_items, - "errors_encountered": self.errors_encountered, - "current_step": current_step, - "progress_percentage": (self.processed_items / self.total_items) * 100, - }) - - # Calculate throughput and ETA - elapsed = time.time() - self.start_time - if elapsed > 0: - session_data["throughput"] = self.processed_items / elapsed - remaining = self.total_items - self.processed_items - session_data["eta_seconds"] = remaining / session_data["throughput"] if session_data["throughput"] > 0 else 0 - - # Send update if enough time has passed - current_time = time.time() - if current_time - self.last_update >= self.update_interval: - self._send_progress_update() - self.last_update = current_time - - def finish(self): - """Mark progress as finished and send final update""" - super().finish() - - # Update session status - session_data = ACTIVE_PROGRESS_SESSIONS.get(self.session_id) - if session_data: - session_data.update({ - "status": "completed", - "progress_percentage": 100, - "eta_seconds": 0 - }) - - # Send final update - self._send_progress_update(final=True) - - def _send_progress_update(self, final: bool = False): - """Send progress update via Server-Sent Events""" - session_data = ACTIVE_PROGRESS_SESSIONS.get(self.session_id, {}) - - event_data = { - "session_id": self.session_id, - "progress": session_data.copy(), - "final": final, - "timestamp": time.time() - } - - ServerSentEventsHandler.send_event(self.session_id, "progress", event_data) - -def run_indexing_with_progress(file_paths: List[str], session_id: str): - """Enhanced indexing function with real-time progress tracking""" - from rag_system.pipelines.indexing_pipeline import IndexingPipeline - from rag_system.utils.ollama_client import OllamaClient - import json - - try: - # Send initial status - ServerSentEventsHandler.send_event(session_id, "status", { - "message": "Initializing indexing pipeline...", - "session_id": session_id - }) - - # Load configuration - config_file = "batch_indexing_config.json" - try: - with open(config_file, 'r') as f: - config = json.load(f) - except FileNotFoundError: - # Fallback to default config - config = { - "embedding_model_name": "Qwen/Qwen3-Embedding-0.6B", - "indexing": { - "embedding_batch_size": 50, - "enrichment_batch_size": 10, - "enable_progress_tracking": True - }, - "contextual_enricher": {"enabled": True, "window_size": 1}, - "retrievers": { - "dense": {"enabled": True, "lancedb_table_name": "default_text_table"}, - "bm25": {"enabled": True, "index_name": "default_bm25_index"} - }, - "storage": { - "chunk_store_path": "./index_store/chunks/chunks.pkl", - "lancedb_uri": "./index_store/lancedb", - "bm25_path": "./index_store/bm25" - } - } - - # Initialize components - ollama_client = OllamaClient() - ollama_config = { - "generation_model": "llama3.2:1b", - "embedding_model": "mxbai-embed-large" - } - - # Create enhanced pipeline - pipeline = IndexingPipeline(config, ollama_client, ollama_config) - - # Create progress tracker for the overall process - total_steps = 6 # Rough estimate of pipeline steps - step_tracker = RealtimeProgressTracker(total_steps, "Document Indexing", session_id) - - with timer("Complete Indexing Pipeline"): - try: - # Step 1: Document Processing - step_tracker.update(1, current_step="Processing documents...") - - # Run the indexing pipeline - pipeline.run(file_paths) - - # Update progress through the steps - step_tracker.update(1, current_step="Chunking completed...") - step_tracker.update(1, current_step="BM25 indexing completed...") - step_tracker.update(1, current_step="Contextual enrichment completed...") - step_tracker.update(1, current_step="Vector embeddings completed...") - step_tracker.update(1, current_step="Indexing finalized...") - - step_tracker.finish() - - # Send completion notification - ServerSentEventsHandler.send_event(session_id, "completion", { - "message": f"Successfully indexed {len(file_paths)} file(s)", - "file_count": len(file_paths), - "session_id": session_id - }) - - except Exception as e: - # Send error notification - ServerSentEventsHandler.send_event(session_id, "error", { - "message": str(e), - "session_id": session_id - }) - raise - - except Exception as e: - logger.error(f"Indexing failed for session {session_id}: {e}") - ServerSentEventsHandler.send_event(session_id, "error", { - "message": str(e), - "session_id": session_id - }) - raise - -class EnhancedRagApiHandler(http.server.BaseHTTPRequestHandler): - """Enhanced API handler with progress tracking support""" - - def do_OPTIONS(self): - """Handle CORS preflight requests for frontend integration.""" - self.send_response(200) - self.send_header('Access-Control-Allow-Origin', '*') - self.send_header('Access-Control-Allow-Methods', 'POST, GET, OPTIONS') - self.send_header('Access-Control-Allow-Headers', 'Content-Type') - self.end_headers() - - def do_GET(self): - """Handle GET requests for progress status and SSE streams""" - parsed_path = urlparse(self.path) - - if parsed_path.path == '/progress': - self.handle_progress_status() - elif parsed_path.path == '/stream': - self.handle_progress_stream() - else: - self.send_json_response({"error": "Not Found"}, status_code=404) - - def do_POST(self): - """Handle POST requests for chat and indexing.""" - parsed_path = urlparse(self.path) - - if parsed_path.path == '/chat': - self.handle_chat() - elif parsed_path.path == '/index': - self.handle_index_with_progress() - else: - self.send_json_response({"error": "Not Found"}, status_code=404) - - def handle_chat(self): - """Handles a chat query by calling the agentic RAG pipeline.""" - try: - content_length = int(self.headers['Content-Length']) - post_data = self.rfile.read(content_length) - data = json.loads(post_data.decode('utf-8')) - - query = data.get('query') - if not query: - self.send_json_response({"error": "Query is required"}, status_code=400) - return - - # Use the single, persistent agent instance to run the query - result = RAG_AGENT.run(query) - - # The result is a dict, so we need to dump it to a JSON string - self.send_json_response(result) - - except json.JSONDecodeError: - self.send_json_response({"error": "Invalid JSON"}, status_code=400) - except Exception as e: - self.send_json_response({"error": f"Server error: {str(e)}"}, status_code=500) - - def handle_index_with_progress(self): - """Triggers the document indexing pipeline with real-time progress tracking.""" - try: - content_length = int(self.headers['Content-Length']) - post_data = self.rfile.read(content_length) - data = json.loads(post_data.decode('utf-8')) - - file_paths = data.get('file_paths') - session_id = data.get('session_id') - - if not file_paths or not isinstance(file_paths, list): - self.send_json_response({ - "error": "A 'file_paths' list is required." - }, status_code=400) - return - - if not session_id: - self.send_json_response({ - "error": "A 'session_id' is required for progress tracking." - }, status_code=400) - return - - # Start indexing in a separate thread to avoid blocking - def run_indexing_thread(): - try: - run_indexing_with_progress(file_paths, session_id) - except Exception as e: - logger.error(f"Indexing thread failed: {e}") - - thread = threading.Thread(target=run_indexing_thread) - thread.daemon = True - thread.start() - - # Return immediate response - self.send_json_response({ - "message": f"Indexing started for {len(file_paths)} file(s)", - "session_id": session_id, - "status": "started", - "progress_stream_url": f"http://localhost:8001/stream?session_id={session_id}" - }) - - except json.JSONDecodeError: - self.send_json_response({"error": "Invalid JSON"}, status_code=400) - except Exception as e: - self.send_json_response({"error": f"Failed to start indexing: {str(e)}"}, status_code=500) - - def handle_progress_status(self): - """Handle GET requests for current progress status""" - parsed_url = urlparse(self.path) - params = parse_qs(parsed_url.query) - session_id = params.get('session_id', [None])[0] - - if not session_id: - self.send_json_response({"error": "session_id is required"}, status_code=400) - return - - progress_data = ACTIVE_PROGRESS_SESSIONS.get(session_id) - if not progress_data: - self.send_json_response({"error": "No active progress for this session"}, status_code=404) - return - - self.send_json_response({ - "session_id": session_id, - "progress": progress_data - }) - - def handle_progress_stream(self): - """Handle Server-Sent Events stream for real-time progress""" - parsed_url = urlparse(self.path) - params = parse_qs(parsed_url.query) - session_id = params.get('session_id', [None])[0] - - if not session_id: - self.send_response(400) - self.end_headers() - return - - # Set up SSE headers - self.send_response(200) - self.send_header('Content-Type', 'text/event-stream') - self.send_header('Cache-Control', 'no-cache') - self.send_header('Connection', 'keep-alive') - self.send_header('Access-Control-Allow-Origin', '*') - self.end_headers() - - # Add this connection to the SSE handler - ServerSentEventsHandler.add_connection(session_id, self) - - # Send initial connection message - initial_message = json.dumps({ - "session_id": session_id, - "message": "Progress stream connected", - "timestamp": time.time() - }) - self.wfile.write(f"event: connected\ndata: {initial_message}\n\n".encode('utf-8')) - self.wfile.flush() - - # Keep connection alive - try: - while session_id in ServerSentEventsHandler.active_connections: - time.sleep(1) - # Send heartbeat - heartbeat = json.dumps({"type": "heartbeat", "timestamp": time.time()}) - self.wfile.write(f"event: heartbeat\ndata: {heartbeat}\n\n".encode('utf-8')) - self.wfile.flush() - except Exception as e: - logger.info(f"SSE connection closed for session {session_id}: {e}") - finally: - ServerSentEventsHandler.remove_connection(session_id) - - def send_json_response(self, data, status_code=200): - """Utility to send a JSON response with CORS headers.""" - self.send_response(status_code) - self.send_header('Content-Type', 'application/json') - self.send_header('Access-Control-Allow-Origin', '*') - self.end_headers() - response = json.dumps(data, indent=2) - self.wfile.write(response.encode('utf-8')) - -def start_enhanced_server(port=8000): - """Start the enhanced API server with a reusable TCP socket.""" - - # Use a custom TCPServer that allows address reuse - class ReusableTCPServer(socketserver.TCPServer): - allow_reuse_address = True - - with ReusableTCPServer(("", port), EnhancedRagApiHandler) as httpd: - print(f"🚀 Starting Enhanced RAG API server on port {port}") - print(f"💬 Chat endpoint: http://localhost:{port}/chat") - print(f"✨ Indexing endpoint: http://localhost:{port}/index") - print(f"📊 Progress endpoint: http://localhost:{port}/progress") - print(f"🌊 Progress stream: http://localhost:{port}/stream") - print(f"📈 Real-time progress tracking enabled via Server-Sent Events!") - httpd.serve_forever() - -if __name__ == '__main__': - # Start the server on a dedicated thread - server_thread = threading.Thread(target=start_enhanced_server) - server_thread.daemon = True - server_thread.start() - - print("🚀 Enhanced RAG API server with progress tracking is running.") - print("Press Ctrl+C to stop.") - - # Keep the main thread alive - try: - while True: - time.sleep(1) - except KeyboardInterrupt: - print("\nStopping server...") \ No newline at end of file diff --git a/rag_system/indexing/embedders.py b/rag_system/indexing/embedders.py index b48648f2..3af248c8 100644 --- a/rag_system/indexing/embedders.py +++ b/rag_system/indexing/embedders.py @@ -1,4 +1,3 @@ -# from rag_system.indexing.representations import BM25Generator import lancedb import pyarrow as pa from typing import List, Dict, Any diff --git a/rag_system/main.py b/rag_system/main.py index a1f50794..3079b7be 100644 --- a/rag_system/main.py +++ b/rag_system/main.py @@ -151,13 +151,6 @@ "enable_progress_tracking": False } }, - "bm25": { - "enabled": True, - "index_name": "rag_bm25_index" - }, - "graph_rag": { - "enabled": False, # Keep disabled for now unless specified - } } # ============================================================================ @@ -280,36 +273,6 @@ def run_chat(query: str): result = agent.run(query) return json.dumps(result, indent=2, ensure_ascii=False) -def show_graph(): - """ - Loads and displays the knowledge graph. - """ - import networkx as nx - import matplotlib.pyplot as plt - - graph_path = PIPELINE_CONFIGS["indexing"]["graph_path"] - if not os.path.exists(graph_path): - print("Knowledge graph not found. Please run the 'index' command first.") - return - - G = nx.read_gml(graph_path) - print("--- Knowledge Graph ---") - print("Nodes:", G.nodes(data=True)) - print("Edges:", G.edges(data=True)) - print("---------------------") - - # Optional: Visualize the graph - try: - pos = nx.spring_layout(G) - nx.draw(G, pos, with_labels=True, node_size=2000, node_color="skyblue", font_size=10, font_weight="bold") - edge_labels = nx.get_edge_attributes(G, 'label') - nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels) - plt.title("Knowledge Graph Visualization") - plt.show() - except Exception as e: - print(f"\nCould not visualize the graph. Matplotlib might not be installed or configured for your environment.") - print(f"Error: {e}") - def run_api_server(): """Starts the advanced RAG API server.""" from rag_system.api_server import start_server @@ -317,7 +280,7 @@ def run_api_server(): def main(): if len(sys.argv) < 2: - print("Usage: python main.py [index|chat|show_graph|api] [query]") + print("Usage: python main.py [index|chat|api] [query]") return command = sys.argv[1] @@ -332,8 +295,6 @@ def main(): query = " ".join(sys.argv[2:]) # 🆕 Print the result for command-line usage print(run_chat(query)) - elif command == "show_graph": - show_graph() elif command == "api": run_api_server() else: diff --git a/rag_system/pipelines/retrieval_pipeline.py b/rag_system/pipelines/retrieval_pipeline.py index f2151273..f8a11f42 100644 --- a/rag_system/pipelines/retrieval_pipeline.py +++ b/rag_system/pipelines/retrieval_pipeline.py @@ -1,4 +1,3 @@ -import pymupdf from typing import List, Dict, Any, Tuple, Optional from PIL import Image import concurrent.futures @@ -17,10 +16,8 @@ from rag_system.indexing.embedders import LanceDBManager from rag_system.rerankers.reranker import QwenReranker from rag_system.rerankers.sentence_pruner import SentencePruner -# from rag_system.indexing.chunk_store import ChunkStore import os -from PIL import Image # --------------------------------------------------------------------------- # Thread-safety helpers @@ -103,20 +100,6 @@ def _get_dense_retriever(self): self.dense_retriever = None return self.dense_retriever - def _get_bm25_retriever(self): - if self.bm25_retriever is None and self.retriever_configs.get("bm25", {}).get("enabled"): - try: - print(f"🔧 Lazily initializing BM25 retriever...") - self.bm25_retriever = BM25Retriever( - index_path=self.storage_config["bm25_path"], - index_name=self.retriever_configs["bm25"]["index_name"] - ) - print("✅ BM25 retriever initialized successfully") - except Exception as e: - print(f"❌ Failed to initialize BM25 retriever on demand: {e}") - # Keep it None so we don't try again - return self.bm25_retriever - def _get_graph_retriever(self): if self._graph_retriever is None and self.retriever_configs.get("graph", {}).get("enabled"): self._graph_retriever = GraphRetriever(graph_path=self.storage_config["graph_path"]) diff --git a/run_system.py b/run_system.py index 8064d6be..f426fc9a 100644 --- a/run_system.py +++ b/run_system.py @@ -499,9 +499,6 @@ def main(): help='Skip frontend startup') parser.add_argument('--health', action='store_true', help='Check health of running services') - parser.add_argument('--stop', action='store_true', - help='Stop all running services') - args = parser.parse_args() # Create service manager @@ -513,12 +510,6 @@ def main(): manager._print_status_summary() return - if args.stop: - # Stop mode - kill any running processes - manager.logger.info("🛑 Stopping all RAG system processes...") - # Implementation for stopping would go here - return - if args.logs_only: # Logs only mode - just tail existing logs manager.logger.info("📋 Showing aggregated logs... (Press Ctrl+C to stop)") diff --git a/src/components/IndexWizard.tsx b/src/components/IndexWizard.tsx deleted file mode 100644 index 8e14a9dd..00000000 --- a/src/components/IndexWizard.tsx +++ /dev/null @@ -1,72 +0,0 @@ -"use client"; -import { useState } from 'react'; -import { ModelSelect } from '@/components/ModelSelect'; - -interface Props { - onClose: () => void; -} - -export function IndexWizard({ onClose }: Props) { - const [files, setFiles] = useState(null); - const [chunkSize, setChunkSize] = useState(512); - const [chunkOverlap, setChunkOverlap] = useState(64); - const [embeddingModel, setEmbeddingModel] = useState(); - // TODO: more params - - const handleFile = (e: React.ChangeEvent) => { - setFiles(e.target.files); - }; - - return ( -
-
-

Create new index

- -
-
- - -
- -
-
- - setChunkSize(parseInt(e.target.value))} - className="w-full bg-gray-800 rounded px-2 py-1" - /> -
-
- - setChunkOverlap(parseInt(e.target.value))} - className="w-full bg-gray-800 rounded px-2 py-1" - /> -
-
- -
- - -
-
- -
- - -
-
-
- ); -} \ No newline at end of file diff --git a/src/components/demo.tsx b/src/components/demo.tsx index 970bf180..cd7bc2b3 100644 --- a/src/components/demo.tsx +++ b/src/components/demo.tsx @@ -1,7 +1,6 @@ "use client"; import { useState, useEffect } from "react" -import { LocalGPTChat } from "@/components/ui/localgpt-chat" import { SessionSidebar } from "@/components/ui/session-sidebar" import { SessionChat } from '@/components/ui/session-chat' import { chatAPI, ChatSession } from "@/lib/api" diff --git a/src/components/ui/GlassSelect.tsx b/src/components/ui/GlassSelect.tsx deleted file mode 100644 index 940975ad..00000000 --- a/src/components/ui/GlassSelect.tsx +++ /dev/null @@ -1,13 +0,0 @@ -"use client"; -import React, { SelectHTMLAttributes } from 'react'; - -export function GlassSelect(props: SelectHTMLAttributes) { - return ( - - ); -} \ No newline at end of file diff --git a/src/components/ui/badge.tsx b/src/components/ui/badge.tsx deleted file mode 100644 index 02054139..00000000 --- a/src/components/ui/badge.tsx +++ /dev/null @@ -1,46 +0,0 @@ -import * as React from "react" -import { Slot } from "@radix-ui/react-slot" -import { cva, type VariantProps } from "class-variance-authority" - -import { cn } from "@/lib/utils" - -const badgeVariants = cva( - "inline-flex items-center justify-center rounded-md border px-2 py-0.5 text-xs font-medium w-fit whitespace-nowrap shrink-0 [&>svg]:size-3 gap-1 [&>svg]:pointer-events-none focus-visible:border-ring focus-visible:ring-ring/50 focus-visible:ring-[3px] aria-invalid:ring-destructive/20 dark:aria-invalid:ring-destructive/40 aria-invalid:border-destructive transition-[color,box-shadow] overflow-hidden", - { - variants: { - variant: { - default: - "border-transparent bg-primary text-primary-foreground [a&]:hover:bg-primary/90", - secondary: - "border-transparent bg-secondary text-secondary-foreground [a&]:hover:bg-secondary/90", - destructive: - "border-transparent bg-destructive text-white [a&]:hover:bg-destructive/90 focus-visible:ring-destructive/20 dark:focus-visible:ring-destructive/40 dark:bg-destructive/60", - outline: - "text-foreground [a&]:hover:bg-accent [a&]:hover:text-accent-foreground", - }, - }, - defaultVariants: { - variant: "default", - }, - } -) - -function Badge({ - className, - variant, - asChild = false, - ...props -}: React.ComponentProps<"span"> & - VariantProps & { asChild?: boolean }) { - const Comp = asChild ? Slot : "span" - - return ( - - ) -} - -export { Badge, badgeVariants } diff --git a/src/components/ui/chat-bubble-demo.tsx b/src/components/ui/chat-bubble-demo.tsx deleted file mode 100644 index 032b63f8..00000000 --- a/src/components/ui/chat-bubble-demo.tsx +++ /dev/null @@ -1,103 +0,0 @@ -"use client" - -import { - ChatBubble, - ChatBubbleAvatar, - ChatBubbleMessage -} from "@/components/ui/chat-bubble" -import { Copy, RefreshCcw } from "lucide-react" - -const messages = [ - { - id: 1, - message: "Help me with my essay.", - sender: "user", - }, - { - id: 2, - message: "I can help you with that. What do you need help with?", - sender: "bot", - }, -] - -const actionIcons = [ - { icon: Copy, type: "Copy" }, - { icon: RefreshCcw, type: "Regenerate" }, -] - -export function ChatBubbleVariants() { - return ( -
- - - - I have a question about the library. - - - - - - - Sure, I'd be happy to help! - - -
- ) -} - -export function ChatBubbleAiLayout() { - return ( -
- {messages.map((message, index) => { - const variant = message.sender === "user" ? "sent" : "received" - return ( -
-
- -
- {message.message} - {message.sender === "bot" && ( -
- {actionIcons.map(({ icon: Icon, type }) => ( - - ))} -
- )} -
-
-
- ) - })} -
- ) -} - -export function ChatBubbleStates() { - return ( -
- - - - - - - - - Error processing request - - -
- ) -} \ No newline at end of file diff --git a/src/components/ui/chat-bubble.tsx b/src/components/ui/chat-bubble.tsx index 87718f4a..d269986b 100644 --- a/src/components/ui/chat-bubble.tsx +++ b/src/components/ui/chat-bubble.tsx @@ -8,14 +8,12 @@ import { MessageLoading } from "@/components/ui/message-loading"; interface ChatBubbleProps { variant?: "sent" | "received" - layout?: "default" | "ai" className?: string children: React.ReactNode } export function ChatBubble({ variant = "received", - layout = "default", // eslint-disable-line @typescript-eslint/no-unused-vars className, children, }: ChatBubbleProps) { diff --git a/src/components/ui/empty-chat-state.tsx b/src/components/ui/empty-chat-state.tsx index 30462d96..02a57fa1 100644 --- a/src/components/ui/empty-chat-state.tsx +++ b/src/components/ui/empty-chat-state.tsx @@ -184,13 +184,6 @@ export function EmptyChatState({
{file.name}
{formatFileSize(file.size)}
- {/* The remove button is commented out as the parent will manage the state now */} - {/* */} ))} diff --git a/src/components/ui/sidebar.tsx b/src/components/ui/sidebar.tsx deleted file mode 100644 index 3b481bca..00000000 --- a/src/components/ui/sidebar.tsx +++ /dev/null @@ -1,260 +0,0 @@ -"use client"; - -import { cn } from "@/lib/utils"; -import { ScrollArea } from "@/components/ui/scroll-area"; -import { motion } from "framer-motion"; -import { - ChevronsUpDown, - LogOut, - MessagesSquare, - Plus, - Settings, - UserCircle, -} from "lucide-react"; -import { Avatar, AvatarFallback } from "@/components/ui/avatar" -import { useState } from "react"; -import { Button } from "@/components/ui/button"; -import { - DropdownMenu, - DropdownMenuContent, - DropdownMenuItem, - DropdownMenuSeparator, - DropdownMenuTrigger, -} from "@/components/ui/dropdown-menu"; -import { Separator } from "@/components/ui/separator"; - -const sidebarVariants = { - open: { - width: "15rem", - }, - closed: { - width: "3.05rem", - }, -}; - -const contentVariants = { - open: { display: "block", opacity: 1 }, - closed: { display: "block", opacity: 1 }, -}; - -const variants = { - open: { - x: 0, - opacity: 1, - transition: { - x: { stiffness: 1000, velocity: -100 }, - }, - }, - closed: { - x: -20, - opacity: 0, - transition: { - x: { stiffness: 100 }, - }, - }, -}; - -const transitionProps = { - type: "tween", - ease: "easeOut", - duration: 0.2, - staggerChildren: 0.1, -}; - -const staggerVariants = { - open: { - transition: { staggerChildren: 0.03, delayChildren: 0.02 }, - }, -}; - -// Mock chat sessions data -const chatSessions = [ - { id: 1, title: "React Component Help", lastMessage: "How to create a sidebar?", timestamp: "2 min ago", isActive: true }, - { id: 2, title: "TypeScript Questions", lastMessage: "Interface vs Type", timestamp: "1 hour ago", isActive: false }, - { id: 3, title: "Next.js Setup", lastMessage: "Setting up shadcn/ui", timestamp: "3 hours ago", isActive: false }, - { id: 4, title: "Tailwind CSS", lastMessage: "Dark mode implementation", timestamp: "1 day ago", isActive: false }, - { id: 5, title: "Database Design", lastMessage: "Schema optimization", timestamp: "2 days ago", isActive: false }, -]; - -export function SessionNavBar() { - const [isCollapsed, setIsCollapsed] = useState(true); - - return ( - setIsCollapsed(false)} - onMouseLeave={() => setIsCollapsed(true)} - > - - -
- {/* Header */} -
-
- - - - - - - Preferences - - - New Chat - - - -
-
- - {/* Chat Sessions */} -
-
- -
- {/* New Chat Button */} - - - - - {/* Chat Sessions List */} - {chatSessions.map((session) => ( -
-
- - - {!isCollapsed && ( -
-

- {session.title} -

-

- {session.lastMessage} -

-

- {session.timestamp} -

-
- )} - -
-
- ))} -
- -
- - {/* Footer */} -
- - - - -
- - - U - - - - {!isCollapsed && ( - <> -

User

- - - )} - -
- - -
- - - U - - -
- - User - - - user@example.com - -
-
- - - Profile - - - Sign out - - - -
-
-
- - - - ); -} \ No newline at end of file diff --git a/src/components/ui/skeleton.tsx b/src/components/ui/skeleton.tsx deleted file mode 100644 index 32ea0ef7..00000000 --- a/src/components/ui/skeleton.tsx +++ /dev/null @@ -1,13 +0,0 @@ -import { cn } from "@/lib/utils" - -function Skeleton({ className, ...props }: React.ComponentProps<"div">) { - return ( -
- ) -} - -export { Skeleton } diff --git a/src/lib/api.ts b/src/lib/api.ts index dcf2bd17..7967d657 100644 --- a/src/lib/api.ts +++ b/src/lib/api.ts @@ -277,22 +277,6 @@ class ChatAPI { } } - async cleanupEmptySessions(): Promise<{ message: string; cleanup_count: number }> { - try { - const response = await fetch(`${API_BASE_URL}/sessions/cleanup`); - - if (!response.ok) { - const errorData = await response.json().catch(() => ({ error: 'Unknown error' })); - throw new Error(`Cleanup sessions error: ${errorData.error || response.statusText}`); - } - - return await response.json(); - } catch (error) { - console.error('Cleanup sessions failed:', error); - throw error; - } - } - async uploadFiles(sessionId: string, files: File[]): Promise<{ message: string; uploaded_files: {filename: string, stored_path: string}[]; @@ -339,59 +323,6 @@ class ChatAPI { } } - // Legacy upload function - can be removed if no longer needed - async uploadPDFs(sessionId: string, files: File[]): Promise<{ - message: string; - uploaded_files: any[]; - processing_results: any[]; - session_documents: any[]; - total_session_documents: number; - }> { - try { - // Test if files have content and show size info - let totalSize = 0; - for (const file of files) { - if (file.size === 0) { - throw new Error(`File ${file.name} is empty (0 bytes)`); - } - totalSize += file.size; - const sizeMB = (file.size / (1024 * 1024)).toFixed(2); - console.log(`📄 File ${file.name}: ${sizeMB}MB (${file.size} bytes), type: ${file.type}`); - } - - const totalSizeMB = (totalSize / (1024 * 1024)).toFixed(2); - console.log(`📄 Total upload size: ${totalSizeMB}MB`); - - if (totalSize > 50 * 1024 * 1024) { // 50MB limit - throw new Error(`Total file size ${totalSizeMB}MB exceeds 50MB limit`); - } - - const formData = new FormData(); - - // Use a generic field name 'file' that the backend expects - let i = 0; - for (const file of files) { - formData.append(`file_${i}`, file, file.name); - i++; - } - - const response = await fetch(`${API_BASE_URL}/sessions/${sessionId}/upload`, { - method: 'POST', - body: formData, - }); - - if (!response.ok) { - const errorData = await response.json().catch(() => ({ error: 'Unknown error' })); - throw new Error(`Upload error: ${errorData.error || response.statusText}`); - } - - return await response.json(); - } catch (error) { - console.error('PDF upload failed:', error); - throw error; - } - } - // Convert database message format to ChatMessage format convertDbMessage(dbMessage: Record): ChatMessage { return { diff --git a/src/utils/textNormalization.ts b/src/utils/textNormalization.ts index 90559abe..48c40c19 100644 --- a/src/utils/textNormalization.ts +++ b/src/utils/textNormalization.ts @@ -39,25 +39,3 @@ export function normalizeStreamingToken(currentText: string, newToken: string): return combined; } -/** - * Check if text contains excessive whitespace that needs normalization - */ -export function hasExcessiveWhitespace(text: string): boolean { - if (!text || typeof text !== 'string') { - return false; - } - - if (/\n{3,}/.test(text)) { - return true; - } - - if (/[ \t]{3,}/.test(text)) { - return true; - } - - if (/[ \t]*\n[ \t]*\n[ \t]*\n/.test(text)) { - return true; - } - - return false; -} From 36f361be3429f66921d7eb6b3fd5c84eaaa84789 Mon Sep 17 00:00:00 2001 From: PromptEngineer <134474669+PromtEngineer@users.noreply.github.com> Date: Tue, 10 Mar 2026 14:32:01 -0700 Subject: [PATCH 2/2] Fix streaming crash: debug logging crashed pipeline, frontend ignored error events MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause: _apply_index_embedding_model() wrote to logs/embedding_debug.log via open(), but the logs/ directory didn't exist. The FileNotFoundError crashed the entire streaming pipeline, which sent an SSE "error" event that the frontend had no handler for — leaving the UI stuck in loading. Backend fix: Replace fragile file-based debug logging with Python's standard logging module (logging.getLogger), which never crashes. Frontend fixes: - session-chat.tsx: Handle "error" SSE events by displaying the error message and stopping the loading state - api.ts: Close the stream reader on "error" events (not just "complete") so the caller unblocks Co-Authored-By: Claude Opus 4.6 --- rag_system/api_server.py | 23 ++++++----------------- src/components/ui/session-chat.tsx | 17 +++++++++++++++++ src/lib/api.ts | 2 +- 3 files changed, 24 insertions(+), 18 deletions(-) diff --git a/rag_system/api_server.py b/rag_system/api_server.py index af0f6c0d..cd984982 100644 --- a/rag_system/api_server.py +++ b/rag_system/api_server.py @@ -39,34 +39,23 @@ # Add helper near top after db & agent init # -------------- Helper ---------------- +_embed_logger = logging.getLogger("embedding_debug") + def _apply_index_embedding_model(idx_ids): """Ensure retrieval pipeline uses the embedding model stored with the first index.""" - debug_info = f"🔧 _apply_index_embedding_model called with idx_ids: {idx_ids}\n" - if not idx_ids: - debug_info += "⚠️ No index IDs provided\n" - with open("logs/embedding_debug.log", "a") as f: - f.write(debug_info) + _embed_logger.debug("No index IDs provided") return try: idx = db.get_index(idx_ids[0]) - debug_info += f"🔧 Retrieved index: {idx.get('id')} with metadata: {idx.get('metadata', {})}\n" model = (idx.get("metadata") or {}).get("embedding_model") - debug_info += f"🔧 Embedding model from metadata: {model}\n" + _embed_logger.debug("Index %s embedding_model=%s", idx_ids[0], model) if model: rp = RAG_AGENT.retrieval_pipeline - current_model = rp.config.get("embedding_model_name") - debug_info += f"🔧 Current embedding model: {current_model}\n" rp.update_embedding_model(model) - debug_info += f"🔧 Updated embedding model to: {model}\n" - else: - debug_info += "⚠️ No embedding model found in metadata\n" + _embed_logger.debug("Updated embedding model to: %s", model) except Exception as e: - debug_info += f"⚠️ Could not apply index embedding model: {e}\n" - - # Write debug info to file - with open("logs/embedding_debug.log", "a") as f: - f.write(debug_info) + _embed_logger.warning("Could not apply index embedding model: %s", e) def _get_table_name_for_session(session_id): """Get the correct vector table name for a session by looking up its linked indexes.""" diff --git a/src/components/ui/session-chat.tsx b/src/components/ui/session-chat.tsx index c14aa1e7..0ca60be2 100644 --- a/src/components/ui/session-chat.tsx +++ b/src/components/ui/session-chat.tsx @@ -442,6 +442,23 @@ export const SessionChat = forwardRef(({ return { ...m, content: { steps }, metadata: { message_type: 'complete' } }; } + if (evt.type === 'error') { + // Server sent an error during the stream + const errMsg = typeof evt.data === 'object' + ? evt.data.error || JSON.stringify(evt.data) + : String(evt.data); + console.error('Stream error from server:', errMsg); + // Show the error in the final-answer step + const finalIdx = steps.findIndex(s => s.key === 'final' || s.key === 'direct'); + if (finalIdx !== -1) { + steps[finalIdx].status = 'done'; + steps[finalIdx].details = `⚠️ Error: ${errMsg}`; + } + // Mark any active steps as done + steps.forEach(s => { if (s.status === 'active' || s.status === 'pending') s.status = 'done'; }); + setIsLoading(false); + return { ...m, content: { steps }, metadata: { message_type: 'complete' } }; + } if (evt.type === 'direct_answer') { const stepsDir: Step[] = [ { key: 'direct', label: 'Answering directly', status: 'active' as const, details: '' } diff --git a/src/lib/api.ts b/src/lib/api.ts index 7967d657..c3323da2 100644 --- a/src/lib/api.ts +++ b/src/lib/api.ts @@ -547,7 +547,7 @@ class ChatAPI { try { const evt = JSON.parse(jsonStr); onEvent(evt); - if (evt.type === 'complete') { + if (evt.type === 'complete' || evt.type === 'error') { // Gracefully close the stream so the caller unblocks try { await reader.cancel(); } catch {} streamClosed = true;