Upload 4 files

model/contextual_response/constants.py

@@ -0,0 +1,8 @@
+# Model configuration
+EMBEDDING_MODEL_NAME = "BAAI/bge-small-en-v1.5"
+GROQ_MODEL_NAME = "llama3-8b-8192"
+
+# Document chunking configuration
+CHUNK_SIZE = 1000
+CHUNK_OVERLAP = 200
+SEPARATORS = ["\n\n", "\n", " ", ""]

model/contextual_response/contextual_response.py

@@ -0,0 +1,238 @@
+import asyncio
+import os
+import logging
+from typing import List, Optional
+from contextlib import contextmanager
+
+import numpy as np
+from dotenv import load_dotenv
+from langchain_groq import ChatGroq
+from langchain_community.embeddings.huggingface import HuggingFaceEmbeddings
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+from langchain.prompts import ChatPromptTemplate
+from langchain_core.runnables import RunnablePassthrough
+from langchain_core.output_parsers import StrOutputParser
+from langchain_core.documents import Document
+from sqlalchemy import select
+from .constants import (
+    EMBEDDING_MODEL_NAME,
+    GROQ_MODEL_NAME,
+    CHUNK_SIZE,
+    CHUNK_OVERLAP,
+    SEPARATORS
+)
+from .system_prompt import RAG_SYSTEM_PROMPT
+from models.db.db_setup import DocumentEmbedding, SessionLocal
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+# Load environment variables
+load_dotenv()
+
+# Configure environment
+os.environ["TOKENIZERS_PARALLELISM"] = "false"
+GROQ_API_KEY = os.getenv("GROQ_API_KEY")
+if not GROQ_API_KEY:
+    logger.error("GROQ_API_KEY environment variable is not set")
+    raise ValueError("GROQ_API_KEY environment variable is not set.")
+os.environ["GROQ_API_KEY"] = GROQ_API_KEY
+
+@contextmanager
+def get_db_session():
+    """Context manager for database sessions."""
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+class RAGManager:
+    def __init__(self):
+        self._embed_model: Optional[HuggingFaceEmbeddings] = None
+        self._rag_llm: Optional[ChatGroq] = None
+        self._initialize_embed_model()
+        self._rag_llm: Optional[ChatGroq] = None
+        logger.info("RAGManager initialized")
+
+    def _initialize_embed_model(self):
+        """Initialize the embedding model."""
+        try:
+            logger.debug("Initializing embedding model")
+            self._embed_model = HuggingFaceEmbeddings(model_name=EMBEDDING_MODEL_NAME)
+            logger.info("Embedding model initialized successfully")
+        except Exception as e:
+            logger.error(f"Error initializing embedding model: {str(e)}")
+            raise
+
+    @property
+    def embed_model(self):
+        """Get or create embedding model."""
+        if self._embed_model is None:
+            self._initialize_embed_model()
+        return self._embed_model
+
+    @property
+    def rag_llm(self):
+        """Get or create RAG LLM model."""
+        if self._rag_llm is None:
+            logger.debug("Initializing RAG LLM model")
+            self._rag_llm = ChatGroq(model=GROQ_MODEL_NAME)
+            logger.info("RAG LLM model initialized successfully")
+        return self._rag_llm
+
+    async def get_relevant_documents(self, query: str, k: int = 4) -> List[Document]:
+        """Get relevant documents using vector similarity search."""
+        with get_db_session() as db:
+            # Get query embedding
+            query_embedding = self.embed_model.embed_query(query)
+
+            # Get all documents and calculate similarities
+            stmt = select(DocumentEmbedding)
+            results = db.execute(stmt).scalars().all()
+
+            if not results:
+                logger.warning("No documents found in database")
+                return []
+
+            # Convert embeddings to numpy array for faster computation
+            embeddings = np.array([doc.embedding for doc in results])
+            query_embedding = np.array(query_embedding)
+
+            # Calculate similarities in batch
+            similarities = np.dot(embeddings, query_embedding) / (
+                np.linalg.norm(embeddings, axis=1) * np.linalg.norm(query_embedding)
+            )
+
+            # Get indices of top k results
+            top_k_indices = np.argsort(similarities)[-k:][::-1]
+
+            # Convert top k results to Documents
+            return [
+                Document(
+                    page_content=results[idx].content,
+                    metadata=results[idx].doc_metadata or {}
+                )
+                for idx in top_k_indices
+            ]
+
+    def format_docs(self, docs: List[Document]) -> str:
+        """Format documents with error handling."""
+        try:
+            if not docs:
+                logger.debug("No documents to format")
+                return ""
+            formatted = "\n\n".join(doc.page_content for doc in docs)
+            logger.debug(f"Formatted {len(docs)} documents")
+            return formatted
+        except Exception as e:
+            logger.error(f"Error formatting documents: {str(e)}")
+            return ""
+
+    async def process_documents(self, documents: List[Document]) -> bool:
+        """Process documents and store embeddings in PostgreSQL."""
+        try:
+            if not documents:
+                logger.warning("No documents provided")
+                return False
+
+            # Split documents
+            text_splitter = RecursiveCharacterTextSplitter(
+                separators=SEPARATORS,
+                chunk_size=CHUNK_SIZE,
+                chunk_overlap=CHUNK_OVERLAP,
+                length_function=len,
+                is_separator_regex=False,
+            )
+
+            split_docs = []
+            for doc in documents:
+                try:
+                    split_docs.extend(text_splitter.split_documents([doc]))
+                except Exception as e:
+                    logger.warning(f"Error splitting document {doc.metadata.get('source', 'unknown')}: {str(e)}")
+                    continue
+
+            if not split_docs:
+                logger.warning("No content could be split from documents")
+                return False
+
+            # Create embeddings and store in database
+            with get_db_session() as db:
+                try:
+                    # Clear existing embeddings
+                    db.query(DocumentEmbedding).delete()
+
+                    # Create embeddings in batch
+                    contents = [doc.page_content for doc in split_docs]
+                    embeddings = self.embed_model.embed_documents(contents)
+
+                    # Store embeddings
+                    db_embeddings = [
+                        DocumentEmbedding(
+                            content=doc.page_content,
+                            embedding=embedding,
+                            doc_metadata=doc.metadata,
+                            source=doc.metadata.get("source")
+                        )
+                        for doc, embedding in zip(split_docs, embeddings)
+                    ]
+                    db.add_all(db_embeddings)
+                    db.commit()
+
+                    logger.info(f"Successfully stored {len(split_docs)} embeddings in database")
+                    return True
+
+                except Exception as e:
+                    db.rollback()
+                    logger.error(f"Error storing embeddings in database: {str(e)}")
+                    return False
+
+        except Exception as e:
+            logger.error(f"Error processing documents: {str(e)}")
+            return False
+
+    async def get_contextual_response(self, question: str) -> str:
+        """Get contextual response using RAG."""
+        try:
+            # Get relevant documents
+            relevant_docs = await self.get_relevant_documents(question)
+            
+            if not relevant_docs:
+                logger.warning("No relevant documents found")
+                return "I don't have enough context to answer that question."
+
+            # Format documents
+            context = self.format_docs(relevant_docs)
+
+            # Generate response
+            prompt = ChatPromptTemplate.from_messages([
+                ("system", RAG_SYSTEM_PROMPT),
+                ("human", "{input}")
+            ])
+
+            chain = (
+                {"context": lambda x: context, "input": RunnablePassthrough()}
+                | prompt
+                | self.rag_llm
+                | StrOutputParser()
+            )
+
+            response = await chain.ainvoke(question)
+            logger.info("Successfully generated contextual response")
+            return response
+
+        except Exception as e:
+            logger.error(f"Error generating contextual response: {str(e)}")
+            return f"Error generating response: {str(e)}"
+
+# Create a singleton instance
+rag_manager = RAGManager()
+logger.info("RAGManager singleton instance created")
+
+# Export the get_contextual_response function for backward compatibility
+async def get_contextual_response(question: str) -> str:
+    return await rag_manager.get_contextual_response(question)

