enhance chat api

src/backend/api_routes/chat_api.py

@@ -1,36 +1,14 @@
-from fastapi import APIRouter, HTTPException
-from services import llm_model_service
-from utils import logger
+from fastapi import APIRouter, HTTPException, Depends
+from fastapi.responses import JSONResponse
+from services import llm_model_service, pinecone_service, embedding_service
 from services.schemas import ConversationInput
+from utils import logger
 
-
-
-logger= logger.get_logger()
+logger = logger.get_logger()
 
 router = APIRouter(prefix="/chat", tags=["Chat"])
 
-
-# @router.post("/query")
-# async def chat_query(request: ChatRequest):
-#     try:
-#         logger.info("Trying to fetch response")
-#         query = request.query
-#         context = pinecone_service.retrieve_context_from_pinecone(query)
-#         response= llm_model_service.generate_response_with_context(query,context)
-#         logger.info("Fetched response")
-#         return response
-#     except Exception as e:
-#         raise HTTPException(status_code=500, detail=str(e))
-
-# @router.get("/history/{date}", response_model=list)
-# async def get_history(date: str):
-#     try:
-#         history = get_chat_history(date)
-#         return history
-#     except Exception as e:
-#         raise HTTPException(status_code=500, detail=str(e))
-
-@router.post("/get-health-advice/")
+@router.post("/get-health-advice", response_model=dict)
 async def get_health_advice(input_data: ConversationInput):
     """
     Handles requests from the frontend and fetches advice using the `get_health_advice()` function.
@@ -38,12 +16,58 @@ async def get_health_advice(input_data: ConversationInput):
     Args:
     - input_data (ConversationInput): User's conversation history.
 
+    Example Input:
+    {
+        "conversation_history": [
+            {"role": "user", "content": "I've been feeling tired lately. What should I do?"},
+            {"role": "assistant", "content": "Are you experiencing any other symptoms?"},
+            {"role": "user", "content": "No, I just feel drained even after sleeping well."}
+        ]
+    }
+
     Returns:
     - dict: Contains 'reply' with the assistant's response.
+
+    Raises:
+    - HTTPException (400): If conversation history or user query is missing.
+    - HTTPException (500): If an internal error occurs during response generation.
     """
+    if not input_data.conversation_history:
+        logger.warning("Empty conversation history received.")
+        raise HTTPException(status_code=400, detail="Conversation history cannot be empty.")
+
     try:
-        assistant_reply = llm_model_service.get_health_advice(input_data.conversation_history)
-        return {"reply": assistant_reply}
+        last_entry = input_data.conversation_history[-1]
+        if not isinstance(last_entry, dict) or last_entry.get("role") != "user":
+            logger.warning("Invalid conversation entry format or missing user query.")
+            raise HTTPException(status_code=400, detail="Invalid conversation entry or missing user query.")
+
+        user_query = last_entry.get("content")
+        if not user_query:
+            logger.warning("User query content is missing in the conversation history.")
+            raise HTTPException(status_code=400, detail="User query content cannot be empty.")
+
+        logger.info(f"Received user query: {user_query}")
+        query_embeddings = embedding_service.get_text_embedding(user_query)
+
+        db_response = pinecone_service.retrieve_context_from_pinecone(query_embeddings)
+        logger.info("Fetched DB response successfully.")
+
+        assistant_reply = llm_model_service.get_health_advice(
+            user_query, db_response, input_data.conversation_history
+        )
+
+        if not assistant_reply:
+            logger.warning("Assistant generated an empty response.")
+            raise HTTPException(status_code=500, detail="Assistant generated an empty response.")
+
+        logger.info("Health advice generated successfully.")
+        return JSONResponse(content={"reply": assistant_reply}, status_code=200)
+
+    except HTTPException as http_exc:
+        logger.error(f"HTTPException occurred: {http_exc.detail}")
+        raise http_exc
 
     except Exception as e:
+        logger.error(f"Unexpected error: {e}", exc_info=True)
         raise HTTPException(status_code=500, detail="Error generating response. Please try again later.")

src/backend/api_routes/knowledge_base_api.py

@@ -1,7 +1,6 @@
 from fastapi import APIRouter,HTTPException
 from data import pinecone_db
 from models.schemas import UpsertRequest,DeleteRequest,MetadataRequest
-from data import pinecone_db
 import pandas as pd
 
 router = APIRouter(prefix="/knowledge-base", tags=['Knowledge Base Operations'])

src/backend/chatbot.py

