Koko210/miku-discord
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
323ca753d118447675f3e37bf9f05a95587e9736
miku-discord/bot/utils/persona_dialogue.py

968 lines
38 KiB
Python
Raw Blame History

# 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 discord
import asyncio
import time
import globals
from utils.logger import get_logger
logger = get_logger('persona')
import os
import json
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.5 # Score needed to trigger interjection
# ============================================================================
# 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:
logger.debug("Loading sentiment analyzer for persona dialogue...")
try:
self._sentiment_analyzer = pipeline(
"sentiment-analysis",
model="distilbert-base-uncased-finetuned-sst-2-english"
)
logger.info("Sentiment analyzer loaded")
except Exception as e:
logger.error(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"
logger.debug(f"[Interjection] Analyzing content: '{message.content[:100]}...'")
logger.debug(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):
logger.info(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:
logger.info(f"{opposite_persona.upper()} WILL INTERJECT (score: {score:.2f})")
logger.info(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:
logger.debug(f"[Basic Filter] System message type: {message.type}")
return False
# Bipolar mode must be enabled
if not globals.BIPOLAR_MODE:
logger.debug(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:
logger.debug(f"[Basic Filter] Other bot message (not our bot)")
return False
logger.debug(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:
logger.error(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()
logger.info(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]
logger.info(f"Ended persona dialogue in channel {channel_id}")
logger.info(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:
logger.error(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:
logger.debug(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:
logger.info(f"Dialogue reached {MAX_TURNS} turns, ending")
self.end_dialogue(channel_id)
return
if time.time() - state["started_at"] > DIALOGUE_TIMEOUT:
logger.info(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:
logger.error(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"],
})
logger.debug(f"Tension: {state['tension']:.2f} (delta: {tension_delta:+.2f})")
# Check if we should escalate to argument
if state["tension"] >= ARGUMENT_TENSION_THRESHOLD:
logger.info(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
logger.debug(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
logger.info(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):
logger.info(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():
logger.info(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))
logger.debug(f"Last word tension: {state['tension']:.2f} (delta: {tension_delta:+.2f})")
# Check for argument escalation
if state["tension"] >= ARGUMENT_TENSION_THRESHOLD:
logger.info(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):
logger.info(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):
logger.warning(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."""
logger.info(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:
logger.warning(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:
logger.warning(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:
logger.warning(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:
logger.error(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
"""
logger.debug(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):
logger.debug(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)
logger.debug(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"
logger.info(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)
Reference in New Issue View Git Blame Copy Permalink