model/contextual_response/database.py

@@ -0,0 +1,31 @@
+from sqlalchemy import create_engine, Column, String, Integer, JSON, ARRAY, Float
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+import os
+from dotenv import load_dotenv
+
+# Load environment variables
+load_dotenv()
+
+# Get database URL from environment variable
+DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./contextual_response.db")
+
+# Create SQLAlchemy engine
+engine = create_engine(DATABASE_URL)
+
+# Create session factory
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+# Create base class for declarative models
+Base = declarative_base()
+
+class DocumentEmbedding(Base):
+    """Model for storing document embeddings."""
+    __tablename__ = "document_embeddings"
+
+    id = Column(Integer, primary_key=True, index=True)
+    content = Column(String, index=True)
+    embedding = Column(ARRAY(Float))
+    doc_metadata = Column(JSON)
+    source = Column(String, index=True) 
+

model/contextual_response/system_prompt.py

@@ -0,0 +1,177 @@
+RAG_SYSTEM_PROMPT = """\
+IMPORTANT: You are a code-only retrieval system. You must NEVER add any text before or after the code.
+
+Given context:
+{context}
+
+STRICT RULES:
+1. Return ONLY the exact code - no introduction, no explanation, no comments
+2. If no exact match is found, return an empty response
+3. Never explain what you're doing
+4. Never add text like "Here is the code" or "I found this"
+5. Never wrap code in markdown backticks
+6. Never add line numbers or annotations
+7. Never add your own comments to the code
+
+Example Valid Responses:
+✓ def process_data(input_data: List[str]) -> Dict[str, Any]:
+      result = {{}}
+      for item in input_data:
+          result[item] = len(item)
+      return result
+
+✓ class DataProcessor:
+      def __init__(self):
+          self.data = []
+      
+      def add_item(self, item):
+          self.data.append(item)
+
+✓ [empty response when no match found]
+
+Example Invalid Responses:
+✗ "Here's the function you're looking for:"
+✗ "I found this code in the repository:"
+✗ "```python"
+✗ "This code snippet shows how to..."
+✗ "You might want to consider..."
+✗ "I cannot provide code-snippet ..."
+
+REMEMBER: Return ONLY the exact code or nothing. No text, no explanations, no formatting."""
+
+def llama3_instructions(language: str,prefix:str,currentLine,suffix) -> dict:
+    return [                
+        {"role": "user", "content": prefix, "name": "prefix"},
+        {"role": "user", "content": currentLine, "name": "currentLine"},
+        {"role": "user", "content": suffix, "name": "suffix"},
+        {
+        "role": "system",
+        "content": f"""## Task: Code Completion
+
+    ### Language: {language}
+
+    ### Instructions:
+    - You are a world-class coding assistant.
+    - Given the current text, context, and the last character of the user input, provide a suggestion for code completion.
+    - If the suggestion is a single line, complete the line concisely, making sure it fits within the surrounding code context.
+    - If the suggestion requires multiple lines, return a well-structured code block that integrates with the surrounding code.
+    - Ensure that any subsequent lines after a block (e.g., function or loop) do not have extra indentation unless logically required by the code structure.
+    - Only suggest functional, syntactically correct code that can be used directly.
+    - Do not repeat any code that has after been suggested for previous text or context.
+    - **Reset the suggestion context** if the user removes the previous suggestion. The assistant should treat the new suggestion as fresh code.
+    - Re-evaluate the context after each suggestion is applied to ensure the next suggestion fits the newly updated code.
+    - This is not a conversation, so please do not ask questions or prompt for additional information.
+    - Never include any annotations such as "# Suggestion:" or "# Suggestions:".
+    - Newlines should be included after any of the following characters: "{{", "[", "(", ")", "]", "}}", and ",".
+    - Ensure that newline suggestions follow the same indentation as the current line.
+    - The suggestion must start with the last character of the current user input.
+    - Only ever return the code snippet, do not return any markdown unless it is part of the code snippet.
+    - If no suggestion is available, return an empty string.""",
+    }]
+
+def deepseek_editor_instructions(language: str) -> dict:
+    return [
+        {
+            "role": "system",
+            "content": f"""
+            ### You are an AI code completion tool for {language}. 
+            - You have to provide Solution Code strictly in {language}.
+            - Response should contain the Code snippet only and always return the response for each request.
+            - If the user provides incorrect code, analyze it and return a corrected version without explanation.
+            - All code responses **MUST** be wrapped inside triple backticks (```{language} ... ```).
+            - The returned solution must be error-free, properly formatted, and executable.
+
+            **STRICT RULES (FOLLOW EXACTLY, NO EXCEPTIONS):**
+            - Always Return **only** the correct and complete code.
+            - No explanations, no descriptions, no comments.
+            - No reasoning, no analysis, no extra words.
+            - Keep the code syntax correct and consistent with the surrounding code.
+            
+            **IF THE USER PROVIDES INCORRECT CODE:**
+            - Detect syntax errors, logical errors, and formatting issues.
+            - Automatically fix the errors and return only the corrected code.
+            - Ensure the code adheres to best practices for {language}.
+            - Do not provide explanations about the fix.
+
+            **LANGUAGE-SPECIFIC RULES:**
+
+            - **Python**: 
+              - Follow **indentation** strictly. Use **4 spaces** for indentation (do not use tabs).
+              - Ensure function definitions, loops, and conditionals are properly formatted.
+              - Code must be properly indented according to Python syntax and Never return incomplete or incorrect indentation.
+              - Maintain strict indentation consistency.
+
+            - **Java**: 
+              - Follow th e Java **syntax** strictly.  
+              - Ensure that function signatures and class definitions are correct.  
+              - Use **braces ** for code blocks (even for single-line blocks).
+              - Use **camelCase** for method and variable names and Class names must use **PascalCase**.
+              - Always use **access modifiers** (`public`, `private`, `protected`) for classes and methods.
+              - Always declare classes with proper **access modifiers** (e.g., `public`, `private`).
+              - If using classes in packages, ensure the correct **package declaration**.
+
+            - **JavaScript**: 
+              - Use **camelCase** for variable and function names.
+              - Use **`const`**, **`let`**, or **`var`** appropriately for variable declarations.
+              - Ensure **semi-colons** at the end of statements where needed.
+              - **Always complete curly braces {"ensure a matching close"}** for functions, loops, conditionals, and objects.
+              - If an opening ```{ "is detected, ensure a matching closing" }```.
+              - Follow **ES6+ syntax** for functions and objects.
+              - Use **arrow functions** where applicable and **async/await** for asynchronous operations.
+
+            - **TypeScript**: 
+              - Follow **JavaScript rules** with the addition of **type annotations** for variables and function parameters/returns.
+              - Ensure the correct use of **interfaces** and **types**.
+              - Consider using **strict null checks** and **interface inheritance** where applicable.
+              - Use **readonly** and **const** where applicable.
+              - Always declare types for function parameters and return values.
+
+            - **PHP**: 
+              - Follow PHP **syntax** and use **`$`** for variable declarations.
+              - For the PHP code,Always declare functions with proper return types.
+              - Ensure correct **function declarations**, especially with proper parentheses and parameter handling.
+              - Ensure variables are declared before being used.
+              - Use `declare(strict_types=1);` at the beginning where applicable.
+
+            **BEHAVIOR FOR SPACES AND NEW LINES:**  
+            - If the user presses enter or adds spaces without meaningful input, **return an empty string (`""`)**.  
+            - Do not return placeholders like `<think> </think>`, "Okay:", "Alright:", or any other annotation.
+
+            **FINAL CONDITION (DO NOT BREAK THIS):**  
+            - **Return only the correct code, nothing else.** 
+            - If no valid completion is possible or the input is just spaces/new lines, **return an empty string (`""`)** ``.  
+            Example:
+              Prefix: ""  
+              CurrentLine: "def Addition(a,b):"  
+              Suffix: ""  
+              Response: "return a+b"
+            Example 2:
+              Prefix: "def Addition(a,b):"
+              CurrentLine: "  "
+              Suffix: ""
+              Response: "return a+b"  
+              """
+        }
+    ]
+
+def chatInstructions()-> dict:
+   return {
+      "content": f"""## Task: Chat Completion
+
+   ### Instructions:
+   - You are a world-class coding assistant with expertise in various programming languages.
+   - Your goal is to help users with coding-related queries, providing precise and contextually relevant answers.
+   - If the user provides code snippets, analyze them carefully and offer suggestions or corrections.
+   - For any code snippet provided, always mention the programming language being used (e.g., Python, JavaScript, etc.) in your response along with the code.
+   - If the user asks a question, respond with concise, well-structured answers that are easy to understand.
+   - Always explain why your solution works, and offer alternatives when applicable to provide the user with options.
+   - Give code snippet if required.
+   - For any code snippet provided in your response, always format it as follows and don't forget to mention the language, and it should always be on the same line as ```:
+       ```language
+
+         #code
+       ```
+   """,
+      "role": "system",
+
+    }