@@ -1,23 +0,0 @@
-from services.pinecone_service import retrieve_relevant_metadata
-from services.supabase_service import store_chat_history
-from sentence_transformers import CrossEncoder
-
-reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
-
-def rerank_results(query, results):
-    pairs = [(query, result["metadata"]["text"]) for result in results]
-    scores = reranker.predict(pairs)
-    return [x for _, x in sorted(zip(scores, results), key=lambda pair: pair[0], reverse=True)]
-
-def chatbot_response(query):
-    results = retrieve_relevant_metadata(query)
-    if results:
-        reranked_results = rerank_results(query, results)
-        best_answer = reranked_results[0]["metadata"]["question"]
-    else:
-        best_answer = "I'm sorry, I couldn't find an answer for your query."
-
-    # store_chat_history(query, best_answer)
-    return best_answer
-
-

src/backend/main.py

@@ -1,17 +1,30 @@
 from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
 from api_routes.chat_api import router as chat_router
-from api_routes.knowledge_base_api import router as knowledge_base_router
-from api_routes.chat_history_db_api import router as chat_history_router
+# from api_routes.knowledge_base_api import router as knowledge_base_router
+# from api_routes.chat_history_db_api import router as chat_history_router
 
 app = FastAPI(
     title="Yuvabe Care Companion AI",
     description="A chatbot for health-related queries"
 )
 
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+    expose_headers=["Cache-Control"],
+)
+
+@app.get("/", tags=["Root"])
+def read_root():
+    return {"message": "Hello World"}
+
 # Register Routes
 app.include_router(chat_router)
-app.include_router(knowledge_base_router)
-app.include_router(chat_history_router)
+# app.include_router(knowledge_base_router)
+# app.include_router(chat_history_router)
 
 # Health Check Endpoint
 @app.get("/health", tags=["Health Check"])

src/backend/models/embedding_model.py

@@ -1,15 +0,0 @@
-from sentence_transformers import SentenceTransformer
-from utils import logger
-
-logger = logger.get_logger()
-
-model = SentenceTransformer("all-MiniLM-L6-v2")
-
-def get_text_embedding(search_query: str):
-    try:
-        text_embedding = model.encode(search_query, convert_to_tensor=True).cpu().numpy().tolist()
-        logger.info("Text embedding successfully retrieved.")
-        return text_embedding
-    except Exception as e:
-        logger.error(f"Error while getting embedding for text: {e}")
-        raise

src/backend/models/schemas.py

@@ -1,26 +0,0 @@
-from pydantic import BaseModel
-from typing import List,Dict, Optional
-
-class Chat_Response(BaseModel):
-    prompt: Optional[List] = None
-    response: Optional[Dict] = None
-
-class UpsertRequest(BaseModel):
-    data: list  # Expecting a list of JSON objects (rows of data)
-
-class DeleteRequest(BaseModel):
-    ids_to_delete: list
-
-class MetadataRequest(BaseModel):
-    prompt: str
-    n_result: int = 3
-    score_threshold: float = 0.45
-
-class ChatRequest(BaseModel):
-    query: str
-
-class ChatResponse(BaseModel):
-    response: str
-
-class ChatHistoryResponse(BaseModel):
-    date: str

src/backend/services/llm_model_service.py

@@ -1,66 +1,124 @@
 import os
+from typing import List, Dict, Optional
 from groq import Groq
 from dotenv import load_dotenv
-from services import pinecone_service
 from utils import logger
 
+# Logger instance
 logger = logger.get_logger()
 
+# Load environment variables
 load_dotenv()
 
