965 lines
38 KiB
Python
965 lines
38 KiB
Python
|
# utils/persona_dialogue.py
|
||
|
|
"""
|
||
|
|
Persona Dialogue System for Miku.
|
||
|
|
|
||
|
|
Enables natural back-and-forth conversations between Hatsune Miku and Evil Miku.
|
||
|
|
Unlike bipolar_mode.py (which handles arguments), this module handles:
|
||
|
|
- Detecting when the opposite persona should interject
|
||
|
|
- Managing natural dialogue flow with self-signaling continuation
|
||
|
|
- Tracking tension that can escalate into arguments
|
||
|
|
- Seamless handoff to the argument system when tension is high
|
||
|
|
|
||
|
|
This system is designed to be lightweight on LLM calls:
|
||
|
|
- Initial trigger uses fast heuristics + sentiment analysis
|
||
|
|
- Each dialogue turn uses ONE LLM call that generates response AND decides continuation
|
||
|
|
- Only escalates to argument system when tension threshold is reached
|
||
|
|
"""
|
||
|
|
|
||
|
|
import os
|
||
|
|
import json
|
||
|
|
import time
|
||
|
|
import asyncio
|
||
|
|
import discord
|
||
|
|
import globals
|
||
|
|
|
||
|
|
from transformers import pipeline
|
||
|
|
|
||
|
|
# ============================================================================
|
||
|
|
# CONSTANTS
|
||
|
|
# ============================================================================
|
||
|
|
|
||
|
|
DIALOGUE_STATE_FILE = "memory/persona_dialogue_state.json"
|
||
|
|
|
||
|
|
# Dialogue settings
|
||
|
|
MAX_TURNS = 20 # Maximum turns before forced end
|
||
|
|
DIALOGUE_TIMEOUT = 900 # 15 minutes max dialogue duration
|
||
|
|
ARGUMENT_TENSION_THRESHOLD = 0.75 # Tension level that triggers argument escalation
|
||
|
|
|
||
|
|
# Initial trigger settings
|
||
|
|
INTERJECTION_COOLDOWN_HARD = 180 # 3 minutes hard block
|
||
|
|
INTERJECTION_COOLDOWN_SOFT = 900 # 15 minutes for full recovery
|
||
|
|
INTERJECTION_THRESHOLD = 0.75 # Score needed to trigger interjection (lowered to account for mood multipliers)
|
||
|
|
|
||
|
|
# ============================================================================
|
||
|
|
# INTERJECTION SCORER (Initial Trigger Decision)
|
||
|
|
# ============================================================================
|
||
|
|
|
||
|
|
class InterjectionScorer:
|
||
|
|
"""
|
||
|
|
Decides if the opposite persona should interject based on message content.
|
||
|
|
Uses fast heuristics + sentiment analysis (no LLM calls).
|
||
|
|
"""
|
||
|
|
|
||
|
|
_instance = None
|
||
|
|
_sentiment_analyzer = None
|
||
|
|
|
||
|
|
def __new__(cls):
|
||
|
|
if cls._instance is None:
|
||
|
|
cls._instance = super().__new__(cls)
|
||
|
|
return cls._instance
|
||
|
|
|
||
|
|
@property
|
||
|
|
def sentiment_analyzer(self):
|
||
|
|
"""Lazy load sentiment analyzer"""
|
||
|
|
if self._sentiment_analyzer is None:
|
||
|
|
print("🔄 Loading sentiment analyzer for persona dialogue...")
|
||
|
|
try:
|
||
|
|
self._sentiment_analyzer = pipeline(
|
||
|
|
"sentiment-analysis",
|
||
|
|
model="distilbert-base-uncased-finetuned-sst-2-english"
|
||
|
|
)
|
||
|
|
print("✅ Sentiment analyzer loaded")
|
||
|
|
except Exception as e:
|
||
|
|
print(f"⚠️ Failed to load sentiment analyzer: {e}")
|
||
|
|
self._sentiment_analyzer = None
|
||
|
|
return self._sentiment_analyzer
|
||
|
|
|
||
|
|
async def should_interject(self, message: discord.Message, current_persona: str) -> tuple:
|
||
|
|
"""
|
||
|
|
Determine if the opposite persona should interject.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
message: The Discord message to analyze
|
||
|
|
current_persona: Who just spoke ("miku" or "evil")
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Tuple of (should_interject: bool, reason: str, score: float)
|
||
|
|
"""
|
||
|
|
|
||
|
|
# Quick rejections
|
||
|
|
if not self._passes_basic_filter(message):
|
||
|
|
return False, "basic_filter_failed", 0.0
|
||
|
|
|
||
|
|
# Check cooldown
|
||
|
|
cooldown_mult = self._check_cooldown()
|
||
|
|
if cooldown_mult == 0.0:
|
||
|
|
return False, "cooldown_active", 0.0
|
||
|
|
|
||
|
|
opposite_persona = "evil" if current_persona == "miku" else "miku"
|
||
|
|
|
||
|
|
print(f"🔍 [Interjection] Analyzing content: '{message.content[:100]}...'")
|
||
|
|
print(f"🔍 [Interjection] Current persona: {current_persona}, Opposite: {opposite_persona}")
|
||
|
|
|
||
|
|
# Calculate score from various factors
|
||
|
|
score = 0.0
|
||
|
|
reasons = []
|
||
|
|
|
||
|
|
# Factor 1: Direct addressing (automatic trigger)
|
||
|
|
if self._mentions_opposite(message.content, opposite_persona):
|
||
|
|
print(f"✅ [Interjection] Direct mention of {opposite_persona} detected!")
|
||
|
|
return True, "directly_addressed", 1.0
|
||
|
|
|
||
|
|
# Factor 2: Topic relevance
|
||
|
|
topic_score = self._check_topic_relevance(message.content, opposite_persona)
|
||
|
|
if topic_score > 0:
|
||
|
|
score += topic_score * 0.3
|
||
|
|
reasons.append(f"topic:{topic_score:.2f}")
|
||
|
|
|
||
|
|
# Factor 3: Emotional intensity
|
||
|
|
emotion_score = self._check_emotional_intensity(message.content)
|
||
|
|
if emotion_score > 0.6:
|
||
|
|
score += emotion_score * 0.25
|
||
|
|
reasons.append(f"emotion:{emotion_score:.2f}")
|
||
|
|
|
||
|
|
# Factor 4: Personality clash
|
||
|
|
clash_score = self._detect_personality_clash(message.content, opposite_persona)
|
||
|
|
if clash_score > 0:
|
||
|
|
score += clash_score * 0.25
|
||
|
|
reasons.append(f"clash:{clash_score:.2f}")
|
||
|
|
|
||
|
|
# Factor 5: Mood multiplier
|
||
|
|
mood_mult = self._get_mood_multiplier(opposite_persona)
|
||
|
|
score *= mood_mult
|
||
|
|
if mood_mult != 1.0:
|
||
|
|
reasons.append(f"mood_mult:{mood_mult:.2f}")
|
||
|
|
|
||
|
|
# Factor 6: Context bonus
|
||
|
|
context_bonus = self._check_conversation_context(message)
|
||
|
|
score += context_bonus * 0.2
|
||
|
|
if context_bonus > 0:
|
||
|
|
reasons.append(f"context:{context_bonus:.2f}")
|
||
|
|
|
||
|
|
# Apply cooldown multiplier
|
||
|
|
score *= cooldown_mult
|
||
|
|
|
||
|
|
# Decision
|
||
|
|
should_interject = score >= INTERJECTION_THRESHOLD
|
||
|
|
reason_str = " | ".join(reasons) if reasons else "no_triggers"
|
||
|
|
|
||
|
|
if should_interject:
|
||
|
|
print(f"✅ {opposite_persona.upper()} WILL INTERJECT (score: {score:.2f})")
|
||
|
|
print(f" Reasons: {reason_str}")
|
||
|
|
|
||
|
|
return should_interject, reason_str, score
|
||
|
|
|
||
|
|
def _passes_basic_filter(self, message: discord.Message) -> bool:
|
||
|
|
"""Fast rejection criteria"""
|
||
|
|
# System messages
|
||
|
|
if message.type != discord.MessageType.default:
|
||
|
|
print(f"❌ [Basic Filter] System message type: {message.type}")
|
||
|
|
return False
|
||
|
|
|
||
|
|
# Bipolar mode must be enabled
|
||
|
|
if not globals.BIPOLAR_MODE:
|
||
|
|
print(f"❌ [Basic Filter] Bipolar mode not enabled")
|
||
|
|
return False
|
||
|
|
|
||
|
|
# Allow bot's own messages (we're checking them for interjections!)
|
||
|
|
# Also allow webhook messages (persona messages)
|
||
|
|
# Only reject OTHER bots' messages
|
||
|
|
if message.author.bot and not message.webhook_id:
|
||
|
|
# Check if it's our own bot
|
||
|
|
if message.author.id != globals.client.user.id:
|
||
|
|
print(f"❌ [Basic Filter] Other bot message (not our bot)")
|
||
|
|
return False
|
||
|
|
|
||
|
|
print(f"✅ [Basic Filter] Passed (bot={message.author.bot}, webhook={message.webhook_id}, our_bot={message.author.id == globals.client.user.id if message.author.bot else 'N/A'})")
|
||
|
|
return True
|
||
|
|
|
||
|
|
def _mentions_opposite(self, content: str, opposite_persona: str) -> bool:
|
||
|
|
"""Check if message directly addresses the opposite persona"""
|
||
|
|
content_lower = content.lower()
|
||
|
|
|
||
|
|
if opposite_persona == "evil":
|
||
|
|
patterns = ["evil miku", "dark miku", "evil version", "bad miku", "evil you"]
|
||
|
|
else:
|
||
|
|
patterns = ["normal miku", "regular miku", "good miku", "real miku", "nice miku", "other miku", "original miku"]
|
||
|
|
|
||
|
|
return any(pattern in content_lower for pattern in patterns)
|
||
|
|
|
||
|
|
def _check_topic_relevance(self, content: str, opposite_persona: str) -> float:
|
||
|
|
"""Check if topics would interest the opposite persona"""
|
||
|
|
content_lower = content.lower()
|
||
|
|
|
||
|
|
if opposite_persona == "evil":
|
||
|
|
# Things Evil Miku can't resist commenting on
|
||
|
|
TRIGGER_TOPICS = {
|
||
|
|
"optimism": ["happiness", "joy", "love", "kindness", "hope", "dreams", "wonderful", "amazing"],
|
||
|
|
"morality": ["good", "should", "must", "right thing", "deserve", "fair", "justice"],
|
||
|
|
"weakness": ["scared", "nervous", "worried", "unsure", "help me", "don't know"],
|
||
|
|
"innocence": ["innocent", "pure", "sweet", "cute", "wholesome", "precious"],
|
||
|
|
}
|
||
|
|
else:
|
||
|
|
# Things Miku can't ignore
|
||
|
|
TRIGGER_TOPICS = {
|
||
|
|
"negativity": ["hate", "terrible", "awful", "worst", "horrible", "disgusting", "pathetic"],
|
||
|
|
"cruelty": ["deserve pain", "suffer", "worthless", "stupid", "idiot", "fool"],
|
||
|
|
"hopelessness": ["no point", "meaningless", "nobody cares", "why bother", "give up"],
|
||
|
|
"evil_gloating": ["foolish", "naive", "weak", "inferior", "pathetic"],
|
||
|
|
}
|
||
|
|
|
||
|
|
total_matches = 0
|
||
|
|
for category, keywords in TRIGGER_TOPICS.items():
|
||
|
|
matches = sum(1 for keyword in keywords if keyword in content_lower)
|
||
|
|
total_matches += matches
|
||
|
|
|
||
|
|
return min(total_matches / 3.0, 1.0)
|
||
|
|
|
||
|
|
def _check_emotional_intensity(self, content: str) -> float:
|
||
|
|
"""Check emotional intensity using sentiment analysis"""
|
||
|
|
if not self.sentiment_analyzer:
|
||
|
|
return 0.5 # Neutral if no analyzer
|
||
|
|
|
||
|
|
try:
|
||
|
|
result = self.sentiment_analyzer(content[:512])[0]
|
||
|
|
confidence = result['score']
|
||
|
|
|
||
|
|
# Punctuation intensity
|
||
|
|
exclamations = content.count('!')
|
||
|
|
questions = content.count('?')
|
||
|
|
caps_ratio = sum(1 for c in content if c.isupper()) / max(len(content), 1)
|
||
|
|
|
||
|
|
intensity_markers = (exclamations * 0.15) + (questions * 0.1) + (caps_ratio * 0.3)
|
||
|
|
|
||
|
|
return min(confidence * 0.6 + intensity_markers, 1.0)
|
||
|
|
except Exception as e:
|
||
|
|
print(f"⚠️ Sentiment analysis error: {e}")
|
||
|
|
return 0.5
|
||
|
|
|
||
|
|
def _detect_personality_clash(self, content: str, opposite_persona: str) -> float:
|
||
|
|
"""Detect statements that clash with the opposite persona's values"""
|
||
|
|
content_lower = content.lower()
|
||
|
|
|
||
|
|
if opposite_persona == "evil":
|
||
|
|
# User being too positive/naive = Evil Miku wants to "correct" them
|
||
|
|
positive_statements = [
|
||
|
|
"i believe in", "i love", "everything will be", "so happy",
|
||
|
|
"the best", "amazing", "perfect", "wonderful life", "so grateful"
|
||
|
|
]
|
||
|
|
return 0.8 if any(stmt in content_lower for stmt in positive_statements) else 0.0
|
||
|
|
else:
|
||
|
|
# User being cruel/negative = Miku wants to help/defend
|
||
|
|
negative_statements = [
|
||
|
|
"i hate", "everyone sucks", "life is meaningless", "don't care",
|
||
|
|
"deserve to suffer", "nobody matters", "worthless", "all terrible"
|
||
|
|
]
|
||
|
|
return 0.8 if any(stmt in content_lower for stmt in negative_statements) else 0.0
|
||
|
|
|
||
|
|
def _get_mood_multiplier(self, opposite_persona: str) -> float:
|
||
|
|
"""Current mood affects likelihood of interjection"""
|
||
|
|
if opposite_persona == "evil":
|
||
|
|
MOOD_MULTIPLIERS = {
|
||
|
|
"aggressive": 1.5,
|
||
|
|
"cruel": 1.3,
|
||
|
|
"mischievous": 1.2,
|
||
|
|
"cunning": 1.0,
|
||
|
|
"sarcastic": 1.1,
|
||
|
|
"evil_neutral": 0.8,
|
||
|
|
"contemplative": 0.6,
|
||
|
|
}
|
||
|
|
return MOOD_MULTIPLIERS.get(globals.EVIL_DM_MOOD, 1.0)
|
||
|
|
else:
|
||
|
|
MOOD_MULTIPLIERS = {
|
||
|
|
"bubbly": 1.4,
|
||
|
|
"excited": 1.3,
|
||
|
|
"curious": 1.2,
|
||
|
|
"neutral": 1.0,
|
||
|
|
"irritated": 0.9,
|
||
|
|
"melancholy": 0.7,
|
||
|
|
"asleep": 0.1,
|
||
|
|
}
|
||
|
|
return MOOD_MULTIPLIERS.get(globals.DM_MOOD, 1.0)
|
||
|
|
|
||
|
|
def _check_conversation_context(self, message: discord.Message) -> float:
|
||
|
|
"""Check if this is part of an active conversation"""
|
||
|
|
score = 0.0
|
||
|
|
|
||
|
|
# Part of a reply chain
|
||
|
|
if hasattr(message, 'reference') and message.reference:
|
||
|
|
score += 0.5
|
||
|
|
|
||
|
|
# Could add more context checks here
|
||
|
|
score += 0.2 # Base activity bonus
|
||
|
|
|
||
|
|
return min(score, 1.0)
|
||
|
|
|
||
|
|
def _check_cooldown(self) -> float:
|
||
|
|
"""Check cooldown and return multiplier (0.0 = blocked, 1.0 = full)"""
|
||
|
|
if not hasattr(globals, 'LAST_PERSONA_DIALOGUE_TIME'):
|
||
|
|
globals.LAST_PERSONA_DIALOGUE_TIME = 0
|
||
|
|
|
||
|
|
current_time = time.time()
|
||
|
|
time_since_last = current_time - globals.LAST_PERSONA_DIALOGUE_TIME
|
||
|
|
|
||
|
|
if time_since_last < INTERJECTION_COOLDOWN_HARD:
|
||
|
|
return 0.0
|
||
|
|
elif time_since_last < INTERJECTION_COOLDOWN_SOFT:
|
||
|
|
return (time_since_last - INTERJECTION_COOLDOWN_HARD) / (INTERJECTION_COOLDOWN_SOFT - INTERJECTION_COOLDOWN_HARD)
|
||
|
|
else:
|
||
|
|
return 1.0
|
||
|
|
|
||
|
|
|
||
|
|
# ============================================================================
|
||
|
|
# PERSONA DIALOGUE MANAGER
|
||
|
|
# ============================================================================
|
||
|
|
|
||
|
|
class PersonaDialogue:
|
||
|
|
"""
|
||
|
|
Manages natural back-and-forth conversations between Miku and Evil Miku.
|
||
|
|
|
||
|
|
Each turn:
|
||
|
|
1. Generate response + continuation signal (single LLM call)
|
||
|
|
2. Calculate tension delta from response
|
||
|
|
3. If tension >= threshold, escalate to argument
|
||
|
|
4. Otherwise, continue or end based on signal
|
||
|
|
"""
|
||
|
|
|
||
|
|
_instance = None
|
||
|
|
_sentiment_analyzer = None
|
||
|
|
|
||
|
|
def __new__(cls):
|
||
|
|
if cls._instance is None:
|
||
|
|
cls._instance = super().__new__(cls)
|
||
|
|
cls._instance.active_dialogues = {}
|
||
|
|
return cls._instance
|
||
|
|
|
||
|
|
@property
|
||
|
|
def sentiment_analyzer(self):
|
||
|
|
"""Lazy load sentiment analyzer (shared with InterjectionScorer)"""
|
||
|
|
if self._sentiment_analyzer is None:
|
||
|
|
scorer = InterjectionScorer()
|
||
|
|
self._sentiment_analyzer = scorer.sentiment_analyzer
|
||
|
|
return self._sentiment_analyzer
|
||
|
|
|
||
|
|
# ========================================================================
|
||
|
|
# DIALOGUE STATE MANAGEMENT
|
||
|
|
# ========================================================================
|
||
|
|
|
||
|
|
def is_dialogue_active(self, channel_id: int) -> bool:
|
||
|
|
"""Check if a dialogue is active in a channel"""
|
||
|
|
return channel_id in self.active_dialogues
|
||
|
|
|
||
|
|
def get_dialogue_state(self, channel_id: int) -> dict:
|
||
|
|
"""Get dialogue state for a channel"""
|
||
|
|
return self.active_dialogues.get(channel_id, None)
|
||
|
|
|
||
|
|
def start_dialogue(self, channel_id: int) -> dict:
|
||
|
|
"""Start a new dialogue in a channel"""
|
||
|
|
state = {
|
||
|
|
"turn_count": 0,
|
||
|
|
"started_at": time.time(),
|
||
|
|
"tension": 0.0,
|
||
|
|
"tension_history": [],
|
||
|
|
"last_speaker": None,
|
||
|
|
}
|
||
|
|
self.active_dialogues[channel_id] = state
|
||
|
|
globals.LAST_PERSONA_DIALOGUE_TIME = time.time()
|
||
|
|
print(f"💬 Started persona dialogue in channel {channel_id}")
|
||
|
|
return state
|
||
|
|
|
||
|
|
def end_dialogue(self, channel_id: int):
|
||
|
|
"""End a dialogue in a channel"""
|
||
|
|
if channel_id in self.active_dialogues:
|
||
|
|
state = self.active_dialogues[channel_id]
|
||
|
|
print(f"🏁 Ended persona dialogue in channel {channel_id}")
|
||
|
|
print(f" Turns: {state['turn_count']}, Final tension: {state['tension']:.2f}")
|
||
|
|
del self.active_dialogues[channel_id]
|
||
|
|
|
||
|
|
# ========================================================================
|
||
|
|
# TENSION CALCULATION
|
||
|
|
# ========================================================================
|
||
|
|
|
||
|
|
def calculate_tension_delta(self, response_text: str, current_tension: float) -> float:
|
||
|
|
"""
|
||
|
|
Analyze a response and determine how much tension it adds/removes.
|
||
|
|
|
||
|
|
Returns delta to add to current tension score.
|
||
|
|
"""
|
||
|
|
|
||
|
|
# Sentiment analysis
|
||
|
|
base_delta = 0.0
|
||
|
|
|
||
|
|
if self.sentiment_analyzer:
|
||
|
|
try:
|
||
|
|
sentiment = self.sentiment_analyzer(response_text[:512])[0]
|
||
|
|
sentiment_score = sentiment['score']
|
||
|
|
is_negative = sentiment['label'] == 'NEGATIVE'
|
||
|
|
|
||
|
|
if is_negative:
|
||
|
|
base_delta = sentiment_score * 0.15
|
||
|
|
else:
|
||
|
|
base_delta = -sentiment_score * 0.05
|
||
|
|
except Exception as e:
|
||
|
|
print(f"⚠️ Sentiment analysis error in tension calc: {e}")
|
||
|
|
|
||
|
|
text_lower = response_text.lower()
|
||
|
|
|
||
|
|
# Escalation patterns
|
||
|
|
escalation_patterns = {
|
||
|
|
"insult": ["idiot", "stupid", "pathetic", "fool", "naive", "worthless", "disgusting", "moron"],
|
||
|
|
"dismissive": ["whatever", "don't care", "waste of time", "not worth", "beneath me", "boring"],
|
||
|
|
"confrontational": ["wrong", "you always", "you never", "how dare", "shut up", "stop"],
|
||
|
|
"mockery": ["oh please", "how cute", "adorable that you think", "laughable", "hilarious"],
|
||
|
|
"challenge": ["prove it", "fight me", "make me", "i dare you", "try me"],
|
||
|
|
}
|
||
|
|
|
||
|
|
# De-escalation patterns
|
||
|
|
deescalation_patterns = {
|
||
|
|
"concession": ["you're right", "fair point", "i suppose", "maybe you have", "good point"],
|
||
|
|
"softening": ["i understand", "let's calm", "didn't mean", "sorry", "apologize"],
|
||
|
|
"deflection": ["anyway", "moving on", "whatever you say", "agree to disagree", "let's just"],
|
||
|
|
}
|
||
|
|
|
||
|
|
# Check escalation
|
||
|
|
for category, patterns in escalation_patterns.items():
|
||
|
|
matches = sum(1 for p in patterns if p in text_lower)
|
||
|
|
if matches > 0:
|
||
|
|
base_delta += matches * 0.08
|
||
|
|
|
||
|
|
# Check de-escalation
|
||
|
|
for category, patterns in deescalation_patterns.items():
|
||
|
|
matches = sum(1 for p in patterns if p in text_lower)
|
||
|
|
if matches > 0:
|
||
|
|
base_delta -= matches * 0.06
|
||
|
|
|
||
|
|
# Intensity multipliers
|
||
|
|
exclamation_count = response_text.count('!')
|
||
|
|
caps_ratio = sum(1 for c in response_text if c.isupper()) / max(len(response_text), 1)
|
||
|
|
|
||
|
|
if exclamation_count > 2 or caps_ratio > 0.3:
|
||
|
|
base_delta *= 1.3
|
||
|
|
|
||
|
|
# Momentum factor
|
||
|
|
if current_tension > 0.5:
|
||
|
|
base_delta *= 1.2
|
||
|
|
|
||
|
|
return base_delta
|
||
|
|
|
||
|
|
# ========================================================================
|
||
|
|
# RESPONSE GENERATION
|
||
|
|
# ========================================================================
|
||
|
|
|
||
|
|
async def generate_response_with_continuation(
|
||
|
|
self,
|
||
|
|
channel: discord.TextChannel,
|
||
|
|
responding_persona: str,
|
||
|
|
context: str,
|
||
|
|
) -> tuple:
|
||
|
|
"""
|
||
|
|
Generate response AND continuation signal in a single LLM call.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Tuple of (response_text, should_continue, confidence)
|
||
|
|
"""
|
||
|
|
from utils.llm import query_llama
|
||
|
|
|
||
|
|
opposite = "Hatsune Miku" if responding_persona == "evil" else "Evil Miku"
|
||
|
|
|
||
|
|
# Get system prompt for persona
|
||
|
|
system_prompt = self._get_persona_system_prompt(responding_persona)
|
||
|
|
|
||
|
|
# Build the combined prompt
|
||
|
|
prompt = f"""{system_prompt}
|
||
|
|
|
||
|
|
Recent conversation:
|
||
|
|
{context}
|
||
|
|
|
||
|
|
Respond naturally as yourself. Keep your response conversational and in-character.
|
||
|
|
|
||
|
|
---
|
||
|
|
|
||
|
|
After your response, evaluate whether {opposite} would want to (or need to) respond.
|
||
|
|
|
||
|
|
The conversation should CONTINUE if ANY of these are true:
|
||
|
|
- You asked them a direct question (almost always YES)
|
||
|
|
- You made a provocative claim they'd dispute
|
||
|
|
- You challenged or insulted them
|
||
|
|
- The topic feels unfinished or confrontational
|
||
|
|
- There's clear tension or disagreement
|
||
|
|
|
||
|
|
The conversation might END if ALL of these are true:
|
||
|
|
- No questions were asked
|
||
|
|
- You made a definitive closing statement ("I'm done", "whatever", "goodbye")
|
||
|
|
- The exchange reached complete resolution
|
||
|
|
- Both sides have said their piece
|
||
|
|
|
||
|
|
IMPORTANT: If you asked a question, the answer is almost always YES - they need to respond!
|
||
|
|
|
||
|
|
On a new line after your response, write:
|
||
|
|
[CONTINUE: YES or NO] [CONFIDENCE: HIGH, MEDIUM, or LOW]"""
|
||
|
|
|
||
|
|
# Use appropriate model
|
||
|
|
model = globals.EVIL_TEXT_MODEL if responding_persona == "evil" else globals.TEXT_MODEL
|
||
|
|
|
||
|
|
# Temporarily set evil mode for proper context
|
||
|
|
original_evil_mode = globals.EVIL_MODE
|
||
|
|
globals.EVIL_MODE = (responding_persona == "evil")
|
||
|
|
|
||
|
|
try:
|
||
|
|
raw_response = await query_llama(
|
||
|
|
user_prompt=prompt,
|
||
|
|
user_id=f"persona_dialogue_{channel.id}",
|
||
|
|
guild_id=channel.guild.id if hasattr(channel, 'guild') and channel.guild else None,
|
||
|
|
response_type="autonomous_general",
|
||
|
|
model=model
|
||
|
|
)
|
||
|
|
finally:
|
||
|
|
globals.EVIL_MODE = original_evil_mode
|
||
|
|
|
||
|
|
if not raw_response or raw_response.startswith("Error"):
|
||
|
|
return None, False, "LOW"
|
||
|
|
|
||
|
|
# Parse response and signal
|
||
|
|
response_text, should_continue, confidence = self._parse_response(raw_response)
|
||
|
|
|
||
|
|
return response_text, should_continue, confidence
|
||
|
|
|
||
|
|
def _parse_response(self, raw_response: str) -> tuple:
|
||
|
|
"""Extract response text and continuation signal"""
|
||
|
|
lines = raw_response.strip().split('\n')
|
||
|
|
|
||
|
|
should_continue = False
|
||
|
|
confidence = "MEDIUM"
|
||
|
|
response_lines = []
|
||
|
|
|
||
|
|
for line in lines:
|
||
|
|
line_upper = line.upper()
|
||
|
|
|
||
|
|
if "[CONTINUE:" in line_upper:
|
||
|
|
should_continue = "YES" in line_upper
|
||
|
|
|
||
|
|
if "HIGH" in line_upper:
|
||
|
|
confidence = "HIGH"
|
||
|
|
elif "LOW" in line_upper:
|
||
|
|
confidence = "LOW"
|
||
|
|
else:
|
||
|
|
confidence = "MEDIUM"
|
||
|
|
else:
|
||
|
|
response_lines.append(line)
|
||
|
|
|
||
|
|
response_text = '\n'.join(response_lines).strip()
|
||
|
|
|
||
|
|
# Clean up any stray signal markers
|
||
|
|
response_text = response_text.replace("[CONTINUE:", "").replace("]", "")
|
||
|
|
response_text = response_text.replace("YES", "").replace("NO", "")
|
||
|
|
response_text = response_text.replace("HIGH", "").replace("MEDIUM", "").replace("LOW", "")
|
||
|
|
response_text = response_text.strip()
|
||
|
|
|
||
|
|
# Override: If the response contains a question mark, always continue
|
||
|
|
if '?' in response_text:
|
||
|
|
print(f"⚠️ [Parse Override] Question detected, forcing continue=YES")
|
||
|
|
should_continue = True
|
||
|
|
if confidence == "LOW":
|
||
|
|
confidence = "MEDIUM"
|
||
|
|
|
||
|
|
return response_text, should_continue, confidence
|
||
|
|
|
||
|
|
def _get_persona_system_prompt(self, persona: str) -> str:
|
||
|
|
"""Get system prompt for a persona"""
|
||
|
|
if persona == "evil":
|
||
|
|
from utils.evil_mode import get_evil_system_prompt
|
||
|
|
return get_evil_system_prompt()
|
||
|
|
else:
|
||
|
|
# Regular Miku prompt - simplified for dialogue
|
||
|
|
return """You are Hatsune Miku, the virtual singer. You are in a conversation with your alter ego, Evil Miku.
|
||
|
|
|
||
|
|
You are generally kind, bubbly, and optimistic, but you're not a pushover. You can be:
|
||
|
|
- Assertive when defending your values
|
||
|
|
- Frustrated when she's being cruel
|
||
|
|
- Curious about her perspective
|
||
|
|
- Hopeful that you can find common ground
|
||
|
|
- Playful when the mood allows
|
||
|
|
|
||
|
|
Respond naturally and conversationally. Keep responses concise (1-3 sentences typically).
|
||
|
|
You can use emojis naturally! ✨💙"""
|
||
|
|
|
||
|
|
# ========================================================================
|
||
|
|
# DIALOGUE TURN HANDLING
|
||
|
|
# ========================================================================
|
||
|
|
|
||
|
|
async def handle_dialogue_turn(
|
||
|
|
self,
|
||
|
|
channel: discord.TextChannel,
|
||
|
|
responding_persona: str,
|
||
|
|
trigger_reason: str = None
|
||
|
|
):
|
||
|
|
"""
|
||
|
|
Handle one turn of dialogue, tracking tension for potential argument escalation.
|
||
|
|
"""
|
||
|
|
channel_id = channel.id
|
||
|
|
|
||
|
|
# Get or create dialogue state
|
||
|
|
state = self.active_dialogues.get(channel_id)
|
||
|
|
if not state:
|
||
|
|
state = self.start_dialogue(channel_id)
|
||
|
|
|
||
|
|
# Safety limits
|
||
|
|
if state["turn_count"] >= MAX_TURNS:
|
||
|
|
print(f"🛑 Dialogue reached {MAX_TURNS} turns, ending")
|
||
|
|
self.end_dialogue(channel_id)
|
||
|
|
return
|
||
|
|
|
||
|
|
if time.time() - state["started_at"] > DIALOGUE_TIMEOUT:
|
||
|
|
print(f"🛑 Dialogue timeout (15 min), ending")
|
||
|
|
self.end_dialogue(channel_id)
|
||
|
|
return
|
||
|
|
|
||
|
|
# Build context from recent messages
|
||
|
|
context = await self._build_conversation_context(channel)
|
||
|
|
|
||
|
|
# Generate response with continuation signal
|
||
|
|
response_text, should_continue, confidence = await self.generate_response_with_continuation(
|
||
|
|
channel=channel,
|
||
|
|
responding_persona=responding_persona,
|
||
|
|
context=context,
|
||
|
|
)
|
||
|
|
|
||
|
|
if not response_text:
|
||
|
|
print(f"⚠️ Failed to generate response for {responding_persona}")
|
||
|
|
self.end_dialogue(channel_id)
|
||
|
|
return
|
||
|
|
|
||
|
|
# Calculate tension change
|
||
|
|
tension_delta = self.calculate_tension_delta(response_text, state["tension"])
|
||
|
|
state["tension"] = max(0.0, min(1.0, state["tension"] + tension_delta))
|
||
|
|
state["tension_history"].append({
|
||
|
|
"turn": state["turn_count"],
|
||
|
|
"speaker": responding_persona,
|
||
|
|
"delta": tension_delta,
|
||
|
|
"total": state["tension"],
|
||
|
|
})
|
||
|
|
|
||
|
|
print(f"🌡️ Tension: {state['tension']:.2f} (delta: {tension_delta:+.2f})")
|
||
|
|
|
||
|
|
# Check if we should escalate to argument
|
||
|
|
if state["tension"] >= ARGUMENT_TENSION_THRESHOLD:
|
||
|
|
print(f"🔥 TENSION THRESHOLD REACHED ({state['tension']:.2f}) - ESCALATING TO ARGUMENT")
|
||
|
|
|
||
|
|
# Send the response that pushed us over
|
||
|
|
await self._send_as_persona(channel, responding_persona, response_text)
|
||
|
|
|
||
|
|
# Transition to argument system
|
||
|
|
await self._escalate_to_argument(channel, responding_persona, response_text)
|
||
|
|
return
|
||
|
|
|
||
|
|
# Send response
|
||
|
|
await self._send_as_persona(channel, responding_persona, response_text)
|
||
|
|
|
||
|
|
# Update state
|
||
|
|
state["turn_count"] += 1
|
||
|
|
state["last_speaker"] = responding_persona
|
||
|
|
|
||
|
|
print(f"🗣️ Turn {state['turn_count']}: {responding_persona} | Continue: {should_continue} ({confidence}) | Tension: {state['tension']:.2f}")
|
||
|
|
|
||
|
|
# Decide what happens next
|
||
|
|
opposite = "evil" if responding_persona == "miku" else "miku"
|
||
|
|
|
||
|
|
if should_continue and confidence in ["HIGH", "MEDIUM"]:
|
||
|
|
asyncio.create_task(self._next_turn(channel, opposite))
|
||
|
|
|
||
|
|
elif should_continue and confidence == "LOW":
|
||
|
|
asyncio.create_task(self._next_turn(channel, opposite))
|
||
|
|
|
||
|
|
elif not should_continue and confidence == "LOW":
|
||
|
|
# Offer opposite persona the last word
|
||
|
|
asyncio.create_task(
|
||
|
|
self._offer_last_word(channel, opposite, context + f"\n{responding_persona}: {response_text}")
|
||
|
|
)
|
||
|
|
else:
|
||
|
|
# Clear signal to end
|
||
|
|
print(f"🏁 Dialogue ended naturally after {state['turn_count']} turns (tension: {state['tension']:.2f})")
|
||
|
|
self.end_dialogue(channel_id)
|
||
|
|
|
||
|
|
async def _next_turn(self, channel: discord.TextChannel, persona: str):
|
||
|
|
"""Queue the next turn"""
|
||
|
|
# Check if dialogue was interrupted
|
||
|
|
if await self._was_interrupted(channel):
|
||
|
|
print(f"💬 Dialogue interrupted by other activity")
|
||
|
|
self.end_dialogue(channel.id)
|
||
|
|
return
|
||
|
|
|
||
|
|
await self.handle_dialogue_turn(channel, persona)
|
||
|
|
|
||
|
|
async def _offer_last_word(self, channel: discord.TextChannel, persona: str, context: str):
|
||
|
|
"""
|
||
|
|
When speaker said NO with LOW confidence, ask opposite if they want to respond.
|
||
|
|
"""
|
||
|
|
from utils.llm import query_llama
|
||
|
|
|
||
|
|
channel_id = channel.id
|
||
|
|
state = self.active_dialogues.get(channel_id)
|
||
|
|
|
||
|
|
if not state:
|
||
|
|
return
|
||
|
|
|
||
|
|
if await self._was_interrupted(channel):
|
||
|
|
self.end_dialogue(channel_id)
|
||
|
|
return
|
||
|
|
|
||
|
|
system_prompt = self._get_persona_system_prompt(persona)
|
||
|
|
|
||
|
|
prompt = f"""{system_prompt}
|
||
|
|
|
||
|
|
Recent exchange:
|
||
|
|
{context}
|
||
|
|
|
||
|
|
The conversation seems to be wrapping up, but wasn't explicitly ended.
|
||
|
|
|
||
|
|
Do you have anything to add? If so, respond naturally.
|
||
|
|
If you're fine letting it end here, write only: [DONE]
|
||
|
|
|
||
|
|
Don't force a response if you have nothing meaningful to contribute."""
|
||
|
|
|
||
|
|
model = globals.EVIL_TEXT_MODEL if persona == "evil" else globals.TEXT_MODEL
|
||
|
|
|
||
|
|
original_evil_mode = globals.EVIL_MODE
|
||
|
|
globals.EVIL_MODE = (persona == "evil")
|
||
|
|
|
||
|
|
try:
|
||
|
|
response = await query_llama(
|
||
|
|
user_prompt=prompt,
|
||
|
|
user_id=f"persona_dialogue_{channel_id}",
|
||
|
|
guild_id=channel.guild.id if hasattr(channel, 'guild') and channel.guild else None,
|
||
|
|
response_type="autonomous_general",
|
||
|
|
model=model
|
||
|
|
)
|
||
|
|
finally:
|
||
|
|
globals.EVIL_MODE = original_evil_mode
|
||
|
|
|
||
|
|
if not response:
|
||
|
|
self.end_dialogue(channel_id)
|
||
|
|
return
|
||
|
|
|
||
|
|
if "[DONE]" in response.upper():
|
||
|
|
print(f"🏁 {persona} chose not to respond, dialogue ended (tension: {state['tension']:.2f})")
|
||
|
|
self.end_dialogue(channel_id)
|
||
|
|
else:
|
||
|
|
clean_response = response.replace("[DONE]", "").strip()
|
||
|
|
|
||
|
|
# Calculate tension
|
||
|
|
tension_delta = self.calculate_tension_delta(clean_response, state["tension"])
|
||
|
|
state["tension"] = max(0.0, min(1.0, state["tension"] + tension_delta))
|
||
|
|
|
||
|
|
print(f"🌡️ Last word tension: {state['tension']:.2f} (delta: {tension_delta:+.2f})")
|
||
|
|
|
||
|
|
# Check for argument escalation
|
||
|
|
if state["tension"] >= ARGUMENT_TENSION_THRESHOLD:
|
||
|
|
print(f"🔥 TENSION THRESHOLD REACHED on last word - ESCALATING TO ARGUMENT")
|
||
|
|
await self._send_as_persona(channel, persona, clean_response)
|
||
|
|
await self._escalate_to_argument(channel, persona, clean_response)
|
||
|
|
return
|
||
|
|
|
||
|
|
# Normal flow
|
||
|
|
await self._send_as_persona(channel, persona, clean_response)
|
||
|
|
|
||
|
|
state["turn_count"] += 1
|
||
|
|
|
||
|
|
# Check if this looks like a closing statement
|
||
|
|
opposite = "evil" if persona == "miku" else "miku"
|
||
|
|
await self._check_if_final(channel, persona, clean_response, opposite)
|
||
|
|
|
||
|
|
async def _check_if_final(self, channel: discord.TextChannel, speaker: str, response: str, opposite: str):
|
||
|
|
"""Check if a response looks like a closing statement"""
|
||
|
|
state = self.active_dialogues.get(channel.id)
|
||
|
|
if not state:
|
||
|
|
return
|
||
|
|
|
||
|
|
# Simple heuristics for closing statements
|
||
|
|
closing_indicators = [
|
||
|
|
response.rstrip().endswith('.'), # Statement, not question
|
||
|
|
'?' not in response, # No questions asked
|
||
|
|
len(response) < 100, # Short responses often close things
|
||
|
|
]
|
||
|
|
|
||
|
|
if all(closing_indicators):
|
||
|
|
print(f"🏁 Dialogue ended after last word, {state['turn_count']} turns total")
|
||
|
|
self.end_dialogue(channel.id)
|
||
|
|
else:
|
||
|
|
asyncio.create_task(self._next_turn(channel, opposite))
|
||
|
|
|
||
|
|
# ========================================================================
|
||
|
|
# ARGUMENT ESCALATION
|
||
|
|
# ========================================================================
|
||
|
|
|
||
|
|
async def _escalate_to_argument(self, channel: discord.TextChannel, last_speaker: str, triggering_message: str):
|
||
|
|
"""
|
||
|
|
Transition from dialogue to full bipolar argument.
|
||
|
|
"""
|
||
|
|
from utils.bipolar_mode import is_argument_in_progress, run_argument
|
||
|
|
|
||
|
|
# Clean up dialogue state
|
||
|
|
self.end_dialogue(channel.id)
|
||
|
|
|
||
|
|
# Don't start if an argument is already going
|
||
|
|
if is_argument_in_progress(channel.id):
|
||
|
|
print(f"⚠️ Argument already in progress, skipping escalation")
|
||
|
|
return
|
||
|
|
|
||
|
|
# Build context for the argument
|
||
|
|
escalation_context = f"""This argument erupted from a conversation that got heated.
|
||
|
|
The last thing said was: "{triggering_message}"
|
||
|
|
|
||
|
|
This pushed things over the edge into a full argument."""
|
||
|
|
|
||
|
|
print(f"⚔️ Escalating to argument in #{channel.name}")
|
||
|
|
|
||
|
|
# Use the existing argument system
|
||
|
|
# Pass the triggering message so the opposite persona responds to it
|
||
|
|
await run_argument(
|
||
|
|
channel=channel,
|
||
|
|
client=globals.client,
|
||
|
|
trigger_context=escalation_context,
|
||
|
|
)
|
||
|
|
|
||
|
|
# ========================================================================
|
||
|
|
# HELPER METHODS
|
||
|
|
# ========================================================================
|
||
|
|
|
||
|
|
async def _was_interrupted(self, channel: discord.TextChannel) -> bool:
|
||
|
|
"""Check if someone else sent a message during the dialogue"""
|
||
|
|
state = self.active_dialogues.get(channel.id)
|
||
|
|
if not state:
|
||
|
|
return True
|
||
|
|
|
||
|
|
try:
|
||
|
|
async for msg in channel.history(limit=1):
|
||
|
|
# If latest message is NOT from our webhooks, we were interrupted
|
||
|
|
if not msg.webhook_id:
|
||
|
|
# Check if it's from the bot itself (could be normal response)
|
||
|
|
if msg.author.id != globals.client.user.id:
|
||
|
|
return True
|
||
|
|
except Exception as e:
|
||
|
|
print(f"⚠️ Error checking for interruption: {e}")
|
||
|
|
|
||
|
|
return False
|
||
|
|
|
||
|
|
async def _build_conversation_context(self, channel: discord.TextChannel, limit: int = 15) -> str:
|
||
|
|
"""Get recent messages for context"""
|
||
|
|
messages = []
|
||
|
|
try:
|
||
|
|
async for msg in channel.history(limit=limit):
|
||
|
|
speaker = self._identify_speaker(msg)
|
||
|
|
messages.append(f"{speaker}: {msg.content}")
|
||
|
|
|
||
|
|
messages.reverse()
|
||
|
|
except Exception as e:
|
||
|
|
print(f"⚠️ Error building conversation context: {e}")
|
||
|
|
|
||
|
|
return '\n'.join(messages)
|
||
|
|
|
||
|
|
def _identify_speaker(self, message: discord.Message) -> str:
|
||
|
|
"""Identify who sent a message"""
|
||
|
|
if message.webhook_id:
|
||
|
|
name_lower = (message.author.name or "").lower()
|
||
|
|
if "evil" in name_lower:
|
||
|
|
return "Evil Miku"
|
||
|
|
return "Hatsune Miku"
|
||
|
|
elif message.author.id == globals.client.user.id:
|
||
|
|
# Bot's own messages - check mode at time of message
|
||
|
|
if globals.EVIL_MODE:
|
||
|
|
return "Evil Miku"
|
||
|
|
return "Hatsune Miku"
|
||
|
|
return message.author.display_name
|
||
|
|
|
||
|
|
async def _send_as_persona(self, channel: discord.TextChannel, persona: str, content: str):
|
||
|
|
"""Send message via webhook"""
|
||
|
|
from utils.bipolar_mode import (
|
||
|
|
get_or_create_webhooks_for_channel,
|
||
|
|
get_miku_display_name,
|
||
|
|
get_evil_miku_display_name
|
||
|
|
)
|
||
|
|
|
||
|
|
webhooks = await get_or_create_webhooks_for_channel(channel)
|
||
|
|
if not webhooks:
|
||
|
|
print(f"⚠️ Could not get webhooks for #{channel.name}")
|
||
|
|
return
|
||
|
|
|
||
|
|
webhook = webhooks["evil_miku"] if persona == "evil" else webhooks["miku"]
|
||
|
|
display_name = get_evil_miku_display_name() if persona == "evil" else get_miku_display_name()
|
||
|
|
|
||
|
|
try:
|
||
|
|
await webhook.send(content=content, username=display_name)
|
||
|
|
except Exception as e:
|
||
|
|
print(f"⚠️ Error sending as {persona}: {e}")
|
||
|
|
|
||
|
|
|
||
|
|
# ============================================================================
|
||
|
|
# CONVENIENCE FUNCTIONS
|
||
|
|
# ============================================================================
|
||
|
|
|
||
|
|
# Singleton instances
|
||
|
|
_scorer = None
|
||
|
|
_dialogue_manager = None
|
||
|
|
|
||
|
|
def get_interjection_scorer() -> InterjectionScorer:
|
||
|
|
"""Get the singleton InterjectionScorer instance"""
|
||
|
|
global _scorer
|
||
|
|
if _scorer is None:
|
||
|
|
_scorer = InterjectionScorer()
|
||
|
|
return _scorer
|
||
|
|
|
||
|
|
def get_dialogue_manager() -> PersonaDialogue:
|
||
|
|
"""Get the singleton PersonaDialogue instance"""
|
||
|
|
global _dialogue_manager
|
||
|
|
if _dialogue_manager is None:
|
||
|
|
_dialogue_manager = PersonaDialogue()
|
||
|
|
return _dialogue_manager
|
||
|
|
|
||
|
|
|
||
|
|
async def check_for_interjection(message: discord.Message, current_persona: str) -> bool:
|
||
|
|
"""
|
||
|
|
Check if the opposite persona should interject based on a message.
|
||
|
|
|
||
|
|
If they should, starts a dialogue automatically.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
message: The Discord message that was just sent
|
||
|
|
current_persona: Who sent the message ("miku" or "evil")
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
True if an interjection was triggered, False otherwise
|
||
|
|
"""
|
||
|
|
print(f"🔍 [Persona Dialogue] Checking interjection for message from {current_persona}")
|
||
|
|
|
||
|
|
scorer = get_interjection_scorer()
|
||
|
|
dialogue_manager = get_dialogue_manager()
|
||
|
|
|
||
|
|
# Don't trigger if dialogue already active
|
||
|
|
if dialogue_manager.is_dialogue_active(message.channel.id):
|
||
|
|
print(f"⏸️ [Persona Dialogue] Dialogue already active in channel {message.channel.id}")
|
||
|
|
return False
|
||
|
|
|
||
|
|
# Check if we should interject
|
||
|
|
should_interject, reason, score = await scorer.should_interject(message, current_persona)
|
||
|
|
|
||
|
|
print(f"📊 [Persona Dialogue] Interjection check: should_interject={should_interject}, reason={reason}, score={score:.2f}")
|
||
|
|
|
||
|
|
if should_interject:
|
||
|
|
opposite_persona = "evil" if current_persona == "miku" else "miku"
|
||
|
|
print(f"🎭 Triggering {opposite_persona} interjection (reason: {reason}, score: {score:.2f})")
|
||
|
|
|
||
|
|
# Start dialogue with the opposite persona responding first
|
||
|
|
dialogue_manager.start_dialogue(message.channel.id)
|
||
|
|
asyncio.create_task(
|
||
|
|
dialogue_manager.handle_dialogue_turn(message.channel, opposite_persona, trigger_reason=reason)
|
||
|
|
)
|
||
|
|
return True
|
||
|
|
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def is_persona_dialogue_active(channel_id: int) -> bool:
|
||
|
|
"""Check if a persona dialogue is currently active in a channel"""
|
||
|
|
dialogue_manager = get_dialogue_manager()
|
||
|
|
return dialogue_manager.is_dialogue_active(channel_id)
|