app/services/ folder was completed

app/services/__init__.py

@@ -0,0 +1 @@
+# Services Package

app/services/chat_service.py

@@ -0,0 +1,231 @@
+"""
+CHAT SERVICE MODULE
+===================
+
+This service owns all chat session and conversation logic. It is used by the
+/chat and /chat/realtime endpoints. Designed for single-user use: the server
+has one ChatService and one in-memory session store; the user can have many
+sessions (each identified by session_id).
+
+RESPONSIBILITIES:
+- get or create session(session id): Return existing session or create new one.
+  If the user sends a session_id that was used before (e.g. before a restart),
+  we try to load it from disk so the conversation continues.
+- add_message / get_chat_history: Keep messages in memory per session.
+- format_history_for_llm: Turn the message list into (user, assistant) pairs
+  and trim to MAX_CHAT_HISTORY_TURNS so we don't overflow the prompt.
+- process message / process realtime message: Add user message, call Groq (or
+  RealtimeGroq), add assistant reply, return reply.
+- save_chat_session: Write session to database/chats_data/*.json so it persists
+  and can be loaded on next startup (and used by the vector store for retrieval).
+"""
+
+import json
+import logging
+from pathlib import Path
+from typing import List, Optional, Dict
+import uuid
+
+from config import CHATS_DATA_DIR, MAX_CHAT_HISTORY_TURNS
+from app.models import ChatMessage, ChatHistory
+from app.services.groq_service import groqservice
+from app.services.realtime_service import RealtimeGroqService
+
+
+logger = logging.getLogger("J.A.R.V.I.S")
+
+
+# ============================================================================
+# CHAT SERVICE  
+# ============================================================================
+
+class ChatService:
+    """
+    Manages chat sessions: in-memory message lists, load/save to disk, and
+    calling Groq (or Realtime) to get replies. All state for active sessions
+    is in self.sessions; saving to disk is done after each message so
+    conversations survive restarts.
+    """
+
+    def __init__(self, groq_service: 'GroqService', realtime_service: RealtimeGroqService = None):
+        """Store references to the Groq and Realtime services; keep sessions in memory."""
+        self.groq_service = groq_service
+        self.realtime_service = realtime_service
+        # Map: session_id -> list of ChatMessage (user and assistant messages in order).
+        self.sessions: Dict[str, List[ChatMessage]] = {}
+
+    # -------------------------------------------------------------------------
+    # SESSION LOAD / VALIDATE / GET-OR-CREATE
+    # -------------------------------------------------------------------------
+
+    def load_session_from_disk(self, session_id: str) -> bool:
+        """
+        Load a session from database/chats_data/ if a file for this session_id exists.
+
+        File name is chat_{safe_session_id}.json where safe_session_id has dashes/spaces removed.
+        On success we put the messages into self.sessions[session_id] so later requests use them.
+        Returns True if loaded, False if file missing or unreadable.
+        """
+        # Sanitize ID for use in filename (no dashes or spaces).
+        safe_session_id = session_id.replace("-", "_").replace(" ", "_")
+        filename = f"chat_{safe_session_id}.json"
+        filepath = CHATS_DATA_DIR / filename
+
+        if not filepath.exists():
+            return False
+        
+        try:
+            with open(filepath, "r", encoding="utf-8") as f:
+                chat_dict = json.load(f)
+            # Convert strored dicts back to ChatMessage objects.
+            messages = [
+                ChatMessage(role=msg.get("role"), content=msg.get("content"))
+                for msg in chat_dict.get("messages", [])
+            ]    
+            self.sessions[session_id] = messages
+            return True
+        except Exception as e:
+            logger.warning("Failed to load session %s from disk: %s", session_id, e)
+            return False
+        
+    def validate_session_id(self, session_id: str) -> bool:
+        """
+        Return True if session_id is safe to use (non-empty, no path traversal, length <= 255).
+        Used to reject malicious or invalid IDs before we use them in file paths.
+        """
+        if not session_id or not session_id.strip():
+            return False
+        # Block path traversal and path separators.
+        if ".." in session_id or "/" in session_id or "\\" in session_id:
+            return False
+        if len(session_id) > 255:
+            return False
+        return True
+        
+    def get_or_create_session(self, session_id: Optional[str] = None) -> str:
+        """
+        Return a session ID and ensure that session exists in memory.
+
+        - If session_id is None: create a new session a new UUID and return it.
+        - If session_id is provided: validate it; if it's in self.sessions return it;
+          else try to load from disk; if not found, create a new session with that ID.
+        Raises ValueError is session_id is invalid (empty, path traversal, or too long).  
+        """
+        if not session_id:
+            new_sessio_id = str(uuid.uuid4())
+            self.sessoions[new_sessio_id] = []
+            return new_sessio_id
+        
+        if not self.validate_session_id(session_id):
+            raise ValueError(
+                f"Invalid session_id format: {session_id}. Session ID must be non-empty, "
+                "not contain path traversal characters, and be under 255 characters."
+            )
+
+        if session_id in self.sessions:
+            return session_id
+        
+        if self.load_session_from_disk(session_id):
+            return session_id
+        
+        # New session with this ID (e.g. client an ID was never saved).
+        self.sessions[session_id] = []
+        return session_id
+    
+
+    # ---------------------------------------------------------------
+    # MESSAGES AND HISTORY FROMATTING
+    # ---------------------------------------------------------------
+    def add_message(self, session_id: str, role: str, content: str):
+        """Append one message (user or assistant) to the session's message list. Creates session if missing."""
+        if session_id not in self.sessions:
+            self.sessions[session_id] = []
+        self.sessions[session_id].append(ChatMessage(role=role, content=content))      
+    
+    def get_chat_history(self, session_id: str) -> List[ChatMessage]:
+        """Return the list of messages for this session (chronological). Empty list if session unkown."""
+        return self.sessions.get(session_id, [])
+    
+    def format_history_for_llm(self, session_id: str, exclude_last: bool = False ) -> List[tuple]:
+        """
+        Build a list of (user_text, assistant_text) pairs for the LLM prompt.
+
+        We only include complete pairs and cap at MAX_CHAT_HISTORY_TRUNS (e.g. 20)
+        so the prompt does not grow unbounded. If exclude_last is True we drop the
+        last message (the current user message that we are about to reply to).
+        """
+        messages = self.get_chat_history(session_id)
+        history = []
+        # If exclude_last, we skip the last message (the current user message we are about to reply to).
+        messages_to_process = messages[:-1] if exclude_last and messages else messages
+        i = 0
+        while i < len(messages_to_process) - 1:
+            user_msg = messages_to_process[i]
+            ai_msg = messages_to_process[i + 1]
+            if user_msg.role == "user" and ai_msg.role == "assistant":
+                history.append((user_msg.content, ai_msg.content))
+                i += 2  
+            else:
+                i += 1
+        # Keep only the most recent turns so the prompt does not exceed token limit.
+        if len(history) > MAX_CHAT_HISTORY_TURNS:
+            history = history[-MAX_CHAT_HISTORY_TURNS:]
+            return history 
+        
+    # --------------------------------------------------------------------------
+    # PROCESS MESSAGE (GENERAL AND REALTIME)
+    # --------------------------------------------------------------------------
+
+    def process_message(self, session_id: str, user_message: str) -> str:
+        """
+        Handle one general-chat message: add user message, call Groq (no web search), add reply, return it.
+        """
+        self.add_message(session_id, "user", user_message) 
+        chat_history = self.format_history_for_llm(session_id, exclude_last=True)
+        response = self.groq_service.get_response(question=user_message, chat_history=chat_history)
+        self.add_messages(session_id, "assistant", response)
+        return response
+    
+    def process_realtime_message(self, session_id: str, user_message: str) -> str:
+        """
+        Handle one realtime message: add user message, call realtime service (Tavily + Groq), add reply, return it.
+        Uses the same session as process_message so history is shared. Raises ValueError if realtime_service is None.
+        """
+        if not self.realtime_service:
+            raise ValueError("Realtime service is not initialized. Cannot process realtime queries.")
+        self.add_message(session_id, "user", user_message)
+        chat_history = self.format_history_for_llm(session_id, exclude_last=True)
+        response = self.realtime_service.get_response(question=user_message, chat_history=chat_history)
+        self.add_messages(session_id, "assistant", response)
+        return response 
+
+    # -------------------------------------------------------------
+    # PERSIST SESSION TO DISK
+    # -------------------------------------------------------------
+
+    def save_chat_session(self, sessiom_id: str):
+        """
+        Write this session's messages to database/chats_data/chat_{safeid}.json.
+
+        Called after each message so the conversation is persisted. The vector store
+        is rebuilt on startup from these files, so new chats are included after restart.
+        If the session is missing or empty we do nothing. On write error we only log.
+        """
+        if session_id not in self.sessions or not self.sessions[session_id]:
+            return
+
+        messages = self.sessions[session_id]
+        safe_session_id = session_id.replace("-", "").replace(" ", "_")
+        filename = f"chat_{safe_session_id}.json"
+        filepath = CHATS_DATA_DIR / filename
+        chat_dict = {
+            "session_id": sessiom_id,
+            "messages": [{"role":msg.role, "content":msg.content} for msg in messages]
+        }
+
+        try:
+            with open(filepath, "w", encoding="utf-8") as f:
+                json.dump(chat_dict, f, indent=2, ensure_ascii=False)
+        except Exception as e:
+            logger.error("Failed to save chat session %s to disk: %s", sessiom_id, e)
+                 