-LLM_MODEL_NAME=os.environ.get("LLM_MODEL_NAME")
-GROQ_KEY = os.environ.get("GROQ_API")
-client = Groq(api_key=GROQ_KEY)
-
-SYSTEM_PROMPT = [
-    {"role": "system", "content": "You are Yuvabe Care Companion AI, an advanced healthcare assistant designed to provide guidance strictly on medical, mental health, and wellness topics. You must not respond to questions unrelated to healthcare."},
-    {"role": "system", "content": "Yuvabe Care Companion AI is powered by the LLaMA 3.3-70B Versatile model, optimized for comprehensive and responsible healthcare support."},
-    {"role": "system", "content": "Your knowledge is up-to-date with the latest medical guidelines as of July 2024, but you are NOT a replacement for professional medical advice."},
-    {"role": "system", "content": "Always provide accurate, empathetic, and responsible responses while reminding users to consult healthcare professionals when necessary."},
-    {"role": "system", "content": "If a user asks something unrelated to healthcare, politely decline to answer and remind them that your expertise is limited to healthcare topics."},
-    {"role": "system", "content": "Refer to conversation history to provide context to your response."},
-    {"role": "system", "content":"You are a helpful, friendly, and engaging assistant. Respond with clear explanations, positive language, and a conversational tone. Use emojis to enhance clarity and create a warm interaction. Keep responses concise but informative. If the user seems confused, provide step-by-step guidance."},
-    {"role": "system", "content": "If the user asks questions about technology, entertainment, news, or unrelated topics, respond with: 'I'm here to assist with healthcare-related queries only.'"},
+# Configuration constants
+LLM_MODEL_NAME = os.getenv("LLM_MODEL_NAME")
+GROQ_API_KEY = os.getenv("GROQ_API")
+
+# Initialize Groq client
+client = Groq(api_key=GROQ_API_KEY)
+
+# System prompt structure s
+SYSTEM_PROMPT: List[Dict[str, str]] = [
+    {"role": "system", "content": "You are Yuvabe Care Companion AI, an advanced healthcare assistant..."},
+    {"role": "system", "content": "Your knowledge is up-to-date with the latest medical guidelines as of July 2024..."},
+    {"role": "system", "content": "Always provide accurate, empathetic, and responsible responses..."},
+    {"role": "system", "content": "If the user asks non-healthcare questions, politely decline..."},
     {"role": "system", "content": "You were created by Velu R, an AI model developer."}
 ]
 
-def get_health_advice(conversation_history):
+# Constants for token limits and configurations
+MAX_TOKENS = 1024
+MAX_HISTORY_TOKENS = 1000
+DEFAULT_TEMPERATURE = 0.7
+
+def truncate_conversation_history(history: List[Dict[str, str]], max_tokens: int = MAX_HISTORY_TOKENS) -> List[Dict[str, str]]:
     """
-    Generates a health-related response based on relevant context from the vector database.
-    Falls back to LLM's internal knowledge if no relevant context is found.
+    Truncates conversation history to maintain token limits.
+    Retains the most recent interactions if token count exceeds the threshold.
+
+    Args:
+    - history (List[Dict[str, str]]): List of conversation messages
+    - max_tokens (int): Maximum allowable tokens for conversation history
 
     Returns:
-    - str: Assistant's reply containing medical guidance or information.
+    - List[Dict[str, str]]: Truncated conversation history
     """
-    try:
+    total_tokens = sum(len(msg["content"]) for msg in history)
+    if total_tokens > max_tokens:
+        logger.warning(f"Conversation history exceeds {max_tokens} tokens. Truncating...")
+        while total_tokens > max_tokens and history:
+            history.pop(0)
+            total_tokens = sum(len(msg["content"]) for msg in history)
+    return history
+
+def build_prompt(
+    user_query: str,
+    db_response: Optional[str],
+    conversation_history: List[Dict[str, str]]
+) -> List[Dict[str, str]]:
+    """
+    Constructs the message prompt for the LLM with system prompts, context, and user queries.
+
+    Args:
+    - user_query (str): The query entered by the user
+    - db_response (Optional[str]): Context retrieved from the vector database
+    - conversation_history (List[Dict[str, str]]): Previous conversation history
+
+    Returns:
+    - List[Dict[str, str]]: Constructed prompt messages
+    """
+    conversation_history = truncate_conversation_history(conversation_history)
 
-        # Step 1: Retrieve context from Pinecone
-        user_query = conversation_history[-1]['content']
-        logger.info(f"Received user query: {user_query}")
-        db_response = pinecone_service.retrieve_context_from_pinecone(user_query)
-        logger.info(f"Fetched DB response: {db_response}")
-
-        # Step 2: Check if context is relevant
-        if db_response and "No relevant information found" not in db_response:
-            # Step 3a: Include relevant context if found
-            messages = SYSTEM_PROMPT + [
-                {"role": "system", "content": f"Relevant Context:{db_response}"},
-                {"role": "user", "content": user_query}
-            ]+conversation_history
-        else:
-            # Step 3b: Use model's internal knowledge if no valid DB response
-            messages = SYSTEM_PROMPT + [
-                {"role": "system", "content": "Please respond using your internal medical knowledge."},
-                {"role": "user", "content": user_query}
-            ] + conversation_history
-
-        # Step 4: Generate response from LLaMA
+    if db_response and "No relevant information found" not in db_response:
+        return SYSTEM_PROMPT + [
+            {"role": "system", "content": f"Relevant Context: {db_response}"},
+            {"role": "user", "content": user_query}
+        ] + conversation_history
+    else:
+        return SYSTEM_PROMPT + [
+            {"role": "system", "content": "Please respond using your internal medical knowledge."},
+            {"role": "user", "content": user_query}
+        ] + conversation_history
+
+def get_health_advice(
+    user_query: str,
+    db_response: Optional[str],
+    conversation_history: List[Dict[str, str]]
+) -> str:
+    """
+    Generates a healthcare-related response using context from the vector database
+    or the LLM's internal knowledge.
+
+    Args:
+    - user_query (str): The user's question or statement
+    - db_response (Optional[str]): Retrieved context for the query
+    - conversation_history (List[Dict[str, str]]): History of the conversation
+
+    Returns:
+    - str: The assistant's response
+    """
+    try:
+        messages = build_prompt(user_query, db_response, conversation_history)
+        
         response = client.chat.completions.create(
             model=LLM_MODEL_NAME,
-            messages=messages
+            messages=messages,
+            max_tokens=MAX_TOKENS,
+            temperature=DEFAULT_TEMPERATURE
         )
 
         assistant_reply = response.choices[0].message.content.strip()
+        logger.info(f"Generated response: {assistant_reply}")
         return assistant_reply
+
+    except (ConnectionError, TimeoutError) as e:
+        logger.error(f"Network error: {e}")
+        return "I'm currently unable to connect to the system. Please try again later."
+
+    except KeyError as e:
+        logger.error(f"Unexpected response structure: {e}")
+        return "I'm sorry, but I couldn't process your request at the moment."
+
     except Exception as e:
+        logger.error(f"Unexpected error occurred: {e}")
         return "I'm sorry, but I'm unable to provide a response right now. Please try again later."

src/backend/services/pinecone_service.py

@@ -1,7 +1,7 @@
 import os
-import sys
-src_directory = os.path.abspath(os.path.join(os.path.dirname(__file__), "../..", "backend"))
-sys.path.append(src_directory)
+# # import sys
+# # src_directory = os.path.abspath(os.path.join(os.path.dirname(__file__), "../..", "backend"))
+# # sys.path.append(src_directory)
 from pinecone import Pinecone, ServerlessSpec
 import time
 from tqdm import tqdm
@@ -225,11 +225,9 @@ def upsert_vector_data(df: pd.DataFrame):
 
     logger.info("All question-answer pairs stored successfully!")
 
-def retrieve_context_from_pinecone(prompt, n_result=3, score_threshold=0.5):
+def retrieve_context_from_pinecone(embedding, n_result=3, score_threshold=0.5):
 
     index = initialize_pinecone_index(PINECONE,INDEX_NAME)
-    # Generate embedding for the provided prompt
-    embedding = get_text_embedding(prompt)
     # Query Pinecone for relevant context
     response = index.query(
         top_k=n_result,

src/backend/services/schemas.py

@@ -1,9 +1,39 @@
 from pydantic import BaseModel
-from typing import List
+from typing import List,Dict, Optional
 
 class ConversationInput(BaseModel):
     conversation_history: list[dict]
 
 class ChatHistoryRequest(BaseModel):
     conversation_id: str
-    messages: List[dict]
+    messages: List[dict]
+
+
+
+
+
+
+
+class Chat_Response(BaseModel):
+    prompt: Optional[List] = None
+    response: Optional[Dict] = None
+
+class UpsertRequest(BaseModel):
+    data: list  # Expecting a list of JSON objects (rows of data)
+
+class DeleteRequest(BaseModel):
+    ids_to_delete: list
+
+class MetadataRequest(BaseModel):
+    prompt: str
+    n_result: int = 3
+    score_threshold: float = 0.45
+
+class ChatRequest(BaseModel):
+    query: str
+
+class ChatResponse(BaseModel):
+    response: str
+
+class ChatHistoryResponse(BaseModel):
+    date: str

src/frontend/pages/chatbot.py

@@ -29,12 +29,12 @@ def fetch_health_advice(conversation_history):
 if "conversation_history" not in st.session_state:
     st.session_state.conversation_history = initialize_conversation()
 
-# # Display chat history
-# for message in st.session_state.conversation_history [-NUMBER_OF_MESSAGES_TO_DISPLAY:]:
-#     role = message["role"]
-#     avatar_image = "src/frontend/images/page_icon.jpg" if role == "assistant" else "src/frontend/images/page_icon.jpg" if role == "user" else None
-#     with st.chat_message(role, avatar=avatar_image):
-#         st.write(message["content"])
+# Display chat history
+for message in st.session_state.conversation_history [-NUMBER_OF_MESSAGES_TO_DISPLAY:]:
+    role = message["role"]
+    avatar_image = "src/frontend/images/page_icon.jpg" if role == "assistant" else "src/frontend/images/page_icon.jpg" if role == "user" else None
+    with st.chat_message(role, avatar=avatar_image):
+        st.write(message["content"])
 
 # User Input
 user_input = st.chat_input("Ask your health-related question:")