Add prompt edit and api key config

.gitignore

@@ -2,5 +2,8 @@
 __pycache__
 .env
 user_media/
+toolkits/
+test*.py
+
 
 

app/api/api_file.py

@@ -0,0 +1,261 @@
+from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Form, Request, Query, status
+from fastapi.responses import StreamingResponse
+import os
+import logging
+import uuid
+from datetime import datetime
+
+from pydantic import BaseModel, Field
+from typing import Optional, List, Any
+from urllib.parse import urlparse
+import shutil
+# from app.wrapper.llm_wrapper import *
+from app.crud.process_file import load_file_with_markitdown, process_uploaded_file
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+def is_url(path: str) -> bool:
+    """
+    Determines if the given path is a URL.
+
+    Args:
+        path (str): The path or URL to check.
+
+    Returns:
+        bool: True if it's a URL, False otherwise.
+    """
+    try:
+        result = urlparse(path)
+        return all([result.scheme, result.netloc])
+    except Exception:
+        return False
+
+
+file_router = APIRouter()
+
+# Configure logging to file with date-based filenames
+log_filename = f"document_logs_{datetime.now().strftime('%Y-%m-%d')}.txt"
+file_handler = logging.FileHandler(log_filename)
+formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
+file_handler.setFormatter(formatter)
+
+# Create a logger for document processing
+doc_logger = logging.getLogger('document_logger')
+doc_logger.setLevel(logging.INFO)
+doc_logger.addHandler(file_handler)
+
+# Also configure the general logger if not already configured
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+from app.search.rag_pipeline import RAGSystem
+from sentence_transformers import SentenceTransformer
+
+
+@file_router.post("/load_file_with_markdown/")
+async def load_file_with_markdown(request: Request, filepaths: List[str]):
+    try:
+        # Ensure RAG system is initialized
+        try:
+            rag_system =  request.app.state.rag_system
+            if rag_system is None:
+                raise AttributeError("RAG system is not initialized in app state")
+        except AttributeError:
+            logger.error("RAG system is not initialized in app state")
+            raise HTTPException(status_code=500, detail="RAG system not initialized in app state")
+
+        
+        processed_files = []
+        pages = []
+        
+         # Process each file path or URL
+        for path in filepaths:
+            if is_url(path):
+                logger.info(f"Processing URL: {path}")
+                try:
+                    # Generate a unique UUID for the document
+                    doc_id = str(uuid.uuid4())
+
+                    # Process the URL
+                    document = await process_uploaded_file(id=doc_id, file_path=path, rag_system=rag_system)
+
+                    # Append the document details to pages
+                    pages.append({
+                        "metadata": {"title": document.title},
+                        "page_content": document.text_content,
+                    })
+
+                    logger.info(f"Successfully processed URL: {path} with ID: {doc_id}")
+
+                    # Log the ID and a 100-character snippet of the document
+                    snippet = document.text_content[:100].replace('\n', ' ').replace('\r', ' ')
+                    # Ensure 'doc_logger' is defined; if not, use 'logger' or define 'doc_logger'
+                    doc_logger.info(f"ID: {doc_id}_{document.title}, Snippet: {snippet}")
+
+
+                except Exception as e:
+                    logger.error(f"Error processing URL {path}: {str(e)}")
+                    processed_files.append({"path": path, "status": "error", "message": str(e)})
+
+            else:
+                logger.info(f"Processing local file: {path}")
+                if os.path.exists(path):
+                    try:
+                        # Generate a unique UUID for the document
+                        doc_id = str(uuid.uuid4())
+
+                        # Process the local file
+                        document = await process_uploaded_file(id=doc_id, file_path=path, rag_system=rag_system)
+
+                        # Append the document details to pages
+                        pages.append({
+                            "metadata": {"title": document.title},
+                            "page_content": document.text_content,
+                        })
+
+                        logger.info(f"Successfully processed file: {path} with ID: {doc_id}")
+
+                        # Log the ID and a 100-character snippet of the document
+                        snippet = document.text_content[:100].replace('\n', ' ').replace('\r', ' ')
+                        # Ensure 'doc_logger' is defined; if not, use 'logger' or define 'doc_logger'
+                        logger.info(f"ID: {doc_id}, Snippet: {snippet}")
+
+                    except Exception as e:
+                        logger.error(f"Error processing file {path}: {str(e)}")
+                        processed_files.append({"path": path, "status": "error", "message": str(e)})
+                else:
+                    logger.error(f"File path does not exist: {path}")
+                    processed_files.append({"path": path, "status": "not found"})
+        
+        # Get total tokens from RAG system
+        total_tokens = rag_system.get_total_tokens() if hasattr(rag_system, "get_total_tokens") else 0
+        
+        return {
+            "message": "File processing completed",
+            "total_tokens": total_tokens,
+            "document_count": len(filepaths),
+            "pages": pages,
+            "errors": processed_files,  # Include details about files that couldn't be processed
+        }
+    
+    except Exception as e:
+        logger.exception("Unexpected error during file processing")
+        raise HTTPException(status_code=500, detail=f"An unexpected error occurred: {str(e)}")
+    
+async def load_file_with_markdown_function(filepaths: List[str],
+    rag_system: Any):
+    try:
+        # Ensure RAG system is initialized
+        try:
+            rag_system =  rag_system
+        except AttributeError:
+            logger.error("RAG system is not initialized in app state")
+            raise HTTPException(status_code=500, detail="RAG system not initialized in app state")
+
+        
+        processed_files = []
+        pages = []
+        
+         # Process each file path or URL
+        for path in filepaths:
+            if is_url(path):
+                logger.info(f"Processing URL: {path}")
+                try:
+                    # Generate a unique UUID for the document
+                    doc_id = str(uuid.uuid4())
+
+                    # Process the URL
+                    document = await process_uploaded_file(id=doc_id, file_path=path, rag_system=rag_system)
+
+                    # Append the document details to pages
+                    pages.append({
+                        "metadata": {"title": document.title},
+                        "page_content": document.text_content,
+                    })
+
+                    logger.info(f"Successfully processed URL: {path} with ID: {doc_id}")
+
+                    # Log the ID and a 100-character snippet of the document
+                    snippet = document.text_content[:100].replace('\n', ' ').replace('\r', ' ')
+                    # Ensure 'doc_logger' is defined; if not, use 'logger' or define 'doc_logger'
+                    doc_logger(f"ID: {doc_id}, Snippet: {snippet}")
+                    logger.info(f"ID: {doc_id}, Snippet: {snippet}")
+
+                except Exception as e:
+                    logger.error(f"Error processing URL {path}: {str(e)}")
+                    processed_files.append({"path": path, "status": "error", "message": str(e)})
+
+            else:
+                logger.info(f"Processing local file: {path}")
+                if os.path.exists(path):
+                    try:
+                        # Generate a unique UUID for the document
+                        doc_id = str(uuid.uuid4())
+
+                        # Process the local file
+                        document = await process_uploaded_file(id=doc_id, file_path=path, rag_system=rag_system)
+
+                        # Append the document details to pages
+                        pages.append({
+                            "metadata": {"title": document.title},
+                            "page_content": document.text_content,
+                        })
+
+                        logger.info(f"Successfully processed file: {path} with ID: {doc_id}")
+
+                        # Log the ID and a 100-character snippet of the document
+                        snippet = document.text_content[:100].replace('\n', ' ').replace('\r', ' ')
+                        # Ensure 'doc_logger' is defined; if not, use 'logger' or define 'doc_logger'
+                        logger.info(f"ID: {doc_id}, Snippet: {snippet}")
+
+                    except Exception as e:
+                        logger.error(f"Error processing file {path}: {str(e)}")
+                        processed_files.append({"path": path, "status": "error", "message": str(e)})
+                else:
+                    logger.error(f"File path does not exist: {path}")
+                    processed_files.append({"path": path, "status": "not found"})
+        
+        # Get total tokens from RAG system
+        total_tokens = rag_system.get_total_tokens() if hasattr(rag_system, "get_total_tokens") else 0
+        
+        return {
+            "message": "File processing completed",
+            "total_tokens": total_tokens,
+            "document_count": len(filepaths),
+            "pages": pages,
+            "errors": processed_files,  # Include details about files that couldn't be processed
+        }
+    
+    except Exception as e:
+        logger.exception("Unexpected error during file processing")
+        raise HTTPException(status_code=500, detail=f"An unexpected error occurred: {str(e)}")
+
+@file_router.get("/document_exists/{doc_id}", status_code=status.HTTP_200_OK)
+async def document_exists(request: Request, doc_id: str):
+    try:
+        rag_system = request.app.state.rag_system
+    except AttributeError:
+        logger.error("RAG system is not initialized in app state")
+        raise HTTPException(status_code=500, detail="RAG system not initialized in app state")
+    
+    exists = doc_id in rag_system.doc_ids
+    return {"document_id": doc_id, "exists": exists}
+
+@file_router.delete("/delete_document/{doc_id}", status_code=status.HTTP_200_OK)
+async def delete_document(request: Request, doc_id: str):
+    try:
+        rag_system = request.app.state.rag_system
+    except AttributeError:
+        logger.error("RAG system is not initialized in app state")
+        raise HTTPException(status_code=500, detail="RAG system not initialized in app state")
+    
+    try:
+        rag_system.delete_document(doc_id)
+        logger.info(f"Deleted document with ID: {doc_id}")
+        return {"message": f"Document with ID {doc_id} has been deleted."}
+    except Exception as e:
+        logger.error(f"Error deleting document with ID {doc_id}: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Failed to delete document: {str(e)}")