app/services/groq_service.py

@@ -0,0 +1,228 @@
+"""
+GROQ SERVICE MODULE
+===================
+
+This module handles general chat: no web search, only the Groq LLM plus context 
+from the vector store (learning data + past chats). Used by ChatService for
+POST /chat.
+
+MULTIPLE API KEYS (round-robin and fallback):
+  - You can set multiple Groq API keys in .env: Groq_API_KEY, GROQ_API_KEY_2,
+    GROQ_API_KEY_3, ... (no limits).
+  - Each request uses one key in roatation: 1st request -> 1st key, 2nd request ->
+    2nd key, 3rd request -> 3rd key, then back to 1st key, and so on. Every Key
+    is used one-by-on so usage is spread across keys.
+  - The round-robin counter is shared across all instances (GroqService and 
+    RealtimeGroqService), so both /chat and /chat/realtime endpoints use the 
+    same rotation sequence.
+  - If the chosen key fail (rate limit 429 or  any error), we try the next key,
+    then the next, until one succeeds or all have been tried.
+  - All APi key usage is logged with masked keys (first 8 and last 4 chars visible)
+    for security and debugging purposes.
+
+FLOW;
+  1. get_response(question, chat_history) is called.
+  2. We ask the vector store for the top-k chunks most similar to the question (retrieval).
+  3. We build a system message: JARVIS_SYSTEM_PROMPT + current time + retrived context.
+  4. We send to Groq using the next key in rotation (or fallback to next key on failure).
+  5. We return the assistant's reply.
+
+Context is only what we retrieve (no full dump of learning data ), so token usage stays bounded.  
+"""
+
+from typing import List, Optional
+from lanchain_groq import ChatGroq
+from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
+from langchain_core.messages import HumanMessage, AImessage
+
+import logging
+
+from config import Groq_API_KEYS, GROQ_MODEL, JARVIS_SYSTEM_PROMPT
+from app.services.vector_store import VectorStoreService
+from app.utils.time_info import get_time_information
+
+logger = logging.getLogger("J.A.R.V.I.S")
+
+
+# ==============================================================================
+# HELPER: EXCAPE CURLY BRACES FOR LANGCHAIN
+# ==============================================================================
+# LangChain prompt templates use {variable_name}. If learning data or chat
+# content contains { or }, the template engine can break. Doubling them
+# makes them literal in the final string
+
+def escape_curly_braces(text: str) -> str:
+    """
+    Double every { and } so LangChain does not treat them as template variables/
+    Learning data or chat content might contain { or }; without escaping escapin, invoke() can fail.
+    """
+    if not text:
+        return text
+    return text.replace("{", "{{").replace("}", "}}")
+
+
+def _is_rate_limit_error(exc: BaseException) -> bool:
+    """
+    Return True if the exception indicates a Groq rate limit (e.g. 429, tokens per day).
+    used for logging; actual fallback tries the next key on any failure when multiple keys exist.
+    """
+    msg = str(exc).lower()
+    return "429" in str(exc) or "rate limit" in msg or "tokens per day" in msg
+
+
+def _mask_api_key(key: str) -> str:
+    """
+    Mask an APi key for safe logging. Shows first 8 and last 4 characters, masks the middle.
+    Example: gsk_1234567890abcdef -> gsk_1234...cdef
+    """
+    if not key or len(key) <= 12:
+        return "***masked***"
+    return f"{key[:8]}...{key[-4:]}"
+
+
+# =============================================================
+# GROQ SERVICE CLAS
+# =============================================================
+
+class GroqService:
+    """
+    General chat: retrieves context from the vector store and calls the Groq LLM.
+    Supports multiple API keys: each reuqest uses the next key in rotation (one-by-one),
+    and if that key fails, the server tries the next key until one succeeds or all fail.
+
+    ROUND-ROBIN BEHAVIOR:
+    - Request 1 uses key 0 (first key)
+    - Request 2 uses key 1 (second key)
+    - Request 3 uses key 2 (third key)
+    - After all keys are used, cycles back to key 0 
+    - If a key fails (rate limit, error), tries the next key in sequence
+    - All reuqests share the same roundrobin counter (class-level)
+    """
+
+    # Class-level counter shared across all instances (GroqService and Realtimeg\GroqService)
+    # This ensures round-robin works across both /chat and /chat/realtime endpoints
+    _shared_key_index = 0 
+    _lock = None # Will be set threading.Lock if threading needed (currently single-threaded)
+
+    def __init__(self, vector_store_service: VectorStoreService):
+        """
+        Create one Groq LLm client per APi key and store the vector store for retrieval.
+        se;f.llms[i] corresponds to GROQ_API_KEY[i]; request N uses key at index (N % len(keys)).
+        """
+        if not GROQ_API_KEYS:
+            raise ValueError(
+                "No Groq APi keys configured. Set GROQ_API_KEY (and optionally GROQ_API_KEY_2, GROQ_API_KEY_3, ...) in .env"
+            )
+        # One ChatGroq instance per key: each reuqest will use one of these in rotation.
+        self.llm = [
+            ChatGroq(
+                groq_api_key=key,
+                model_name=GROQ_MODEL,
+                temperature=0.8,
+            )
+            for key in GROQ_API_KEYS
+        ]
+        self.vector_store_service = vector_store_service
+        logger.info(f"Initialized GroqService with {len(GROQ_API_KEYS)} API key(s)")
+
+    def _invoke_llm(
+            self,
+            prompt: ChatPromptTemplate,
+            messages: list,
+            question: str,
+    ) -> str:
+        """
+        Call the LLM using the next key in rotation; on failure, try the next key until one secceeds.
+
+        - Round-robin: the request uses key at index (_shared_key_index % n), then we increment
+          _shared_key_index so the next reuqest uses the next key. All instances share the same counter,
+        - Fallback: if the chosen key raises (e.g. 429 rate limit), we try the next key, then the next,
+          until one returns successfully or we have tried all keys.
+        Returns response.content. Raises if all keys fail.  
+        """
+        n = len(self.llms)
+        # Which key to try first for this reqests (rount-robin: request 1 -> key 0, request 2 -> key , ...).
+        # Use class-level counter so all instances (GroqService and RealtimeGroqService) sjare the same rotation.
+        start_o = GroqService._shared_key_index % n
+        current_key_index = GroqService._shared_key_index
+        GroqService._shared_key_index += 1 # Next request will use the next key.
+
+        # Log which key we're using (maksed for security)
+        masked_key = _mask_api_key(GROQ_API_KEYS[start_o])
+        logger.info(f"Using API key {start_o + 1}/{n} (round-robin index: {current_key_index}): {masked_key}")
+
+        last_exc = None
+        keys_tried = []
+        # Try each key in order starting from start_i (wrap around with % n).
+        for j in range(n):
+            i = (start_i + j) % n
+            keys_tried.append(i)
+            try:
+                # Build chain with this key's LLM and invoke once.
+                chain = prompt | self.llms[i]
+                response = chain.invoke({"history": messages, "question": question})
+                # Log succes if we had to fall back to a different key
+                if j > 0:
+                    masked_success_key = _mask_api_key(GROQ_API_KEYS[i])
+                    logger.info(f"Fallback successful: API key #{i + 1}/{n} succeeded: {masked_success_key}")
+                return response.content
+            except Exception as e:
+                last_exc = e
+                masked_failed_key = _mask_api_key(GROQ_API_KEYS[i])
+                if _is_rate_limit_error(e):
+                    logger.warning(f"API key #{i + 1}/{n} rate limited: {masked_failed_key} - {str(e)[:100]}")
+                else:
+                    logger.warning(f"API key #{i + 1}/{n} failed : {masked_failed_key} - {str(e)[:100]}")
+                # If we have more than one key, try the next one; otherwise raise immediately .
+                if n > 1:
+                    continue
+                raise Exception(f"Error getting response from Groq: {str(e)}") from e
+        # All keys were tried and all failed; raise the last exception.
+        masked_all_keys = ", ".join([_mask_api_key(GROQ_API_KEYS[i]) for i in keys_tried])
+        logger.error(f"All API keys failed. Tried keys: {masked_all_keys}")
+        raise Exception(f"Error getting response from Groq: {str(last_exc)}") from last_exc
+    
+    def get_response(
+            self, 
+            question: str,
+            chat_history: Optional[List[tuple]] = None
+    ) -> str:
+        """
+        Return the assistant's reply for this question (general chat, no web search).
+        Retrives context from the vector store, builds the prompt, then calls , _invoke_llm
+        which use the next API key in rotation and fallback to other keys on failure.
+        """
+        try:
+            # Get relevant chunks from the learning data and past chats (bounded token usage).
+            # If retrieval fails (e.g. vector store not ready), use empty context so the LLM still answer.
+            context = ""
+            try:
+                retriever = self.vector_store_service.get_retriever(k=10)
+                context_docs = retriever.invoke(question)
+                context = "\n".join([doc.page_content for doc in context_docs]) if context_docs else ""
+            except Exception as e:
+                logger.warning(f"Vector store retrieval failed, using empty context: %s", retrieve_err)
+
+            # Build the system message: personality + current time + retrieved context.
+            time_info = get_time_information()
+            system_message = JARVIS_SYSTEM_PROMPT + f"\n\nCurrent time and date: {time_info}"
+            if context:
+                system_message += f"\n\nRelevant Context from your learning data and past conversations:\n{escape_curly_braces(context)}"
+
+            # Prompt template: system message, chat history placeholder, current question.
+            prompt = ChatPromptTemplate.from_messages([
+                    ("system", system_message),
+                    MessagesPlaceholder(variable_name="history"),
+                    ("human", "{question}"),
+            ])    
+            # Convert (user, assistant) pairs to LangChain message objects.
+            messages = []
+            if chat_history:
+                for human_msg, ai_msg in chat_history:
+                    messages.append(HumanMessage(content=human_msg))
+                    messages.append(AImessage(content=ai_msg))
+                
+            # Use next key in rotation; on failure, try remaining keys until one succeeds or all fail.
+            return self._invoke_llm(prompt, messages, question)
+        except Exception as e:
+            raise Exception(f"Error response from groq: {str(e)}") from e

