') and value.endswith(''):
try:
json_str = value[5:-6] # Remove andtags return json.loads(json_str) except: return {} elif value.startswith('{') and value.endswith('}'): try: return json.loads(value) except: return {} else: return {"value": value} elif isinstance(value, (list, tuple)): return {"items": list(value)} elif isinstance(value, (int, float, bool)): return {"value": value} else: return {"data": str(value)} def dict_to_html_json(value: Any) -> str: """Convert a value to HTML-formatted JSON for display.""" try: safe_dict = ensure_dict_safe(value) return f"
{json.dumps(safe_dict, indent=2)}"
except Exception:
return "{}"
from .themes import get_theme, get_custom_css
from .components import (
create_header, create_file_upload_section, create_search_interface,
create_results_display, create_document_management, create_system_status,
create_analytics_dashboard, format_document_list, format_search_results,
create_analytics_charts, format_system_overview, create_error_display,
create_success_display, create_loading_display
)
from .conversation_components import (
create_chat_interface, create_chat_history_display, create_chat_input, create_conversation_controls,
create_conversation_state, create_conversation_info_panel, create_typing_indicator,
create_conversation_analytics, update_conversation_analytics, update_chat_history,
format_conversation_response, extract_conversation_stats, get_conversation_css,
get_conversation_javascript, create_chat_message
)
from .utils import (
save_uploaded_files, validate_file_types, cleanup_temp_files,
generate_session_id, format_file_size, parse_search_filters
)
sys.path.append(str(Path(__file__).parent.parent))
from src.rag_system import RAGSystem, EnhancedRAGSystem
from src.error_handler import RAGError
from src.logging_utils import setup_logging, get_safe_logger, sanitize_log_message
class RAGInterface:
"""Main interface class for the RAG system."""
def __init__(self, config_path: str = None, use_enhanced_rag: bool = True):
"""Initialize the RAG interface."""
self.rag_system: Optional[Union[RAGSystem, EnhancedRAGSystem]] = None
self.config_path = config_path or "config.yaml"
self.use_enhanced_rag = use_enhanced_rag
self.active_sessions: Dict[str, Dict[str, Any]] = {}
self._initialization_lock = threading.Lock()
self._initialized = False
self._conversation_enabled = False
# Setup logging with Unicode safety
self.logger = get_safe_logger(__name__)
# Initialize RAG system
self._initialize_rag_system()
def _initialize_rag_system(self) -> None:
"""Initialize the RAG system."""
try:
with self._initialization_lock:
if not self._initialized:
if self.use_enhanced_rag:
print("Initializing Enhanced RAG system with conversation capabilities...")
try:
self.rag_system = EnhancedRAGSystem(config_path=self.config_path)
self._conversation_enabled = hasattr(self.rag_system, 'process_conversation')
print(f"Enhanced RAG system initialized. Conversation enabled: {self._conversation_enabled}")
except Exception as e:
print(f"Failed to initialize Enhanced RAG system: {e}")
print("Falling back to standard RAG system...")
self.rag_system = RAGSystem(config_path=self.config_path)
self._conversation_enabled = False
else:
print("Initializing standard RAG system...")
self.rag_system = RAGSystem(config_path=self.config_path)
self._conversation_enabled = False
# Warm up the system
warmup_result = self.rag_system.warmup()
if warmup_result.get("success"):
system_type = "Enhanced RAG" if self._conversation_enabled else "Standard RAG"
print(f"{system_type} system initialized and warmed up successfully")
self._initialized = True
else:
print(f"RAG system warmup failed: {warmup_result.get('error', {}).get('message')}")
except Exception as e:
print(f"Failed to initialize RAG system: {e}")
self.rag_system = None
def get_system_status(self) -> str:
"""Get current system status HTML."""
if not self.rag_system or not self._initialized:
return create_error_display("System not initialized. Please check configuration and restart.")
try:
stats_response = self.rag_system.get_system_stats()
if not stats_response.get("success"):
return create_error_display(f"Failed to get system status: {stats_response.get('error', {}).get('message')}")
stats = stats_response["data"]
status_info = stats.get("status", {})
if status_info.get("ready"):
status_message = f"System ready - {status_info.get('documents_indexed', 0)} documents indexed"
return create_success_display(status_message)
else:
return create_error_display("System not ready")
except Exception as e:
return create_error_display(f"Error getting system status: {str(e)}")
def process_documents(
self,
files: List[gr.File],
session_id: str,
progress=gr.Progress()
) -> Tuple[str, str, bool, str]:
"""Process uploaded documents."""
if not files:
return (
create_error_display("No files uploaded"),
create_error_display("Please select files to upload"),
False, # upload button disabled
"No documents uploaded yet."
)
if not self.rag_system or not self._initialized:
return (
create_error_display("System not initialized"),
create_error_display("Please restart the application"),
False,
"No documents uploaded yet."
)
try:
self.logger.info(f"Starting document upload process - {len(files)} files received")
# Validate file types
allowed_extensions = [".pdf", ".docx", ".txt"]
valid_files, validation_errors = validate_file_types(files, allowed_extensions)
if validation_errors:
self.logger.warning(f"File validation errors: {validation_errors}")
error_html = create_error_display("\\n".join(validation_errors))
return error_html, error_html, len(valid_files) > 0, self.get_document_list()
self.logger.info(f"File validation passed - {len(valid_files)} valid files")
# Save uploaded files
progress(0.1, desc="Saving uploaded files...")
self.logger.info("Saving uploaded files to temporary directory...")
saved_files = save_uploaded_files(valid_files)
for file_path, original_name in saved_files:
file_size = Path(file_path).stat().st_size / (1024 * 1024) # Size in MB
self.logger.info(f"Saved file: {original_name} ({file_size:.2f} MB) -> {file_path}")
if not saved_files:
return (
create_error_display("No valid files to process"),
create_error_display("Please check your files and try again"),
False,
self.get_document_list()
)
# Process each file with timeout
processed_count = 0
total_files = len(saved_files)
results = []
timeout_seconds = 600 # 10 minutes
def process_single_file(file_path, original_name, session_id):
"""Process a single file - to be run with timeout."""
self.logger.info(f"Processing file: {original_name}")
start_time = time.time()
result = self.rag_system.add_document(
file_path=file_path,
filename=original_name,
user_session=session_id
)
processing_time = time.time() - start_time
self.logger.info(f"File processing completed: {original_name} (took {processing_time:.2f}s)")
return result
self.logger.info(f"Starting processing of {total_files} files with {timeout_seconds//60}-minute timeout per file")
with ThreadPoolExecutor(max_workers=1) as executor:
for i, (file_path, original_name) in enumerate(saved_files):
progress((i + 1) / total_files * 0.8 + 0.1, desc=f"Processing {original_name}...")
self.logger.info(f"Processing file {i+1}/{total_files}: {original_name}")
try:
# Submit the task with timeout
future = executor.submit(process_single_file, file_path, original_name, session_id)
result = future.result(timeout=timeout_seconds)
if result.get("success"):
processed_count += 1
chunks_created = result['data']['chunks_created']
# Log detailed success info
self.logger.info(f"SUCCESS: {original_name} - {chunks_created} chunks created")
# Log sample chunk info if available
if 'sample_chunks' in result['data']:
sample_chunks = result['data']['sample_chunks']
self.logger.info(f"Sample chunks from {original_name}:")
for idx, chunk in enumerate(sample_chunks[:3]): # Show first 3 chunks
chunk_preview = chunk['content'][:100] + "..." if len(chunk['content']) > 100 else chunk['content']
self.logger.info(f" Chunk {idx}: {chunk_preview}")
results.append(f"✅ {original_name}: {chunks_created} chunks created")
else:
error_msg = result.get("error", {}).get("message", "Unknown error")
self.logger.error(f"FAILED: {original_name} - {error_msg}")
results.append(f"❌ {original_name}: {error_msg}")
except FutureTimeoutError:
self.logger.error(f"TIMEOUT: {original_name} exceeded {timeout_seconds//60} minute limit")
results.append(f"⏰ {original_name}: Processing timed out after {timeout_seconds//60} minutes")
future.cancel() # Cancel the task if possible
except Exception as e:
self.logger.error(f"EXCEPTION: {original_name} - {str(e)}")
results.append(f"❌ {original_name}: {str(e)}")
progress(1.0, desc="Cleaning up...")
self.logger.info("Cleaning up temporary files...")
# Clean up temporary files
cleanup_temp_files([fp for fp, _ in saved_files])
# Log final summary
total_processing_time = time.time() - time.time() # This will be updated properly
self.logger.info(f"Document upload process completed:")
self.logger.info(f" - Total files: {total_files}")
self.logger.info(f" - Successfully processed: {processed_count}")
self.logger.info(f" - Failed: {total_files - processed_count}")
self.logger.info(f" - Success rate: {(processed_count/total_files*100):.1f}%")
# Create result message
if processed_count == total_files:
self.logger.info(f"[SUCCESS] ALL DOCUMENTS PROCESSED SUCCESSFULLY ({processed_count}/{total_files})")
status_html = create_success_display(
f"Successfully processed {processed_count} documents:\\n" + "\\n".join(results)
)
upload_status = create_success_display(f"All {processed_count} documents processed successfully!")
elif processed_count > 0:
self.logger.warning(f"[PARTIAL] PARTIAL SUCCESS ({processed_count}/{total_files} documents processed)")
status_html = f"""
Ask questions about your documents or start a conversation. I can help you find information, explain concepts, and provide detailed answers based on your uploaded documents.
Ask questions about your documents or start a conversation. I can help you find information, explain concepts, and provide detailed answers based on your uploaded documents.
🎯 What I can do for you:
💡 New here? Start by uploading documents in the "📁 Document Upload" tab, then come back to chat!