monoai.chat

 21class Chat():
 22    """
 23    Chat class is responsible for handling the chat interface and messages history.
 24    
 25    Examples
 26    --------
 27    Basic usage:
 28    ```
 29    from monoai.models import Model
 30    from monoai.chat import Chat
 31    model = Model(provider="openai", model="gpt-4o-mini")
 32    chat = Chat(model=model)
 33    response = chat.ask("2+2") # {"chat_id": "...", "response": "4", "model": {...}}
 34    response = chat.ask("+2") # {"chat_id": "...", "response": "6", "model": {...}}
 35    ```    
 36
 37    With history:
 38    ```
 39    from monoai.models import Model
 40    from monoai.chat import Chat
 41    model = Model(provider="openai", model="gpt-4o-mini")
 42    
 43    # Create a new chat with JSON history
 44    chat = Chat(model=model, history="json")
 45    print(chat.chat_id) # 8cc2bfa3-e9a0-4b82-b46e-3376cd220dd3
 46    response = chat.ask("Hello! I'm Giuseppe") # {"chat_id": "...", "response": "Hello!", "model": {...}}
 47
 48    # Load a chat with JSON history
 49    chat = Chat(model=model, history="json", chat_id="8cc2bfa3-e9a0-4b82-b46e-3376cd220dd3")
 50    response = chat.ask("What's my name?") # {"chat_id": "...", "response": "Your name is Giuseppe", "model": {...}}
 51
 52    # Create a new chat with in-memory dictionary history
 53    chat = Chat(model=model, history="dict")
 54    print(chat.chat_id) # 8cc2bfa3-e9a0-4b82-b46e-3376cd220dd3
 55    response = chat.ask("Hello! I'm Giuseppe") # {"chat_id": "...", "response": "Hello!", "model": {...}}
 56    ```
 57
 58    With history summarizer:
 59
 60    ```
 61    from monoai.models import Model
 62    
 63    model = Model(provider="openai", model="gpt-4o-mini")
 64    chat = Chat(
 65        model=model,
 66        history="json", 
 67        history_summarizer_provider="openai", 
 68        history_summarizer_model="gpt-4o-mini", 
 69        history_summarizer_max_tokens=100
 70    )
 71                
 72    response = chat.ask("Hello! I'm Giuseppe") # {"chat_id": "...", "response": "Hello!", "model": {...}}
 73    response = chat.ask("What's my name?") # {"chat_id": "...", "response": "Your name is Giuseppe", "model": {...}}
 74    ```
 75
 76    With metadata for observability:
 77
 78    ```
 79    from monoai.models import Model
 80    
 81    model = Model(provider="openai", model="gpt-4o-mini")
 82    chat = Chat(model=model)
 83    
 84    # Pass metadata for tracking
 85    response = chat.ask(
 86        "Hello! I'm Giuseppe", 
 87        metadata={
 88            "user_id": "12345",
 89            "session_id": "abc-def-ghi",
 90            "feature": "chat"
 91        }
 92    )
 93    
 94    # Streaming with metadata
 95    async for chunk in chat.ask_stream(
 96        "Tell me a story",
 97        metadata={"user_id": "12345", "request_type": "story_generation"}
 98    ):
 99        print(chunk)
100    ```
101    """
102
103    _HISTORY_MAP = {
104        "json": JSONHistory,
105        "sqlite": SQLiteHistory,
106        "mongodb": MongoDBHistory,
107        "dict": DictHistory
108    }
109
110    def __init__(self, 
111                 model,
112                 system_prompt: Optional[Union[Prompt, str]] = None,
113                 max_tokens: Optional[int] = None,
114                 history: Union[str, BaseHistory] = "dict", 
115                 history_last_n: Optional[int] = None,
116                 history_path: Optional[str] = "histories",
117                 history_summarizer_provider: Optional[str] = None, 
118                 history_summarizer_model: Optional[str] = None,
119                 history_summarizer_max_tokens: Optional[int] = None,
120                 chat_id: Optional[str] = None) -> None:
121
122        """
123        Initialize a new Chat instance.
124
125        Parameters
126        ----------
127        model : Model
128            Model instance to use for the chat
129        system_prompt : str | Prompt, optional
130            System prompt or Prompt object. If string ends with '.prompt', 
131            it will be treated as a file path to load the prompt from.
132        max_tokens : int, optional
133            Maximum number of tokens for each request
134        history : str | BaseHistory, optional
135            The type of history to use for the chat. Options: "json", "sqlite", "mongodb", "dict".
136            Default is "dict" for in-memory storage.
137        history_last_n : int, optional
138            The last n messages to keep in the history. If None, keeps all messages.
139        history_path : str, optional
140            The path to the history storage (not used for "dict" history type).
141            Default is "histories".
142        history_summarizer_provider : str, optional
143            The provider of the history summarizer model.
144        history_summarizer_model : str, optional
145            The model name for the history summarizer.
146        history_summarizer_max_tokens : int, optional
147            The maximum number of tokens for the history summarizer.
148        chat_id : str, optional
149            The id of the chat to load. If not provided, a new chat will be created.
150            If provided but chat doesn't exist, a new chat will be created with this ID.
151
152        Raises
153        ------
154        ChatError
155            If invalid parameters are provided or initialization fails
156        """
157        try:
158            
159            self._max_tokens = max_tokens
160            self._model = model
161            
162            self._history_summarizer = None
163
164            # Initialize history
165            self._initialize_history(history, history_last_n, history_path)
166            
167            # Initialize history summarizer
168            self._initialize_history_summarizer(
169                history_summarizer_provider, 
170                history_summarizer_model, 
171                history_summarizer_max_tokens
172            )
173
174            # Process system prompt
175            processed_system_prompt = self._process_system_prompt(system_prompt)
176            print(processed_system_prompt)
177            # Initialize chat
178            if chat_id is None:
179                self.chat_id = self._history.new(processed_system_prompt)
180            else:
181                # Check if chat exists, if not create it
182                self.chat_id = self._initialize_or_create_chat(chat_id, processed_system_prompt)
183            
184            self._metadata = {"session_id": self.chat_id}
185
186        except Exception as e:
187            logger.error(f"Failed to initialize Chat: {e}")
188            raise ChatError(f"Chat initialization failed: {e}")
189
190    def _initialize_or_create_chat(self, chat_id: str, system_prompt: str) -> str:
191        """Initialize existing chat or create new one with specified ID.
192        
193        Parameters
194        ----------
195        chat_id : str
196            The chat ID to check/create
197        system_prompt : str
198            System prompt to use if creating new chat
199            
200        Returns
201        -------
202        str
203            The chat ID (either existing or newly created)
204        """
205        try:
206            # Try to load existing chat
207            existing_messages = self._history.load(chat_id)
208            
209            # Check if chat exists and has messages
210            if existing_messages and len(existing_messages) > 0:
211                logger.info(f"Loaded existing chat: {chat_id}")
212                return chat_id
213            else:
214                # Chat doesn't exist, create it with the specified ID
215                logger.info(f"Creating new chat with ID: {chat_id}")
216                return self._create_chat_with_id(chat_id, system_prompt)
217                
218        except Exception as e:
219            # If loading fails, assume chat doesn't exist and create it
220            logger.info(f"Failed to load chat {chat_id}, creating new one: {e}")
221            return self._create_chat_with_id(chat_id, system_prompt)
222    
223    def _create_chat_with_id(self, chat_id: str, system_prompt: str) -> str:
224        """Create a new chat with a specific ID.
225        
226        Parameters
227        ----------
228        chat_id : str
229            The specific chat ID to use
230        system_prompt : str
231            System prompt for the new chat
232            
233        Returns
234        -------
235        str
236            The chat ID
237        """
238        try:
239            # Create system message
240            system_message = {"role": "system", "content": system_prompt}
241            
242            # Store the system message with the specific chat_id
243            self._history.store(chat_id, [system_message])
244            
245            return chat_id
246            
247        except Exception as e:
248            logger.error(f"Failed to create chat with ID {chat_id}: {e}")
249            raise ChatError(f"Failed to create chat with ID {chat_id}: {e}")
250
251
252    def _initialize_history(self, history: Union[str, BaseHistory], history_last_n: Optional[int], history_path: Optional[str]) -> None:
253        """Initialize the history system.
254        
255        Parameters
256        ----------
257        history : Union[str, BaseHistory]
258            History type or instance
259        history_last_n : Optional[int]
260            Number of last messages to keep
261        history_path : Optional[str]
262            Path for history storage
263            
264        Raises
265        ------
266        ChatError
267            If history initialization fails
268        """
269        try:
270            if isinstance(history, str):
271                if history == "dict":
272                    # DictHistory doesn't need db_path
273                    self._history = self._HISTORY_MAP[history](last_n=history_last_n)
274                else:
275                    self._history = self._HISTORY_MAP[history](last_n=history_last_n, path=history_path)
276            else:
277                self._history = history
278        except Exception as e:
279            raise ChatError(f"Failed to initialize history: {e}")
280
281    def _initialize_history_summarizer(self, provider: Optional[str], model: Optional[str], max_tokens: Optional[int]) -> None:
282        """Initialize the history summarizer.
283        
284        Parameters
285        ----------
286        provider : Optional[str]
287            Provider for the summarizer model
288        model : Optional[str]
289            Model name for the summarizer
290        max_tokens : Optional[int]
291            Maximum tokens for summarization
292        """
293        if provider is not None and model is not None:
294            try:
295                self._history_summarizer = HistorySummarizer(
296                    Model(provider=provider, model=model, max_tokens=max_tokens)
297                )
298            except Exception as e:
299                logger.warning(f"Failed to initialize history summarizer: {e}")
300
301    def _process_system_prompt(self, system_prompt: Optional[Union[Prompt, str]]) -> str:
302        """Process and load the system prompt.
303        
304        Parameters
305        ----------
306        system_prompt : Optional[Union[Prompt, str]]
307            System prompt to process
308            
309        Returns
310        -------
311        str
312            Processed system prompt string
313        """
314        try:
315            if system_prompt is None:
316                return self._load_default_system_prompt()
317            elif isinstance(system_prompt, str) and system_prompt.endswith(".prompt"):
318                return self._load_prompt_file(system_prompt)
319            elif isinstance(system_prompt, Prompt):
320                return str(system_prompt)
321            else:
322                return system_prompt
323        except Exception as e:
324            logger.warning(f"Failed to process system prompt: {e}")
325            return ""
326
327    def _load_default_system_prompt(self) -> str:
328        """Load the default system prompt from file.
329        
330        Returns
331        -------
332        str
333            Default system prompt content
334        """
335        try:
336            prompt_path = Conf()["prompts_path"]
337            system_prompt_path = Path(prompt_path) / "system.prompt"
338            
339            if system_prompt_path.exists():
340                with open(system_prompt_path, "r", encoding="utf-8") as f:
341                    return f.read()
342            else:
343                logger.info("Default system prompt file not found, using empty prompt")
344                return ""
345        except Exception as e:
346            logger.warning(f"Failed to load default system prompt: {e}")
347            return ""
348
349    def _load_prompt_file(self, prompt_file: str) -> str:
350        """Load a prompt from a file.
351        
352        Parameters
353        ----------
354        prompt_file : str
355            Path to the prompt file
356            
357        Returns
358        -------
359        str
360            Content of the prompt file
361            
362        Raises
363        ------
364        ChatError
365            If file cannot be loaded
366        """
367        try:
368            prompt_path = Conf()["prompts_path"]
369            full_path = Path(prompt_path) / prompt_file
370            
371            if not full_path.exists():
372                raise FileNotFoundError(f"Prompt file not found: {full_path}")
373                
374            with open(full_path, "r", encoding="utf-8") as f:
375                return f.read()
376        except Exception as e:
377            raise ChatError(f"Failed to load prompt file {prompt_file}: {e}")
378
379    def _process_file_attachment(self, prompt: str, file: Optional[Union[str, bytes]], file_type: Optional[str]) -> Union[str, List[Dict[str, Any]]]:
380        """Process file attachment and return modified prompt.
381        
382        Parameters
383        ----------
384        prompt : str
385            Original prompt text
386        file : Optional[Union[str, bytes]]
387            File to attach (path, URL, or bytes)
388        file_type : Optional[str]
389            Type of the file
390            
391        Returns
392        -------
393        Union[str, List[Dict[str, Any]]]
394            Modified prompt with file content or multimodal content
395            
396        Raises
397        ------
398        ChatError
399            If file processing fails
400        """
401        if file is None:
402            return prompt
403            
404        try:
405            # Determine file type
406            if file_type is None and isinstance(file, str):
407                file_type = file.split(".")[-1].lower()
408                
409            if not file_type:
410                raise ChatError("File type could not be determined")
411                
412            conf = Conf()
413            
414            # Handle text files
415            if file_type in conf["supported_files"]["text"]:
416                return self._process_text_file(prompt, file)
417            # Handle image files
418            elif file_type in conf["supported_files"]["image"]:
419                return self._process_image_file(prompt, file)
420            else:
421                raise ChatError(f"Unsupported file type: {file_type}")
422                
423        except Exception as e:
424            raise ChatError(f"Failed to process file attachment: {e}")
425
426    def _process_text_file(self, prompt: str, file: Union[str, bytes]) -> str:
427        """Process text file attachment.
428        
429        Parameters
430        ----------
431        prompt : str
432            Original prompt text
433        file : Union[str, bytes]
434            Text file to process
435            
436        Returns
437        -------
438        str
439            Prompt with file content appended
440            
441        Raises
442        ------
443        ChatError
444            If text file processing fails
445        """
446        try:
447            if isinstance(file, str):
448                # Handle as file path or URL
449                if os.path.exists(file):
450                    with open(file, "r", encoding="utf-8") as f:
451                        file_content = f.read()
452                else:
453                    # Assume it's a URL or remote file
454                    file_content = file
455            else:
456                # Handle as bytes
457                file_content = file.decode("utf-8")
458                
459            return prompt + Conf()["default_prompt"]["file"] + file_content
460            
461        except Exception as e:
462            raise ChatError(f"Failed to process text file: {e}")
463
464    def _process_image_file(self, prompt: str, file: Union[str, bytes]) -> List[Dict[str, Any]]:
465        """Process image file attachment.
466        
467        Parameters
468        ----------
469        prompt : str
470            Original prompt text
471        file : Union[str, bytes]
472            Image file to process
473            
474        Returns
475        -------
476        List[Dict[str, Any]]
477            Multimodal content with text and image
478            
479        Raises
480        ------
481        ChatError
482            If image file processing fails
483        """
484        try:
485            if isinstance(file, str):
486                # Handle as file path or URL
487                if os.path.exists(file):
488                    # Convert local file to base64
489                    with open(file, "rb") as f:
490                        file_bytes = f.read()
491                    image_url = f"data:image/jpeg;base64,{base64.b64encode(file_bytes).decode('utf-8')}"
492                else:
493                    # Assume it's already a URL
494                    image_url = file
495            else:
496                # Handle as bytes
497                image_url = f"data:image/jpeg;base64,{base64.b64encode(file).decode('utf-8')}"
498                
499            return [
500                {"type": "text", "text": prompt},
501                {"type": "image_url", "image_url": image_url}
502            ]
503            
504        except Exception as e:
505            raise ChatError(f"Failed to process image file: {e}")
506
507    def _prepare_messages(self, prompt: Union[str, List[Dict[str, Any]]]) -> List[Dict[str, Any]]:
508        """Prepare messages for the API call.
509        
510        Parameters
511        ----------
512        prompt : Union[str, List[Dict[str, Any]]]
513            User prompt or multimodal content
514            
515        Returns
516        -------
517        List[Dict[str, Any]]
518            Formatted messages for API call
519            
520        Raises
521        ------
522        ChatError
523            If message preparation fails
524        """
525        try:
526            messages = self._history.load(self.chat_id)
527            messages = [{'role': d.get('role'), 'content': d.get('content')} for d in messages]
528
529            messages.append({"role": "user", "content": prompt})
530
531            # Apply history summarization if enabled
532            if self._history_summarizer is not None:
533                return self._apply_history_summarization(messages)
534            else:
535                return messages
536                
537        except Exception as e:
538            raise ChatError(f"Failed to prepare messages: {e}")
539
540    def _apply_history_summarization(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
541        """Apply history summarization to messages.
542        
543        Parameters
544        ----------
545        messages : List[Dict[str, Any]]
546            Full message history
547            
548        Returns
549        -------
550        List[Dict[str, Any]]
551            Summarized messages for API call
552        """
553        try:
554            summarized = self._history_summarizer.summarize(messages)
555            ask_messages = [messages[0], messages[-1]]
556            ask_messages[0]["content"] += Conf()["default_prompt"]["summary"] + summarized
557            return ask_messages
558        except Exception as e:
559            logger.warning(f"History summarization failed, using full history: {e}")
560            return messages
561
562    def _store_messages(self, messages: List[Dict[str, Any]], response_content: str) -> None:
563        """Store messages in history.
564        
565        Parameters
566        ----------
567        messages : List[Dict[str, Any]]
568            Messages to store
569        response_content : str
570            Assistant response content
571        """
572        try:
573            messages.append({"role": "assistant", "content": response_content})
574            self._history.store(self.chat_id, messages[-2:])
575        except Exception as e:
576            logger.error(f"Failed to store messages: {e}")
577
578
579    def ask(self, prompt: str, file: Optional[Union[str, bytes]] = None, file_type: Optional[str] = None, return_history: bool = False, metadata: Optional[Dict[str, Any]] = {}) -> Union[str, List[Dict[str, Any]]]:
580        """
581        Ask the model a question.
582
583        Parameters
584        ----------
585        prompt : str
586            The question to ask the model
587        file : str | bytes, optional
588            The file to attach to the message, if it's a string it will be treated as a local path or remote url to a file
589        file_type : str, optional
590            The type of the file
591        return_history : bool, optional
592            Whether to return the full history of the chat or only the response
593        metadata : Dict[str, Any], optional
594            Metadata to pass to the model for observability and tracking
595
596        Returns
597        -------
598        Dict[str, Any]
599            Dictionary containing:
600            - chat_id: str - The chat identifier
601            - response: str - The model response
602            - model: Dict[str, str] - Model provider and name
603            - history: List[Dict[str, Any]], optional - Full chat history if return_history is True
604
605        Raises
606        ------
607        ChatError
608            If the request fails or parameters are invalid
609        """
610
611        metadata = self._metadata | metadata
612
613        try:
614            # Process file attachment
615            processed_prompt = self._process_file_attachment(prompt, file, file_type)
616            
617            # Prepare messages
618            messages = self._prepare_messages(processed_prompt)
619            
620            # Make API call
621            response = self._model.ask(messages, metadata=metadata)
622            response_content = response["response"]
623            
624            # Store messages
625            self._store_messages(messages, response_content)
626            
627            response = {
628                "chat_id": self.chat_id,
629                "response": response_content,
630                "model": {
631                    "provider": self._model.provider,
632                    "name": self._model.model
633                }
634            }
635
636            if return_history:
637                response["history"] = messages
638            
639            return response
640                
641        except Exception as e:
642            logger.error(f"Ask request failed: {e}")
643            raise ChatError(f"Ask request failed: {e}")
644
645    async def ask_stream(self, prompt: str, file: Optional[Union[str, bytes]] = None, file_type: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> AsyncGenerator[str, None]:
646        """
647        Ask the model a question and stream the response.
648
649        Parameters
650        ----------
651        prompt : str
652            The question to ask the model
653        file : str | bytes, optional
654            The file to attach to the message
655        file_type : str, optional
656            The type of the file
657        metadata : Dict[str, Any], optional
658            Metadata to pass to the model for observability and tracking
659
660        Yields
661        ------
662        Dict[str, Any]
663            Streaming response chunks containing:
664            - chat_id: str - The chat identifier (first chunk)
665            - delta: str - Response text chunk (subsequent chunks)
666            - Other streaming metadata
667
668        Raises
669        ------
670        ChatError
671            If the request fails or parameters are invalid
672        """
673
674        metadata = self._metadata | metadata
675
676        try:
677            # Process file attachment
678            processed_prompt = self._process_file_attachment(prompt, file, file_type)
679            
680            # Prepare messages
681            messages = self._prepare_messages(processed_prompt)
682            
683            # Make streaming API call
684            response = self._model.ask_stream(messages, metadata=metadata)
685            yield {"chat_id": self.chat_id}
686            response_text = ""
687            async for chunk in response: 
688                if "delta" in chunk:
689                    response_text += chunk["delta"]
690                yield chunk
691
692            # Store messages
693            self._store_messages(messages, response_text)
694            
695        except Exception as e:
696            logger.error(f"Stream request failed: {e}")
697            raise ChatError(f"Stream request failed: {e}")
698
699    async def ask_async(self, prompt: str, file: Optional[Union[str, bytes]] = None, file_type: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> AsyncGenerator[str, None]:
700        """
701        Ask the model a question and stream the response asynchronously.
702
703        Parameters
704        ----------
705        prompt : str
706            The question to ask the model
707        file : str | bytes, optional
708            The file to attach to the message
709        file_type : str, optional
710            The type of the file
711        metadata : Dict[str, Any], optional
712            Metadata to pass to the model for observability and tracking
713
714        Yields
715        ------
716        str
717            Response text chunks as strings
718
719        Raises
720        ------
721        ChatError
722            If the request fails or parameters are invalid
723        """
724
725        metadata = self._metadata | metadata
726
727        try:
728            # Process file attachment
729            processed_prompt = self._process_file_attachment(prompt, file, file_type)
730            
731            # Prepare messages
732            messages = self._prepare_messages(processed_prompt)
733            
734            # Make streaming API call
735            response = await acompletion(
736                model=self._model,
737                messages=messages,
738                stream=True,
739                max_tokens=self._max_tokens,
740                metadata=metadata
741            )
742
743            response_text = ""
744            async for part in response:
745                content = part["choices"][0]["delta"]["content"] or ""
746                response_text += content
747                yield content
748                
749            # Store messages
750            self._store_messages(messages, response_text)
751            
752        except Exception as e:
753            logger.error(f"Async request failed: {e}")
754            raise ChatError(f"Async request failed: {e}")
755
756    def get_chat_info(self) -> Dict[str, Any]:
757        """
758        Get information about the current chat.
759
760        Returns
761        -------
762        Dict[str, Any]
763            Dictionary containing chat information
764        """
765        return {
766            "chat_id": self.chat_id,
767            "provider": self._model.provider,
768            "model": self._model.model,
769            "max_tokens": self._max_tokens,
770            "has_history_summarizer": self._history_summarizer is not None
771        }
772
773    def clear_history(self) -> None:
774        """
775        Clear the chat history.
776
777        Raises
778        ------
779        ChatError
780            If clearing history fails
781        """
782        try:
783            self._history.clear(self.chat_id)
784            logger.info(f"Cleared history for chat {self.chat_id}")
785        except Exception as e:
786            logger.error(f"Failed to clear history: {e}")
787            raise ChatError(f"Failed to clear history: {e}")