app/api/api_prompt.py

@@ -0,0 +1,32 @@
+from fastapi import FastAPI, HTTPException, APIRouter
+from pydantic import BaseModel
+from app.utils.system_prompt import system_prompt, agentic_prompt
+
+prompt_router = APIRouter()
+
+# Define a model for the prompts
+class Prompt(BaseModel):
+    system_prompt: str = None
+    agentic_prompt: str = None
+
+# API endpoint to get the current prompts
+@prompt_router.get("/prompts")
+def get_prompts():
+    return {
+        "system_prompt": system_prompt,
+        "agentic_prompt": agentic_prompt,
+    }
+
+# API endpoint to update the prompts
+@prompt_router.put("/prompts")
+def update_prompts(prompts: Prompt):
+    global system_prompt, agentic_prompt
+    if prompts.system_prompt is not None:
+        system_prompt = prompts.system_prompt
+    if prompts.agentic_prompt is not None:
+        agentic_prompt = prompts.agentic_prompt
+    return {
+        "message": "Prompts updated successfully",
+        "system_prompt": system_prompt,
+        "agentic_prompt": agentic_prompt,
+    }

app/app.py

@@ -0,0 +1,153 @@
+from fastapi import FastAPI, Request, status
+from fastapi.responses import JSONResponse
+from fastapi.responses import Response
+from fastapi.exceptions import HTTPException
+from slowapi import Limiter, _rate_limit_exceeded_handler
+from slowapi.errors import RateLimitExceeded
+from slowapi.util import get_remote_address
+from slowapi.middleware import SlowAPIMiddleware
+from typing import Dict, List
+from prometheus_client import Counter, Histogram, start_http_server
+from pydantic import BaseModel, ValidationError
+from app.services.message import generate_reply, send_reply
+import logging
+from datetime import datetime
+from sentence_transformers import SentenceTransformer
+
+from contextlib import asynccontextmanager
+# from app.db.database import create_indexes, init_db
+from app.services.webhook_handler import verify_webhook
+from app.handlers.message_handler import MessageHandler
+from app.handlers.webhook_handler import WebhookHandler
+from app.handlers.media_handler import WhatsAppMediaHandler
+from app.services.cache import MessageCache
+from app.services.chat_manager import ChatManager
+from app.api.api_file import file_router
+from app.utils.load_env import ACCESS_TOKEN
+from app.search.rag_pipeline import RAGSystem
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+
+logger = logging.getLogger(__name__)
+
+# Initialize handlers at startup
+message_handler = None
+webhook_handler = None
+
+
+async def setup_message_handler():
+    logger = logging.getLogger(__name__)
+    message_cache = MessageCache()
+    chat_manager = ChatManager()
+    media_handler = WhatsAppMediaHandler()
+    
+    return MessageHandler(
+        message_cache=message_cache,
+        chat_manager=chat_manager,
+        media_handler=media_handler,
+        logger=logger
+    )
+
+async def setup_rag_system():
+    embedding_model = SentenceTransformer('all-MiniLM-L6-v2')  # Replace with your model if different
+    rag_system = RAGSystem(embedding_model)
+
+    return rag_system
+
+# Initialize FastAPI app
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+   
+    
+    try:
+        # await init_db()
+
+        logger.info("Connected to the MongoDB database!")
+
+        rag_system = await setup_rag_system()
+
+        app.state.rag_system = rag_system
+
+        global message_handler, webhook_handler
+        message_handler = await setup_message_handler()
+        webhook_handler = WebhookHandler(message_handler)
+        # collections = app.database.list_collection_names()
+        # print(f"Collections in {db_name}: {collections}")
+        yield
+    except Exception as e:
+        logger.error(e)
+
+# Initialize Limiter and Prometheus Metrics
+limiter = Limiter(key_func=get_remote_address)
+app = FastAPI(lifespan=lifespan)
+app.state.limiter = limiter
+app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
+
+# Add SlowAPI Middleware
+app.add_middleware(SlowAPIMiddleware)
+
+# app.include_router(users.router, prefix="/users", tags=["Users"])
+app.include_router(file_router, prefix="/file_load", tags=["File Load"])
+
+# Prometheus metrics
+webhook_requests = Counter('webhook_requests_total', 'Total webhook requests')
+webhook_processing_time = Histogram('webhook_processing_seconds', 'Time spent processing webhook')
+
+# Pseudocode: You may have a database or environment variable to validate API keys
+VALID_API_KEYS = {"!@#$%^"}
+class WebhookPayload(BaseModel):
+    entry: List[Dict]
+
+@app.post("/api/v1/messages")
+@limiter.limit("20/minute")
+async def process_message(request: Request):
+    try:
+        # Validate developer’s API key
+        api_key = request.headers.get("Authorization")
+        if not api_key or not api_key.startswith("Bearer "):
+            raise HTTPException(status_code=401, detail="Missing or invalid API key")
+        
+        api_key_value = api_key.replace("Bearer ", "")
+        if api_key_value not in VALID_API_KEYS:
+            raise HTTPException(status_code=403, detail="Forbidden")
+        
+        payload = await request.json()
+
+          # Extract needed credentials from query params or request body
+        # e.g., whatsapp_token, verify_token, llm_api_key, llm_model
+        whatsapp_token = request.query_params.get("whatsapp_token")
+        whatsapp_url = request.query_params.get("whatsapp_url")
+        gemini_api = request.query_params.get("gemini_api")
+        llm_model = request.query_params.get("cx_code")
+
+        print(f"payload: {payload}")
+        response = await webhook_handler.process_webhook(
+            payload=payload,
+            whatsapp_token=whatsapp_token,
+            whatsapp_url=whatsapp_url,
+            gemini_api=gemini_api,
+        )
+        
+        return JSONResponse(
+            content=response.__dict__,
+            status_code=status.HTTP_200_OK
+        )
+        
+    except ValidationError as ve:
+            logger.error(f"Validation error: {ve}")
+            return JSONResponse(
+                content={"status": "error", "detail": ve.errors()},
+                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY
+            )
+    except Exception as e:
+            logger.error(f"Unexpected error: {str(e)}")
+            return JSONResponse(
+                content={"status": "error", "detail": str(e)},
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
+            )
+    
+app.get("/webhook")(verify_webhook)

app/crud/process_file.py

