From dd2eda6df881d56a26b1207c78066f39f6eb7ab0 Mon Sep 17 00:00:00 2001 From: Madhu Date: Wed, 4 Dec 2024 03:01:26 +0530 Subject: [PATCH] new project - llm hybrid search + RAG claude (unfinished) --- .../llm_app_hybrid_RAG_claude/.gitignore | 52 +++ .../llm_app_hybrid_RAG_claude/README.md | 99 +++++ .../llm_app_hybrid_RAG_claude/chainlit.yaml | 70 ++++ .../llm_app_hybrid_RAG_claude/main.py | 348 ++++++++++++++++++ .../requirements.txt | 12 + 5 files changed, 581 insertions(+) create mode 100644 rag_tutorials/llm_app_hybrid_RAG_claude/.gitignore create mode 100644 rag_tutorials/llm_app_hybrid_RAG_claude/README.md create mode 100644 rag_tutorials/llm_app_hybrid_RAG_claude/chainlit.yaml create mode 100644 rag_tutorials/llm_app_hybrid_RAG_claude/main.py create mode 100644 rag_tutorials/llm_app_hybrid_RAG_claude/requirements.txt diff --git a/rag_tutorials/llm_app_hybrid_RAG_claude/.gitignore b/rag_tutorials/llm_app_hybrid_RAG_claude/.gitignore new file mode 100644 index 0000000..c7e7a21 --- /dev/null +++ b/rag_tutorials/llm_app_hybrid_RAG_claude/.gitignore @@ -0,0 +1,52 @@ +# Python virtual environment +venv/ +myrag/ +env/ +.env + +# Python cache files +__pycache__/ +*.py[cod] +*$py.class + +# Distribution / packaging +dist/ +build/ +*.egg-info/ + +# IDE specific files +.idea/ +.vscode/ +*.swp +*.swo + +# Local development files +*.log +.DS_Store + +# Test files +test.txt +test.pdf + +# Database files +*.db +*.sqlite3 + +# Environment variables +.env +.env.local +.env.*.local + +# Chainlit specific +.chainlit/ +chainlit.md + +# Temporary files +tmp/ +temp/ + +.files +.flashrank_cache +.env + +raglite.sqlite \ No newline at end of file diff --git a/rag_tutorials/llm_app_hybrid_RAG_claude/README.md b/rag_tutorials/llm_app_hybrid_RAG_claude/README.md new file mode 100644 index 0000000..726aef3 --- /dev/null +++ b/rag_tutorials/llm_app_hybrid_RAG_claude/README.md @@ -0,0 +1,99 @@ +# Hybrid RAG Claude Chat 🤖 + +A powerful document Q&A application that combines Hybrid Search (RAG) with Claude's general knowledge. This is built on the RAGLite framework and Chainlit for the UI. + +## Features + +- **Hybrid Question Answering** + - RAG-based answers for document-specific queries + - Fallback to Claude for general knowledge questions + - Seamless switching between modes + +- **Document Processing**: + - PDF document upload and processing + - Automatic text chunking and embedding + - Hybrid search combining semantic and keyword matching + - Reranking for better context selection + +- **Interactive Chat Interface**: + - Real-time streaming responses + - Chat history preservation + - Error handling with retry options + - File upload validation + +- **Multi-Model Integration**: + - Claude for text generation + - OpenAI for embeddings + - Cohere for reranking (tried using the new Cohere 3.5 reranker) + +## Prerequisites + +You'll need the following API keys and database setup: + +1. **Database**: Create a free PostgreSQL database at [Neon](https://neon.tech): + - Sign up/Login at Neon + - Create a new project + - Copy the connection string (looks like: `postgresql://user:pass@ep-xyz.region.aws.neon.tech/dbname`) + +2. **API Keys**: + - [OpenAI API key](https://platform.openai.com/api-keys) for embeddings + - [Anthropic API key](https://console.anthropic.com/settings/keys) for Claude + - [Cohere API key](https://dashboard.cohere.com/api-keys) for reranking + +## Installation + +1. **Clone the Repository**: + ```bash + git clone + cd rag_tutorials/llm_app_hybrid_RAG_claude + ``` + +2. **Install Dependencies**: + ```bash + pip install -r requirements.txt + ``` + +3. **Install spaCy Model**: + ```bash + pip install https://github.com/explosion/spacy-models/releases/download/xx_sent_ud_sm-3.7.0/xx_sent_ud_sm-3.7.0-py3-none-any.whl + ``` + +4. **Run the Application**: + ```bash + chainlit run main.py + ``` + +## Usage + +1. Start the application +2. When prompted, enter your: + - OpenAI API key + - Anthropic API key + - Cohere API key + - Neon PostgreSQL URL +3. Upload PDF documents +4. Start asking questions! + - Document-specific questions will use RAG + - General questions will use Claude directly + +## Database Options + +The application supports multiple database backends: + +- **PostgreSQL** (Recommended): + - Create a free serverless PostgreSQL database at [Neon](https://neon.tech) + - Get instant provisioning and scale-to-zero capability + - Connection string format: `postgresql://user:pass@ep-xyz.region.aws.neon.tech/dbname` + +- **MySQL**: + ``` + mysql://user:pass@host:port/db + ``` +- **SQLite** (Local development): + ``` + sqlite:///path/to/db.sqlite + ``` + +## Contributing + +Contributions are welcome! Please feel free to submit a Pull Request. diff --git a/rag_tutorials/llm_app_hybrid_RAG_claude/chainlit.yaml b/rag_tutorials/llm_app_hybrid_RAG_claude/chainlit.yaml new file mode 100644 index 0000000..8135e83 --- /dev/null +++ b/rag_tutorials/llm_app_hybrid_RAG_claude/chainlit.yaml @@ -0,0 +1,70 @@ +chainlit_server: + cors: + allowed_origins: ["*"] + +project: + # Enable user environment variables + user_env: + - OPENAI_API_KEY + - ANTHROPIC_API_KEY + - COHERE_API_KEY + - DB_URL + +ui: + name: "LLM App with Hybrid Search and RAG - Claude" + description: "by unwind ai" + hide_cot: false + default_collapse_content: true + default_expand_messages: true + input_box: + width: "100%" + max_width: "600px" + placeholder: "Enter your message..." + layout: + mode: "wide" + centered: true + show_sidebar: true + sidebar_position: "left" + +# Configure the environment variables UI +user_env: + OPENAI_API_KEY: + type: "secret" # This will mask the input + description: "Your OpenAI API key for embeddings" + placeholder: "sk-..." + required: true + + ANTHROPIC_API_KEY: + type: "secret" + description: "Your Anthropic API key for Claude" + placeholder: "sk-ant-..." + required: true + + COHERE_API_KEY: + type: "secret" + description: "Your Cohere API key for reranking" + placeholder: "co-..." + required: true + + DB_URL: + type: "string" + description: "Your PostgreSQL database connection URL" + placeholder: "postgresql://user:pass@host:port/db" + required: true + +message_display: + timeout: 180 + +features: + prompt_playground: false + secure_credentials: true + +theme: + light: + primary: "#2563eb" + background: "#ffffff" + sidebar: "#f7f7f8" + dark: + primary: "#2563eb" + background: "#1a1a1a" + sidebar: "#1e1e1e" \ No newline at end of file diff --git a/rag_tutorials/llm_app_hybrid_RAG_claude/main.py b/rag_tutorials/llm_app_hybrid_RAG_claude/main.py new file mode 100644 index 0000000..d4ff142 --- /dev/null +++ b/rag_tutorials/llm_app_hybrid_RAG_claude/main.py @@ -0,0 +1,348 @@ +import os +import logging +import chainlit as cl +from raglite import RAGLiteConfig, insert_document, hybrid_search, retrieve_chunks, rerank_chunks, rag +from rerankers import Reranker +from typing import List +from pathlib import Path +from chainlit.action import Action +from chainlit.input_widget import TextInput +from chainlit import AskUserMessage +import anthropic + +# Configure logging +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +# Initialize global config variable +my_config = None + +# Define RAG system prompt +RAG_SYSTEM_PROMPT = """ +You are a friendly and knowledgeable assistant that provides complete and insightful answers. +Answer the user's question using only the context below. +When responding, you MUST NOT reference the existence of the context, directly or indirectly. +Instead, you MUST treat the context as if its contents are entirely part of your working memory. +""".strip() + +def initialize_config(user_env: dict) -> RAGLiteConfig: + """Initialize RAGLite configuration with user-provided keys.""" + return RAGLiteConfig( + db_url=user_env["DB_URL"], + llm="claude-3-opus-20240229", + embedder="text-embedding-3-large", + embedder_normalize=True, + chunk_max_size=2000, + embedder_sentence_window_size=2, + reranker=Reranker( + "cohere", + api_key=user_env["COHERE_API_KEY"], + lang="en" + ) + ) + +def process_document(file_path: str) -> None: + """Process and embed a document into the database.""" + logger.info(f"Starting to process document: {file_path}") + try: + import time + start_time = time.time() + + # Insert document into PostgreSQL database + insert_document(Path(file_path), config=my_config) + + processing_time = time.time() - start_time + logger.info(f"Document processed and embedded in {processing_time:.2f} seconds") + + except Exception as e: + logger.error(f"Error processing document: {str(e)}") + raise + +def perform_search(query: str) -> List[dict]: + """Perform hybrid search and reranking on the query.""" + logger.info(f"Performing hybrid search for: {query}") + try: + # First try hybrid search in the database + chunk_ids, scores = hybrid_search(query, num_results=10, config=my_config) + logger.debug(f"Found {len(chunk_ids)} chunks with scores: {scores}") + + if not chunk_ids: + logger.info("No relevant chunks found in database") + return [] + + # Retrieve and rerank chunks + chunks = retrieve_chunks(chunk_ids, config=my_config) + reranked_chunks = rerank_chunks(query, chunks, config=my_config) + + return reranked_chunks + except Exception as e: + logger.error(f"Search error: {str(e)}") + return [] + +@cl.on_chat_start +async def start() -> None: + try: + logger.info("Chat session started") + cl.user_session.set("chat_history", []) + + # Helper function to validate and get API key - so that if the user enters the wrong key, they can try again + async def get_valid_api_key(key_type: str, validation_prefix: tuple) -> str: + while True: + key_response = await cl.AskUserMessage( + content=f"Please enter your {key_type} API key:", + timeout=180 + ).send() + + if not key_response or 'output' not in key_response: + await cl.Message(content=f"❌ {key_type} API key is required").send() + continue + + key_value = key_response['output'].strip() + if not any(key_value.startswith(prefix) for prefix in validation_prefix): + await cl.Message( + content=f"❌ Invalid {key_type} API key format. Please try again." + ).send() + continue + + return key_value + + # this is the helper function to validate and get DB URL, so that if the user enters the wrong URL, they can try again + async def get_valid_db_url() -> str: + valid_db_prefixes = ('postgresql://', 'mysql://', 'sqlite:///') + while True: + db_url_response = await cl.AskUserMessage( + content="Please enter your Database URL:\nSupported formats:\n- PostgreSQL: postgresql://user:pass@host:port/db\n- MySQL: mysql://user:pass@host:port/db\n- SQLite: sqlite:///path/to/db.sqlite", + timeout=180 + ).send() + + if not db_url_response or 'output' not in db_url_response: + await cl.Message(content="❌ Database URL is required").send() + continue + + db_url_value = db_url_response['output'].strip() + if not any(db_url_value.startswith(prefix) for prefix in valid_db_prefixes): + await cl.Message( + content="❌ Invalid database URL format. Must start with one of:\n- postgresql://\n- mysql://\n- sqlite://\nPlease try again." + ).send() + continue + + return db_url_value + + # Get and validate API keys with retry + openai_key = await get_valid_api_key("OpenAI", ("sk-", "sk-proj-")) + anthropic_key = await get_valid_api_key("Anthropic", ("sk-ant-",)) + cohere_key = await get_valid_api_key("Cohere", ("",)) # Cohere keys don't have a specific prefix + + # Get and validate DB URL with retry + db_url = await get_valid_db_url() + + # Store validated values in user_env + user_env = { + "OPENAI_API_KEY": openai_key, + "ANTHROPIC_API_KEY": anthropic_key, + "COHERE_API_KEY": cohere_key, + "DB_URL": db_url + } + + # Store in user session + cl.user_session.set("env", user_env) + logger.info("API keys stored in user session") + + # Initialize RAGLite config with retry + while True: + try: + global my_config + my_config = initialize_config(user_env) + await cl.Message(content="✅ Successfully configured with your API keys!").send() + break + except Exception as e: + logger.error(f"Configuration error: {str(e)}") + await cl.Message(content=f"❌ Error configuring with provided keys: {str(e)}\nPlease check your credentials and try again.").send() + # Retry getting all credentials + openai_key = await get_valid_api_key("OpenAI", ("sk-", "sk-proj-")) + anthropic_key = await get_valid_api_key("Anthropic", ("sk-ant-",)) + cohere_key = await get_valid_api_key("Cohere", ("",)) + db_url = await get_valid_db_url() + user_env.update({ + "OPENAI_API_KEY": openai_key, + "ANTHROPIC_API_KEY": anthropic_key, + "COHERE_API_KEY": cohere_key, + "DB_URL": db_url + }) + cl.user_session.set("env", user_env) + + async def get_valid_documents() -> List[cl.File]: + while True: + files = await cl.AskFileMessage( + content="Please upload one or more PDF documents to begin!", + accept=["application/pdf"], + max_size_mb=20, + max_files=5 + ).send() + + if not files: + await cl.Message(content="❌ No files were uploaded. Please try again.").send() + continue + + return files + + # Process documents with retry for each file + async def process_documents(files: List[cl.File]) -> bool: + """Process uploaded documents with retry functionality for failed files.""" + processed_files = set() + files_to_process = files.copy() + + while files_to_process: + current_file = files_to_process.pop(0) + + if current_file.name in processed_files: + continue + + logger.info(f"Processing file: {current_file.name}") + step = cl.Step(name=f"Processing {current_file.name}...") + async with step: # Use step as context manager + try: + process_document(current_file.path) + processed_files.add(current_file.name) + await cl.Message( + content=f"✅ The Document '{current_file.name}' is processed successfully." + ).send() + except Exception as e: + logger.error(f"Failed to process '{current_file.name}': {str(e)}") + error_message = f"❌ Failed to process '{current_file.name}': {str(e)}" + + # Ask user if they want to retry this file + retry = await cl.AskUserMessage( + content=f"{error_message}\nWould you like to try uploading this file again? (yes/no)", + timeout=180 + ).send() + + if retry and retry['output'].lower().strip() == 'yes': + new_file = await cl.AskFileMessage( + content=f"Please upload '{current_file.name}' again:", + accept=["application/pdf"], + max_size_mb=20, + max_files=1 + ).send() + + if new_file: + files_to_process.append(new_file[0]) + + # Ensure step is completed + await step.end() + + if not processed_files: + await cl.Message(content="❌ No documents were processed successfully. Please try uploading new documents.").send() + return False + + # Send final success message and return to chat + final_msg = cl.Message(content="✅ Document processing completed. You can now ask questions!") + await final_msg.send() + return True + + # Main document processing loop + while True: + files = await get_valid_documents() + if await process_documents(files): + break + + # Ask if user wants to try uploading different documents + retry = await cl.AskUserMessage( + content="Would you like to try uploading different documents? (yes/no)", + timeout=180 + ).send() + + if not retry or retry['output'].lower().strip() != 'yes': + await cl.Message(content="❌ Stopping due to document processing failures.").send() + return + + except Exception as e: + logger.error(f"Error in chat start: {str(e)}") + await cl.Message(content=f"Error initializing chat: {str(e)}").send() + +@cl.on_message +async def message_handler(message: cl.Message) -> None: + try: + msg = cl.Message(content="Thinking...") + await msg.send() + + query = message.content.strip() + chat_history = cl.user_session.get("chat_history", []) + + # Search for relevant chunks using global config + reranked_chunks = perform_search(query) + + if reranked_chunks: + logger.info("Using RAG for response generation") + try: + # Convert chat history to proper format for RAG + formatted_messages = [] + for user_msg, assistant_msg in chat_history: + formatted_messages.append({"role": "user", "content": user_msg}) + formatted_messages.append({"role": "assistant", "content": assistant_msg}) + + response_stream = rag( + prompt=query, + system_prompt=RAG_SYSTEM_PROMPT, + search=hybrid_search, + messages=formatted_messages, + max_contexts=5, + config=my_config + ) + + full_response = "" + for chunk in response_stream: + full_response += chunk + await msg.stream_token(chunk) + + await msg.send() + + except Exception as e: + logger.error(f"RAG error: {str(e)}") + # If RAG fails, fall back to general Claude + await handle_fallback(query, msg) + return + + else: + logger.info("No relevant chunks found, falling back to general Claude response") + await handle_fallback(query, msg) + return + + # Update chat history + chat_history.append((query, full_response)) + cl.user_session.set("chat_history", chat_history) + + except Exception as e: + error_msg = f"Error processing your question: {str(e)}" + logger.error(error_msg) + await msg.send(content=error_msg) # Use send instead of update + +async def handle_fallback(query: str, msg: cl.Message) -> None: + """Handle fallback to Claude when RAG is not available or fails.""" + try: + user_env = cl.user_session.get("env") + client = anthropic.Anthropic(api_key=user_env["ANTHROPIC_API_KEY"]) + + response = client.messages.create( + model="claude-3-5-sonnet-20241022", + max_tokens=1024, + messages=[ + {"role": "user", "content": query} + ] + ) + + full_response = response.content[0].text + await msg.send(content=full_response) + + # Update chat history + chat_history = cl.user_session.get("chat_history", []) + chat_history.append((query, full_response)) + cl.user_session.set("chat_history", chat_history) + + except Exception as e: + error_msg = f"Fallback error: {str(e)}" + logger.error(error_msg) + await msg.send(content=error_msg) + +if __name__ == "__main__": + cl.run() diff --git a/rag_tutorials/llm_app_hybrid_RAG_claude/requirements.txt b/rag_tutorials/llm_app_hybrid_RAG_claude/requirements.txt new file mode 100644 index 0000000..f5c403a --- /dev/null +++ b/rag_tutorials/llm_app_hybrid_RAG_claude/requirements.txt @@ -0,0 +1,12 @@ +chainlit==1.3.2 +anthropic==0.8.1 +raglite==0.2.1 +pydantic==2.10.1 +sqlalchemy>=2.0.0 +psycopg2-binary>=2.9.9 +openai>=1.0.0 +cohere>=4.37 +pypdf>=3.0.0 +python-dotenv>=1.0.0 +rerankers==0.6.0 +spacy>=3.7.0