"""Conversation context management for RAG chatbot.""" from typing import Optional class ConversationContext: """Manages conversation context for multi-turn interactions.""" def __init__(self, max_history_length: int = 10): """Initialize conversation context manager. Args: max_history_length: Maximum number of messages to retain. """ self.max_history_length = max_history_length def build_query( self, question: str, conversation_history: Optional[list] = None, selected_text: Optional[str] = None, ) -> str: """Build an enhanced query with context. Args: question: Current user question. conversation_history: Previous conversation messages. selected_text: Optional selected text for context. Returns: Enhanced query string. """ query_parts = [] # Add selected text context if provided if selected_text: query_parts.append(f"Context from selected text: {selected_text}") # Add relevant conversation context if conversation_history: recent_context = self._extract_relevant_context( conversation_history, question, ) if recent_context: query_parts.append(f"Previous context: {recent_context}") # Add the current question query_parts.append(question) return " | ".join(query_parts) def _extract_relevant_context( self, conversation_history: list, current_question: str, ) -> str: """Extract relevant context from conversation history. Args: conversation_history: List of previous messages. current_question: Current question for relevance. Returns: Relevant context string. """ if not conversation_history: return "" # Get last few messages for context recent_messages = conversation_history[-self.max_history_length:] # Extract key topics from recent messages context_parts = [] for msg in recent_messages: if msg.get("role") == "user": content = msg.get("content", "")[:200] # Limit length context_parts.append(content) return " ... ".join(context_parts[-3:]) # Last 3 user messages def format_history_for_prompt( self, conversation_history: list, max_messages: int = 6, ) -> list: """Format conversation history for prompt. Args: conversation_history: Full conversation history. max_messages: Maximum messages to include. Returns: Formatted history list. """ if not conversation_history: return [] # Take only recent messages recent = conversation_history[-max_messages:] # Format for OpenAI API formatted = [] for msg in recent: role = msg.get("role", "user") content = msg.get("content", "") if role in ["user", "assistant", "system"]: formatted.append({"role": role, "content": content}) return formatted def summarize_context( self, conversation_history: list, ) -> dict: """Summarize conversation context. Args: conversation_history: Full conversation history. Returns: Context summary dict. """ if not conversation_history: return { "message_count": 0, "topics": [], "last_topic": None, } # Extract topics from user messages topics = [] for msg in conversation_history: if msg.get("role") == "user": content = msg.get("content", "") # Simple topic extraction - first few words words = content.split()[:5] if words: topics.append(" ".join(words)) return { "message_count": len(conversation_history), "topics": topics[-5:], # Last 5 topics "last_topic": topics[-1] if topics else None, }