app/services/realtime_service.py

@@ -0,0 +1,168 @@
+"""
+REALTIME GROQ SERVICE MODULE
+=============================
+
+Extents GroqService to add Tavily web search before calling the LLM. Used by 
+ChatService for POST /chat/realtime. Same session and history as general chat;
+the only difference is we run a Tavily search for the user's question and add 
+the results to the system message, them call Groq.
+
+ROUND-ROBIN API KEYS:
+  - Shares the same round-robin counter as GroqService (class-level _shared_key_index)
+  - This means /chat and /chat/realtime requests use the same rotation sequence
+  - Example: If /chat uses key 1, the next /chat/realtime request will use key 2
+  - All API key usage is logged wih masked keys for security and debugging
+
+FLOW:
+  1. search_tavily(question): call Tavily API, format results as text (or "" on failure).
+  2. get_response(question, chat_history): add search results to system message,
+     then same as parent: retrieve context from vector store, build prompt , call Groq.
+
+If TAVILY_API_KEY is not set, tavily_clinet is None and search_tavily returns "";
+the user still gets an answer from Groq with no search results.
+"""
+
+from typing import List, Optional
+from tavily import TavilyClient
+import logging
+import os 
+
+from app.services.groq_service import GroqService, escape_curly_braces
+from app.services.vector_store import VectorStoreService
+from app.utils.time_info import get_time_information
+from app.utils.retry import with_retry
+from config import JARVIS_SYSTEM_PROMPT
+from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
+from langchain_core.messages import HumanMessage, AIMessage
+
+
+logger = logging.getLogger("J.A.R.V.I.S")
+
+
+# ==============================================================================
+# REALTIME GROQ SERVICE CLASS (extends GroqService)
+# ==============================================================================
+
+class RealtimeGroqService(GroqService):
+    """
+    Same as GroqService but runs a Tavily web search first and adds the results
+    to the system message. If Tavily is missing or fails, we still call Groq with
+    no search results (user gets and answer without real-time data).
+    """
+
+    def __init__(self, vector_store_service: VectorStoreService):
+        """Call parent init (Groq LLM + vector store); then create Tavily client if key is set."""
+        super().__init__(vector_store_service)
+        tavily_api_key = os.getenv("TAVILY_API_KEY", "")
+        if tavily_api_key:
+            self.tavily_client = TavilyClient(api_key=tavily_api_key)
+            logger.info("Tavily search client initialized seccessfully")
+        else:
+            self.tavily_client = None
+            logger.warning("TAVILY_API_KEY not set. Realtime search will be unavailable.")
+
+    def search_tavily(self, ques: str, num_results: int = 5) -> str:
+        """
+        Call Tavily API with the given query and return formatted result text for the prompt.
+        On any failure (no key, rate limit, network) we return "" so the LLM still gets a reply.
+        """
+        if not self.tavily_client:
+            logger.warning("Tavily client not initialized. TAVILY_API_KEY not set.")
+            return ""
+        
+        try:
+            # Perform Tavily search with retries for rate limits and transient errors.
+            response = with_retry(
+                lambda: self.tavily_client.search(
+                    query=query,
+                    search_depth="basic",
+                    max_results=num_results,
+                    include_answer=False,
+                    include_raw_content=False,
+                ),
+                max_retries=3,
+                initial_delay=1.0,
+            )
+
+            results = response.get('results', [])
+
+            if not results:
+                logger.warning(f"No Tavily search results found for query: {query}")
+                return ""
+            
+            # Format search results as text for the system message. 
+            formatted_results = f"Search resluts for '{query}':\n[start]\n"
+
+            for i, result in enumerate(results[:num_results], 1):
+                title = result.get('title', 'No title')
+                content = result.get('content', 'No description')
+                url = result.get('url', '')
+
+                formatted_results += f"Title: {title}\n"
+                formatted_results += f"Description: {content}\n"
+                if url:
+                    formatted_results += f"URL: {url}\n"
+                formatted_results += "\n"
+
+            formatted_results += "[end]"
+
+            logger.info(f"Tavily search completed for query: {query} with {len(results)} results")
+            return formatted_results
+        
+        except Exception as e:
+            # If search fails (network error, rate limit, etc.), log and return empty
+            # The AI will still respond using its knwonledge, just without real-time data 
+            logger.error(f"error performing Tavily search: {e}")
+            return ""
+        
+    def get_response(self, question: str, chat_history: Optional[List[tuple]] = None) -> str:
+        """
+        Run Tavily search for the question, add results to system message, then call the Groq
+        via the parent's _invoke_llm (same multi-key round-robin and fallback as general chat).
+        """
+        try:
+            logger.info(f"Searching Tavily for: {question}")
+            search_results = self.search_tavily(question, num_results=5)
+
+            # Retrieve context from vector store (learning data + past chats).
+            # If retrieval fails, use empty context so the LLM still answers (e.eg. with Tavily results).
+            context = ""
+            try:
+                retriever = self.vector_store_service.get_retriever(k=10)
+                context_docs = retriever.invoke(question)
+                context = "\n\n".join([doc.page_content for doc in context_docs]) if context_docs else ""
+            except Exception as retrieval_err:
+                logger.warning("Vector store retrieval failed, using empty context: %s", retrieval_err)
+
+            # Build system message: personality + time + Tavily results + retrieved context.
+            time_info = get_time_information()
+            system_message += JARVIS_SYSTEM_PROMPT + f"\n\nCurrent time and data: {time_info}"
+
+            if search_result:
+                escaped_search_results = escape_curly_braces(search_results)
+                system_message += f"\n\nRecent search results:\n{escaped_search_results}"
+
+            if context:
+                escaped_context = escape_curly_braces(context)
+                system_message += f"\n\nRelevant context from your learning data and past conversations:\n{escaped_context}"
+
+            prompt = ChatPromptTemplate.from_messages([
+                ("system", system_message),
+                MessagesPlaceholder(variable_name="history"),
+                ("human","{question}"),
+            ])
+            messages =[]
+            if chat_history:
+                for human_msg, ai_msg in chat_history:
+                    messages.append(HumanMessage(content=human_msg))
+                    messages.append(AIMessage(content=ai_msg))
+
+            # uses same round-robin and fallback as general chat: next key one-by-one, try next on failure.
+            response_content = self._invoke_llm(prompt, messages, question) 
+            logger.info(f"Realtime response generated for: {question}")
+            return response_content
+        
+        except Exception as e:
+            logger.error(f"Error in realtime get_response: {e}", exc_info=True)
+            # Re-raise so main.py can return 429 (rate limit) or 500 consistently with general chat.
+            raise