@@ -0,0 +1,152 @@
+# app/crud.py
+from langchain_community.document_loaders import PyPDFLoader, Docx2txtLoader, CSVLoader, UnstructuredExcelLoader
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+from sqlalchemy.future import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from markitdown import MarkItDown
+import os
+import logging
+
+
+
+from typing import List, Optional
+# from app.db.models.docs import *
+# from app.schemas.schemas import DocumentCreate, DocumentUpdate
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+async def load_file_with_markitdown(file_path:str, llm_client:str=None, model:str=None):
+    
+    if llm_client and model:
+        markitdown = MarkItDown(llm_client, model)
+        documents = markitdown.convert(file_path)
+    else:
+        markitdown = MarkItDown()
+        documents = markitdown.convert(file_path)
+
+    return documents
+
+
+
+async def load_pdf_with_langchain(file_path):
+    """
+    Loads and extracts text from a PDF file using LangChain's PyPDFLoader.
+
+    Parameters:
+        file_path (str): Path to the PDF file.
+
+    Returns:
+        List[Document]: A list of LangChain Document objects with metadata.
+    """
+
+    loader = PyPDFLoader(file_path, extract_images=True)
+   
+    documents = loader.load()
+
+    return documents  # Returns a list of Document objects
+
+async def load_file_with_langchain(file_path: str):
+    """
+    Loads and extracts text from a PDF or DOCX file using LangChain's appropriate loader.
+
+    Parameters:
+        file_path (str): Path to the file (PDF or DOCX).
+
+    Returns:
+        List[Document]: A list of LangChain Document objects with metadata.
+    """
+    # Determine the file extension
+    _, file_extension = os.path.splitext(file_path)
+    
+    # Choose the loader based on file extension
+    if file_extension.lower() == '.pdf':
+        loader = PyPDFLoader(file_path)
+    elif file_extension.lower() == '.docx':
+        loader = Docx2txtLoader(file_path)
+    elif file_extension.lower() == '.csv':
+        loader = CSVLoader(file_path)
+    elif file_extension.lower() == '.xlsx':
+        loader = UnstructuredExcelLoader(file_path)
+    else:
+        raise ValueError("Unsupported file format. Please provide a PDF or DOCX file.")
+    
+    # Load the documents
+    documents = loader.load()
+
+    return documents
+
+async def split_documents(documents, chunk_size=10000, chunk_overlap=1000):
+    """
+    Splits documents into smaller chunks with overlap.
+
+    Parameters:
+        documents (List[Document]): List of LangChain Document objects.
+        chunk_size (int): The maximum size of each chunk.
+        chunk_overlap (int): The number of characters to overlap between chunks.
+
+    Returns:
+        List[Document]: List of chunked Document objects.
+    """
+    text_splitter = RecursiveCharacterTextSplitter(
+        chunk_size=chunk_size,
+        chunk_overlap=chunk_overlap,
+    )
+    split_docs = text_splitter.split_documents(documents)
+    return split_docs
+
+async def process_uploaded_file(
+    id, file_path,
+    rag_system=None, 
+    llm_client=None, 
+    llm_model=None
+):
+  
+    try:
+        # Load the document using LangChain
+        documents = await load_file_with_markitdown(file_path, llm_client=llm_client, model=llm_model)
+        logger.info(f"Loaded document: {file_path}")
+
+         # Concatenate all pages to get the full document text for context generation
+        # whole_document_content = "\n".join([doc.page_content for doc in documents])
+
+    except Exception as e:
+        logger.error(f"Failed to load document {file_path}: {e}")
+        raise RuntimeError(f"Error loading document: {file_path}") from e
+
+    # # Generate context for each chunk if llm is provided
+    # if llm:
+    #     for doc in split_docs:
+    #         try:
+    #             context = await llm.generate_context(doc, whole_document_content=whole_document_content)
+    #             # Add context to the beginning of the page content
+    #             doc.page_content = f"{context.replace('<|eot_id|>', '')}\n\n{doc.page_content}"
+    #             logger.info(f"Context generated and added for chunk {split_docs.index(doc)}")
+    #         except Exception as e:
+    #             logger.error(f"Failed to generate context for chunk {split_docs.index(doc)}: {e}")
+    #             raise RuntimeError(f"Error generating context for chunk {split_docs.index(doc)}") from e
+            
+    # Add to RAG system if rag_system is provided and load_only is False
+    if rag_system:
+        try:
+            rag_system.add_document(doc_id = f"{id}_{documents.title}", text = documents.text_content)
+           
+            print(f"doc_id: {id}_{documents.title}")
+            print(f"content: {documents.text_content}")
+
+                        # print(f"New Page Content: {doc.page_content}")
+            logger.info(f"Document chunks successfully added to RAG system for file {file_path}")
+
+        except Exception as e:
+            logger.error(f"Failed to add document chunks to RAG system for {file_path}: {e}")
+            raise RuntimeError(f"Error adding document to RAG system: {file_path}") from e
+    else:
+        logger.info(f"Loaded document {file_path}, but not added to RAG system")
+
+    return documents
+
+
+
+
+

app/handlers/media_handler.py

@@ -7,9 +7,9 @@ logger = logging.getLogger(__name__)
 
 class MediaHandler(ABC):
     @abstractmethod
-    async def download(self, media_id: str, access_token: str, file_path: str) -> str:
+    async def download(self, media_id: str, whatsapp_token: str, file_path: str) -> str:
         pass
 
 class WhatsAppMediaHandler(MediaHandler):
-    async def download(self, media_id: str, access_token: str, file_path: str) -> str:
-        return await download_whatsapp_media(media_id, access_token, file_path)
+    async def download(self, media_id: str, whatsapp_token: str, file_path: str) -> str:
+        return await download_whatsapp_media(media_id, whatsapp_token, file_path)

app/handlers/message_handler.py

@@ -26,7 +26,7 @@ class MessageHandler:
         self.media_handler = media_handler
         self.logger = logger
 
-    async def handle(self, raw_message: dict, access_token: str) -> dict:
+    async def handle(self, raw_message: dict, whatsapp_token: str, whatsapp_url:str,gemini_api:str) -> dict:
         try:
             # Parse message
             message = MessageParser.parse(raw_message)
@@ -36,7 +36,7 @@ class MessageHandler:
                 return {"status": "duplicate", "message_id": message.id}
 
             # Download media
-            media_paths = await self._process_media(message, access_token)
+            media_paths = await self._process_media(message, whatsapp_token)
             
             self.chat_manager.initialize_chat(message.sender_id)
 
@@ -45,7 +45,9 @@ class MessageHandler:
             result = await process_message_with_llm(
                 message.sender_id,
                 message.content,
-                self.chat_manager.get_chat_history(message.sender_id),
+                self.chat_manager.get_chat_history(message.sender_id), 
+                whatsapp_token=whatsapp_token, 
+                whatsapp_url=whatsapp_url,
                 **media_paths
             )
 
@@ -60,7 +62,7 @@ class MessageHandler:
         except Exception as e:
             return {"status": "error", "message_id": raw_message.get("id"), "error": str(e)}
 
