""" Service for chat functionality. """ import os import json import uuid from datetime import datetime from typing import List, Dict, Any, Optional from ai_service.config import config from ai_service.models.model_service import model_service from ai_service.models.model_parameters import ModelParameters class ChatService: """Service for chat functionality.""" def __init__(self): """Initialize the chat service.""" # Ensure data directory exists os.makedirs(os.path.dirname(config.SQLITE_DB_PATH), exist_ok=True) # For now, we'll store chat data in a simple JSON file self.chats_file = os.path.join(os.path.dirname(config.SQLITE_DB_PATH), 'chats.json') self._load_chats() def _load_chats(self): """Load chats from file.""" if os.path.exists(self.chats_file): try: with open(self.chats_file, 'r') as f: self.chats = json.load(f) except Exception as e: print(f"Error loading chats: {str(e)}") self.chats = {} else: self.chats = {} def _save_chats(self): """Save chats to file.""" try: with open(self.chats_file, 'w') as f: json.dump(self.chats, f, indent=2) except Exception as e: print(f"Error saving chats: {str(e)}") def create_chat(self, user_id: str, title: Optional[str] = None, model_id: Optional[str] = None, is_team_chat: bool = False) -> str: """ Create a new chat. Args: user_id: ID of the user creating the chat. title: Optional title for the chat. model_id: Optional model ID to use for this chat. is_team_chat: Whether this is a team chat. Returns: ID of the created chat. """ # Generate a unique ID for the chat chat_id = str(uuid.uuid4()) # Create chat data self.chats[chat_id] = { 'id': chat_id, 'title': title or f"Chat {len(self.chats) + 1}", 'user_id': user_id, 'model_id': model_id or config.DEFAULT_MODEL, 'is_team_chat': is_team_chat, 'created_at': datetime.utcnow().isoformat(), 'updated_at': datetime.utcnow().isoformat(), 'messages': [], 'team_members': [user_id] if is_team_chat else [] } # Save chats to file self._save_chats() return chat_id def add_message(self, chat_id: str, content: str, user_id: str, is_user_message: bool = True) -> Dict[str, Any]: """ Add a message to a chat. Args: chat_id: ID of the chat. content: Message content. user_id: ID of the user sending the message. is_user_message: Whether this is a user message (vs. bot message). Returns: Added message. """ if chat_id not in self.chats: raise ValueError(f"Chat with ID {chat_id} not found") # Create message data message = { 'id': str(uuid.uuid4()), 'content': content, 'user_id': user_id if is_user_message else None, 'is_user_message': is_user_message, 'timestamp': datetime.utcnow().isoformat() } # Add message to chat self.chats[chat_id]['messages'].append(message) # Update chat timestamp self.chats[chat_id]['updated_at'] = datetime.utcnow().isoformat() # Save chats to file self._save_chats() return message def get_chat(self, chat_id: str) -> Optional[Dict[str, Any]]: """ Get a chat by ID. Args: chat_id: ID of the chat. Returns: Chat data if found, None otherwise. """ return self.chats.get(chat_id) def get_user_chats(self, user_id: str) -> List[Dict[str, Any]]: """ Get all chats for a user. Args: user_id: ID of the user. Returns: List of chat data. """ user_chats = [] for chat_id, chat in self.chats.items(): # Include private chats owned by the user if chat['user_id'] == user_id and not chat['is_team_chat']: user_chats.append(chat) # Include team chats where the user is a member elif chat['is_team_chat'] and user_id in chat['team_members']: user_chats.append(chat) # Sort by updated_at (newest first) user_chats.sort(key=lambda x: x['updated_at'], reverse=True) return user_chats def add_team_member(self, chat_id: str, user_id: str) -> bool: """ Add a user to a team chat. Args: chat_id: ID of the team chat. user_id: ID of the user to add. Returns: True if addition was successful, False otherwise. """ if chat_id not in self.chats: return False chat = self.chats[chat_id] if not chat['is_team_chat']: return False if user_id not in chat['team_members']: chat['team_members'].append(user_id) self._save_chats() return True def remove_team_member(self, chat_id: str, user_id: str) -> bool: """ Remove a user from a team chat. Args: chat_id: ID of the team chat. user_id: ID of the user to remove. Returns: True if removal was successful, False otherwise. """ if chat_id not in self.chats: return False chat = self.chats[chat_id] if not chat['is_team_chat']: return False if user_id in chat['team_members']: chat['team_members'].remove(user_id) self._save_chats() return True def delete_chat(self, chat_id: str) -> bool: """ Delete a chat. Args: chat_id: ID of the chat to delete. Returns: True if deletion was successful, False otherwise. """ if chat_id not in self.chats: return False del self.chats[chat_id] self._save_chats() return True def get_chat_response(self, chat_id: str, message: str, user_id: str, use_rag: bool = False, temperature: Optional[float] = None, max_tokens: Optional[int] = None, top_p: Optional[float] = None, frequency_penalty: Optional[float] = None, presence_penalty: Optional[float] = None, stop_sequences: Optional[List[str]] = None, system_prompt: Optional[str] = None, min_p: Optional[float] = None, top_k: Optional[int] = None, repeat_penalty: Optional[float] = None, function_calling: Optional[bool] = None) -> Dict[str, Any]: """ Get a response from the chatbot. Args: chat_id: ID of the chat. message: User message. user_id: ID of the user sending the message. use_rag: Whether to use RAG (Retrieval Augmented Generation). temperature: Controls randomness in the response. max_tokens: Maximum number of tokens to generate. top_p: Nucleus sampling parameter. frequency_penalty: Penalizes repeated tokens. presence_penalty: Penalizes repeated topics. stop_sequences: Sequences where the API will stop generating. system_prompt: System prompt to guide the model's behavior. min_p: Minimum probability threshold for token selection. top_k: Only sample from the top k tokens. repeat_penalty: Penalty for repeating tokens. function_calling: Whether to enable function calling. Returns: Bot response message. """ if chat_id not in self.chats: raise ValueError(f"Chat with ID {chat_id} not found") chat = self.chats[chat_id] # Add user message to chat self.add_message(chat_id, message, user_id, is_user_message=True) # Prepare conversation context for the model context = [] for msg in chat['messages'][-10:]: # Use last 10 messages as context role = "user" if msg['is_user_message'] else "assistant" context.append({ "role": role, "content": msg['content'] }) # Create model parameters model_params = ModelParameters( temperature=temperature, max_tokens=max_tokens, top_p=top_p, frequency_penalty=frequency_penalty, presence_penalty=presence_penalty, stop_sequences=stop_sequences, system_prompt=system_prompt, min_p=min_p, top_k=top_k, repeat_penalty=repeat_penalty, function_calling=function_calling ) # Get response from model model_id = chat['model_id'] response_text = model_service.generate_response( model_id=model_id, prompt=message, context=context, use_rag=use_rag, model_params=model_params ) # Add bot response to chat response_message = self.add_message( chat_id=chat_id, content=response_text, user_id=user_id, is_user_message=False ) return response_message # Create a singleton instance chat_service = ChatService()