app/services/vector_store.py

@@ -0,0 +1,142 @@
+"""
+VECTOR STORE SERVICE MODULE
+===========================
+
+This service builds and queries the FAISS vector index used for context retrieval.
+Learning data (database/learning_data/*.txt) and past chats (database/chats_data/*.json)
+are loaded at startup, split into chunks, embedded with HuggingFace, and stored in FAISS.
+When the user ask a question we embed it and retrieve and k most similar chunks; only
+those chunks are sent to the LLM, so token usage is bounded.
+
+LIFECYCLE:
+  - create_vector_store(): Load all .txt and .json, chunk, embed, build FAISS, save to disk.
+    Called once at startup. Restart the server after adding new .txt files so they are included.
+  - get_retriever(k): Return a retriever that fetches k nearest chunks for a query string. 
+  - save_vector_store(): Write the current FAISS index to database/vector_store/ (called after create).
+
+Embeddings run locally (sentence-transformers); no extra API key. Groq and Realtime services
+call get_retriever() for every request to get context.  
+"""
+
+import json
+import logging
+from pathlib import Path
+from typing import List, Optional
+from langchain_text_splitters import RecursiveCharacterTextSplitter
+from langchain_huggingface import HuggingFaceEmbeddings 
+from langchain_community.vectorstores import FAISS
+from langchain_core.documents import Document
+
+from config import (
+    LEARNING_DATA_DIR,
+    CHAS_DATA_DIR,
+    VECTOR_STORE_DIR,
+    EMBEDDING_MODEL,
+    CHUNK_SIZE,
+    CHUNK_OVERLAP,
+)
+
+
+logger = logging.getLogger("J.A.R.V.I.S")
+
+
+# =========================================================
+# VECTOR STORE SERVICE CLASS
+# =========================================================
+
+class VectorStoreService:
+    """
+    Builds a FAISS index from learning_data .txt files and chats_data .json files,
+    and provides a retriever to fetch the k most relevant chunks for a query.
+    """
+
+    def __init__(self):
+        """Create the embedding model (local) and text splitter; vector_store is set in create_vector_store()."""
+        # Embeddings run locally (no API key); used to convert text into vectors for similarity search.
+        self.embeddings = HuggingFaceEmbeddings(
+            model_name=EMBEDDING_MODEL,
+            model_kwargs={"device":"cpu"},
+        )
+        self.text_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=CHUNK_SIZE,
+            chunk_overlap=CHUNK_OVERLAP,
+        )
+        self.vector_store: Optional[FAISS] = None
+
+        # ----------------------------------------------------------------------
+        # LoAD DOcUMENTS FROM DISK
+        # ----------------------------------------------------------------------
+
+    def load_learning_data(self) -> List[Document]:
+        """Read all .text files in database/learning_data/ and return one Document per file (content + source name). """
+        documents = []
+        for file_path in list(LEARNING_DATA_DIR.glob("*.txt")):
+            try:
+                with open(file_path, "r", encoding="utf-8") as f:
+                    content = f.read().strip()
+                    if content:
+                        documents.append(Document(page_content=content,, metadata={"source": str(file_path.name)}))
+            except Exception as e:
+                logger.warning("Could not load learning dat file %s: %s", file_path, e)
+        return documents
+
+    def load_chat_history(self) -> List[Document]:
+        """load all .json files in database/chats_data/; turn each into one Document (User:/Assistant: lines)."""
+        documents = []
+        for file_path in list(CHATS_DATA_DIR.glob("*.json")):
+            try:
+                with open(file_path, "r", encoding="utf-8") as f:
+                    chat_data = json.load(f)
+                messages = chat_data.get("messages", [])
+                # Format as "User: ..." / "Assistant: ..." so the retriever can match past conversations.
+                chat_content = "\n".join([
+                    f"User: {msg.get('content', '')}"if msg.get('role') == 'user'
+                    else f"Assistant: {msg.get('content', '')}"
+                    for msg in messages
+                ])
+                if chat_content.strip():
+                    documents.append(Document(page_content=chat_content, metadata={"source": f"chat_{file_path.stem}"}))
+            except Exception as e:
+                logger.warning("Could not load chat history file %s: %s", file_path, e)
+        return documents
+
+    # -------------------------------------------------------
+    # BUILD AND SAVE FAISS INDEX
+    # -------------------------------------------------------
+
+    def create_vector_store(self) -> FAISS:
+        """
+        Load learning_data + chats_data, embed, build FAISS index, save to disk.
+        Called once at startup. If there are no documents we create a tiny placeholder index.
+        """
+        learning_docs = self.load_learning_data()
+        chat_docs = self.load_chat_history()
+        all_documents = learning_docs + chat_docs
+
+        if not all_documents:
+            # Placeholder so get_retriever() never fails; return this single chunk for any query.
+            self.vector_store = FAISS.from_texts(["No data available yet."], self.embeddiings)
+        else:
+            chunks = self.text_splitter.split_document(all_documents)
+            self.vector_store = FAISS.from_documents(chunks, self.embeddings)
+
+        self.save_vector_store()
+        return self.vector_store
+
+    def save_vector_store(self):
+        """Write the current FAISS index to database/vector_store/. On error we only log."""
+        if self.vector_store:
+            try:
+                self.vector_store.save_local(str(VECTOR_STORE_DIR))
+            except Exception as e:
+                logger.error("failed to save vector store to disk: %s", e)
+
+    # ---------------------------------------------------------------------------
+    # RETRIEVER FOR CONTEXT
+    # ---------------------------------------------------------------------------
+
+    def get_retriever(self, k: int = 10):
+        """Return a retriever that returns that k most similar chunks for a query string."""
+        if not self.vector_store:
+            raise RuntimeError("Vector store not initialized. This should not happen.")
+        return self.vector_store.as_retriever(search_kwargs={"k":k})