-    async def _process_media(self, message: Message, access_token: str) -> Dict[str, Optional[str]]:
+    async def _process_media(self, message: Message, whatsapp_token: str) -> Dict[str, Optional[str]]:
         media_paths = {
             "image_file_path": None,
             "doc_path": None,
@@ -74,7 +76,7 @@ class MessageHandler:
             self.logger.info(f"Processing {media_type.value}: {content.file_path}")
             file_path = await self.media_handler.download(
                 content.id,
-                access_token,
+                whatsapp_token,
                 content.file_path
             )
             self.logger.info(f"{media_type.value} file_path: {file_path}")

app/handlers/webhook_handler.py

@@ -18,7 +18,7 @@ class WebhookHandler:
         self.message_handler = message_handler
         self.logger = logging.getLogger(__name__)
 
-    async def process_webhook(self, payload: dict, access_token: str) -> WebhookResponse:
+    async def process_webhook(self, payload: dict, whatsapp_token: str, whatsapp_url:str,gemini_api:str) -> WebhookResponse:
         request_id = f"req_{int(time.time()*1000)}"
         results = []
 
@@ -37,7 +37,9 @@ class WebhookHandler:
                         self.logger.info(f"Processing message: {message}")
                         response = await self.message_handler.handle(
                             raw_message=message,
-                            access_token=access_token
+                            whatsapp_token=whatsapp_token, 
+                            whatsapp_url=whatsapp_url,
+                            gemini_api=gemini_api,
                         )
                         results.append(response)
 

app/main.py

@@ -21,7 +21,7 @@ from app.handlers.webhook_handler import WebhookHandler
 from app.handlers.media_handler import WhatsAppMediaHandler
 from app.services.cache import MessageCache
 from app.services.chat_manager import ChatManager
-
+from app.api.api_prompt import prompt_router
 from app.utils.load_env import ACCESS_TOKEN
 
 # Configure logging
@@ -79,7 +79,7 @@ app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
 app.add_middleware(SlowAPIMiddleware)
 
 # app.include_router(users.router, prefix="/users", tags=["Users"])
-
+app.include_router(prompt_router, prefix="/prompts", tags=["Prompts"])
 
 # Prometheus metrics
 webhook_requests = Counter('webhook_requests_total', 'Total webhook requests')
@@ -105,11 +105,19 @@ async def webhook(request: Request):
             # Process the webhook payload here
             # For example:
         # results = process_webhook_entries(validated_payload.entry)
+         # e.g., whatsapp_token, verify_token, llm_api_key, llm_model
+        whatsapp_token = request.query_params.get("whatsapp_token")
+        whatsapp_url = request.query_params.get("whatsapp_url")
+        gemini_api = request.query_params.get("gemini_api")
+        llm_model = request.query_params.get("cx_code")
+
+        print(f"payload: {payload}")
         response = await webhook_handler.process_webhook(
             payload=payload,
-            access_token=ACCESS_TOKEN
+            whatsapp_token=whatsapp_token,
+            whatsapp_url=whatsapp_url,
+            gemini_api=gemini_api,
         )
-        
         return JSONResponse(
             content=response.__dict__,
             status_code=status.HTTP_200_OK

app/search/bm25_search.py

@@ -0,0 +1,159 @@
+# bm25_search.py
+import asyncio
+from rank_bm25 import BM25Okapi
+import nltk
+import string
+from typing import List, Set, Optional
+from nltk.corpus import stopwords
+from nltk.stem import WordNetLemmatizer
+
+
+def download_nltk_resources():
+    """
+    Downloads required NLTK resources synchronously.
+    """
+    resources = ['punkt', 'stopwords', 'wordnet', 'omw-1.4']
+    for resource in resources:
+        try:
+            nltk.download(resource, quiet=True)
+        except Exception as e:
+            print(f"Error downloading {resource}: {str(e)}")
+
+class BM25_search:
+    # Class variable to track if resources have been downloaded
+    nltk_resources_downloaded = False
+
+    def __init__(self, remove_stopwords: bool = True, perform_lemmatization: bool = False):
+        """
+        Initializes the BM25search.
+
+        Parameters:
+        - remove_stopwords (bool): Whether to remove stopwords during preprocessing.
+        - perform_lemmatization (bool): Whether to perform lemmatization on tokens.
+        """
+        # Ensure NLTK resources are downloaded only once
+        if not BM25_search.nltk_resources_downloaded:
+            download_nltk_resources()
+            BM25_search.nltk_resources_downloaded = True  # Mark as downloaded
+
+        self.documents: List[str] = []
+        self.doc_ids: List[str] = []
+        self.tokenized_docs: List[List[str]] = []
+        self.bm25: Optional[BM25Okapi] = None
+        self.remove_stopwords = remove_stopwords
+        self.perform_lemmatization = perform_lemmatization
+        self.stop_words: Set[str] = set(stopwords.words('english')) if remove_stopwords else set()
+        self.lemmatizer = WordNetLemmatizer() if perform_lemmatization else None
+
+    def preprocess(self, text: str) -> List[str]:
+        """
+        Preprocesses the input text by lowercasing, removing punctuation,
+        tokenizing, removing stopwords, and optionally lemmatizing.
+        """
+        text = text.lower().translate(str.maketrans('', '', string.punctuation))
+        tokens = nltk.word_tokenize(text)
+        if self.remove_stopwords:
+            tokens = [token for token in tokens if token not in self.stop_words]
+        if self.perform_lemmatization and self.lemmatizer:
+            tokens = [self.lemmatizer.lemmatize(token) for token in tokens]
+        return tokens
+
+    def add_document(self, doc_id: str, new_doc: str) -> None:
+        """
+        Adds a new document to the corpus and updates the BM25 index.
+        """
+        processed_tokens = self.preprocess(new_doc)
+        
+        self.documents.append(new_doc)
+        self.doc_ids.append(doc_id)
+        self.tokenized_docs.append(processed_tokens)
+        # Ensure update_bm25 is awaited if required in async context
+        self.update_bm25()
+        print(f"Added document ID: {doc_id}")
+
+    async def remove_document(self, index: int) -> None:
+        """
+        Removes a document from the corpus based on its index and updates the BM25 index.
+        """
+        if 0 <= index < len(self.documents):
+            removed_doc_id = self.doc_ids[index]
+            del self.documents[index]
+            del self.doc_ids[index]
+            del self.tokenized_docs[index]
+            self.update_bm25()
+            print(f"Removed document ID: {removed_doc_id}")
+        else:
+            print(f"Index {index} is out of bounds.")
+
+    def update_bm25(self) -> None:
+        """
+        Updates the BM25 index based on the current tokenized documents.
+        """
+        if self.tokenized_docs:
+            self.bm25 = BM25Okapi(self.tokenized_docs)
+            print("BM25 index has been initialized.")
+        else:
+            print("No documents to initialize BM25.")
+
+
+    def get_scores(self, query: str) -> List[float]:
+        """
+        Computes BM25 scores for all documents based on the given query.
+        """
+        processed_query = self.preprocess(query)
+        print(f"Tokenized Query: {processed_query}")
+        
+        if self.bm25:
+            return self.bm25.get_scores(processed_query)
+        else:
+            print("BM25 is not initialized.")
+            return []
+
+    def get_top_n_docs(self, query: str, n: int = 5) -> List[str]:
+        """
+        Returns the top N documents for a given query.
+        """
+        processed_query = self.preprocess(query)
+        if self.bm25:
+            return self.bm25.get_top_n(processed_query, self.documents, n)
+        else:
+            print("initialized.")
+            return []
+    
+    def clear_documents(self) -> None:
+        """
+        Clears all documents from the BM25 index.
+        """
+        self.documents = []
+        self.doc_ids = []
+        self.tokenized_docs = []
+        self.bm25 = None  # Reset BM25 index
+        print("BM25 documents cleared and index reset.")
+    
+    def get_document(self, doc_id: str) -> str:
+        """
+        Retrieves a document by its document ID.
+        
+        Parameters:
+        - doc_id (str): The ID of the document to retrieve.
+
+        Returns:
+        - str: The document text if found, otherwise an empty string.
+        """
+        try:
+            index = self.doc_ids.index(doc_id)
+            return self.documents[index]
+        except ValueError:
+            print(f"Document ID {doc_id} not found.")
+            return ""
+
+
+async def initialize_bm25_search(remove_stopwords: bool = True, perform_lemmatization: bool = False) -> BM25_search:
+    """
+    Initializes the BM25search with proper NLTK resource downloading.
+    """
+    loop = asyncio.get_running_loop()
+    await loop.run_in_executor(None, download_nltk_resources)
+    return BM25_search(remove_stopwords, perform_lemmatization)
+
+

app/search/faiss_search.py

@@ -0,0 +1,76 @@
+# faiss_wrapper.py
+import faiss
+import numpy as np
+
+class FAISS_search:
+    def __init__(self, embedding_model):
+        self.documents = []
+        self.doc_ids = []
+        self.embedding_model = embedding_model
+        self.dimension = len(embedding_model.encode("embedding"))
+        self.index = faiss.IndexIDMap(faiss.IndexFlatL2(self.dimension))
+
+    def add_document(self, doc_id, new_doc):
+        self.documents.append(new_doc)
+        self.doc_ids.append(doc_id)
+        # Encode and add document with its index as ID
+        embedding = self.embedding_model.encode([new_doc], convert_to_numpy=True).astype('float32')
+
+        if embedding.size == 0:
+            print("No documents to add to FAISS index.")
+            return
+
+        idx = len(self.documents) - 1
+        id_array = np.array([idx]).astype('int64')
+        self.index.add_with_ids(embedding, id_array)
+
+    def remove_document(self, index):
+        if 0 <= index < len(self.documents):
+            del self.documents[index]
+            del self.doc_ids[index]
+            # Rebuild the index
+            self.build_index()
+        else:
+            print(f"Index {index} is out of bounds.")
+
+    def build_index(self):
+        embeddings = self.embedding_model.encode(self.documents, convert_to_numpy=True).astype('float32')
+        idx_array = np.arange(len(self.documents)).astype('int64')
+        self.index = faiss.IndexIDMap(faiss.IndexFlatL2(self.dimension))
+        self.index.add_with_ids(embeddings, idx_array)
+
+    def search(self, query, k):
+        if self.index.ntotal == 0:
+            # No documents in the index
+            print("FAISS index is empty. No results can be returned.")
+            return np.array([]), np.array([])  # Return empty arrays for distances and indices
+        query_embedding = self.embedding_model.encode([query], convert_to_numpy=True).astype('float32')
+        distances, indices = self.index.search(query_embedding, k)
+        return distances[0], indices[0]
+    
+    def clear_documents(self) -> None:
+        """
+        Clears all documents from the FAISS index.
+        """
+        self.documents = []
+        self.doc_ids = []
+        # Reset the FAISS index
+        self.index = faiss.IndexIDMap(faiss.IndexFlatL2(self.dimension))
+        print("FAISS documents cleared and index reset.")
+    
+    def get_document(self, doc_id: str) -> str:
+        """
+        Retrieves a document by its document ID.
+        
+        Parameters:
+        - doc_id (str): The ID of the document to retrieve.
+
+        Returns:
+        - str: The document text if found, otherwise an empty string.
+        """
+        try:
+            index = self.doc_ids.index(doc_id)
+            return self.documents[index]
+        except ValueError:
+            print(f"Document ID {doc_id} not found.")
+            return ""

app/search/hybrid_search.py

@@ -0,0 +1,247 @@
+import numpy as np
+import logging, torch
+from sklearn.preprocessing import MinMaxScaler
+from sentence_transformers import CrossEncoder
+# from FlagEmbedding import FlagReranker
+
+
+class Hybrid_search:
+    def __init__(self, bm25_search, faiss_search, reranker_model_name="BAAI/bge-reranker-v2-gemma", initial_bm25_weight=0.5):
+        self.bm25_search = bm25_search
+        self.faiss_search = faiss_search
+        self.bm25_weight = initial_bm25_weight
+        # self.reranker = FlagReranker(reranker_model_name, use_fp16=True)
+        self.logger = logging.getLogger(__name__)
+
+    async def advanced_search(self, query, keywords, top_n=5, threshold=0.53, prefixes=None):
+        # Dynamic BM25 weighting
+        self._dynamic_weighting(len(query.split()))
+        keywords = f"{' '.join(keywords)}"
+        self.logger.info(f"Query: {query}")
+        self.logger.info(f"Keywords: {keywords}")
+
+        # Get BM25 scores and doc_ids
+        bm25_scores, bm25_doc_ids = self._get_bm25_results(keywords, top_n = top_n)
+        # self.logger.info(f"BM25 Scores: {bm25_scores}, BM25 doc_ids: {bm25_doc_ids}")
+        # Get FAISS distances, indices, and doc_ids
+        faiss_distances, faiss_indices, faiss_doc_ids = self._get_faiss_results(query)
+        try:
+            faiss_distances, indices, faiss_doc_ids = self._get_faiss_results(query, top_n = top_n)
+            # for dist, idx, doc_id in zip(faiss_distances, indices, faiss_doc_ids):
+            #     print(f"Distance: {dist:.4f}, Index: {idx}, Doc ID: {doc_id}")
+        except Exception as e:
+            self.logger.error(f"Search failed: {str(e)}")
+        # Map doc_ids to scores
+        bm25_scores_dict, faiss_scores_dict = self._map_scores_to_doc_ids(
+            bm25_doc_ids, bm25_scores, faiss_doc_ids, faiss_distances
+        )
+        # Create a unified set of doc IDs
+        all_doc_ids = sorted(set(bm25_doc_ids).union(faiss_doc_ids))
+        # print(f"All doc_ids: {all_doc_ids}, BM25 doc_ids: {bm25_doc_ids}, FAISS doc_ids: {faiss_doc_ids}")
+
+        # Filter doc_ids based on prefixes
+        filtered_doc_ids = self._filter_doc_ids_by_prefixes(all_doc_ids, prefixes)
+        # self.logger.info(f"Filtered doc_ids: {filtered_doc_ids}")
+
+        if not filtered_doc_ids:
+            self.logger.info("No documents match the prefixes.")
+            return []
+
+        # Prepare score lists
+        filtered_bm25_scores, filtered_faiss_scores = self._get_filtered_scores(
+            filtered_doc_ids, bm25_scores_dict, faiss_scores_dict
+        )
+        # self.logger.info(f"Filtered BM25 scores: {filtered_bm25_scores}")
+        # self.logger.info(f"Filtered FAISS scores: {filtered_faiss_scores}")
+       
+
+        # Normalize scores
+        bm25_scores_normalized, faiss_scores_normalized = self._normalize_scores(
+            filtered_bm25_scores, filtered_faiss_scores
+        )
+
+        # Calculate hybrid scores
+        hybrid_scores = self._calculate_hybrid_scores(bm25_scores_normalized, faiss_scores_normalized)
+
+        # Display hybrid scores
+        for idx, doc_id in enumerate(filtered_doc_ids):
+            print(f"Hybrid Score: {hybrid_scores[idx]:.4f}, Doc ID: {doc_id}")
+
+        # Apply threshold and get top_n results
+        results = self._get_top_n_results(filtered_doc_ids, hybrid_scores, top_n, threshold)
+        # self.logger.info(f"Results before reranking: {results}")
+
+        # If results exist, apply re-ranking
+        # if results:
+        #     re_ranked_results = self._rerank_results(query, results)
+        #     self.logger.info(f"Results after reranking: {re_ranked_results}")
+        #     return re_ranked_results
+
+        return results
+    
+
+    def _dynamic_weighting(self, query_length):
+        if query_length <= 5:
+            self.bm25_weight = 0.7
+        else:
+            self.bm25_weight = 0.5
+        self.logger.info(f"Dynamic BM25 weight set to: {self.bm25_weight}")
+
+    def _get_bm25_results(self, keywords, top_n:int = None):
+        # Get BM25 scores
+        bm25_scores = np.array(self.bm25_search.get_scores(keywords))
+        bm25_doc_ids = np.array(self.bm25_search.doc_ids)  # Assuming doc_ids is a list of document IDs
+
+        # Log the scores and IDs before filtering
+        # self.logger.info(f"BM25 scores: {bm25_scores}")
+        # self.logger.info(f"BM25 doc_ids: {bm25_doc_ids}")
+
+        # Get the top k indices based on BM25 scores
+        top_k_indices = np.argsort(bm25_scores)[-top_n:][::-1]
+
+        # Retrieve top k scores and corresponding document IDs
+        top_k_scores = bm25_scores[top_k_indices]
+        top_k_doc_ids = bm25_doc_ids[top_k_indices]
+
+        # Return top k scores and document IDs
+        return top_k_scores, top_k_doc_ids
+
+    def _get_faiss_results(self, query, top_n: int = None) -> tuple[np.ndarray, np.ndarray, list[str]]:
+    
+        try:
+            # If top_k is not specified, use all documents
+            if top_n is None:
+                top_n = len(self.faiss_search.doc_ids)
+                
+            # Use the search's search method which handles the embedding
+            distances, indices = self.faiss_search.search(query, k=top_n) 
+            
+            if len(distances) == 0 or len(indices) == 0:
+                # Handle case where FAISS returns empty results
+                self.logger.info("FAISS search returned no results.")
+                return np.array([]), np.array([]), []
+            
+            # Filter out invalid indices (-1)
+            valid_mask = indices != -1
+            filtered_distances = distances[valid_mask]
+            filtered_indices = indices[valid_mask]
+            
+            # Map indices to doc_ids
+            doc_ids = [self.faiss_search.doc_ids[idx] for idx in filtered_indices 
+                    if 0 <= idx < len(self.faiss_search.doc_ids)]
+            
+            # self.logger.info(f"FAISS distances: {filtered_distances}")
+            # self.logger.info(f"FAISS indices: {filtered_indices}")
+            # self.logger.info(f"FAISS doc_ids: {doc_ids}")
+            
+            return filtered_distances, filtered_indices, doc_ids
+            
+        except Exception as e:
+            self.logger.error(f"Error in FAISS search: {str(e)}")
+            raise
+
+    def _map_scores_to_doc_ids(self, bm25_doc_ids, bm25_scores, faiss_doc_ids, faiss_scores):
+        bm25_scores_dict = dict(zip(bm25_doc_ids, bm25_scores))
+        faiss_scores_dict = dict(zip(faiss_doc_ids, faiss_scores))
+        # self.logger.info(f"BM25 scores dict: {bm25_scores_dict}")
+        # self.logger.info(f"FAISS scores dict: {faiss_scores_dict}")
+        return bm25_scores_dict, faiss_scores_dict
+
+    def _filter_doc_ids_by_prefixes(self, all_doc_ids, prefixes):
+        if prefixes:
+            filtered_doc_ids = [
+                doc_id
+                for doc_id in all_doc_ids
+                if any(doc_id.startswith(prefix) for prefix in prefixes)
+            ]
+        else:
+            filtered_doc_ids = list(all_doc_ids)
+        return filtered_doc_ids
+
+    def _get_filtered_scores(self, filtered_doc_ids, bm25_scores_dict, faiss_scores_dict):
+         # Initialize lists to hold scores in the unified doc ID order
+        bm25_aligned_scores = []
+        faiss_aligned_scores = []
+
+        # Populate aligned score lists, filling missing scores with neutral values
+        for doc_id in filtered_doc_ids:
+            bm25_aligned_scores.append(bm25_scores_dict.get(doc_id, 0))  # Use 0 if not found in BM25
+            faiss_aligned_scores.append(faiss_scores_dict.get(doc_id, max(faiss_scores_dict.values()) + 1))  # Use a high distance if not found in FAISS
+        
+        # Invert the FAISS scores
+        faiss_aligned_scores = [1 / score if score != 0 else 0 for score in faiss_aligned_scores]
+
+        return bm25_aligned_scores, faiss_aligned_scores
+
+    def _normalize_scores(self, filtered_bm25_scores, filtered_faiss_scores):
+        scaler_bm25 = MinMaxScaler()
+        bm25_scores_normalized = self._normalize_array(filtered_bm25_scores, scaler_bm25)
+
+        scaler_faiss = MinMaxScaler()
+        faiss_scores_normalized = self._normalize_array(filtered_faiss_scores, scaler_faiss)
+
+        # self.logger.info(f"Normalized BM25 scores: {bm25_scores_normalized}")
+        # self.logger.info(f"Normalized FAISS scores: {faiss_scores_normalized}")
+        return bm25_scores_normalized, faiss_scores_normalized
+
+    def _normalize_array(self, scores, scaler):
+        scores_array = np.array(scores)
+        if np.ptp(scores_array) > 0:
+            normalized_scores = scaler.fit_transform(scores_array.reshape(-1, 1)).flatten()
+        else:
+            # Handle identical scores with a fallback to uniform 0.5
+            normalized_scores = np.full_like(scores_array, 0.5, dtype=float)
+        return normalized_scores
+
+    def _calculate_hybrid_scores(self, bm25_scores_normalized, faiss_scores_normalized):
+        hybrid_scores = self.bm25_weight * bm25_scores_normalized + (1 - self.bm25_weight) * faiss_scores_normalized
+        # self.logger.info(f"Hybrid scores: {hybrid_scores}")
+        return hybrid_scores
+
+    def _get_top_n_results(self, filtered_doc_ids, hybrid_scores, top_n, threshold):
+        hybrid_scores = np.array(hybrid_scores)
+        threshold_indices = np.where(hybrid_scores >= threshold)[0]
+        if len(threshold_indices) == 0:
+            self.logger.info("No documents meet the threshold.")
+            return []
+
+        sorted_indices = threshold_indices[np.argsort(hybrid_scores[threshold_indices])[::-1]]
+        top_indices = sorted_indices[:top_n]
+
+        results = [(filtered_doc_ids[idx], hybrid_scores[idx]) for idx in top_indices]
+        self.logger.info(f"Top {top_n} results: {results}")
+        return results
+
+    def _rerank_results(self, query, results):
+        """
+        Re-rank the retrieved documents using FlagReranker with normalized scores.
+
+        Parameters:
+        - query (str): The search query.
+        - results (List[Tuple[str, float]]): A list of (doc_id, score) tuples.
+
+        Returns:
+        - List[Tuple[str, float]]: Re-ranked list of (doc_id, score) tuples with normalized scores.
+        """
+        # Prepare input for the re-ranker
+        document_texts = [self.bm25_search.get_document(doc_id) for doc_id, _ in results]
+        doc_ids = [doc_id for doc_id, _ in results]
+
+        # Generate pairwise scores using the FlagReranker
+        rerank_inputs = [[query, doc] for doc in document_texts]
+        with torch.no_grad():
+            rerank_scores = self.reranker.compute_score(rerank_inputs, normalize=True)
+
+        # rerank_scores = self.reranker.compute_score(rerank_inputs, normalize=True)
+
+        # Combine doc_ids with normalized re-rank scores and sort by scores
+        reranked_results = sorted(
+            zip(doc_ids, rerank_scores),
+            key=lambda x: x[1],
+            reverse=True
+        )
+
+        # Log and return results
+        # self.logger.info(f"Re-ranked results with normalized scores: {reranked_results}")
+        return reranked_results
+

app/search/rag_pipeline.py

@@ -0,0 +1,147 @@
+# rag_pipeline.py
+
+import numpy as np
+
+import pickle
+import os
+import logging
+import asyncio
+
+from app.search.bm25_search import BM25_search
+from app.search.faiss_search import FAISS_search
+from app.search.hybrid_search import Hybrid_search
+from app.utils.token_counter import TokenCounter
+
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# rag.py
+class RAGSystem:
+    def __init__(self, embedding_model):
+        self.token_counter = TokenCounter()
+        self.documents = []
+        self.doc_ids = []
+        self.results = []
+        self.meta_data = []
+        self.embedding_model = embedding_model
+        self.bm25_wrapper = BM25_search()
+        self.faiss_wrapper = FAISS_search(embedding_model)
+        self.hybrid_search = Hybrid_search(self.bm25_wrapper, self.faiss_wrapper)
+
+    def add_document(self, doc_id, text, metadata=None):
+        self.token_counter.add_document(doc_id, text)
+        self.doc_ids.append(doc_id)
+        self.documents.append(text)
+        self.meta_data.append(metadata)
+        self.bm25_wrapper.add_document(doc_id, text)
+        self.faiss_wrapper.add_document(doc_id, text)
+
+    def delete_document(self, doc_id):
+        try:
+            index = self.doc_ids.index(doc_id)
+            del self.doc_ids[index]
+            del self.documents[index]
+            self.bm25_wrapper.remove_document(index)
+            self.faiss_wrapper.remove_document(index)
+            self.token_counter.remove_document(doc_id)
+        except ValueError:
+            logging.warning(f"Document ID {doc_id} not found.")
+
+    async def adv_query(self, query_text, keywords, top_k=15, prefixes=None):
+        results = await self.hybrid_search.advanced_search(
+            query_text,
+            keywords=keywords,
+            top_n=top_k,
+            threshold=0.43,
+            prefixes=prefixes
+        )
+        retrieved_docs = []
+        if results:
+            seen_docs = set()
+            for doc_id, score in results:
+                if doc_id not in seen_docs:
+                     # Check if the doc_id exists in self.doc_ids
+                    if doc_id not in self.doc_ids:
+                        logger.error(f"doc_id {doc_id} not found in self.doc_ids")
+                    seen_docs.add(doc_id)
+                  
+                    # Fetch the index of the document
+                    try:
+                        index = self.doc_ids.index(doc_id)
+                    except ValueError as e:
+                        logger.error(f"Error finding index for doc_id {doc_id}: {e}")
+                        continue
+
+                     # Validate index range
+                    if index >= len(self.documents) or index >= len(self.meta_data):
+                        logger.error(f"Index {index} out of range for documents or metadata")
+                        continue
+
+                    doc = self.documents[index]
+                    
+                    # meta_data = self.meta_data[index]
+                    # Extract the file name and page number
+                    # file_name = meta_data['source'].split('/')[-1]  # Extracts 'POJK 31 - 2018.pdf'
+                    # page_number = meta_data.get('page', 'unknown')
+                    # url = meta_data['source']
+                    # file_name = meta_data.get('source', 'unknown_source').split('/')[-1]  # Safe extraction
+                    # page_number = meta_data.get('page', 'unknown')  # Default to 'unknown' if 'page' is missing
+                    # url = meta_data.get('source', 'unknown_url')  # Default URL fallback
+
+                    # logger.info(f"file_name: {file_name}, page_number: {page_number}, url: {url}")
+
+                    # Format as a single string
+                    # content_string = f"'{file_name}', 'page': {page_number}"
+                    # doc_name = f"{file_name}"
+                  
+                    self.results.append(doc)
+                    retrieved_docs.append({"text": doc})
+            return retrieved_docs
+        else:
+            return [{"name": "No relevant documents found.", "text": None}]
+
+    def get_total_tokens(self):
+        return self.token_counter.get_total_tokens()
+    def get_context(self):
+        context = "\n".join(self.results)
+        return context
+
+    def save_state(self, path):
+    # Save doc_ids, documents, and token counter state
+        with open(f"{path}_state.pkl", 'wb') as f:
+            pickle.dump({
+                "doc_ids": self.doc_ids,
+                "documents": self.documents,
+                "meta_data": self.meta_data,
+                "token_counts": self.token_counter.doc_tokens
+            }, f)
+
+    def load_state(self, path):
+        if os.path.exists(f"{path}_state.pkl"):
+            with open(f"{path}_state.pkl", 'rb') as f:
+                state_data = pickle.load(f)
+                self.doc_ids = state_data["doc_ids"]
+                self.documents = state_data["documents"]
+                self.meta_data = state_data["meta_data"]
+                self.token_counter.doc_tokens = state_data["token_counts"]
+
+            # Clear and rebuild BM25 and FAISS
+            self.bm25_wrapper.clear_documents()
+            self.faiss_wrapper.clear_documents()
+            for doc_id, document in zip(self.doc_ids, self.documents):
+                self.bm25_wrapper.add_document(doc_id, document)
+                self.faiss_wrapper.add_document(doc_id, document)
+
+            self.token_counter.total_tokens = sum(self.token_counter.doc_tokens.values())
+            logging.info("System state loaded successfully with documents and indices rebuilt.")
+        else:
+            logging.info("No previous state found, initializing fresh state.")
+            self.doc_ids = []
+            self.documents = []
+            self.meta_data = []  # Reset meta_data
+            self.token_counter = TokenCounter()
+            self.bm25_wrapper = BM25_search()
+            self.faiss_wrapper = FAISS_search(self.embedding_model)
+            self.hybrid_search = Hybrid_search(self.bm25_wrapper, self.faiss_wrapper)

app/services/message.py

@@ -6,7 +6,9 @@ from datetime import datetime
 import logging
 import asyncio
 from openai import AsyncOpenAI
+import json
 import google.generativeai as genai
+
 import PIL.Image
 from typing import List, Dict, Any, Optional
 
@@ -17,6 +19,29 @@ from app.services.search_engine import google_search
 # Load environment variables
 load_dotenv()
 
+# Define function specifications for Gemini
+function_declarations = [
+    {
+        "name": "google_search",
+        "description": "Perform a Google search and retrieve search results",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "query": {
+                    "type": "string",
+                    "description": "The search query to perform"
+                },
+                "num_results": {
+                    "type": "string",
+                    "description": "Number of search results to retrieve (1-10)",
+                    "default": "3"
+                }
+            },
+            "required": ["query"]
+        }
+    }
+]
+
 genai.configure(api_key=GEMNI_API)
 client = AsyncOpenAI(api_key = OPENAI_API)
 # Configure logging
