""" Utility functions for handling AI conversations about divination records (BaZi, LiuYao, etc.). """ import logging from typing import Dict, Any, Optional, List from django.utils import timezone from django.conf import settings from ai.services.factory import LLMServiceFactory from ai.utils.config import get_ai_config from ai.utils.bazi import prepare_bazi_prompt from ai.utils.liuyao_analysis import prepare_liuyao_prompt from ai.models import Conversation, Message, ConversationSubject, ConversationConfig from bazi.models import Person logger = logging.getLogger(__name__) def calculate_context_size( divination_prompt: str, conversation_history: List[Dict[str, str]], current_message: str, context_summary: Optional[str] = None ) -> int: """ Calculate the total context size in characters. Args: divination_prompt: Full divination prompt (BaZi, LiuYao, etc.) conversation_history: List of previous messages current_message: Current user message context_summary: Optional summary of older messages Returns: Total size in characters (bytes) """ # Divination prompt size prompt_size = len(divination_prompt.encode('utf-8')) # Context summary size summary_size = len(context_summary.encode('utf-8')) if context_summary else 0 # Conversation history size history_size = sum( len(msg.get('content', '').encode('utf-8')) for msg in conversation_history ) # Current message size message_size = len(current_message.encode('utf-8')) # Add overhead for formatting (role labels, separators, etc.) # Estimate ~50 bytes per message for formatting formatting_overhead = len(conversation_history) * 50 # Add overhead for summary section (~100 bytes) summary_overhead = 100 if context_summary else 0 total_size = prompt_size + summary_size + history_size + message_size + formatting_overhead + summary_overhead return total_size def generate_conversation_summary( conversation: Conversation, messages_to_summarize: List[Message], language: str = 'zh-hans', existing_summary: Optional[str] = None ) -> str: """ Generate a summary of conversation messages using AI. Uses incremental summarization: combines existing summary with new messages. Args: conversation: Conversation object messages_to_summarize: List of Message objects to summarize (new messages since last summarization) language: Language for summary ('zh-hans' or 'en') existing_summary: Optional existing summary to combine with new summary Returns: Combined summary text """ if not messages_to_summarize: return existing_summary or "" # Get divination context for summary prompt based on conversation subject type subject_obj = conversation.get_subject_object() if not subject_obj: logger.error(f"Conversation {conversation.id} has no valid subject object") return existing_summary or "" if conversation.subject.content_type == 'bazi': divination_prompt = prepare_bazi_prompt(subject_obj, language=language) divination_type_name = "BaZi (Four Pillars of Destiny)" if language == 'en' else "八字（四柱命理）" elif conversation.subject.content_type == 'liuyao': divination_prompt = prepare_liuyao_prompt(subject_obj, language=language) divination_type_name = "LiuYao (Six Lines Divination)" if language == 'en' else "六爻" else: logger.error(f"Unsupported content type: {conversation.subject.content_type}") return existing_summary or "" # Format messages for summarization conversation_text = "" for msg in messages_to_summarize: role_label = "User" if msg.role == 'user' else "Assistant" conversation_text += f"{role_label}: {msg.content}\n\n" # Create summarization prompt - incremental approach if language == 'en': if existing_summary: summary_prompt = f"""You are summarizing a conversation about a {divination_type_name} chart. {divination_type_name} Chart Context: {divination_prompt[:2000]}... (truncated for summary) Previous Conversation Summary: {existing_summary} New Conversation Messages to Add to Summary: {conversation_text} Please provide a concise summary that: 1. Combines the previous summary with the new messages 2. Preserves key questions asked by the user (from both old and new messages) 3. Preserves important insights and analysis provided (from both old and new messages) 4. Maintains context about specific BaZi elements discussed 5. Is concise but comprehensive (aim for 1000-2000 words total) 6. Integrates the new information naturally with the existing summary Updated Summary:""" else: summary_prompt = f"""You are summarizing a conversation about a {divination_type_name} chart. {divination_type_name} Chart Context: {divination_prompt[:2000]}... (truncated for summary) Conversation History to Summarize: {conversation_text} Please provide a concise summary of this conversation that: 1. Preserves key questions asked by the user 2. Preserves important insights and analysis provided 3. Maintains context about specific BaZi elements discussed 4. Is concise but comprehensive (aim for 1000-2000 words) Summary:""" else: if existing_summary: summary_prompt = f"""你正在总结一段关于{divination_type_name}命盘的对话。 {divination_type_name}命盘上下文： {divination_prompt[:2000]}...（为摘要而截断） 之前的对话摘要： {existing_summary} 需要添加到摘要的新对话消息： {conversation_text} 请提供一段简洁的摘要，要求： 1. 将之前的摘要与新消息结合起来 2. 保留用户提出的关键问题（包括旧消息和新消息中的） 3. 保留提供的重要见解和分析（包括旧消息和新消息中的） 4. 保持讨论的具体八字元素的上下文 5. 简洁但全面（目标1000-2000字） 6. 自然地将新信息与现有摘要整合 更新的摘要：""" else: summary_prompt = f"""你正在总结一段关于{divination_type_name}命盘的对话。 {divination_type_name}命盘上下文： {divination_prompt[:2000]}...（为摘要而截断） 需要总结的对话历史： {conversation_text} 请提供这段对话的简洁摘要，要求： 1. 保留用户提出的关键问题 2. 保留提供的重要见解和分析 3. 保持讨论的具体八字元素的上下文 4. 简洁但全面（目标1000-2000字） 摘要：""" # Get AI configuration - use conversation-specific config (e.g., 'bazi_conversation' instead of 'bazi') conversation_config_type = f"{conversation.subject.content_type}_conversation" config = get_ai_config(conversation_config_type) provider = config['provider'] model = config['model'] # Use the same model as the conversation llm_service = LLMServiceFactory.get_service(provider) if model: llm_service.change_model(model) # Generate summary summary = llm_service.get_completion( prompt=summary_prompt, temperature=0.3, # Lower temperature for more consistent summaries max_tokens=4000 # Allow for 1000-2000 word summaries ) return summary.strip() if summary else (existing_summary or "") def manage_conversation_context( conversation: Conversation, language: str = 'zh-hans' ) -> None: """ Check context size and generate summary if needed. Called before sending each new message. Args: conversation: Conversation object language: Language for summary generation """ # Get all messages all_messages = list(conversation.messages.order_by('created_at')) # Get configurable message limit max_messages = ConversationConfig.get_max_messages() if len(all_messages) <= max_messages: # Not enough messages to need summarization return # Get divination prompt size based on conversation subject type if not conversation.subject: logger.error(f"Conversation {conversation.id} has no subject") return subject_obj = conversation.get_subject_object() if not subject_obj: logger.error(f"Conversation {conversation.id} has no valid subject object") return if conversation.subject.content_type == 'bazi': divination_prompt = prepare_bazi_prompt(subject_obj, language=language) elif conversation.subject.content_type == 'liuyao': divination_prompt = prepare_liuyao_prompt(subject_obj, language=language) else: logger.error(f"Unsupported content type: {conversation.subject.content_type}") return # Calculate size with last N messages (configurable) last_messages = all_messages[-max_messages:] recent_history = [ {'role': msg.role, 'content': msg.content} for msg in last_messages ] # Estimate current message size (average) avg_message_size = sum(len(msg.content) for msg in all_messages) / len(all_messages) if all_messages else 0 # Calculate total context size with existing summary context_size = calculate_context_size( divination_prompt=divination_prompt, conversation_history=recent_history, current_message=" " * int(avg_message_size), # Estimate context_summary=conversation.context_summary ) # Get threshold from settings (default 100KB) threshold = getattr(settings, 'CONVERSATION_CONTEXT_THRESHOLD', 100000) # 100KB default # Check if we need to summarize if context_size > threshold: # Determine which messages to summarize # Only messages that haven't been summarized yet (incremental approach) # This avoids exceeding context limit when summarizing messages_to_summarize = [] start_id = conversation.last_summarized_message_id or 0 for msg in all_messages: if msg.id > start_id and msg not in last_messages: messages_to_summarize.append(msg) if messages_to_summarize: try: # Incremental summarization: combine existing summary with new messages # This approach: # - Avoids exceeding context limit (only summarizes new messages, not all) # - More efficient token usage # - Maintains context continuity by combining summaries # - Only done when threshold exceeded (infrequent) new_summary = generate_conversation_summary( conversation=conversation, messages_to_summarize=messages_to_summarize, language=language, existing_summary=conversation.context_summary ) # Update conversation with combined summary conversation.context_summary = new_summary conversation.context_summary_updated_at = timezone.now() conversation.last_summarized_message_id = messages_to_summarize[-1].id if messages_to_summarize else None conversation.save(update_fields=[ 'context_summary', 'context_summary_updated_at', 'last_summarized_message_id' ]) logger.info(f"Generated incremental context summary for conversation {conversation.id} (summarized {len(messages_to_summarize)} new messages)") except Exception as e: logger.error(f"Failed to generate summary for conversation {conversation.id}: {str(e)}") # Don't raise - will fallback to sending only last n messages that fit def prepare_conversation_prompt( conversation: Conversation, conversation_history: List[Dict[str, str]], user_message: str, language: Optional[str] = 'zh-hans' ) -> str: """ Prepare a prompt for conversation with divination context and managed history. Supports BaZi, LiuYao, and other divination types via ConversationSubject. Args: conversation: Conversation object (for accessing summary and subject) conversation_history: List of recent messages (last N, configurable) user_message: The current user message language: Language for the response ('zh-hans' or 'en') Returns: Formatted prompt string """ # Get subject object subject_obj = conversation.get_subject_object() if not subject_obj: raise ValueError(f"Conversation {conversation.id} has no valid subject object") # Get divination prompt based on content type if conversation.subject.content_type == 'bazi': divination_prompt = prepare_bazi_prompt(subject_obj, language=language) subject_name = subject_obj.name if hasattr(subject_obj, 'name') else 'Unknown' if language == 'en': system_prompt = f"""You are an expert in BaZi (Four Pillars of Destiny) analysis. You are having a conversation with a user about {subject_name}'s BaZi chart. Full BaZi Chart Information: {divination_prompt} Please provide helpful, accurate analysis and answer questions about this BaZi chart. Be conversational and engaging while maintaining accuracy. Reference specific pillars and elements when relevant.""" else: system_prompt = f"""你是一位八字（四柱命理）分析专家。你正在与用户讨论关于{subject_name}的八字命盘。 完整八字信息： {divination_prompt} 请提供有帮助、准确的分析，并回答关于这个八字命盘的问题。保持对话性和吸引力，同时保持准确性。在相关时引用具体的柱和五行。""" elif conversation.subject.content_type == 'liuyao': divination_prompt = prepare_liuyao_prompt(subject_obj, language=language) subject_name = subject_obj.question if hasattr(subject_obj, 'question') else 'Unknown' if language == 'en': system_prompt = f"""You are an expert in LiuYao (Six Lines Divination) analysis. You are having a conversation with a user about their LiuYao divination. Question: {subject_name} Full LiuYao Divination Information: {divination_prompt} Please provide helpful, accurate analysis and answer questions about this LiuYao divination. Be conversational and engaging while maintaining accuracy. Reference specific hexagrams, lines, and elements when relevant.""" else: system_prompt = f"""你是一位六爻分析专家。你正在与用户讨论关于他们的六爻占卜。 问题：{subject_name} 完整六爻信息： {divination_prompt} 请提供有帮助、准确的分析，并回答关于这个六爻占卜的问题。保持对话性和吸引力，同时保持准确性。在相关时引用具体的卦象、爻位和五行。""" else: raise ValueError(f"Unsupported content type: {conversation.subject.content_type}") # Add context summary if exists context_section = "" if conversation.context_summary: if language == 'en': context_section = f"\n\nPrevious Conversation Summary:\n{conversation.context_summary}\n" else: context_section = f"\n\n之前的对话摘要：\n{conversation.context_summary}\n" # Build recent conversation history (last N messages, configurable) conversation_text = "" if conversation_history: for msg in conversation_history: role = msg.get('role', 'user') content = msg.get('content', '') if role == 'user': conversation_text += f"User: {content}\n\n" else: conversation_text += f"Assistant: {content}\n\n" # Combine all parts full_prompt = f"{system_prompt}{context_section}\n\nRecent Conversation:\n{conversation_text}User: {user_message}\n\nAssistant:" return full_prompt def send_conversation_message( conversation: Conversation, user_message: str, provider: Optional[str] = None, model: Optional[str] = None, language: Optional[str] = 'zh-hans', existing_user_message: Optional[Message] = None ) -> Message: """ Send a user message and get AI response in a conversation. Includes context management. Args: conversation: Conversation object user_message: User message content provider: Optional AI provider override model: Optional AI model override language: Language for the conversation ('zh-hans' or 'en') existing_user_message: Optional existing Message object to reuse (for retries) Returns: Assistant Message object """ user_msg = None try: # Check and manage context before sending manage_conversation_context(conversation, language=language) # Reuse existing message if provided (for retries), otherwise create new one if existing_user_message: user_msg = existing_user_message # Update content if changed and set status to pending if user_msg.content != user_message: user_msg.content = user_message user_msg.status = 'pending' user_msg.error_message = None # Clear previous error user_msg.save(update_fields=['content', 'status', 'error_message']) else: # Save user message with pending status user_msg = Message.objects.create( conversation=conversation, role='user', content=user_message, status='pending' ) # Get AI configuration if not provided (use conversation-specific config) if not provider or not model: # Use conversation-specific config type (e.g., 'bazi_conversation' instead of 'bazi') conversation_config_type = f"{conversation.subject.content_type}_conversation" config = get_ai_config(conversation_config_type) provider = provider or config['provider'] model = model or config['model'] # Get last N messages (excluding the one we just created, configurable limit) max_messages = ConversationConfig.get_max_messages() all_messages = conversation.messages.exclude(pk=user_msg.pk).order_by('created_at') last_messages = list(all_messages)[-max_messages:] conversation_history = [ {'role': msg.role, 'content': msg.content} for msg in last_messages ] # Prepare prompt with context management # If summary generation failed, fallback to sending only last n messages that fit try: prompt = prepare_conversation_prompt( conversation=conversation, conversation_history=conversation_history, user_message=user_message, language=language ) except Exception as e: logger.warning(f"Failed to prepare prompt with summary, falling back to limited history: {str(e)}") # Fallback: send only last n messages that fit within context limit subject_obj = conversation.get_subject_object() if not subject_obj: raise ValueError(f"Conversation {conversation.id} has no valid subject object") if conversation.subject.content_type == 'bazi': divination_prompt = prepare_bazi_prompt(subject_obj, language=language) elif conversation.subject.content_type == 'liuyao': divination_prompt = prepare_liuyao_prompt(subject_obj, language=language) else: raise ValueError(f"Unsupported content type: {conversation.subject.content_type}") threshold = getattr(settings, 'CONVERSATION_CONTEXT_THRESHOLD', 100000) # Find how many messages fit limited_history = [] for msg in reversed(conversation_history): test_history = [msg] + limited_history test_size = calculate_context_size( divination_prompt=divination_prompt, conversation_history=test_history, current_message=user_message ) if test_size <= threshold: limited_history.insert(0, msg) else: break prompt = prepare_conversation_prompt( conversation=conversation, conversation_history=limited_history, user_message=user_message, language=language ) # Get LLM service llm_service = LLMServiceFactory.get_service(provider) if model: llm_service.change_model(model) # Get AI response logger.info(f"Sending conversation message for conversation {conversation.id}") ai_response = llm_service.get_completion( prompt=prompt, temperature=0.7, max_tokens=4096 ) if not ai_response: raise ValueError("Empty response from LLM service") # Save AI response assistant_msg = Message.objects.create( conversation=conversation, role='assistant', content=ai_response.strip(), provider=provider, model=model, meta={'language': language} ) # Update user message status to 'sent' (successful) user_msg.status = 'sent' user_msg.save(update_fields=['status']) # Update conversation timestamp conversation.updated_at = timezone.now() conversation.save(update_fields=['updated_at']) logger.info(f"Successfully created assistant message for conversation {conversation.id}") return assistant_msg except Exception as e: # Mark user message as failed if it exists if user_msg: error_msg = str(e) user_msg.status = 'failed' user_msg.error_message = error_msg user_msg.save(update_fields=['status', 'error_message']) logger.error(f"Marked user message {user_msg.id} as failed: {error_msg}") logger.error(f"Error sending conversation message: {str(e)}") raise def create_conversation( user, content_type: str, object_id: int, title: Optional[str] = None ) -> Conversation: """ Create a new conversation thread. Args: user: User object content_type: Type of divination ('bazi' or 'liuyao') object_id: ID of the divination record (Person ID for BaZi, LiuYao ID for LiuYao) title: Optional title for the conversation Returns: Conversation object """ # Get or create conversation subject subject = ConversationSubject.get_or_create_subject(content_type, object_id) # Get subject object for title subject_obj = subject.get_object() if not title: if content_type == 'bazi' and subject_obj: title = f"Conversation about {subject_obj.name}" elif content_type == 'liuyao' and subject_obj: title = f"Conversation about {subject_obj.question[:50]}" else: title = f"Conversation about {content_type} #{object_id}" conversation = Conversation.objects.create( user=user, subject=subject, title=title ) logger.info(f"Created conversation {conversation.id} for user {user.id} about {content_type} #{object_id}") return conversation def create_bazi_conversation( user, person: Person, title: Optional[str] = None ) -> Conversation: """ Create a new BaZi conversation thread (convenience function). Args: user: User object person: Person (BaZi chart) object title: Optional title for the conversation Returns: Conversation object """ return create_conversation(user, 'bazi', person.id, title) def create_liuyao_conversation( user, liuyao, title: Optional[str] = None ) -> Conversation: """ Create a new LiuYao conversation thread (convenience function). Args: user: User object liuyao: LiuYao object title: Optional title for the conversation Returns: Conversation object """ return create_conversation(user, 'liuyao', liuyao.id, title)