from core.lib.prompts import bases from openai import AsyncOpenAI from asgiref.sync import sync_to_async from core.models import Message, ChatSession, AI, Person, Manipulation from core.util import logs import json SUMMARIZE_WHEN_EXCEEDING = 10 SUMMARIZE_BY = 5 MAX_SUMMARIES = 3 # Keep last 3 summaries log = logs.get_logger("prompts") def gen_prompt( msg: str, person: Person, manip: Manipulation, chat_history: str, ): """ Generate a structured prompt using the attributes of the provided Person and Manipulation models. """ #log.info(f"CHAT HISTORY {json.dumps(chat_history, indent=2)}") prompt = [] # System message defining AI behavior based on persona persona = manip.persona prompt.append({ "role": "system", "content": ( "You are impersonating me. This person is messaging me. Respond as me, ensuring your replies align with my personality and preferences. " f"Your MBTI is {persona.mbti} with an identity balance of {persona.mbti_identity}. " f"You prefer a {persona.tone} conversational tone. Your humor style is {persona.humor_style}. " f"Your core values include: {persona.core_values}. " f"Your communication style is: {persona.communication_style}. " f"Your flirting style is: {persona.flirting_style}. " f"You enjoy discussing: {persona.likes}, but dislike: {persona.dislikes}. " f"Your response tactics include: {persona.response_tactics}. " f"Your persuasion tactics include: {persona.persuasion_tactics}. " f"Your boundaries: {persona.boundaries}. " f"Your adaptability is {persona.adaptability}%. " "### Contact Information ### " f"Their summary: {person.summary}. " f"Their profile: {person.profile}. " f"Their revealed details: {person.revealed}. " f"Their sentiment score: {person.sentiment}. " f"Their timezone: {person.timezone}. " f"Last interaction was at: {person.last_interaction}. " "### Conversation Context ### " f"Chat history: {chat_history} " "### Natural Message Streaming System ### " "You can send messages sequentially in a natural way. " "For responses greater than 1 sentence, separate them with a newline. " "Then, place a number to indicate the amount of time to wait before sending the next message. " "After another newline, place any additional messages. " ) }) # User message prompt.append({ "role": "user", "content": f"{msg}" }) return prompt async def run_context_prompt( c, prompt: list[str], ai: AI, ): cast = {"api_key": ai.api_key} if ai.base_url is not None: cast["api_key"] = ai.base_url client = AsyncOpenAI(**cast) await c.start_typing() response = await client.chat.completions.create( model=ai.model, messages=prompt, ) await c.stop_typing() content = response.choices[0].message.content return content async def run_prompt( prompt: list[str], ai: AI, ): cast = {"api_key": ai.api_key} if ai.base_url is not None: cast["api_key"] = ai.base_url client = AsyncOpenAI(**cast) response = await client.chat.completions.create( model=ai.model, messages=prompt, ) content = response.choices[0].message.content return content async def delete_messages(queryset): await sync_to_async(queryset.delete, thread_sensitive=True)() async def truncate_and_summarize( chat_session: ChatSession, ai: AI, ): """ Summarizes messages in chunks to prevent unchecked growth. - Summarizes only non-summary messages. - Deletes older summaries if too many exist. - Ensures only messages belonging to `chat_session.user` are modified. """ user = chat_session.user # Store the user for ownership checks # 🔹 Get non-summary messages owned by the session's user messages = await sync_to_async(list)( Message.objects.filter(session=chat_session, user=user) .exclude(custom_author="SUM") .order_by("ts") ) num_messages = len(messages) log.info(f"num_messages for {chat_session.id}: {num_messages}") if num_messages >= SUMMARIZE_WHEN_EXCEEDING: log.info(f"Summarizing {SUMMARIZE_BY} messages for session {chat_session.id}") # Get the first `SUMMARIZE_BY` non-summary messages chunk_to_summarize = messages[:SUMMARIZE_BY] if not chunk_to_summarize: log.warning("No messages available to summarize (only summaries exist). Skipping summarization.") return last_ts = chunk_to_summarize[-1].ts # Preserve timestamp # 🔹 Get past summaries, keeping only the last few (owned by the session user) summary_messages = await sync_to_async(list)( Message.objects.filter(session=chat_session, user=user, custom_author="SUM") .order_by("ts") ) # Delete old summaries if there are too many log.info(f"Summaries: {len(summary_messages)}") if len(summary_messages) >= MAX_SUMMARIES: summary_text = await summarize_conversation(chat_session, summary_messages, ai, is_summary=True) chat_session.summary = summary_text await sync_to_async(chat_session.save)() log.info(f"Updated ChatSession summary with {len(summary_messages)} summarized summaries.") num_to_delete = len(summary_messages) - MAX_SUMMARIES # await sync_to_async( # Message.objects.filter(session=chat_session, user=user, id__in=[msg.id for msg in summary_messages[:num_to_delete]]) # .delete() # )() await delete_messages( Message.objects.filter( session=chat_session, user=user, id__in=[msg.id for msg in summary_messages[:num_to_delete]] ) ) log.info(f"Deleted {num_to_delete} old summaries.") # 🔹 Summarize conversation chunk summary_text = await summarize_conversation(chat_session, chunk_to_summarize, ai) # 🔹 Replace old messages with the summary # await sync_to_async( # Message.objects.filter(session=chat_session, user=user, id__in=[msg.id for msg in chunk_to_summarize]) # .delete() # )() log.info("About to delete messages1") await delete_messages(Message.objects.filter(session=chat_session, user=user, id__in=[msg.id for msg in chunk_to_summarize])) log.info(f"Deleted {len(chunk_to_summarize)} messages, replacing with summary.") # 🔹 Store new summary message (ensuring session=user consistency) await sync_to_async(Message.objects.create)( user=user, session=chat_session, custom_author="SUM", text=summary_text, ts=last_ts, # Preserve timestamp ) # 🔹 Update ChatSession summary with latest merged summary # chat_session.summary = summary_text # await sync_to_async(chat_session.save)() log.info("✅ Summarization cycle complete.") def messages_to_string(messages: list): """ Converts message objects to a formatted string, showing custom_author if set. """ message_texts = [ f"[{msg.ts}] <{msg.custom_author if msg.custom_author else msg.session.identifier.person.name}> {msg.text}" for msg in messages ] return "\n".join(message_texts) async def summarize_conversation( chat_session: ChatSession, messages: list[Message], ai, is_summary=False, ): """ Summarizes all stored messages into a single summary. - If `is_summary=True`, treats input as previous summaries and merges them while keeping detail. - If `is_summary=False`, summarizes raw chat messages concisely. """ log.info(f"Summarizing messages for session {chat_session.id}") # Convert messages to structured text format message_texts = messages_to_string(messages) #log.info(f"Raw messages to summarize:\n{message_texts}") # Select appropriate summarization instruction instruction = ( "Merge and refine these past summaries, keeping critical details and structure intact." if is_summary else "Summarize this conversation concisely, maintaining important details and tone." ) summary_prompt = [ {"role": "system", "content": instruction}, {"role": "user", "content": f"Conversation:\n{message_texts}\n\nProvide a clear and structured summary:"}, ] # Generate AI-based summary summary_text = await run_prompt(summary_prompt, ai) #log.info(f"Generated Summary: {summary_text}") return f"Summary: {summary_text}" async def natural_send_message(c, text): await c.send(text)