@@ -31,9 +56,9 @@ if not WHATSAPP_API_URL or not ACCESS_TOKEN:
     logger.warning("Environment variables for WHATSAPP_API_URL or ACCESS_TOKEN are not set!")
 
 # Helper function to send a reply
-async def send_reply(to: str, body: str) -> Dict[str, Any]:
+async def send_reply(to: str, body: str, whatsapp_token: str, whatsapp_url:str) -> Dict[str, Any]:
     headers = {
-        "Authorization": f"Bearer {ACCESS_TOKEN}",
+        "Authorization": f"Bearer {whatsapp_token}",
         "Content-Type": "application/json"
     }
     data = {
@@ -46,7 +71,7 @@ async def send_reply(to: str, body: str) -> Dict[str, Any]:
     }
 
     async with httpx.AsyncClient() as client:
-        response = await client.post(WHATSAPP_API_URL, json=data, headers=headers)
+        response = await client.post(whatsapp_url, json=data, headers=headers)
 
     if response.status_code != 200:
         error_detail = response.json()
@@ -74,7 +99,9 @@ async def generate_reply(sender: str, content: str, timestamp: int) -> str:
 async def process_message_with_llm(
     sender_id: str,
     content: str,
-    history: List[Dict[str, str]],
+    history: List[Dict[str, str]], 
+    whatsapp_token: str, 
+    whatsapp_url:str,
     image_file_path: Optional[str] = None,
     doc_path: Optional[str] = None,
     video_file_path: Optional[str] = None,
@@ -92,7 +119,7 @@ async def process_message_with_llm(
         )
         logger.info(f"Generated reply: {generated_reply}")
 
-        response = await send_reply(sender_id, generated_reply)
+        response = await send_reply(sender_id, generated_reply, whatsapp_token, whatsapp_url)
 #         return generated_reply
         return generated_reply
     except Exception as e:
@@ -140,14 +167,55 @@ async def generate_response_from_gemini(
             pass  # Placeholder for video processing logic
 
         # Send the user's message
-        response = await chat.send_message_async(content, tools=[google_search])
+        response = await chat.send_message_async(content)
+        # response = await handle_function_call(response)
         return response.text
 
     except Exception as e:
         logger.error("Error in generate_response_from_gemini:", exc_info=True)
         return "Sorry, I couldn't generate a response at this time."
 
-
+async def handle_function_call(chat):
+    """
+    Handle function calls from the Gemini API.
+    
+    Args:
+        chat (ChatSession): The current chat session.
+    
+    Returns:
+        The response after resolving function calls.
+    """
+    # Continue the conversation and handle any function calls
+    while True:
+        response = chat.send_message_async(chat.history[-1])
+        
+        # Check if there are any function calls to handle
+        if response.candidates[0].content.parts[0].function_call:
+            function_call = response.candidates[0].content.parts[0].function_call
+            function_name = function_call.name
+            function_args = json.loads(function_call.args)
+            
+            # Dispatch to the appropriate function
+            if function_name == "google_search":
+                # Handle async function call
+                result = await google_search(
+                    query=function_args['query'], 
+                    num_results=function_args.get('num_results', '3')
+                )
+           
+            
+            # Send the function result back to continue the conversation
+            response = chat.send_message_async(
+                part={
+                    "function_response": {
+                        "name": function_name,
+                        "response": result
+                    }
+                }
+            )
+        else:
+            # No more function calls, return the final response
+            return response
 
 # Process message with retry logic
 # async def process_message_with_retry(

app/services/search_engine.py

@@ -60,7 +60,8 @@ async def google_search(query: str, num_results: str = "3") -> Optional[List[Dic
         if response.status_code == 200:
             results = response.json()
             items = results.get("items", [])
-            return [{"title": item["title"], "link": item["link"]} for item in items]
+      
+            return [{"title": item["title"], "link": item["link"], "snippet": item["snippet"]} for item in items] 
 
         else:
             logger.error(f"Google Search API error: {response.status_code} - {response.text}")
@@ -70,3 +71,18 @@ async def google_search(query: str, num_results: str = "3") -> Optional[List[Dic
         logger.error("A network error occurred while performing the Google search.")
         logger.error(f"Error details: {e}")
         return None
+
+def set_light_values(brightness: str, color_temp: str) -> Dict[str, str]:
+    """Set the brightness and color temperature of a room light. (mock API).
+
+    Args:
+        brightness: Light level from 0 to 100. Zero is off and 100 is full brightness
+        color_temp: Color temperature of the light fixture, which can be `daylight`, `cool` or `warm`.
+
+    Returns:
+        A dictionary containing the set brightness and color temperature.
+    """
+    return {
+        "brightness": brightness,
+        "colorTemperature": color_temp
+    }

app/utils/load_env.py

@@ -20,8 +20,8 @@ CX_CODE = os.getenv("CX_CODE")
 CUSTOM_SEARCH_API_KEY = os.getenv("CUSTOM_SEARCH_API_KEY")
 
 # Debugging: Print the retrieved ACCESS_TOKEN (for development only)
-if ENV == "development":
-    print(f"ACCESS_TOKEN loaded: {ACCESS_TOKEN}")
+# if ENV == "development":
+#     print(f"ACCESS_TOKEN loaded: {ACCESS_TOKEN}")
 
 
 

app/utils/system_prompt.py

@@ -22,4 +22,10 @@ Example Interactions:
     If a user asks, “Are there any issues with the city government's policies?” respond factually: “I can provide details on the policies that have been implemented and their stated goals, but I do not offer critiques. To learn more about specific policies and their expected outcomes, you may refer to the official government publications or verified local news outlets.”
 
 By following these guidelines, you will serve as a reliable, respectful, and informative resource for users looking to understand the latest happenings in Surabaya without engaging in criticism of the government.
+"""
+
+agentic_prompt = """ You are a helpful assistant and have capabilities to search the web. 
+When you the links are given, you should summarize the content of the link and give a short summary.
+You should also include the source of the link in the summary.
+
 """

app/utils/token_counter.py

@@ -0,0 +1,29 @@
+# token_counter.py
+
+import os
+import tiktoken
+
+# Choose the encoding based on your model, e.g., 'cl100k_base' for OpenAI models
+encoding = tiktoken.get_encoding("cl100k_base")
+
+def count_tokens(text):
+    tokens = encoding.encode(text)
+    return len(tokens)
+    
+class TokenCounter:
+    def __init__(self):
+        self.total_tokens = 0
+        self.doc_tokens = {}
+
+    def add_document(self, doc_id, text):
+        num_tokens = count_tokens(text)
+        self.doc_tokens[doc_id] = num_tokens
+        self.total_tokens += num_tokens
+
+    def remove_document(self, doc_id):
+        if doc_id in self.doc_tokens:
+            self.total_tokens -= self.doc_tokens[doc_id]
+            del self.doc_tokens[doc_id]
+
+    def get_total_tokens(self):
+        return self.total_tokens

app/utils/tool_call_extractor.py

@@ -0,0 +1,149 @@
+import re
+import json
+from typing import List, Dict, Any, Optional
+
+class ToolCallExtractor:
+    def __init__(self):
+        # Existing regex patterns (retain if needed for other formats)
+        self.complete_pattern = re.compile(r'<\|python_tag\|>(.*?)<\|eom_id\|>', re.DOTALL)
+        self.partial_pattern = re.compile(r'(.*?)<\|eom_id\|>', re.DOTALL)
+        
+    def _extract_function_args(self, args) -> Dict[str, Any]:
+        """
+        Flatten the nested function args structure for Google AI protobuf types.
+        """
+        flattened_args = {}
+        
+        try:
+            # Explicitly check for fields
+            if hasattr(args, 'fields'):
+                # Iterate through fields using to_dict() to convert protobuf to dict
+                for field in args.fields:
+                    key = field.key
+                    value = field.value
+                    
+                    # Additional debugging
+                    print(f"Field key: {key}")
+                    print(f"Field value type: {type(value)}")
+                    print(f"Field value: {value}")
+                    
+                    # Extract string value
+                    if hasattr(value, 'string_value'):
+                        flattened_args[key] = value.string_value
+                        print(f"Extracted string value: {value.string_value}")
+                    elif hasattr(value, 'number_value'):
+                        flattened_args[key] = value.number_value
+                    elif hasattr(value, 'bool_value') and value.bool_value is not None:
+                        flattened_args[key] = value.bool_value
+            
+            # Added additional debug information
+            print(f"Final flattened args: {flattened_args}")
+        
+        except Exception as e:
+            print(f"Error extracting function args: {e}")
+        
+        return flattened_args
+
+    def extract_tool_calls(self, input_string: str) -> List[Dict[str, Any]]:
+        """
+        Extract tool calls from input string, handling various inconsistent formats.
+        
+        Args:
+            input_string (str): The input string containing tool calls.
+        
+        Returns:
+            list: A list of dictionaries representing the parsed tool calls.
+        """
+        tool_calls = []
+        
+        # Existing tag-based extraction (retain if needed)
+        complete_matches = self.complete_pattern.findall(input_string)
+        if complete_matches:
+            for match in complete_matches:
+                tool_calls.extend(self._extract_json_objects(match))
+            return tool_calls
+            
+        partial_matches = self.partial_pattern.findall(input_string)
+        if partial_matches:
+            for match in partial_matches:
+                tool_calls.extend(self._extract_json_objects(match))
+            return tool_calls
+            
+        # Fallback: Attempt to parse the entire string
+        tool_calls.extend(self._extract_json_objects(input_string))
+        
+        return tool_calls
+
+    def _extract_json_objects(self, text: str) -> List[Dict[str, Any]]:
+        """
+        Extract and parse multiple JSON objects from a string.
+        """
+        json_objects = []
+        potential_jsons = text.split(';')
+        
+        for json_str in potential_jsons:
+            parsed_obj = self._clean_and_parse_json(json_str)
+            if parsed_obj:
+                json_objects.append(parsed_obj)
+                
+        return json_objects
+
+    def _clean_and_parse_json(self, json_str: str) -> Optional[Dict[str, Any]]:
+        """
+        Clean and parse a JSON string, handling common formatting issues.
+        """
+        try:
+            json_str = json_str.strip()
+            if json_str.startswith('{') or json_str.startswith('['):
+                return json.loads(json_str)
+            return None
+        except json.JSONDecodeError:
+            return None
+
+    def validate_tool_call(self, tool_call: Dict[str, Any]) -> bool:
+        """
+        Validate if a tool call has the required fields.
+        """
+        return (
+            isinstance(tool_call, dict) and
+            'name' in tool_call and
+            isinstance(tool_call['name'], str)
+        )
+    
+    def extract_function_call(self, response_parts: List[Any]) -> Dict[str, Any]:
+        """
+        Extract function call details from the response parts.
+        
+        Args:
+            response_parts (list): The list of response parts from the chat model.
+        
+        Returns:
+            dict: A dictionary containing the function name and flattened arguments.
+        """
+        for part in response_parts:
+            # Debug print
+            print(f"Examining part: {part}")
+            print(f"Part type: {type(part)}")
+            
+            # Check for function_call attribute
+            if hasattr(part, 'function_call') and part.function_call:
+                function_call = part.function_call
+                
+                # Debug print
+                print(f"Function call: {function_call}")
+                print(f"Function call type: {type(function_call)}")
+                print(f"Function args: {function_call.args}")
+                
+                # Extract function name
+                function_name = getattr(function_call, 'name', None)
+                if not function_name:
+                    continue  # Skip if function name is missing
+                
+                # Extract function arguments
+                function_args = self._extract_function_args(function_call.args)
+                
+                return {
+                    "name": function_name,
+                    "args": function_args
+                }
+        return {}

document_logs_2024-12-20.txt

@@ -0,0 +1 @@
+2024-12-20 12:49:01,713 - INFO - ID: ea205193-4582-44bc-ab71-80176aac7aef, Snippet:  [SSW](https://sswalfa.surabaya.go.id/home)  [![](https://sswalfa.surabaya.go.id/assets/images/logo-

testcode.py

@@ -1,4 +1,140 @@
+import os
+import httpx
+from dotenv import load_dotenv
+from typing import Dict, Any, Optional, List, Iterable
+from datetime import datetime
+import logging
+import asyncio
+import json
+import google.generativeai as genai
 import PIL.Image
 
-organ = PIL.Image.open("organ.jpg")
-print(organ)
+# Import custom modules
+from app.utils.load_env import ACCESS_TOKEN, WHATSAPP_API_URL, GEMNI_API, OPENAI_API
+from app.utils.system_prompt import system_prompt, agentic_prompt
+from google.generativeai.types import content_types
+from testtool import ToolCallParser, FunctionExecutor
+from app.services.search_engine import google_search, set_light_values
+
+# Load environment variables
+load_dotenv()
+
+# Set up logging
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
+logger = logging.getLogger(__name__)
+
+def tool_config_from_mode(mode: str, fns: Iterable[str] = ()):
+    """
+    Create a tool config with the specified function calling mode.
+    """
+    return content_types.to_tool_config(
+        {"function_calling_config": {"mode": mode, "allowed_function_names": fns}}
+    )
+
+def transform_result_to_response(results: List[Dict[str, Any]]) -> Dict[str, Any]:
+    """
+    Transform a list of result objects into a structured response dictionary.
+    """
+    response = {}
+    for res in results:
+        if res.get("status") == "success":
+            function_name = res.get("function")
+            function_result = res.get("result")
+            response[function_name] = function_result
+        else:
+            # Handle individual failures if necessary
+            response[res.get("function", "unknown_function")] = {
+                "error": "Function execution failed."
+            }
+    return response
+
+async def process_tool_calls(input_string: str) -> List[Dict[str, Any]]:
+    """
+    Processes all tool calls extracted from the input string and executes them.
+    """
+    tool_calls = ToolCallParser.extract_tool_calls(input_string)
+    logger.info(f"Extracted tool_calls: {tool_calls}")
+    results = []
+    for tool_call in tool_calls:
+        result = await FunctionExecutor.call_function(tool_call)
+        results.append(result)
+    return results
+
+async def main():
+    # Define available functions and tool configuration
+    available_functions = ["google_search", "set_light_values"]
+    config = tool_config_from_mode("any", fns=available_functions)
+
+    # Define chat history
+    history = [{"role": "user", "parts": "This is the chat history so far"}]
+
+    # Configure the Gemini API
+    genai.configure(api_key=GEMNI_API)
+    model = genai.GenerativeModel(
+        "gemini-1.5-pro-002",
+        system_instruction=agentic_prompt,
+        tools=[google_search, set_light_values]
+    )
+
+    # Start chat with history
+    chat = model.start_chat(history=history)
+
+    # Send the user's message and await the response
+    try:
+        response = chat.send_message(
+            "find the cheapest flight price from Medan to Jakarta on 1st January 2025",
+            tool_config=config
+        )
+    except Exception as e:
+        logger.error(f"Error sending message: {e}")
+        return
+
+    # Ensure that response.parts exists and is iterable
+    if not hasattr(response, 'parts') or not isinstance(response.parts, Iterable):
+        logger.error("Invalid response format: 'parts' attribute is missing or not iterable.")
+        return
+
+    # Convert response parts to a single input string
+    input_string = "\n".join(str(part) for part in response.parts)
+    logger.info(f"Input string for tool processing: {input_string}")
+
+    # Process tool calls
+    try:
+        results = await process_tool_calls(input_string)
+    except Exception as e:
+        logger.error(f"Error processing tool calls: {e}")
+        return
+
+    # Log and print the results
+    logger.info("Results from tool calls:")
+    for result in results:
+        logger.info(json.dumps(result, indent=4))
+        print(json.dumps(result, indent=4))
+
+    # Transform the results into the desired response format
+    responses = transform_result_to_response(results)
+
+    # Build the response parts for the chat
+    try:
+        response_parts = [
+            genai.protos.Part(
+                function_response=genai.protos.FunctionResponse(
+                    name=fn,
+                    response={"result": val}
+                )
+            )
+            for fn, val in responses.items()
+        ]
+    except Exception as e:
+        logger.error(f"Error building response parts: {e}")
+        return
+
+    # Send the function responses back to the chat
+    try:
+        final_response = chat.send_message(response_parts)
+        print(final_response.text)
+    except Exception as e:
+        logger.error(f"Error sending final response: {e}")
+
+if __name__ == "__main__":
+    asyncio.run(main())