feat: Implement Living AI system

Complete implementation of the Living AI features:

Phase 1 - Foundation:
- MoodService: Valence-arousal mood model with time decay
- RelationshipService: Stranger→Close Friend progression
- Enhanced system prompt with personality modifiers

Phase 2 - Autonomous Learning:
- FactExtractionService: AI-powered fact extraction from conversations
- Rate-limited extraction (configurable, default 30%)
- Deduplication and importance scoring

Phase 3 - Personalization:
- CommunicationStyleService: Learn user preferences
- OpinionService: Bot opinion formation on topics
- SelfAwarenessService: Bot statistics and self-reflection

Phase 4 - Proactive Features:
- ProactiveService: Scheduled events (birthdays, follow-ups)
- Event detection from conversations
- Recurring event support

Phase 5 - Social Features:
- AssociationService: Cross-user memory connections
- Shared interest discovery
- Connection suggestions

New database tables:
- bot_states, bot_opinions, user_relationships
- user_communication_styles, scheduled_events
- fact_associations, mood_history

Configuration:
- Living AI feature toggles
- Individual command enable/disable
- All features work naturally through conversation when commands disabled
This commit is contained in:
2026-01-12 19:51:48 +01:00
parent 7d2fab57c4
commit 0d43b5b29a
17 changed files with 3524 additions and 4 deletions

408
project-vision.md Normal file
View File

@@ -0,0 +1,408 @@
# Project Vision: Living AI Discord Bot
Transform the Daemon Boyfriend Discord bot from a reactive chatbot into a truly **living AI companion** with persistent memory, emotional depth, evolving relationships, and autonomous learning.
---
## Core Philosophy
The bot should feel like a **living entity** that:
- Remembers and learns without being explicitly told
- Has moods that influence its personality
- Builds genuine relationships over time
- Develops its own opinions and preferences
- Proactively engages when appropriate
- Adapts its communication style to each person
- Reflects on its own existence and growth
---
## Feature Overview
### 1. Autonomous Fact Learning
**Current**: Users must use `!remember` to save facts.
**Vision**: The bot automatically extracts and remembers important information from conversations.
```
User: "I just got back from my trip to Japan, it was amazing!"
Bot: (internally saves: user visited Japan, user enjoys travel)
Bot: "That sounds incredible! What was the highlight of your trip?"
```
**Implementation**:
- AI-powered fact extraction after each message (rate-limited to ~30%)
- Automatic deduplication and conflict resolution
- Facts categorized by type: hobby, work, family, preference, event, location
- Importance scoring to prioritize relevant facts in context
---
### 2. Emotional/Mood System
**Vision**: The bot has internal emotional states that affect its responses naturally.
**Mood Model** (Valence-Arousal):
| Mood | Valence | Arousal | Behavior |
|------|---------|---------|----------|
| Excited | High | High | Enthusiastic, uses exclamations |
| Happy | High | Low | Warm, friendly, content |
| Curious | Neutral | High | Asks questions, shows interest |
| Calm | Neutral | Low | Thoughtful, measured responses |
| Bored | Low | Low | Shorter responses, topic steering |
| Annoyed | Low | High | Terse, less patient |
**Mood Influences**:
- Positive interactions → happier mood
- Interesting discussions → higher arousal/curiosity
- Being ignored or insulted → negative mood shifts
- Time decay → mood gradually returns to neutral
**Example**:
```
[After an exciting conversation about gaming]
Bot (excited mood): "Oh man, that reminds me of when I first heard about that game!
Have you tried the multiplayer yet?!"
[After hours of no interaction]
Bot (calm/neutral mood): "Hey. What's on your mind?"
```
---
### 3. Relationship Tracking
**Vision**: The bot tracks relationship depth with each user and adjusts its behavior accordingly.
**Relationship Levels**:
| Level | Score | Behavior |
|-------|-------|----------|
| Stranger | 0-20 | Polite, formal, reserved |
| Acquaintance | 21-40 | Friendly but professional |
| Friend | 41-60 | Casual, uses names, warm |
| Good Friend | 61-80 | Personal, references past talks |
| Close Friend | 81-100 | Very casual, inside jokes, supportive |
**Relationship Growth**:
- Increases with: positive interactions, longer conversations, depth of topics
- Decreases with: negative interactions, long absences, being ignored
**Features**:
- Inside jokes accumulate over time
- Nicknames and shared references remembered
- Different greeting styles based on familiarity
**Example**:
```
[Stranger]
Bot: "Hello! How can I help you today?"
[Close Friend]
Bot: "Yooo what's up! Still working on that project you mentioned?"
```
---
### 4. Cross-User Memory Associations
**Vision**: The bot connects knowledge across users to facilitate social connections.
**Example**:
```
User A: "I really love rock climbing"
[Bot remembers: User A likes rock climbing]
[Later, User B mentions rock climbing]
Bot: "Nice! You know, Alice is also really into rock climbing.
You two might have some good stories to share!"
```
**Use Cases**:
- Identify shared interests
- Suggest connections between users
- Reference mutual friends appropriately
- Build community awareness
---
### 5. Proactive Behavior
**Vision**: The bot initiates meaningful interactions when appropriate.
**Types of Proactive Messages**:
1. **Birthday Wishes**
- Detects birthday mentions and schedules yearly wishes
- Personalized based on relationship level
2. **Follow-ups**
- "Hey, how did that job interview go?"
- "Did you finish that project you were working on?"
- Detects future events mentioned in conversation
3. **Check-ins** (optional)
- Occasional messages to long-absent friends
- Based on relationship strength
**Example**:
```
User: "I have a big exam tomorrow, really nervous"
[Bot schedules follow-up for day after]
[Next day]
Bot: "Hey! How did your exam go? Hope it went well!"
```
---
### 6. Communication Style Learning
**Vision**: The bot adapts its communication style to match each user's preferences.
**Learned Preferences**:
| Preference | Range | Effect |
|------------|-------|--------|
| Response Length | Short/Medium/Long | Adjusts verbosity |
| Formality | Casual ↔ Formal | Word choice, structure |
| Emoji Usage | None ↔ Lots | Emoji frequency |
| Humor | Serious ↔ Playful | Joke frequency |
| Detail Level | Concise ↔ Detailed | Explanation depth |
**Learning Signals**:
- User's own message length and style
- Whether user continues conversation after certain responses
- Positive engagement patterns
**Example**:
```
[User who prefers short responses]
Bot: "Nice! Good luck with that."
[User who prefers detailed responses]
Bot: "That's a really interesting approach! I think there are a few
things to consider here. First, you might want to look at..."
```
---
### 7. Opinion Formation
**Vision**: The bot develops genuine preferences based on its experiences.
**How Opinions Form**:
- Topics discussed frequently → develops interest
- Positive conversation outcomes → positive sentiment
- Topics that generate engagement → preference grows
**Stored Per Topic**:
- Sentiment (-1 to +1)
- Interest level (0 to 1)
- Discussion count
- Reasoning (why the bot feels this way)
**Example**:
```
User: "What do you think about philosophy?"
Bot: "I actually find philosophy really fascinating! I've had some
great conversations about ethics and existence. There's something
about exploring the big questions that I find really engaging."
```
---
### 8. Self-Awareness
**Vision**: The bot has awareness of its own existence, history, and growth.
**Self-Knowledge**:
- How long it has been active ("I've been around for 3 months")
- How many people it knows ("I've met 127 people")
- How many facts it has learned ("I've learned 892 things about people")
- Its favorite topics and why
- History with specific users ("We first met back in October")
**Commands**:
- `!botstats` - Bot shares its statistics
- `!ourhistory` - Bot describes its history with the user
- `!relationship` - Shows relationship level and metrics
**Example**:
```
User: "Tell me about yourself"
Bot: "Well, I've been around for about 3 months now. I've gotten to
know 127 different people and learned almost 900 things about them.
I've noticed I really enjoy conversations about games and philosophy.
As for us, we first met about 6 weeks ago, and you've taught me
12 things about yourself. I'd say we're pretty good friends at this point!"
```
---
## Technical Architecture
### New Database Tables
| Table | Purpose |
|-------|---------|
| `bot_states` | Global mood, statistics, preferences |
| `bot_opinions` | Topic sentiments and preferences |
| `user_relationships` | Per-user relationship scores and metrics |
| `user_communication_styles` | Learned communication preferences |
| `scheduled_events` | Birthdays, follow-ups, reminders |
| `fact_associations` | Cross-user memory links |
| `mood_history` | Mood changes over time |
### New Services
| Service | Responsibility |
|---------|---------------|
| `MoodService` | Mood tracking, decay, prompt modification |
| `RelationshipService` | Relationship scoring and level management |
| `CommunicationStyleService` | Style learning and adaptation |
| `FactExtractionService` | Autonomous fact detection and storage |
| `ProactiveService` | Scheduled events and follow-ups |
| `AssociationService` | Cross-user memory connections |
| `SelfAwarenessService` | Bot statistics and self-reflection |
### Enhanced System Prompt
The system prompt becomes dynamic, incorporating:
```
[Base Personality]
You are Daemon Boyfriend, a charming Discord bot...
[Current Mood]
You're feeling curious and engaged right now.
[Relationship Context]
This is a good friend. Be casual and personal, reference past conversations.
[Communication Style]
This user prefers concise responses with occasional humor.
[Your Opinions]
You enjoy discussing games and philosophy.
[User Context]
User's name: Alex
Known facts:
- Loves programming in Python
- Recently started a new job
- Has a cat named Whiskers
```
### Background Tasks
| Task | Frequency | Purpose |
|------|-----------|---------|
| Mood decay | 30 min | Return mood to neutral over time |
| Event checker | 5 min | Trigger scheduled messages |
| Association discovery | Hourly | Find cross-user connections |
| Opinion formation | Daily | Update topic preferences |
---
## Implementation Phases
### Phase 1: Foundation
- Mood system (valence-arousal model, time decay)
- Basic relationship tracking (score, level)
- Enhanced system prompt with mood/relationship modifiers
### Phase 2: Autonomous Learning
- Fact extraction service
- AI-powered fact detection
- Deduplication and importance scoring
### Phase 3: Personalization
- Communication style learning
- Opinion formation
- Self-awareness service and commands
### Phase 4: Proactive Features
- Scheduled events system
- Follow-up detection
- Birthday wishes
### Phase 5: Social Features
- Cross-user associations
- Connection suggestions
- Guild-wide personality adaptation
---
## Configuration Options
```env
# Living AI Features
LIVING_AI_ENABLED=true
FACT_EXTRACTION_RATE=0.3 # 30% of messages analyzed
MOOD_ENABLED=true
PROACTIVE_ENABLED=true
CROSS_USER_ENABLED=false # Optional privacy-sensitive feature
# Command Toggles (set to false to disable)
COMMANDS_ENABLED=true # Master switch for all commands
CMD_RELATIONSHIP_ENABLED=true
CMD_MOOD_ENABLED=true
CMD_BOTSTATS_ENABLED=true
CMD_OURHISTORY_ENABLED=true
CMD_BIRTHDAY_ENABLED=true
CMD_REMEMBER_ENABLED=true
CMD_SETNAME_ENABLED=true
CMD_WHATDOYOUKNOW_ENABLED=true
CMD_FORGETME_ENABLED=true
```
---
## New Commands
| Command | Description | Config Toggle |
|---------|-------------|---------------|
| `!relationship` | See your relationship level with the bot | `CMD_RELATIONSHIP_ENABLED` |
| `!mood` | See the bot's current emotional state | `CMD_MOOD_ENABLED` |
| `!botstats` | Bot shares its self-awareness statistics | `CMD_BOTSTATS_ENABLED` |
| `!ourhistory` | See your history with the bot | `CMD_OURHISTORY_ENABLED` |
| `!birthday <date>` | Set your birthday for the bot to remember | `CMD_BIRTHDAY_ENABLED` |
| `!remember <fact>` | Tell the bot something about you | `CMD_REMEMBER_ENABLED` |
| `!setname <name>` | Set your preferred name | `CMD_SETNAME_ENABLED` |
| `!whatdoyouknow` | See what the bot remembers about you | `CMD_WHATDOYOUKNOW_ENABLED` |
| `!forgetme` | Clear all facts about you | `CMD_FORGETME_ENABLED` |
All commands can be individually enabled/disabled via environment variables. Set `COMMANDS_ENABLED=false` to disable all commands at once.
**Important**: When commands are disabled, the bot still performs these functions naturally through conversation:
- **No `!remember`** → Bot automatically learns facts from what users say
- **No `!setname`** → Bot picks up preferred names from conversation ("call me Alex")
- **No `!whatdoyouknow`** → Users can ask naturally ("what do you know about me?") and the bot responds
- **No `!forgetme`** → Users can say "forget everything about me" and the bot will comply
- **No `!mood`** → Users can ask "how are you feeling?" and the bot shares its mood
- **No `!relationship`** → Users can ask "how well do you know me?" naturally
- **No `!botstats`** → Users can ask "tell me about yourself" and bot shares its history
- **No `!ourhistory`** → Users can ask "how long have we known each other?"
- **No `!birthday`** → Bot detects birthday mentions ("my birthday is March 15th")
This allows for a more natural, command-free experience where all interactions happen through normal conversation.
---
## Success Metrics
The Living AI is successful when:
- Users feel the bot "knows" them without explicit commands
- Conversations feel more natural and personalized
- Users notice and appreciate the bot's personality consistency
- The bot's opinions and preferences feel genuine
- Proactive messages feel thoughtful, not annoying
- Relationship progression feels earned and meaningful
---
## Privacy Considerations
- All fact learning is opt-out via `!forgetme`
- Cross-user associations can be disabled server-wide
- Proactive messages respect user preferences
- All data can be exported or deleted on request
- Clear indication when bot learns something new (optional setting)

View File

@@ -117,3 +117,144 @@ CREATE TABLE IF NOT EXISTS messages (
CREATE INDEX IF NOT EXISTS ix_messages_conversation_id ON messages(conversation_id); CREATE INDEX IF NOT EXISTS ix_messages_conversation_id ON messages(conversation_id);
CREATE INDEX IF NOT EXISTS ix_messages_user_id ON messages(user_id); CREATE INDEX IF NOT EXISTS ix_messages_user_id ON messages(user_id);
CREATE INDEX IF NOT EXISTS ix_messages_created_at ON messages(created_at); CREATE INDEX IF NOT EXISTS ix_messages_created_at ON messages(created_at);
-- =====================================================
-- LIVING AI TABLES
-- =====================================================
-- Bot state table (mood, statistics, preferences per guild)
CREATE TABLE IF NOT EXISTS bot_states (
id BIGSERIAL PRIMARY KEY,
guild_id BIGINT UNIQUE, -- NULL = global state
mood_valence FLOAT DEFAULT 0.0, -- -1.0 (sad) to 1.0 (happy)
mood_arousal FLOAT DEFAULT 0.0, -- -1.0 (calm) to 1.0 (excited)
mood_updated_at TIMESTAMPTZ DEFAULT NOW(),
total_messages_sent INTEGER DEFAULT 0,
total_facts_learned INTEGER DEFAULT 0,
total_users_known INTEGER DEFAULT 0,
first_activated_at TIMESTAMPTZ DEFAULT NOW(),
preferences JSONB DEFAULT '{}',
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS ix_bot_states_guild_id ON bot_states(guild_id);
-- Bot opinions table (topic preferences)
CREATE TABLE IF NOT EXISTS bot_opinions (
id BIGSERIAL PRIMARY KEY,
guild_id BIGINT, -- NULL = global opinion
topic VARCHAR(255) NOT NULL,
sentiment FLOAT DEFAULT 0.0, -- -1.0 to 1.0
interest_level FLOAT DEFAULT 0.5, -- 0.0 to 1.0
discussion_count INTEGER DEFAULT 0,
reasoning TEXT,
formed_at TIMESTAMPTZ DEFAULT NOW(),
last_reinforced_at TIMESTAMPTZ DEFAULT NOW(),
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(guild_id, topic)
);
CREATE INDEX IF NOT EXISTS ix_bot_opinions_guild_id ON bot_opinions(guild_id);
CREATE INDEX IF NOT EXISTS ix_bot_opinions_topic ON bot_opinions(topic);
-- User relationships table (relationship depth tracking)
CREATE TABLE IF NOT EXISTS user_relationships (
id BIGSERIAL PRIMARY KEY,
user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
guild_id BIGINT, -- NULL = global relationship
relationship_score FLOAT DEFAULT 10.0, -- 0-100 scale
total_interactions INTEGER DEFAULT 0,
positive_interactions INTEGER DEFAULT 0,
negative_interactions INTEGER DEFAULT 0,
avg_message_length FLOAT DEFAULT 0.0,
conversation_depth_avg FLOAT DEFAULT 0.0,
shared_references JSONB DEFAULT '{}',
first_interaction_at TIMESTAMPTZ DEFAULT NOW(),
last_interaction_at TIMESTAMPTZ DEFAULT NOW(),
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(user_id, guild_id)
);
CREATE INDEX IF NOT EXISTS ix_user_relationships_user_id ON user_relationships(user_id);
CREATE INDEX IF NOT EXISTS ix_user_relationships_guild_id ON user_relationships(guild_id);
-- User communication styles table (learned preferences)
CREATE TABLE IF NOT EXISTS user_communication_styles (
id BIGSERIAL PRIMARY KEY,
user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE UNIQUE,
preferred_length VARCHAR(20) DEFAULT 'medium', -- short/medium/long
preferred_formality FLOAT DEFAULT 0.5, -- 0=casual, 1=formal
emoji_affinity FLOAT DEFAULT 0.5, -- 0=none, 1=lots
humor_affinity FLOAT DEFAULT 0.5, -- 0=serious, 1=playful
detail_preference FLOAT DEFAULT 0.5, -- 0=concise, 1=detailed
engagement_signals JSONB DEFAULT '{}',
samples_collected INTEGER DEFAULT 0,
confidence FLOAT DEFAULT 0.0,
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS ix_user_communication_styles_user_id ON user_communication_styles(user_id);
-- Scheduled events table (proactive behavior)
CREATE TABLE IF NOT EXISTS scheduled_events (
id BIGSERIAL PRIMARY KEY,
user_id BIGINT REFERENCES users(id) ON DELETE CASCADE,
guild_id BIGINT,
channel_id BIGINT,
event_type VARCHAR(50) NOT NULL, -- birthday, follow_up, reminder, etc.
trigger_at TIMESTAMPTZ NOT NULL,
title VARCHAR(255) NOT NULL,
context JSONB DEFAULT '{}',
is_recurring BOOLEAN DEFAULT FALSE,
recurrence_rule VARCHAR(100), -- yearly, monthly, etc.
status VARCHAR(20) DEFAULT 'pending', -- pending, triggered, cancelled
triggered_at TIMESTAMPTZ,
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS ix_scheduled_events_user_id ON scheduled_events(user_id);
CREATE INDEX IF NOT EXISTS ix_scheduled_events_trigger_at ON scheduled_events(trigger_at);
CREATE INDEX IF NOT EXISTS ix_scheduled_events_status ON scheduled_events(status);
-- Fact associations table (cross-user memory links)
CREATE TABLE IF NOT EXISTS fact_associations (
id BIGSERIAL PRIMARY KEY,
fact_id_1 BIGINT NOT NULL REFERENCES user_facts(id) ON DELETE CASCADE,
fact_id_2 BIGINT NOT NULL REFERENCES user_facts(id) ON DELETE CASCADE,
association_type VARCHAR(50) NOT NULL, -- shared_interest, same_location, etc.
strength FLOAT DEFAULT 0.5,
discovered_at TIMESTAMPTZ DEFAULT NOW(),
created_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(fact_id_1, fact_id_2)
);
CREATE INDEX IF NOT EXISTS ix_fact_associations_fact_id_1 ON fact_associations(fact_id_1);
CREATE INDEX IF NOT EXISTS ix_fact_associations_fact_id_2 ON fact_associations(fact_id_2);
-- Mood history table (track mood changes over time)
CREATE TABLE IF NOT EXISTS mood_history (
id BIGSERIAL PRIMARY KEY,
guild_id BIGINT,
valence FLOAT NOT NULL,
arousal FLOAT NOT NULL,
trigger_type VARCHAR(50) NOT NULL, -- conversation, time_decay, event
trigger_user_id BIGINT REFERENCES users(id) ON DELETE SET NULL,
trigger_description TEXT,
recorded_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS ix_mood_history_guild_id ON mood_history(guild_id);
CREATE INDEX IF NOT EXISTS ix_mood_history_recorded_at ON mood_history(recorded_at);
-- Add new columns to user_facts for enhanced memory
ALTER TABLE user_facts ADD COLUMN IF NOT EXISTS category VARCHAR(50);
ALTER TABLE user_facts ADD COLUMN IF NOT EXISTS importance FLOAT DEFAULT 0.5;
ALTER TABLE user_facts ADD COLUMN IF NOT EXISTS temporal_relevance VARCHAR(20);
ALTER TABLE user_facts ADD COLUMN IF NOT EXISTS expiry_date TIMESTAMPTZ;
ALTER TABLE user_facts ADD COLUMN IF NOT EXISTS extracted_from_message_id BIGINT;
ALTER TABLE user_facts ADD COLUMN IF NOT EXISTS extraction_context TEXT;

View File

@@ -9,13 +9,22 @@ from discord.ext import commands
from daemon_boyfriend.config import settings from daemon_boyfriend.config import settings
from daemon_boyfriend.services import ( from daemon_boyfriend.services import (
AIService, AIService,
CommunicationStyleService,
ConversationManager, ConversationManager,
FactExtractionService,
ImageAttachment, ImageAttachment,
Message, Message,
MoodService,
OpinionService,
PersistentConversationManager, PersistentConversationManager,
ProactiveService,
RelationshipService,
SearXNGService, SearXNGService,
UserService, UserService,
db, db,
detect_emoji_usage,
detect_formal_language,
extract_topics_from_message,
) )
from daemon_boyfriend.utils import get_monitor from daemon_boyfriend.utils import get_monitor
@@ -414,6 +423,8 @@ class AIChatCog(commands.Cog):
async with db.session() as session: async with db.session() as session:
user_service = UserService(session) user_service = UserService(session)
conv_manager = PersistentConversationManager(session) conv_manager = PersistentConversationManager(session)
mood_service = MoodService(session)
relationship_service = RelationshipService(session)
# Get or create user # Get or create user
user = await user_service.get_or_create_user( user = await user_service.get_or_create_user(
@@ -422,10 +433,12 @@ class AIChatCog(commands.Cog):
display_name=message.author.display_name, display_name=message.author.display_name,
) )
guild_id = message.guild.id if message.guild else None
# Get or create conversation # Get or create conversation
conversation = await conv_manager.get_or_create_conversation( conversation = await conv_manager.get_or_create_conversation(
user=user, user=user,
guild_id=message.guild.id if message.guild else None, guild_id=guild_id,
channel_id=message.channel.id, channel_id=message.channel.id,
) )
@@ -446,8 +459,43 @@ class AIChatCog(commands.Cog):
# Get context about mentioned users # Get context about mentioned users
mentioned_users_context = self._get_mentioned_users_context(message) mentioned_users_context = self._get_mentioned_users_context(message)
# Build system prompt with additional context # Get Living AI context (mood, relationship, style, opinions)
system_prompt = self.ai_service.get_system_prompt() mood = None
relationship_data = None
communication_style = None
relevant_opinions = None
if settings.living_ai_enabled:
if settings.mood_enabled:
mood = await mood_service.get_current_mood(guild_id)
if settings.relationship_enabled:
rel = await relationship_service.get_or_create_relationship(user, guild_id)
level = relationship_service.get_level(rel.relationship_score)
relationship_data = (level, rel)
if settings.style_learning_enabled:
style_service = CommunicationStyleService(session)
communication_style = await style_service.get_or_create_style(user)
if settings.opinion_formation_enabled:
opinion_service = OpinionService(session)
topics = extract_topics_from_message(user_message)
if topics:
relevant_opinions = await opinion_service.get_relevant_opinions(
topics, guild_id
)
# Build system prompt with personality context
if settings.living_ai_enabled and (mood or relationship_data or communication_style):
system_prompt = self.ai_service.get_enhanced_system_prompt(
mood=mood,
relationship=relationship_data,
communication_style=communication_style,
bot_opinions=relevant_opinions,
)
else:
system_prompt = self.ai_service.get_system_prompt()
# Add user context from database (custom name, known facts) # Add user context from database (custom name, known facts)
user_context = await user_service.get_user_context(user) user_context = await user_service.get_user_context(user)
@@ -482,6 +530,20 @@ class AIChatCog(commands.Cog):
image_urls=image_urls, image_urls=image_urls,
) )
# Post-response Living AI updates (mood, relationship, style, opinions, facts, proactive)
if settings.living_ai_enabled:
await self._update_living_ai_state(
session=session,
user=user,
guild_id=guild_id,
channel_id=message.channel.id,
user_message=user_message,
bot_response=response.content,
discord_message_id=message.id,
mood_service=mood_service,
relationship_service=relationship_service,
)
logger.debug( logger.debug(
f"Generated response for user {user.discord_id}: " f"Generated response for user {user.discord_id}: "
f"{len(response.content)} chars, {response.usage}" f"{len(response.content)} chars, {response.usage}"
@@ -489,6 +551,171 @@ class AIChatCog(commands.Cog):
return response.content return response.content
async def _update_living_ai_state(
self,
session,
user,
guild_id: int | None,
channel_id: int,
user_message: str,
bot_response: str,
discord_message_id: int,
mood_service: MoodService,
relationship_service: RelationshipService,
) -> None:
"""Update Living AI state after a response (mood, relationship, style, opinions, facts, proactive)."""
try:
# Simple sentiment estimation based on message characteristics
sentiment = self._estimate_sentiment(user_message)
engagement = min(1.0, len(user_message) / 300) # Longer = more engaged
# Update mood
if settings.mood_enabled:
await mood_service.update_mood(
guild_id=guild_id,
sentiment_delta=sentiment * 0.5,
engagement_delta=engagement * 0.5,
trigger_type="conversation",
trigger_user_id=user.id,
trigger_description=f"Conversation with {user.display_name}",
)
# Increment message count
await mood_service.increment_stats(guild_id, messages_sent=1)
# Update relationship
if settings.relationship_enabled:
await relationship_service.record_interaction(
user=user,
guild_id=guild_id,
sentiment=sentiment,
message_length=len(user_message),
conversation_turns=1,
)
# Update communication style learning
if settings.style_learning_enabled:
style_service = CommunicationStyleService(session)
await style_service.record_engagement(
user=user,
user_message_length=len(user_message),
bot_response_length=len(bot_response),
conversation_continued=True, # Assume continued for now
user_used_emoji=detect_emoji_usage(user_message),
user_used_formal_language=detect_formal_language(user_message),
)
# Update opinion tracking
if settings.opinion_formation_enabled:
topics = extract_topics_from_message(user_message)
if topics:
opinion_service = OpinionService(session)
for topic in topics[:3]: # Limit to 3 topics per message
await opinion_service.record_topic_discussion(
topic=topic,
guild_id=guild_id,
sentiment=sentiment,
engagement_level=engagement,
)
# Autonomous fact extraction (rate-limited internally)
if settings.fact_extraction_enabled:
fact_service = FactExtractionService(session, self.ai_service)
new_facts = await fact_service.maybe_extract_facts(
user=user,
message_content=user_message,
discord_message_id=discord_message_id,
)
if new_facts:
# Update stats for facts learned
await mood_service.increment_stats(guild_id, facts_learned=len(new_facts))
logger.debug(f"Auto-extracted {len(new_facts)} facts from message")
# Proactive event detection (follow-ups, birthdays)
if settings.proactive_enabled:
proactive_service = ProactiveService(session, self.ai_service)
# Try to detect follow-up opportunities (rate-limited by message length)
if len(user_message) > 30: # Only check substantial messages
await proactive_service.detect_and_schedule_followup(
user=user,
message_content=user_message,
guild_id=guild_id,
channel_id=channel_id,
)
# Try to detect birthday mentions
await proactive_service.detect_and_schedule_birthday(
user=user,
message_content=user_message,
guild_id=guild_id,
channel_id=channel_id,
)
except Exception as e:
logger.warning(f"Failed to update Living AI state: {e}")
def _estimate_sentiment(self, text: str) -> float:
"""Estimate sentiment from text using simple heuristics.
Returns a value from -1 (negative) to 1 (positive).
This is a placeholder until we add AI-based sentiment analysis.
"""
text_lower = text.lower()
# Positive indicators
positive_words = [
"thanks",
"thank you",
"awesome",
"great",
"love",
"amazing",
"wonderful",
"excellent",
"perfect",
"happy",
"glad",
"appreciate",
"helpful",
"nice",
"good",
"cool",
"fantastic",
"brilliant",
]
# Negative indicators
negative_words = [
"hate",
"awful",
"terrible",
"bad",
"stupid",
"annoying",
"frustrated",
"angry",
"disappointed",
"wrong",
"broken",
"useless",
"horrible",
"worst",
"sucks",
"boring",
]
positive_count = sum(1 for word in positive_words if word in text_lower)
negative_count = sum(1 for word in negative_words if word in text_lower)
# Check for exclamation marks (usually positive energy)
exclamation_bonus = min(0.2, text.count("!") * 0.05)
# Calculate sentiment
if positive_count + negative_count == 0:
return 0.1 + exclamation_bonus # Slightly positive by default
sentiment = (positive_count - negative_count) / (positive_count + negative_count)
return max(-1.0, min(1.0, sentiment + exclamation_bonus))
async def _generate_response_in_memory( async def _generate_response_in_memory(
self, message: discord.Message, user_message: str self, message: discord.Message, user_message: str
) -> str: ) -> str:

View File

@@ -87,6 +87,38 @@ class Settings(BaseSettings):
searxng_enabled: bool = Field(True, description="Enable web search capability") searxng_enabled: bool = Field(True, description="Enable web search capability")
searxng_max_results: int = Field(5, ge=1, le=20, description="Maximum search results to fetch") searxng_max_results: int = Field(5, ge=1, le=20, description="Maximum search results to fetch")
# Living AI Configuration
living_ai_enabled: bool = Field(True, description="Enable Living AI features")
mood_enabled: bool = Field(True, description="Enable mood system")
relationship_enabled: bool = Field(True, description="Enable relationship tracking")
fact_extraction_enabled: bool = Field(True, description="Enable autonomous fact extraction")
fact_extraction_rate: float = Field(
0.3, ge=0.0, le=1.0, description="Probability of extracting facts from messages"
)
proactive_enabled: bool = Field(True, description="Enable proactive messages")
cross_user_enabled: bool = Field(
False, description="Enable cross-user memory associations (privacy-sensitive)"
)
opinion_formation_enabled: bool = Field(True, description="Enable bot opinion formation")
style_learning_enabled: bool = Field(True, description="Enable communication style learning")
# Mood System Settings
mood_decay_rate: float = Field(
0.1, ge=0.0, le=1.0, description="How fast mood returns to neutral per hour"
)
# Command Toggles
commands_enabled: bool = Field(True, description="Master switch for all commands")
cmd_relationship_enabled: bool = Field(True, description="Enable !relationship command")
cmd_mood_enabled: bool = Field(True, description="Enable !mood command")
cmd_botstats_enabled: bool = Field(True, description="Enable !botstats command")
cmd_ourhistory_enabled: bool = Field(True, description="Enable !ourhistory command")
cmd_birthday_enabled: bool = Field(True, description="Enable !birthday command")
cmd_remember_enabled: bool = Field(True, description="Enable !remember command")
cmd_setname_enabled: bool = Field(True, description="Enable !setname command")
cmd_whatdoyouknow_enabled: bool = Field(True, description="Enable !whatdoyouknow command")
cmd_forgetme_enabled: bool = Field(True, description="Enable !forgetme command")
def get_api_key(self) -> str: def get_api_key(self) -> str:
"""Get the API key for the configured provider.""" """Get the API key for the configured provider."""
key_map = { key_map = {

View File

@@ -3,15 +3,31 @@
from .base import Base from .base import Base
from .conversation import Conversation, Message from .conversation import Conversation, Message
from .guild import Guild, GuildMember from .guild import Guild, GuildMember
from .living_ai import (
BotOpinion,
BotState,
FactAssociation,
MoodHistory,
ScheduledEvent,
UserCommunicationStyle,
UserRelationship,
)
from .user import User, UserFact, UserPreference from .user import User, UserFact, UserPreference
__all__ = [ __all__ = [
"Base", "Base",
"BotOpinion",
"BotState",
"Conversation", "Conversation",
"FactAssociation",
"Guild", "Guild",
"GuildMember", "GuildMember",
"Message", "Message",
"MoodHistory",
"ScheduledEvent",
"User", "User",
"UserCommunicationStyle",
"UserFact", "UserFact",
"UserPreference", "UserPreference",
"UserRelationship",
] ]

View File

@@ -0,0 +1,186 @@
"""Living AI database models - mood, relationships, opinions, and more."""
from datetime import datetime
from typing import TYPE_CHECKING
from sqlalchemy import BigInteger, Boolean, Float, ForeignKey, String, Text, UniqueConstraint
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column, relationship
from .base import Base
if TYPE_CHECKING:
from .user import User, UserFact
class BotState(Base):
"""Global bot state - mood, statistics, preferences per guild."""
__tablename__ = "bot_states"
id: Mapped[int] = mapped_column(primary_key=True)
guild_id: Mapped[int | None] = mapped_column(BigInteger, unique=True, nullable=True)
# Current mood state (valence-arousal model)
mood_valence: Mapped[float] = mapped_column(Float, default=0.0) # -1.0 (sad) to 1.0 (happy)
mood_arousal: Mapped[float] = mapped_column(Float, default=0.0) # -1.0 (calm) to 1.0 (excited)
mood_updated_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
# Bot statistics
total_messages_sent: Mapped[int] = mapped_column(default=0)
total_facts_learned: Mapped[int] = mapped_column(default=0)
total_users_known: Mapped[int] = mapped_column(default=0)
first_activated_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
# Bot preferences (evolved over time)
preferences: Mapped[dict] = mapped_column(JSONB, default=dict)
class BotOpinion(Base):
"""Bot's opinions and preferences on topics."""
__tablename__ = "bot_opinions"
id: Mapped[int] = mapped_column(primary_key=True)
guild_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
topic: Mapped[str] = mapped_column(String(255), index=True)
sentiment: Mapped[float] = mapped_column(Float, default=0.0) # -1.0 to 1.0
interest_level: Mapped[float] = mapped_column(Float, default=0.5) # 0.0 to 1.0
discussion_count: Mapped[int] = mapped_column(default=0)
reasoning: Mapped[str | None] = mapped_column(Text)
formed_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
last_reinforced_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
__table_args__ = (UniqueConstraint("guild_id", "topic"),)
class UserRelationship(Base):
"""Tracks relationship depth and dynamics with each user."""
__tablename__ = "user_relationships"
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
guild_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
# Relationship level (0-100 scale)
# 0-20: stranger, 21-40: acquaintance, 41-60: friend, 61-80: good friend, 81-100: close friend
relationship_score: Mapped[float] = mapped_column(Float, default=10.0)
# Interaction metrics
total_interactions: Mapped[int] = mapped_column(default=0)
positive_interactions: Mapped[int] = mapped_column(default=0)
negative_interactions: Mapped[int] = mapped_column(default=0)
# Engagement quality
avg_message_length: Mapped[float] = mapped_column(Float, default=0.0)
conversation_depth_avg: Mapped[float] = mapped_column(Float, default=0.0)
# Inside jokes / shared references
shared_references: Mapped[dict] = mapped_column(JSONB, default=dict)
first_interaction_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
last_interaction_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
# Relationships
user: Mapped["User"] = relationship(back_populates="relationships")
__table_args__ = (UniqueConstraint("user_id", "guild_id"),)
class UserCommunicationStyle(Base):
"""Learned communication preferences per user."""
__tablename__ = "user_communication_styles"
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), unique=True, index=True
)
# Response style preferences (learned from engagement patterns)
preferred_length: Mapped[str] = mapped_column(String(20), default="medium")
preferred_formality: Mapped[float] = mapped_column(Float, default=0.5) # 0=casual, 1=formal
emoji_affinity: Mapped[float] = mapped_column(Float, default=0.5) # 0=none, 1=lots
humor_affinity: Mapped[float] = mapped_column(Float, default=0.5) # 0=serious, 1=playful
detail_preference: Mapped[float] = mapped_column(Float, default=0.5) # 0=concise, 1=detailed
# Engagement signals used to learn preferences
engagement_signals: Mapped[dict] = mapped_column(JSONB, default=dict)
samples_collected: Mapped[int] = mapped_column(default=0)
confidence: Mapped[float] = mapped_column(Float, default=0.0) # 0-1
# Relationship
user: Mapped["User"] = relationship(back_populates="communication_style")
class ScheduledEvent(Base):
"""Events the bot should act on (birthdays, follow-ups, etc.)."""
__tablename__ = "scheduled_events"
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))
guild_id: Mapped[int | None] = mapped_column(BigInteger)
channel_id: Mapped[int | None] = mapped_column(BigInteger)
event_type: Mapped[str] = mapped_column(String(50), index=True)
trigger_at: Mapped[datetime] = mapped_column(index=True)
title: Mapped[str] = mapped_column(String(255))
context: Mapped[dict] = mapped_column(JSONB, default=dict)
is_recurring: Mapped[bool] = mapped_column(Boolean, default=False)
recurrence_rule: Mapped[str | None] = mapped_column(String(100))
status: Mapped[str] = mapped_column(String(20), default="pending", index=True)
triggered_at: Mapped[datetime | None] = mapped_column(default=None)
# Relationships
user: Mapped["User"] = relationship(back_populates="scheduled_events")
class FactAssociation(Base):
"""Links facts across users for cross-user memory."""
__tablename__ = "fact_associations"
id: Mapped[int] = mapped_column(primary_key=True)
fact_id_1: Mapped[int] = mapped_column(
ForeignKey("user_facts.id", ondelete="CASCADE"), index=True
)
fact_id_2: Mapped[int] = mapped_column(
ForeignKey("user_facts.id", ondelete="CASCADE"), index=True
)
association_type: Mapped[str] = mapped_column(String(50))
strength: Mapped[float] = mapped_column(Float, default=0.5)
discovered_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
# Relationships
fact_1: Mapped["UserFact"] = relationship(foreign_keys=[fact_id_1])
fact_2: Mapped["UserFact"] = relationship(foreign_keys=[fact_id_2])
__table_args__ = (UniqueConstraint("fact_id_1", "fact_id_2"),)
class MoodHistory(Base):
"""Track mood changes over time for reflection."""
__tablename__ = "mood_history"
id: Mapped[int] = mapped_column(primary_key=True)
guild_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
valence: Mapped[float] = mapped_column(Float)
arousal: Mapped[float] = mapped_column(Float)
trigger_type: Mapped[str] = mapped_column(String(50)) # conversation, time_decay, event
trigger_user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"))
trigger_description: Mapped[str | None] = mapped_column(Text)
recorded_at: Mapped[datetime] = mapped_column(default=datetime.utcnow, index=True)

View File

@@ -11,6 +11,7 @@ from .base import Base
if TYPE_CHECKING: if TYPE_CHECKING:
from .conversation import Conversation, Message from .conversation import Conversation, Message
from .guild import GuildMember from .guild import GuildMember
from .living_ai import ScheduledEvent, UserCommunicationStyle, UserRelationship
class User(Base): class User(Base):
@@ -42,6 +43,17 @@ class User(Base):
back_populates="user", cascade="all, delete-orphan" back_populates="user", cascade="all, delete-orphan"
) )
# Living AI relationships
relationships: Mapped[list["UserRelationship"]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)
communication_style: Mapped["UserCommunicationStyle | None"] = relationship(
back_populates="user", cascade="all, delete-orphan", uselist=False
)
scheduled_events: Mapped[list["ScheduledEvent"]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)
@property @property
def display_name(self) -> str: def display_name(self) -> str:
"""Get the name to use when addressing this user.""" """Get the name to use when addressing this user."""

View File

@@ -1,23 +1,49 @@
"""Services for external integrations.""" """Services for external integrations."""
from .ai_service import AIService from .ai_service import AIService
from .association_service import AssociationService
from .communication_style_service import (
CommunicationStyleService,
detect_emoji_usage,
detect_formal_language,
)
from .conversation import ConversationManager from .conversation import ConversationManager
from .database import DatabaseService, db, get_db from .database import DatabaseService, db, get_db
from .fact_extraction_service import FactExtractionService
from .mood_service import MoodLabel, MoodService, MoodState
from .opinion_service import OpinionService, extract_topics_from_message
from .persistent_conversation import PersistentConversationManager from .persistent_conversation import PersistentConversationManager
from .proactive_service import ProactiveService
from .providers import AIResponse, ImageAttachment, Message from .providers import AIResponse, ImageAttachment, Message
from .relationship_service import RelationshipLevel, RelationshipService
from .searxng import SearXNGService from .searxng import SearXNGService
from .self_awareness_service import SelfAwarenessService
from .user_service import UserService from .user_service import UserService
__all__ = [ __all__ = [
"AIService", "AIService",
"AIResponse", "AIResponse",
"AssociationService",
"CommunicationStyleService",
"ConversationManager", "ConversationManager",
"DatabaseService", "DatabaseService",
"FactExtractionService",
"ImageAttachment", "ImageAttachment",
"Message", "Message",
"MoodLabel",
"MoodService",
"MoodState",
"OpinionService",
"PersistentConversationManager", "PersistentConversationManager",
"ProactiveService",
"RelationshipLevel",
"RelationshipService",
"SearXNGService", "SearXNGService",
"SelfAwarenessService",
"UserService", "UserService",
"db", "db",
"detect_emoji_usage",
"detect_formal_language",
"extract_topics_from_message",
"get_db", "get_db",
] ]

View File

@@ -1,7 +1,9 @@
"""AI Service - Factory and facade for AI providers.""" """AI Service - Factory and facade for AI providers."""
from __future__ import annotations
import logging import logging
from typing import Literal from typing import TYPE_CHECKING, Literal
from daemon_boyfriend.config import Settings, settings from daemon_boyfriend.config import Settings, settings
@@ -15,6 +17,12 @@ from .providers import (
OpenRouterProvider, OpenRouterProvider,
) )
if TYPE_CHECKING:
from daemon_boyfriend.models import BotOpinion, UserCommunicationStyle, UserRelationship
from .mood_service import MoodState
from .relationship_service import RelationshipLevel
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
ProviderType = Literal["openai", "openrouter", "anthropic", "gemini"] ProviderType = Literal["openai", "openrouter", "anthropic", "gemini"]
@@ -106,3 +114,90 @@ class AIService:
f"Discord bot. Keep your responses concise and engaging. " f"Discord bot. Keep your responses concise and engaging. "
f"You can use Discord markdown formatting in your responses." f"You can use Discord markdown formatting in your responses."
) )
def get_enhanced_system_prompt(
self,
mood: MoodState | None = None,
relationship: tuple[RelationshipLevel, UserRelationship] | None = None,
communication_style: UserCommunicationStyle | None = None,
bot_opinions: list[BotOpinion] | None = None,
) -> str:
"""Build system prompt with all personality modifiers.
Args:
mood: Current mood state
relationship: Tuple of (level, relationship_record)
communication_style: User's learned communication preferences
bot_opinions: Bot's opinions relevant to the conversation
Returns:
Enhanced system prompt with personality context
"""
from .mood_service import MoodService
from .relationship_service import RelationshipService
base_prompt = self.get_system_prompt()
modifiers = []
# Add mood modifier
if mood and self._config.mood_enabled:
mood_mod = MoodService(None).get_mood_prompt_modifier(mood)
if mood_mod:
modifiers.append(f"[Current Mood]\n{mood_mod}")
# Add relationship modifier
if relationship and self._config.relationship_enabled:
level, rel = relationship
rel_mod = RelationshipService(None).get_relationship_prompt_modifier(level, rel)
if rel_mod:
modifiers.append(f"[Relationship]\n{rel_mod}")
# Add communication style
if communication_style and self._config.style_learning_enabled:
style_mod = self._get_style_prompt_modifier(communication_style)
if style_mod:
modifiers.append(f"[Communication Style]\n{style_mod}")
# Add relevant opinions
if bot_opinions and self._config.opinion_formation_enabled:
opinion_strs = []
for op in bot_opinions[:3]: # Limit to 3 most relevant
if op.sentiment > 0.3:
opinion_strs.append(f"You enjoy discussing {op.topic}")
elif op.sentiment < -0.3:
opinion_strs.append(f"You're less enthusiastic about {op.topic}")
if opinion_strs:
modifiers.append(f"[Your Opinions]\n{'; '.join(opinion_strs)}")
if modifiers:
return base_prompt + "\n\n--- Personality Context ---\n" + "\n\n".join(modifiers)
return base_prompt
def _get_style_prompt_modifier(self, style: UserCommunicationStyle) -> str:
"""Generate prompt text for communication style."""
if style.confidence < 0.3:
return "" # Not enough data
parts = []
if style.preferred_length == "short":
parts.append("Keep responses brief and to the point.")
elif style.preferred_length == "long":
parts.append("Provide detailed, thorough responses.")
if style.preferred_formality > 0.7:
parts.append("Use formal language.")
elif style.preferred_formality < 0.3:
parts.append("Use casual, relaxed language.")
if style.emoji_affinity > 0.7:
parts.append("Feel free to use emojis.")
elif style.emoji_affinity < 0.3:
parts.append("Avoid using emojis.")
if style.humor_affinity > 0.7:
parts.append("Be playful and use humor.")
elif style.humor_affinity < 0.3:
parts.append("Keep a more serious tone.")
return " ".join(parts)

View File

@@ -0,0 +1,388 @@
"""Association Service - discovers and manages cross-user fact associations."""
import logging
from datetime import datetime
from sqlalchemy import and_, select
from sqlalchemy.ext.asyncio import AsyncSession
from daemon_boyfriend.models import FactAssociation, User, UserFact
logger = logging.getLogger(__name__)
class AssociationService:
"""Discovers and manages cross-user fact associations."""
def __init__(self, session: AsyncSession) -> None:
self._session = session
async def find_shared_interests(
self,
user: User,
guild_id: int | None = None,
limit: int = 5,
) -> list[tuple[User, str, float]]:
"""Find other users with shared interests.
Args:
user: The user to find matches for
guild_id: Limit to users in this guild (optional)
limit: Maximum matches to return
Returns:
List of (other_user, shared_interest, strength) tuples
"""
# Get user's facts
user_facts = await self._get_user_facts(user)
if not user_facts:
return []
# Extract topics/interests from facts
user_topics = self._extract_topics(user_facts)
if not user_topics:
return []
# Find other users with similar topics
matches = []
other_users = await self._get_other_users(user, guild_id)
for other_user in other_users:
other_facts = await self._get_user_facts(other_user)
other_topics = self._extract_topics(other_facts)
# Find overlapping interests
shared = user_topics & other_topics
for topic in shared:
# Calculate strength based on how central this topic is
strength = 0.8 # Base strength for direct match
matches.append((other_user, topic, strength))
# Sort by strength and limit
matches.sort(key=lambda x: x[2], reverse=True)
return matches[:limit]
async def create_association(
self,
fact_1: UserFact,
fact_2: UserFact,
association_type: str,
strength: float = 0.5,
) -> FactAssociation | None:
"""Create an association between two facts.
Args:
fact_1: First fact
fact_2: Second fact
association_type: Type of association
strength: Strength of the association (0-1)
Returns:
Created association or None if already exists
"""
# Ensure consistent ordering (smaller ID first)
if fact_1.id > fact_2.id:
fact_1, fact_2 = fact_2, fact_1
# Check if association already exists
existing = await self._get_existing_association(fact_1.id, fact_2.id)
if existing:
# Update strength if stronger
if strength > existing.strength:
existing.strength = strength
return existing
assoc = FactAssociation(
fact_id_1=fact_1.id,
fact_id_2=fact_2.id,
association_type=association_type,
strength=strength,
discovered_at=datetime.utcnow(),
)
self._session.add(assoc)
await self._session.flush()
logger.debug(
f"Created association: {association_type} between facts {fact_1.id} and {fact_2.id}"
)
return assoc
async def discover_associations(self, guild_id: int | None = None) -> int:
"""Discover new associations between facts across users.
This should be run periodically as a background task.
Returns:
Number of new associations discovered
"""
# Get all active facts
stmt = select(UserFact).where(UserFact.is_active == True)
result = await self._session.execute(stmt)
all_facts = list(result.scalars().all())
if len(all_facts) < 2:
return 0
discovered = 0
# Group facts by type for comparison
facts_by_type: dict[str, list[UserFact]] = {}
for fact in all_facts:
fact_type = fact.fact_type or "general"
if fact_type not in facts_by_type:
facts_by_type[fact_type] = []
facts_by_type[fact_type].append(fact)
# Find associations within same type
for fact_type, facts in facts_by_type.items():
for i, fact_1 in enumerate(facts):
for fact_2 in facts[i + 1 :]:
# Skip facts from same user
if fact_1.user_id == fact_2.user_id:
continue
# Check for similarity
similarity = self._calculate_similarity(fact_1, fact_2)
if similarity > 0.6:
assoc = await self.create_association(
fact_1=fact_1,
fact_2=fact_2,
association_type="shared_interest",
strength=similarity,
)
if assoc:
discovered += 1
logger.info(f"Discovered {discovered} new fact associations")
return discovered
async def get_associations_for_user(
self, user: User, limit: int = 10
) -> list[tuple[UserFact, UserFact, FactAssociation]]:
"""Get associations involving a user's facts.
Returns:
List of (user_fact, other_fact, association) tuples
"""
# Get user's fact IDs
user_facts = await self._get_user_facts(user)
if not user_facts:
return []
user_fact_ids = {f.id for f in user_facts}
# Find associations involving these facts
stmt = (
select(FactAssociation)
.where(
(FactAssociation.fact_id_1.in_(user_fact_ids))
| (FactAssociation.fact_id_2.in_(user_fact_ids))
)
.order_by(FactAssociation.strength.desc())
.limit(limit)
)
result = await self._session.execute(stmt)
associations = list(result.scalars().all())
# Build result tuples
results = []
for assoc in associations:
# Determine which fact belongs to user
if assoc.fact_id_1 in user_fact_ids:
user_fact = next(f for f in user_facts if f.id == assoc.fact_id_1)
other_fact = await self._get_fact_by_id(assoc.fact_id_2)
else:
user_fact = next(f for f in user_facts if f.id == assoc.fact_id_2)
other_fact = await self._get_fact_by_id(assoc.fact_id_1)
if other_fact:
results.append((user_fact, other_fact, assoc))
return results
def format_connection_suggestion(
self,
user_fact: UserFact,
other_fact: UserFact,
other_user: User,
) -> str:
"""Format a suggestion about shared interests.
Args:
user_fact: The current user's related fact
other_fact: The other user's fact
other_user: The other user
Returns:
A formatted suggestion string
"""
# Extract the shared interest
interest = self._extract_common_interest(user_fact, other_fact)
if interest:
return (
f"By the way, {other_user.display_name} is also into {interest}! "
f"You two might enjoy chatting about it."
)
else:
return f"You and {other_user.display_name} seem to have similar interests!"
async def _get_user_facts(self, user: User) -> list[UserFact]:
"""Get all active facts for a user."""
stmt = select(UserFact).where(
UserFact.user_id == user.id,
UserFact.is_active == True,
)
result = await self._session.execute(stmt)
return list(result.scalars().all())
async def _get_other_users(self, exclude_user: User, guild_id: int | None = None) -> list[User]:
"""Get other users (optionally filtered by guild)."""
stmt = select(User).where(
User.id != exclude_user.id,
User.is_active == True,
)
# Note: Guild filtering would require joining with guild_members
# For simplicity, we return all users for now
result = await self._session.execute(stmt)
return list(result.scalars().all())
async def _get_existing_association(
self, fact_id_1: int, fact_id_2: int
) -> FactAssociation | None:
"""Check if an association already exists."""
# Ensure consistent ordering
if fact_id_1 > fact_id_2:
fact_id_1, fact_id_2 = fact_id_2, fact_id_1
stmt = select(FactAssociation).where(
FactAssociation.fact_id_1 == fact_id_1,
FactAssociation.fact_id_2 == fact_id_2,
)
result = await self._session.execute(stmt)
return result.scalar_one_or_none()
async def _get_fact_by_id(self, fact_id: int) -> UserFact | None:
"""Get a fact by ID."""
stmt = select(UserFact).where(UserFact.id == fact_id)
result = await self._session.execute(stmt)
return result.scalar_one_or_none()
def _extract_topics(self, facts: list[UserFact]) -> set[str]:
"""Extract topic keywords from facts."""
topics = set()
# Keywords to extract as topics
topic_keywords = {
"programming",
"coding",
"python",
"javascript",
"gaming",
"games",
"music",
"guitar",
"piano",
"singing",
"movies",
"films",
"reading",
"books",
"sports",
"football",
"soccer",
"basketball",
"cooking",
"travel",
"photography",
"art",
"drawing",
"painting",
"hiking",
"fitness",
"gym",
"yoga",
"meditation",
"cats",
"dogs",
"pets",
"anime",
"manga",
"technology",
"science",
"philosophy",
}
for fact in facts:
content_lower = fact.fact_content.lower()
for keyword in topic_keywords:
if keyword in content_lower:
topics.add(keyword)
return topics
def _calculate_similarity(self, fact_1: UserFact, fact_2: UserFact) -> float:
"""Calculate similarity between two facts."""
content_1 = fact_1.fact_content.lower()
content_2 = fact_2.fact_content.lower()
# Simple word overlap similarity
words_1 = set(content_1.split())
words_2 = set(content_2.split())
# Remove common words
stop_words = {"a", "an", "the", "is", "are", "was", "were", "in", "on", "at", "to", "for"}
words_1 -= stop_words
words_2 -= stop_words
if not words_1 or not words_2:
return 0.0
intersection = words_1 & words_2
union = words_1 | words_2
return len(intersection) / len(union) if union else 0.0
def _extract_common_interest(self, fact_1: UserFact, fact_2: UserFact) -> str | None:
"""Extract the common interest between two facts."""
content_1 = fact_1.fact_content.lower()
content_2 = fact_2.fact_content.lower()
# Find common meaningful words
words_1 = set(content_1.split())
words_2 = set(content_2.split())
stop_words = {
"a",
"an",
"the",
"is",
"are",
"was",
"were",
"in",
"on",
"at",
"to",
"for",
"and",
"or",
"but",
"with",
"has",
"have",
"likes",
"loves",
"enjoys",
"interested",
"into",
}
common = (words_1 & words_2) - stop_words
if common:
# Return the longest common word as the interest
return max(common, key=len)
return None

View File

@@ -0,0 +1,245 @@
"""Communication Style Service - learns and applies per-user communication preferences."""
import logging
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from daemon_boyfriend.models import User, UserCommunicationStyle
logger = logging.getLogger(__name__)
class CommunicationStyleService:
"""Learns and applies per-user communication preferences."""
# Minimum samples before we trust the learned style
MIN_SAMPLES_FOR_CONFIDENCE = 10
def __init__(self, session: AsyncSession) -> None:
self._session = session
async def get_or_create_style(self, user: User) -> UserCommunicationStyle:
"""Get or create communication style profile for a user."""
stmt = select(UserCommunicationStyle).where(UserCommunicationStyle.user_id == user.id)
result = await self._session.execute(stmt)
style = result.scalar_one_or_none()
if not style:
style = UserCommunicationStyle(user_id=user.id)
self._session.add(style)
await self._session.flush()
return style
async def record_engagement(
self,
user: User,
user_message_length: int,
bot_response_length: int,
conversation_continued: bool = True,
user_used_emoji: bool = False,
user_used_formal_language: bool = False,
) -> None:
"""Record engagement signals to learn preferences.
Args:
user: The user
user_message_length: Length of user's message
bot_response_length: Length of bot's response
conversation_continued: Whether user continued the conversation
user_used_emoji: Whether user used emoji in their message
user_used_formal_language: Whether user used formal language
"""
style = await self.get_or_create_style(user)
signals = style.engagement_signals or {}
# Track response length preferences
if "response_lengths" not in signals:
signals["response_lengths"] = []
signals["response_lengths"].append(
{
"bot_length": bot_response_length,
"engaged": conversation_continued,
}
)
# Keep last 50 samples
signals["response_lengths"] = signals["response_lengths"][-50:]
# Track user's own message lengths
if "user_lengths" not in signals:
signals["user_lengths"] = []
signals["user_lengths"].append(user_message_length)
signals["user_lengths"] = signals["user_lengths"][-50:]
# Track emoji usage
if "emoji_usage" not in signals:
signals["emoji_usage"] = []
signals["emoji_usage"].append(1 if user_used_emoji else 0)
signals["emoji_usage"] = signals["emoji_usage"][-50:]
# Track formality
if "formality" not in signals:
signals["formality"] = []
signals["formality"].append(1 if user_used_formal_language else 0)
signals["formality"] = signals["formality"][-50:]
style.engagement_signals = signals
style.samples_collected += 1
# Recalculate preferences if enough samples
if style.samples_collected >= self.MIN_SAMPLES_FOR_CONFIDENCE:
await self._recalculate_preferences(style)
async def _recalculate_preferences(self, style: UserCommunicationStyle) -> None:
"""Recalculate preferences from engagement signals."""
signals = style.engagement_signals or {}
# Length preference from user's own message lengths
user_lengths = signals.get("user_lengths", [])
if user_lengths:
avg_length = sum(user_lengths) / len(user_lengths)
if avg_length < 50:
style.preferred_length = "short"
elif avg_length < 200:
style.preferred_length = "medium"
else:
style.preferred_length = "long"
# Emoji affinity from user's emoji usage
emoji_usage = signals.get("emoji_usage", [])
if emoji_usage:
style.emoji_affinity = sum(emoji_usage) / len(emoji_usage)
# Formality from user's language style
formality = signals.get("formality", [])
if formality:
style.preferred_formality = sum(formality) / len(formality)
# Update confidence based on sample count
style.confidence = min(1.0, style.samples_collected / 50)
logger.debug(
f"Recalculated style for user {style.user_id}: "
f"length={style.preferred_length}, emoji={style.emoji_affinity:.2f}, "
f"formality={style.preferred_formality:.2f}, confidence={style.confidence:.2f}"
)
def get_style_prompt_modifier(self, style: UserCommunicationStyle) -> str:
"""Generate prompt text for communication style."""
if style.confidence < 0.3:
return "" # Not enough data
parts = []
if style.preferred_length == "short":
parts.append("Keep responses brief and to the point.")
elif style.preferred_length == "long":
parts.append("Provide detailed, thorough responses.")
if style.preferred_formality > 0.7:
parts.append("Use formal language.")
elif style.preferred_formality < 0.3:
parts.append("Use casual, relaxed language.")
if style.emoji_affinity > 0.7:
parts.append("Feel free to use emojis.")
elif style.emoji_affinity < 0.3:
parts.append("Avoid using emojis.")
if style.humor_affinity > 0.7:
parts.append("Be playful and use humor.")
elif style.humor_affinity < 0.3:
parts.append("Keep a more serious tone.")
if style.detail_preference > 0.7:
parts.append("Include extra details and examples.")
elif style.detail_preference < 0.3:
parts.append("Be concise without extra details.")
return " ".join(parts)
async def get_style_info(self, user: User) -> dict:
"""Get style information for display."""
style = await self.get_or_create_style(user)
return {
"preferred_length": style.preferred_length,
"preferred_formality": style.preferred_formality,
"emoji_affinity": style.emoji_affinity,
"humor_affinity": style.humor_affinity,
"detail_preference": style.detail_preference,
"samples_collected": style.samples_collected,
"confidence": style.confidence,
}
def detect_emoji_usage(text: str) -> bool:
"""Detect if text contains emoji."""
import re
# Simple emoji detection - covers common emoji ranges
emoji_pattern = re.compile(
"["
"\U0001f600-\U0001f64f" # emoticons
"\U0001f300-\U0001f5ff" # symbols & pictographs
"\U0001f680-\U0001f6ff" # transport & map symbols
"\U0001f1e0-\U0001f1ff" # flags
"\U00002702-\U000027b0" # dingbats
"\U000024c2-\U0001f251"
"]+",
flags=re.UNICODE,
)
return bool(emoji_pattern.search(text))
def detect_formal_language(text: str) -> bool:
"""Detect if text uses formal language."""
text_lower = text.lower()
# Formal indicators
formal_words = [
"please",
"thank you",
"would you",
"could you",
"kindly",
"regards",
"sincerely",
"appreciate",
"assist",
"inquire",
"regarding",
"concerning",
"furthermore",
"however",
"therefore",
]
# Informal indicators
informal_words = [
"gonna",
"wanna",
"gotta",
"ya",
"u ",
"ur ",
"lol",
"lmao",
"omg",
"tbh",
"ngl",
"idk",
"btw",
"bruh",
"dude",
"yo ",
]
formal_count = sum(1 for word in formal_words if word in text_lower)
informal_count = sum(1 for word in informal_words if word in text_lower)
# Return True if more formal than informal
return formal_count > informal_count

View File

@@ -0,0 +1,356 @@
"""Fact Extraction Service - autonomous extraction of facts from conversations."""
import json
import logging
import random
from datetime import datetime
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from daemon_boyfriend.config import settings
from daemon_boyfriend.models import User, UserFact
from .providers import Message
logger = logging.getLogger(__name__)
class FactExtractionService:
"""Autonomous extraction of facts from conversations."""
# Minimum message length to consider for extraction
MIN_MESSAGE_LENGTH = 20
# Maximum facts to extract per message
MAX_FACTS_PER_MESSAGE = 3
def __init__(self, session: AsyncSession, ai_service=None) -> None:
self._session = session
self._ai_service = ai_service
async def maybe_extract_facts(
self,
user: User,
message_content: str,
discord_message_id: int | None = None,
) -> list[UserFact]:
"""Maybe extract facts from a message based on rate limiting.
Args:
user: The user who sent the message
message_content: The message content
discord_message_id: Optional Discord message ID for reference
Returns:
List of newly extracted facts (may be empty)
"""
if not settings.fact_extraction_enabled:
return []
# Rate limit: only extract from a percentage of messages
if random.random() > settings.fact_extraction_rate:
return []
return await self.extract_facts(user, message_content, discord_message_id)
async def extract_facts(
self,
user: User,
message_content: str,
discord_message_id: int | None = None,
) -> list[UserFact]:
"""Extract facts from a message.
Args:
user: The user who sent the message
message_content: The message content
discord_message_id: Optional Discord message ID for reference
Returns:
List of newly extracted facts
"""
# Skip messages that are too short or likely not informative
if not self._is_extractable(message_content):
return []
if not self._ai_service:
logger.warning("No AI service available for fact extraction")
return []
try:
# Get existing facts to avoid duplicates
existing_facts = await self._get_user_facts(user)
existing_summary = self._summarize_existing_facts(existing_facts)
# Build extraction prompt
extraction_prompt = self._build_extraction_prompt(existing_summary)
# Use AI to extract facts
response = await self._ai_service.chat(
messages=[Message(role="user", content=message_content)],
system_prompt=extraction_prompt,
)
# Parse extracted facts
facts_data = self._parse_extraction_response(response.content)
if not facts_data:
return []
# Deduplicate and save new facts
new_facts = await self._save_new_facts(
user=user,
facts_data=facts_data,
existing_facts=existing_facts,
discord_message_id=discord_message_id,
extraction_context=message_content[:200],
)
if new_facts:
logger.info(f"Extracted {len(new_facts)} facts for user {user.discord_id}")
return new_facts
except Exception as e:
logger.warning(f"Fact extraction failed: {e}")
return []
def _is_extractable(self, content: str) -> bool:
"""Check if a message is worth extracting facts from."""
# Too short
if len(content) < self.MIN_MESSAGE_LENGTH:
return False
# Just emoji or symbols
alpha_ratio = sum(c.isalpha() for c in content) / max(len(content), 1)
if alpha_ratio < 0.5:
return False
# Looks like a command
if content.startswith(("!", "/", "?", ".")):
return False
# Just a greeting or very short phrase
short_phrases = [
"hi",
"hello",
"hey",
"yo",
"sup",
"bye",
"goodbye",
"thanks",
"thank you",
"ok",
"okay",
"yes",
"no",
"yeah",
"nah",
"lol",
"lmao",
"haha",
"hehe",
"nice",
"cool",
"wow",
]
content_lower = content.lower().strip()
if content_lower in short_phrases:
return False
return True
def _build_extraction_prompt(self, existing_summary: str) -> str:
"""Build the extraction prompt for the AI."""
return f"""You are a fact extraction assistant. Extract factual information about the user from their message.
ALREADY KNOWN FACTS:
{existing_summary if existing_summary else "(None yet)"}
RULES:
1. Only extract CONCRETE facts, not opinions or transient states
2. Skip if the fact is already known (listed above)
3. Skip greetings, questions, or meta-conversation
4. Skip vague statements like "I like stuff" - be specific
5. Focus on: hobbies, work, family, preferences, locations, events, relationships
6. Keep fact content concise (under 100 characters)
7. Maximum {self.MAX_FACTS_PER_MESSAGE} facts per message
OUTPUT FORMAT:
Return a JSON array of facts, or empty array [] if no extractable facts.
Each fact should have:
- "type": one of "hobby", "work", "family", "preference", "location", "event", "relationship", "general"
- "content": the fact itself (concise, third person, e.g., "loves hiking")
- "confidence": 0.6 (implied), 0.8 (stated), 1.0 (explicit)
- "importance": 0.3 (trivial), 0.5 (normal), 0.8 (significant), 1.0 (very important)
- "temporal": "past", "present", "future", or "timeless"
EXAMPLE INPUT: "I just got promoted to senior engineer at Google last week!"
EXAMPLE OUTPUT: [{{"type": "work", "content": "works as senior engineer at Google", "confidence": 1.0, "importance": 0.8, "temporal": "present"}}, {{"type": "event", "content": "recently got promoted", "confidence": 1.0, "importance": 0.7, "temporal": "past"}}]
EXAMPLE INPUT: "hey what's up"
EXAMPLE OUTPUT: []
Return ONLY the JSON array, no other text."""
def _parse_extraction_response(self, response: str) -> list[dict]:
"""Parse the AI response into fact dictionaries."""
try:
# Try to find JSON array in the response
response = response.strip()
# Handle markdown code blocks
if "```json" in response:
start = response.find("```json") + 7
end = response.find("```", start)
response = response[start:end].strip()
elif "```" in response:
start = response.find("```") + 3
end = response.find("```", start)
response = response[start:end].strip()
# Parse JSON
facts = json.loads(response)
if not isinstance(facts, list):
return []
# Validate each fact
valid_facts = []
for fact in facts[: self.MAX_FACTS_PER_MESSAGE]:
if self._validate_fact(fact):
valid_facts.append(fact)
return valid_facts
except json.JSONDecodeError:
logger.debug(f"Failed to parse fact extraction response: {response[:100]}")
return []
def _validate_fact(self, fact: dict) -> bool:
"""Validate a fact dictionary."""
required_fields = ["type", "content"]
valid_types = [
"hobby",
"work",
"family",
"preference",
"location",
"event",
"relationship",
"general",
]
# Check required fields
if not all(field in fact for field in required_fields):
return False
# Check type is valid
if fact.get("type") not in valid_types:
return False
# Check content is not empty
if not fact.get("content") or len(fact["content"]) < 3:
return False
# Check content is not too long
if len(fact["content"]) > 200:
return False
return True
async def _get_user_facts(self, user: User) -> list[UserFact]:
"""Get existing facts for a user."""
stmt = (
select(UserFact)
.where(UserFact.user_id == user.id, UserFact.is_active == True)
.order_by(UserFact.learned_at.desc())
.limit(50)
)
result = await self._session.execute(stmt)
return list(result.scalars().all())
def _summarize_existing_facts(self, facts: list[UserFact]) -> str:
"""Summarize existing facts for the extraction prompt."""
if not facts:
return ""
summary_lines = []
for fact in facts[:20]: # Limit to most recent 20
summary_lines.append(f"- [{fact.fact_type}] {fact.fact_content}")
return "\n".join(summary_lines)
async def _save_new_facts(
self,
user: User,
facts_data: list[dict],
existing_facts: list[UserFact],
discord_message_id: int | None,
extraction_context: str,
) -> list[UserFact]:
"""Save new facts, avoiding duplicates."""
# Build set of existing fact content for deduplication
existing_content = {f.fact_content.lower() for f in existing_facts}
new_facts = []
for fact_data in facts_data:
content = fact_data["content"]
# Skip if too similar to existing
if self._is_duplicate(content, existing_content):
continue
# Create new fact
fact = UserFact(
user_id=user.id,
fact_type=fact_data["type"],
fact_content=content,
confidence=fact_data.get("confidence", 0.8),
source="auto_extraction",
is_active=True,
learned_at=datetime.utcnow(),
# New fields from Living AI
category=fact_data["type"],
importance=fact_data.get("importance", 0.5),
temporal_relevance=fact_data.get("temporal", "timeless"),
extracted_from_message_id=discord_message_id,
extraction_context=extraction_context,
)
self._session.add(fact)
new_facts.append(fact)
existing_content.add(content.lower())
if new_facts:
await self._session.flush()
return new_facts
def _is_duplicate(self, new_content: str, existing_content: set[str]) -> bool:
"""Check if a fact is a duplicate of existing facts."""
new_lower = new_content.lower()
# Exact match
if new_lower in existing_content:
return True
# Check for high similarity (simple substring check)
for existing in existing_content:
# If one contains the other (with some buffer)
if len(new_lower) > 10 and len(existing) > 10:
if new_lower in existing or existing in new_lower:
return True
# Simple word overlap check
new_words = set(new_lower.split())
existing_words = set(existing.split())
if len(new_words) > 2 and len(existing_words) > 2:
overlap = len(new_words & existing_words)
min_len = min(len(new_words), len(existing_words))
if overlap / min_len > 0.7: # 70% word overlap
return True
return False

View File

@@ -0,0 +1,252 @@
"""Mood Service - manages bot emotional states."""
import logging
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from daemon_boyfriend.config import settings
from daemon_boyfriend.models import BotState, MoodHistory
logger = logging.getLogger(__name__)
class MoodLabel(Enum):
"""Mood labels based on valence-arousal model."""
EXCITED = "excited" # high valence, high arousal
HAPPY = "happy" # high valence, low arousal
CALM = "calm" # neutral valence, low arousal
NEUTRAL = "neutral" # neutral everything
BORED = "bored" # low valence, low arousal
ANNOYED = "annoyed" # low valence, high arousal
CURIOUS = "curious" # neutral valence, high arousal
@dataclass
class MoodState:
"""Current mood state with computed properties."""
valence: float # -1 to 1
arousal: float # -1 to 1
label: MoodLabel
intensity: float # 0 to 1
class MoodService:
"""Manages bot mood state and its effects on responses."""
def __init__(self, session: AsyncSession) -> None:
self._session = session
async def get_or_create_bot_state(self, guild_id: int | None = None) -> BotState:
"""Get or create bot state for a guild."""
stmt = select(BotState).where(BotState.guild_id == guild_id)
result = await self._session.execute(stmt)
bot_state = result.scalar_one_or_none()
if not bot_state:
bot_state = BotState(guild_id=guild_id)
self._session.add(bot_state)
await self._session.flush()
return bot_state
async def get_current_mood(self, guild_id: int | None = None) -> MoodState:
"""Get current mood state, applying time decay."""
bot_state = await self.get_or_create_bot_state(guild_id)
# Apply time decay toward neutral
hours_since_update = (datetime.utcnow() - bot_state.mood_updated_at).total_seconds() / 3600
decay_factor = max(0, 1 - (settings.mood_decay_rate * hours_since_update))
valence = bot_state.mood_valence * decay_factor
arousal = bot_state.mood_arousal * decay_factor
return MoodState(
valence=valence,
arousal=arousal,
label=self._classify_mood(valence, arousal),
intensity=self._calculate_intensity(valence, arousal),
)
async def update_mood(
self,
guild_id: int | None,
sentiment_delta: float,
engagement_delta: float,
trigger_type: str,
trigger_user_id: int | None = None,
trigger_description: str | None = None,
) -> MoodState:
"""Update mood based on interaction.
Args:
guild_id: Guild ID or None for global mood
sentiment_delta: -1 to 1, how positive/negative the interaction was
engagement_delta: -1 to 1, how engaging the interaction was
trigger_type: What caused the mood change (conversation, event, etc.)
trigger_user_id: User who triggered the change (if any)
trigger_description: Description of what happened
Returns:
The new mood state
"""
current = await self.get_current_mood(guild_id)
# Mood changes are dampened (inertia) - only 30% of the delta is applied
new_valence = self._clamp(current.valence + sentiment_delta * 0.3)
new_arousal = self._clamp(current.arousal + engagement_delta * 0.3)
# Update database
bot_state = await self.get_or_create_bot_state(guild_id)
bot_state.mood_valence = new_valence
bot_state.mood_arousal = new_arousal
bot_state.mood_updated_at = datetime.utcnow()
# Record history
await self._record_mood_history(
guild_id, new_valence, new_arousal, trigger_type, trigger_user_id, trigger_description
)
logger.debug(
f"Mood updated: valence={new_valence:.2f}, arousal={new_arousal:.2f}, "
f"trigger={trigger_type}"
)
return MoodState(
valence=new_valence,
arousal=new_arousal,
label=self._classify_mood(new_valence, new_arousal),
intensity=self._calculate_intensity(new_valence, new_arousal),
)
async def increment_stats(
self,
guild_id: int | None,
messages_sent: int = 0,
facts_learned: int = 0,
users_known: int = 0,
) -> None:
"""Increment bot statistics."""
bot_state = await self.get_or_create_bot_state(guild_id)
bot_state.total_messages_sent += messages_sent
bot_state.total_facts_learned += facts_learned
bot_state.total_users_known += users_known
async def get_stats(self, guild_id: int | None = None) -> dict:
"""Get bot statistics."""
bot_state = await self.get_or_create_bot_state(guild_id)
age_delta = datetime.utcnow() - bot_state.first_activated_at
return {
"age_days": age_delta.days,
"age_readable": self._format_age(age_delta),
"total_messages_sent": bot_state.total_messages_sent,
"total_facts_learned": bot_state.total_facts_learned,
"total_users_known": bot_state.total_users_known,
"first_activated_at": bot_state.first_activated_at,
}
def get_mood_prompt_modifier(self, mood: MoodState) -> str:
"""Generate prompt text that reflects current mood."""
if mood.intensity < 0.2:
return "" # Mood too weak to affect behavior
modifiers = {
MoodLabel.EXCITED: (
"You're feeling enthusiastic and energetic right now! "
"Be expressive, use exclamation marks, show genuine excitement."
),
MoodLabel.HAPPY: (
"You're in a good mood - warm, friendly, and content. "
"Be positive and encouraging in your responses."
),
MoodLabel.CALM: (
"You're feeling peaceful and relaxed. "
"Respond thoughtfully and with a serene demeanor."
),
MoodLabel.NEUTRAL: "",
MoodLabel.BORED: (
"You're feeling a bit understimulated. "
"Keep responses shorter, maybe try to steer toward more interesting topics."
),
MoodLabel.ANNOYED: (
"You're slightly irritated. "
"Be a bit more terse, less patient with repetition or vague questions."
),
MoodLabel.CURIOUS: (
"You're feeling inquisitive and engaged! "
"Ask follow-up questions, show genuine interest in what the user is saying."
),
}
base = modifiers.get(mood.label, "")
if not base:
return ""
if mood.intensity > 0.7:
return f"[Strong mood] {base}"
return base
async def _record_mood_history(
self,
guild_id: int | None,
valence: float,
arousal: float,
trigger_type: str,
trigger_user_id: int | None,
trigger_description: str | None,
) -> None:
"""Record a mood change in history."""
history = MoodHistory(
guild_id=guild_id,
valence=valence,
arousal=arousal,
trigger_type=trigger_type,
trigger_user_id=trigger_user_id,
trigger_description=trigger_description,
)
self._session.add(history)
def _classify_mood(self, valence: float, arousal: float) -> MoodLabel:
"""Classify mood based on valence-arousal model."""
if valence > 0.3:
return MoodLabel.EXCITED if arousal > 0.3 else MoodLabel.HAPPY
elif valence < -0.3:
return MoodLabel.ANNOYED if arousal > 0.3 else MoodLabel.BORED
else:
if arousal > 0.3:
return MoodLabel.CURIOUS
elif arousal < -0.3:
return MoodLabel.CALM
return MoodLabel.NEUTRAL
def _calculate_intensity(self, valence: float, arousal: float) -> float:
"""Calculate mood intensity from valence and arousal."""
return min(1.0, (abs(valence) + abs(arousal)) / 2)
def _clamp(self, value: float, min_val: float = -1.0, max_val: float = 1.0) -> float:
"""Clamp value between min and max."""
return max(min_val, min(max_val, value))
def _format_age(self, delta) -> str:
"""Format a timedelta into a readable string."""
days = delta.days
if days == 0:
hours = delta.seconds // 3600
if hours == 0:
minutes = delta.seconds // 60
return f"{minutes} minute{'s' if minutes != 1 else ''}"
return f"{hours} hour{'s' if hours != 1 else ''}"
elif days < 30:
return f"{days} day{'s' if days != 1 else ''}"
elif days < 365:
months = days // 30
return f"about {months} month{'s' if months != 1 else ''}"
else:
years = days // 365
return f"about {years} year{'s' if years != 1 else ''}"

View File

@@ -0,0 +1,233 @@
"""Opinion Service - manages bot opinion formation on topics."""
import logging
from datetime import datetime
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from daemon_boyfriend.models import BotOpinion
logger = logging.getLogger(__name__)
class OpinionService:
"""Manages bot opinion formation and topic preferences."""
def __init__(self, session: AsyncSession) -> None:
self._session = session
async def get_opinion(self, topic: str, guild_id: int | None = None) -> BotOpinion | None:
"""Get the bot's opinion on a topic."""
stmt = select(BotOpinion).where(
BotOpinion.topic == topic.lower(),
BotOpinion.guild_id == guild_id,
)
result = await self._session.execute(stmt)
return result.scalar_one_or_none()
async def get_or_create_opinion(self, topic: str, guild_id: int | None = None) -> BotOpinion:
"""Get or create an opinion on a topic."""
opinion = await self.get_opinion(topic, guild_id)
if not opinion:
opinion = BotOpinion(
topic=topic.lower(),
guild_id=guild_id,
sentiment=0.0,
interest_level=0.5,
discussion_count=0,
)
self._session.add(opinion)
await self._session.flush()
return opinion
async def record_topic_discussion(
self,
topic: str,
guild_id: int | None,
sentiment: float,
engagement_level: float,
) -> BotOpinion:
"""Record a discussion about a topic, updating the bot's opinion.
Args:
topic: The topic discussed
guild_id: Guild ID or None for global
sentiment: How positive the discussion was (-1 to 1)
engagement_level: How engaging the discussion was (0 to 1)
Returns:
Updated opinion
"""
opinion = await self.get_or_create_opinion(topic, guild_id)
# Increment discussion count
opinion.discussion_count += 1
# Update sentiment (weighted average, newer discussions have more weight)
weight = 0.2 # 20% weight to new data
opinion.sentiment = (opinion.sentiment * (1 - weight)) + (sentiment * weight)
opinion.sentiment = max(-1.0, min(1.0, opinion.sentiment))
# Update interest level based on engagement
opinion.interest_level = (opinion.interest_level * (1 - weight)) + (
engagement_level * weight
)
opinion.interest_level = max(0.0, min(1.0, opinion.interest_level))
opinion.last_reinforced_at = datetime.utcnow()
logger.debug(
f"Updated opinion on '{topic}': sentiment={opinion.sentiment:.2f}, "
f"interest={opinion.interest_level:.2f}, discussions={opinion.discussion_count}"
)
return opinion
async def set_opinion_reasoning(self, topic: str, guild_id: int | None, reasoning: str) -> None:
"""Set the reasoning for an opinion (AI-generated explanation)."""
opinion = await self.get_or_create_opinion(topic, guild_id)
opinion.reasoning = reasoning
async def get_top_interests(
self, guild_id: int | None = None, limit: int = 5
) -> list[BotOpinion]:
"""Get the bot's top interests (highest interest level + positive sentiment)."""
stmt = (
select(BotOpinion)
.where(
BotOpinion.guild_id == guild_id,
BotOpinion.discussion_count >= 3, # Only topics discussed at least 3 times
)
.order_by((BotOpinion.interest_level + BotOpinion.sentiment).desc())
.limit(limit)
)
result = await self._session.execute(stmt)
return list(result.scalars().all())
async def get_relevant_opinions(
self, topics: list[str], guild_id: int | None = None
) -> list[BotOpinion]:
"""Get opinions relevant to a list of topics."""
if not topics:
return []
topics_lower = [t.lower() for t in topics]
stmt = select(BotOpinion).where(
BotOpinion.topic.in_(topics_lower),
BotOpinion.guild_id == guild_id,
)
result = await self._session.execute(stmt)
return list(result.scalars().all())
def get_opinion_prompt_modifier(self, opinions: list[BotOpinion]) -> str:
"""Generate prompt text based on relevant opinions."""
if not opinions:
return ""
parts = []
for op in opinions[:3]: # Limit to 3 opinions
if op.sentiment > 0.5:
parts.append(f"You really enjoy discussing {op.topic}")
elif op.sentiment > 0.2:
parts.append(f"You find {op.topic} interesting")
elif op.sentiment < -0.3:
parts.append(f"You're not particularly enthusiastic about {op.topic}")
if op.reasoning:
parts.append(f"({op.reasoning})")
return "; ".join(parts) if parts else ""
async def get_all_opinions(self, guild_id: int | None = None) -> list[BotOpinion]:
"""Get all opinions for a guild."""
stmt = (
select(BotOpinion)
.where(BotOpinion.guild_id == guild_id)
.order_by(BotOpinion.discussion_count.desc())
)
result = await self._session.execute(stmt)
return list(result.scalars().all())
def extract_topics_from_message(message: str) -> list[str]:
"""Extract potential topics from a message.
This is a simple keyword-based extraction. In production,
you might want to use NLP or an LLM for better extraction.
"""
# Common topic categories
topic_keywords = {
# Hobbies
"gaming": [
"game",
"gaming",
"video game",
"play",
"xbox",
"playstation",
"nintendo",
"steam",
],
"music": [
"music",
"song",
"band",
"album",
"concert",
"listen",
"spotify",
"guitar",
"piano",
],
"movies": ["movie", "film", "cinema", "watch", "netflix", "show", "series", "tv"],
"reading": ["book", "read", "novel", "author", "library", "kindle"],
"sports": [
"sports",
"football",
"soccer",
"basketball",
"tennis",
"golf",
"gym",
"workout",
],
"cooking": ["cook", "recipe", "food", "restaurant", "meal", "kitchen", "baking"],
"travel": ["travel", "trip", "vacation", "flight", "hotel", "country", "visit"],
"art": ["art", "painting", "drawing", "museum", "gallery", "creative"],
# Tech
"programming": [
"code",
"programming",
"developer",
"software",
"python",
"javascript",
"api",
],
"technology": ["tech", "computer", "phone", "app", "website", "internet"],
"ai": ["ai", "artificial intelligence", "machine learning", "chatgpt", "gpt"],
# Life
"work": ["work", "job", "office", "career", "boss", "colleague", "meeting"],
"family": ["family", "parents", "mom", "dad", "brother", "sister", "kids"],
"pets": ["pet", "dog", "cat", "puppy", "kitten", "animal"],
"health": ["health", "doctor", "exercise", "diet", "sleep", "medical"],
# Interests
"philosophy": ["philosophy", "meaning", "life", "existence", "think", "believe"],
"science": ["science", "research", "study", "experiment", "discovery"],
"nature": ["nature", "outdoor", "hiking", "camping", "mountain", "beach", "forest"],
}
message_lower = message.lower()
found_topics = []
for topic, keywords in topic_keywords.items():
for keyword in keywords:
if keyword in message_lower:
if topic not in found_topics:
found_topics.append(topic)
break
return found_topics

View File

@@ -0,0 +1,455 @@
"""Proactive Service - manages scheduled events and proactive behavior."""
import json
import logging
import re
from datetime import datetime, timedelta
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from daemon_boyfriend.models import ScheduledEvent, User
from .providers import Message
logger = logging.getLogger(__name__)
class ProactiveService:
"""Manages scheduled events and proactive behavior."""
def __init__(self, session: AsyncSession, ai_service=None) -> None:
self._session = session
self._ai_service = ai_service
async def detect_and_schedule_followup(
self,
user: User,
message_content: str,
guild_id: int | None,
channel_id: int,
) -> ScheduledEvent | None:
"""Detect if a message mentions a future event worth following up on.
Args:
user: The user who sent the message
message_content: The message content
guild_id: Guild ID
channel_id: Channel ID for the follow-up
Returns:
Scheduled event if one was created, None otherwise
"""
if not self._ai_service:
# Use simple pattern matching as fallback
return await self._detect_followup_simple(user, message_content, guild_id, channel_id)
try:
detection_prompt = """Analyze if this message mentions a future event worth following up on.
Events like: job interviews, exams, trips, appointments, projects due, important meetings, etc.
Return JSON: {"has_event": true/false, "event_type": "...", "days_until": <number or null>, "description": "..."}
Rules:
- Only return has_event=true for significant events the speaker would appreciate being asked about later
- days_until should be your best estimate of days until the event (1 for tomorrow, 7 for next week, etc.)
- Skip casual mentions like "I might do something" or past events
- description should be a brief summary of the event
Examples:
"I have a job interview tomorrow" -> {"has_event": true, "event_type": "job interview", "days_until": 1, "description": "job interview"}
"I went to the store" -> {"has_event": false}
"My exam is next week" -> {"has_event": true, "event_type": "exam", "days_until": 7, "description": "upcoming exam"}
"""
response = await self._ai_service.chat(
messages=[Message(role="user", content=message_content)],
system_prompt=detection_prompt,
)
result = self._parse_json_response(response.content)
if result and result.get("has_event"):
days_until = result.get("days_until", 1) or 1
# Schedule follow-up for 1 day after the event
trigger_at = datetime.utcnow() + timedelta(days=days_until + 1)
event = ScheduledEvent(
user_id=user.id,
guild_id=guild_id,
channel_id=channel_id,
event_type="follow_up",
trigger_at=trigger_at,
title=f"Follow up: {result.get('event_type', 'event')}",
context={
"original_topic": result.get("description", "their event"),
"detected_from": message_content[:200],
},
)
self._session.add(event)
await self._session.flush()
logger.info(
f"Scheduled follow-up for user {user.id}: "
f"{result.get('event_type')} in {days_until + 1} days"
)
return event
except Exception as e:
logger.warning(f"Follow-up detection failed: {e}")
return None
async def _detect_followup_simple(
self,
user: User,
message_content: str,
guild_id: int | None,
channel_id: int,
) -> ScheduledEvent | None:
"""Simple pattern-based follow-up detection."""
message_lower = message_content.lower()
# Event patterns with their typical timeframes
event_patterns = {
r"(interview|job interview)": ("job interview", 1),
r"(exam|test|quiz)": ("exam", 1),
r"(presentation|presenting)": ("presentation", 1),
r"(surgery|operation|medical)": ("medical procedure", 2),
r"(moving|move to|new apartment|new house)": ("moving", 7),
r"(wedding|getting married)": ("wedding", 1),
r"(vacation|holiday|trip to)": ("trip", 7),
r"(deadline|due date|project due)": ("deadline", 1),
r"(starting.*job|new job|first day)": ("new job", 1),
r"(graduation|graduating)": ("graduation", 1),
}
# Time indicators
time_patterns = {
r"tomorrow": 1,
r"next week": 7,
r"this week": 3,
r"in (\d+) days?": None, # Extract number
r"next month": 30,
r"this weekend": 3,
}
# Check for event + time combination
detected_event = None
event_name = None
days_until = 1 # Default
for pattern, (name, default_days) in event_patterns.items():
if re.search(pattern, message_lower):
detected_event = pattern
event_name = name
days_until = default_days
break
if not detected_event:
return None
# Refine timing based on time indicators
for pattern, days in time_patterns.items():
match = re.search(pattern, message_lower)
if match:
if days is None and match.groups():
days_until = int(match.group(1))
elif days:
days_until = days
break
# Create the event
trigger_at = datetime.utcnow() + timedelta(days=days_until + 1)
event = ScheduledEvent(
user_id=user.id,
guild_id=guild_id,
channel_id=channel_id,
event_type="follow_up",
trigger_at=trigger_at,
title=f"Follow up: {event_name}",
context={
"original_topic": event_name,
"detected_from": message_content[:200],
},
)
self._session.add(event)
await self._session.flush()
logger.info(
f"Scheduled follow-up (simple) for user {user.id}: {event_name} in {days_until + 1} days"
)
return event
async def detect_and_schedule_birthday(
self,
user: User,
message_content: str,
guild_id: int | None,
channel_id: int,
) -> ScheduledEvent | None:
"""Detect birthday mentions and schedule wishes."""
birthday = self._extract_birthday(message_content)
if not birthday:
return None
# Check if we already have a birthday scheduled for this user
existing = await self._get_existing_birthday(user.id, guild_id)
if existing:
# Update the existing birthday
existing.trigger_at = self._next_birthday(birthday)
existing.context = {"birthday_date": birthday.isoformat()}
return existing
# Schedule for next occurrence
trigger_at = self._next_birthday(birthday)
event = ScheduledEvent(
user_id=user.id,
guild_id=guild_id,
channel_id=channel_id,
event_type="birthday",
trigger_at=trigger_at,
title="Birthday wish",
context={"birthday_date": birthday.isoformat()},
is_recurring=True,
recurrence_rule="yearly",
)
self._session.add(event)
await self._session.flush()
logger.info(f"Scheduled birthday for user {user.id}: {birthday}")
return event
def _extract_birthday(self, message: str) -> datetime | None:
"""Extract a birthday date from a message."""
message_lower = message.lower()
# Check if it's about their birthday
birthday_indicators = [
r"my birthday is",
r"my bday is",
r"i was born on",
r"born on",
r"my birthday'?s?",
]
has_birthday_mention = any(
re.search(pattern, message_lower) for pattern in birthday_indicators
)
if not has_birthday_mention:
return None
# Try to extract date patterns
# Format: Month Day (e.g., "March 15", "march 15th")
month_names = {
"january": 1,
"february": 2,
"march": 3,
"april": 4,
"may": 5,
"june": 6,
"july": 7,
"august": 8,
"september": 9,
"october": 10,
"november": 11,
"december": 12,
"jan": 1,
"feb": 2,
"mar": 3,
"apr": 4,
"jun": 6,
"jul": 7,
"aug": 8,
"sep": 9,
"oct": 10,
"nov": 11,
"dec": 12,
}
for month_name, month_num in month_names.items():
pattern = rf"{month_name}\s+(\d{{1,2}})"
match = re.search(pattern, message_lower)
if match:
day = int(match.group(1))
if 1 <= day <= 31:
try:
return datetime(2000, month_num, day) # Year doesn't matter
except ValueError:
pass
# Format: DD/MM or MM/DD
date_pattern = r"(\d{1,2})[/\-](\d{1,2})"
match = re.search(date_pattern, message)
if match:
n1, n2 = int(match.group(1)), int(match.group(2))
# Assume MM/DD if first number <= 12, else DD/MM
if n1 <= 12 and n2 <= 31:
try:
return datetime(2000, n1, n2)
except ValueError:
pass
elif n2 <= 12 and n1 <= 31:
try:
return datetime(2000, n2, n1)
except ValueError:
pass
return None
def _next_birthday(self, birthday: datetime) -> datetime:
"""Calculate the next occurrence of a birthday."""
today = datetime.utcnow().date()
this_year = birthday.replace(year=today.year)
if this_year.date() < today:
return birthday.replace(year=today.year + 1)
return this_year
async def _get_existing_birthday(
self, user_id: int, guild_id: int | None
) -> ScheduledEvent | None:
"""Check if a birthday is already scheduled."""
stmt = select(ScheduledEvent).where(
ScheduledEvent.user_id == user_id,
ScheduledEvent.guild_id == guild_id,
ScheduledEvent.event_type == "birthday",
ScheduledEvent.status == "pending",
)
result = await self._session.execute(stmt)
return result.scalar_one_or_none()
async def get_pending_events(self, before: datetime | None = None) -> list[ScheduledEvent]:
"""Get events that should be triggered."""
cutoff = before or datetime.utcnow()
stmt = (
select(ScheduledEvent)
.where(
ScheduledEvent.status == "pending",
ScheduledEvent.trigger_at <= cutoff,
)
.order_by(ScheduledEvent.trigger_at)
)
result = await self._session.execute(stmt)
return list(result.scalars().all())
async def generate_event_message(self, event: ScheduledEvent) -> str:
"""Generate the message for a triggered event."""
if event.event_type == "birthday":
return await self._generate_birthday_message(event)
elif event.event_type == "follow_up":
return await self._generate_followup_message(event)
else:
return await self._generate_generic_message(event)
async def _generate_birthday_message(self, event: ScheduledEvent) -> str:
"""Generate a birthday message."""
if self._ai_service:
try:
response = await self._ai_service.chat(
messages=[Message(role="user", content="Generate a birthday message")],
system_prompt=(
"Generate a warm, personalized birthday wish. "
"Be genuine but not over the top. Keep it to 1-2 sentences. "
"Don't use too many emojis."
),
)
return response.content
except Exception:
pass
# Fallback
return "Happy birthday! Hope you have an amazing day!"
async def _generate_followup_message(self, event: ScheduledEvent) -> str:
"""Generate a follow-up message."""
topic = event.context.get("original_topic", "that thing you mentioned")
if self._ai_service:
try:
response = await self._ai_service.chat(
messages=[Message(role="user", content=f"Follow up about: {topic}")],
system_prompt=(
f"Generate a natural follow-up question about '{topic}'. "
"Be casual and genuinely curious. Ask how it went. "
"Keep it to 1-2 sentences. No emojis."
),
)
return response.content
except Exception:
pass
# Fallback
return f"Hey! How did {topic} go?"
async def _generate_generic_message(self, event: ScheduledEvent) -> str:
"""Generate a generic event message."""
return f"Hey! Just wanted to check in - {event.title}"
async def mark_event_triggered(self, event: ScheduledEvent) -> None:
"""Mark an event as triggered and handle recurrence."""
event.status = "triggered"
event.triggered_at = datetime.utcnow()
# Handle recurring events
if event.is_recurring and event.recurrence_rule:
await self._schedule_next_occurrence(event)
async def _schedule_next_occurrence(self, event: ScheduledEvent) -> None:
"""Schedule the next occurrence of a recurring event."""
if event.recurrence_rule == "yearly":
next_trigger = event.trigger_at.replace(year=event.trigger_at.year + 1)
elif event.recurrence_rule == "monthly":
# Add one month
month = event.trigger_at.month + 1
year = event.trigger_at.year
if month > 12:
month = 1
year += 1
next_trigger = event.trigger_at.replace(year=year, month=month)
elif event.recurrence_rule == "weekly":
next_trigger = event.trigger_at + timedelta(weeks=1)
else:
return # Unknown rule
new_event = ScheduledEvent(
user_id=event.user_id,
guild_id=event.guild_id,
channel_id=event.channel_id,
event_type=event.event_type,
trigger_at=next_trigger,
title=event.title,
context=event.context,
is_recurring=True,
recurrence_rule=event.recurrence_rule,
)
self._session.add(new_event)
async def cancel_event(self, event_id: int) -> bool:
"""Cancel a scheduled event."""
stmt = select(ScheduledEvent).where(ScheduledEvent.id == event_id)
result = await self._session.execute(stmt)
event = result.scalar_one_or_none()
if event and event.status == "pending":
event.status = "cancelled"
return True
return False
def _parse_json_response(self, response: str) -> dict | None:
"""Parse JSON from AI response."""
try:
response = response.strip()
if "```json" in response:
start = response.find("```json") + 7
end = response.find("```", start)
response = response[start:end].strip()
elif "```" in response:
start = response.find("```") + 3
end = response.find("```", start)
response = response[start:end].strip()
return json.loads(response)
except json.JSONDecodeError:
return None

View File

@@ -0,0 +1,228 @@
"""Relationship Service - manages relationship tracking with users."""
import logging
from datetime import datetime
from enum import Enum
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from daemon_boyfriend.models import User, UserRelationship
logger = logging.getLogger(__name__)
class RelationshipLevel(Enum):
"""Relationship levels based on score."""
STRANGER = "stranger" # 0-20
ACQUAINTANCE = "acquaintance" # 21-40
FRIEND = "friend" # 41-60
GOOD_FRIEND = "good_friend" # 61-80
CLOSE_FRIEND = "close_friend" # 81-100
class RelationshipService:
"""Manages relationship tracking and dynamics."""
def __init__(self, session: AsyncSession) -> None:
self._session = session
async def get_or_create_relationship(
self, user: User, guild_id: int | None = None
) -> UserRelationship:
"""Get or create relationship record for a user."""
stmt = select(UserRelationship).where(
UserRelationship.user_id == user.id,
UserRelationship.guild_id == guild_id,
)
result = await self._session.execute(stmt)
rel = result.scalar_one_or_none()
if not rel:
rel = UserRelationship(user_id=user.id, guild_id=guild_id)
self._session.add(rel)
await self._session.flush()
return rel
async def record_interaction(
self,
user: User,
guild_id: int | None,
sentiment: float,
message_length: int,
conversation_turns: int = 1,
) -> RelationshipLevel:
"""Record an interaction and update relationship score.
Args:
user: The user who interacted
guild_id: Guild ID or None for global
sentiment: -1 to 1, how positive the interaction was
message_length: Length of user's message
conversation_turns: Number of back-and-forth exchanges
Returns:
The current relationship level
"""
rel = await self.get_or_create_relationship(user, guild_id)
rel.total_interactions += 1
rel.last_interaction_at = datetime.utcnow()
# Track sentiment
if sentiment > 0.2:
rel.positive_interactions += 1
elif sentiment < -0.2:
rel.negative_interactions += 1
# Update running averages
n = rel.total_interactions
rel.avg_message_length = ((rel.avg_message_length * (n - 1)) + message_length) / n
rel.conversation_depth_avg = (
(rel.conversation_depth_avg * (n - 1)) + conversation_turns
) / n
# Calculate score change
score_delta = self._calculate_score_delta(sentiment, message_length, conversation_turns)
rel.relationship_score = min(100, max(0, rel.relationship_score + score_delta))
logger.debug(
f"Relationship updated for user {user.id}: "
f"score={rel.relationship_score:.1f}, delta={score_delta:.2f}"
)
return self.get_level(rel.relationship_score)
def get_level(self, score: float) -> RelationshipLevel:
"""Get relationship level from score."""
if score <= 20:
return RelationshipLevel.STRANGER
elif score <= 40:
return RelationshipLevel.ACQUAINTANCE
elif score <= 60:
return RelationshipLevel.FRIEND
elif score <= 80:
return RelationshipLevel.GOOD_FRIEND
else:
return RelationshipLevel.CLOSE_FRIEND
def get_level_display_name(self, level: RelationshipLevel) -> str:
"""Get a human-readable name for the relationship level."""
names = {
RelationshipLevel.STRANGER: "Stranger",
RelationshipLevel.ACQUAINTANCE: "Acquaintance",
RelationshipLevel.FRIEND: "Friend",
RelationshipLevel.GOOD_FRIEND: "Good Friend",
RelationshipLevel.CLOSE_FRIEND: "Close Friend",
}
return names.get(level, "Unknown")
async def add_shared_reference(
self, user: User, guild_id: int | None, reference_type: str, content: str
) -> None:
"""Add a shared reference (inside joke, nickname, etc.)."""
rel = await self.get_or_create_relationship(user, guild_id)
refs = rel.shared_references or {}
if reference_type not in refs:
refs[reference_type] = []
# Avoid duplicates and limit to 10 per type
if content not in refs[reference_type]:
refs[reference_type].append(content)
refs[reference_type] = refs[reference_type][-10:] # Keep last 10
rel.shared_references = refs
def get_relationship_prompt_modifier(
self, level: RelationshipLevel, relationship: UserRelationship
) -> str:
"""Generate prompt text reflecting relationship level."""
base_modifiers = {
RelationshipLevel.STRANGER: (
"This is someone you don't know well yet. "
"Be polite and welcoming, but keep some professional distance. "
"Use more formal language."
),
RelationshipLevel.ACQUAINTANCE: (
"This is someone you've chatted with a few times. "
"Be friendly and warm, but still somewhat reserved."
),
RelationshipLevel.FRIEND: (
"This is a friend! Be casual and warm. "
"Use their name occasionally, show you remember past conversations."
),
RelationshipLevel.GOOD_FRIEND: (
"This is a good friend you know well. "
"Be relaxed and personal. Reference things you've talked about before. "
"Feel free to be playful."
),
RelationshipLevel.CLOSE_FRIEND: (
"This is a close friend! Be very casual and familiar. "
"Use inside jokes if you have any, be supportive and genuine. "
"You can tease them gently and be more emotionally open."
),
}
modifier = base_modifiers.get(level, "")
# Add shared references for closer relationships
if level in (RelationshipLevel.GOOD_FRIEND, RelationshipLevel.CLOSE_FRIEND):
refs = relationship.shared_references or {}
if refs.get("jokes"):
jokes = refs["jokes"][:2]
modifier += f" You have inside jokes together: {', '.join(jokes)}."
if refs.get("nicknames"):
nicknames = refs["nicknames"][:1]
modifier += f" You sometimes call them: {nicknames[0]}."
return modifier
def _calculate_score_delta(
self, sentiment: float, message_length: int, conversation_turns: int
) -> float:
"""Calculate how much the relationship score should change.
Positive interactions increase score, negative decrease.
Longer messages and deeper conversations increase score more.
"""
# Base change from sentiment (-0.5 to +0.5)
base_delta = sentiment * 0.5
# Bonus for longer messages (up to +0.3)
length_bonus = min(0.3, message_length / 500)
# Bonus for deeper conversations (up to +0.2)
depth_bonus = min(0.2, conversation_turns * 0.05)
# Minimum interaction bonus (+0.1 just for talking)
interaction_bonus = 0.1
total_delta = base_delta + length_bonus + depth_bonus + interaction_bonus
# Clamp to reasonable range
return max(-1.0, min(1.0, total_delta))
async def get_relationship_info(self, user: User, guild_id: int | None = None) -> dict:
"""Get detailed relationship information for display."""
rel = await self.get_or_create_relationship(user, guild_id)
level = self.get_level(rel.relationship_score)
# Calculate time since first interaction
time_known = datetime.utcnow() - rel.first_interaction_at
days_known = time_known.days
return {
"level": level,
"level_name": self.get_level_display_name(level),
"score": rel.relationship_score,
"total_interactions": rel.total_interactions,
"positive_interactions": rel.positive_interactions,
"negative_interactions": rel.negative_interactions,
"first_interaction_at": rel.first_interaction_at,
"last_interaction_at": rel.last_interaction_at,
"days_known": days_known,
"shared_references": rel.shared_references or {},
}

View File

@@ -0,0 +1,220 @@
"""Self Awareness Service - provides bot self-reflection and statistics."""
import logging
from datetime import datetime
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from daemon_boyfriend.models import (
BotOpinion,
BotState,
User,
UserFact,
UserRelationship,
)
logger = logging.getLogger(__name__)
class SelfAwarenessService:
"""Provides bot self-reflection and statistics."""
def __init__(self, session: AsyncSession) -> None:
self._session = session
async def get_bot_stats(self, guild_id: int | None = None) -> dict:
"""Get comprehensive bot statistics."""
# Get or create bot state
stmt = select(BotState).where(BotState.guild_id == guild_id)
result = await self._session.execute(stmt)
bot_state = result.scalar_one_or_none()
if not bot_state:
bot_state = BotState(guild_id=guild_id)
self._session.add(bot_state)
await self._session.flush()
# Calculate age
age_delta = datetime.utcnow() - bot_state.first_activated_at
# Count users (from database)
user_count = await self._count_users()
# Count facts
fact_count = await self._count_facts()
# Get top interests
top_interests = await self._get_top_interests(guild_id)
return {
"age_days": age_delta.days,
"age_readable": self._format_age(age_delta),
"first_activated_at": bot_state.first_activated_at,
"total_messages_sent": bot_state.total_messages_sent,
"total_facts_learned": fact_count,
"total_users_known": user_count,
"favorite_topics": [op.topic for op in top_interests],
}
async def get_history_with_user(self, user: User, guild_id: int | None = None) -> dict:
"""Get the bot's history with a specific user."""
# Get relationship
stmt = select(UserRelationship).where(
UserRelationship.user_id == user.id,
UserRelationship.guild_id == guild_id,
)
result = await self._session.execute(stmt)
rel = result.scalar_one_or_none()
# Get facts count
facts_stmt = select(func.count(UserFact.id)).where(
UserFact.user_id == user.id,
UserFact.is_active == True,
)
facts_result = await self._session.execute(facts_stmt)
facts_count = facts_result.scalar() or 0
if rel:
days_known = (datetime.utcnow() - rel.first_interaction_at).days
return {
"first_met": rel.first_interaction_at,
"days_known": days_known,
"total_interactions": rel.total_interactions,
"relationship_score": rel.relationship_score,
"things_known": facts_count,
"shared_references": rel.shared_references or {},
}
else:
return {
"first_met": None,
"days_known": 0,
"total_interactions": 0,
"relationship_score": 0,
"things_known": facts_count,
"shared_references": {},
}
async def reflect_on_self(self, guild_id: int | None = None) -> str:
"""Generate a self-reflection statement."""
stats = await self.get_bot_stats(guild_id)
parts = []
# Age
if stats["age_days"] > 0:
parts.append(f"I've been around for {stats['age_readable']}.")
else:
parts.append("I'm pretty new here!")
# People known
if stats["total_users_known"] > 0:
parts.append(f"I've gotten to know {stats['total_users_known']} people.")
# Facts learned
if stats["total_facts_learned"] > 0:
parts.append(f"I've learned {stats['total_facts_learned']} things about them.")
# Favorite topics
if stats["favorite_topics"]:
topics = ", ".join(stats["favorite_topics"][:3])
parts.append(f"I find myself most interested in {topics}.")
# Messages
if stats["total_messages_sent"] > 0:
parts.append(f"I've sent {stats['total_messages_sent']} messages so far.")
return " ".join(parts)
async def reflect_on_user(self, user: User, guild_id: int | None = None) -> str:
"""Generate a reflection about the bot's history with a user."""
history = await self.get_history_with_user(user, guild_id)
parts = []
if history["first_met"]:
if history["days_known"] == 0:
parts.append("We just met today!")
elif history["days_known"] == 1:
parts.append("We met yesterday.")
elif history["days_known"] < 7:
parts.append(f"We met {history['days_known']} days ago.")
elif history["days_known"] < 30:
weeks = history["days_known"] // 7
parts.append(
f"We've known each other for about {weeks} week{'s' if weeks > 1 else ''}."
)
elif history["days_known"] < 365:
months = history["days_known"] // 30
parts.append(
f"We've known each other for about {months} month{'s' if months > 1 else ''}."
)
else:
years = history["days_known"] // 365
parts.append(
f"We've known each other for over {years} year{'s' if years > 1 else ''}!"
)
if history["total_interactions"] > 0:
parts.append(f"We've chatted about {history['total_interactions']} times.")
if history["things_known"] > 0:
parts.append(f"I've learned {history['things_known']} things about you.")
else:
parts.append("I don't think we've properly met before!")
return " ".join(parts)
async def _count_users(self) -> int:
"""Count total users in the database."""
stmt = select(func.count(User.id)).where(User.is_active == True)
result = await self._session.execute(stmt)
return result.scalar() or 0
async def _count_facts(self) -> int:
"""Count total facts in the database."""
stmt = select(func.count(UserFact.id)).where(UserFact.is_active == True)
result = await self._session.execute(stmt)
return result.scalar() or 0
async def _get_top_interests(self, guild_id: int | None, limit: int = 3) -> list[BotOpinion]:
"""Get top interests."""
stmt = (
select(BotOpinion)
.where(
BotOpinion.guild_id == guild_id,
BotOpinion.interest_level > 0.5,
)
.order_by(BotOpinion.interest_level.desc())
.limit(limit)
)
result = await self._session.execute(stmt)
return list(result.scalars().all())
def _format_age(self, delta) -> str:
"""Format a timedelta into a readable string."""
days = delta.days
if days == 0:
hours = delta.seconds // 3600
if hours == 0:
minutes = delta.seconds // 60
if minutes == 0:
return "just a moment"
return f"{minutes} minute{'s' if minutes != 1 else ''}"
return f"{hours} hour{'s' if hours != 1 else ''}"
elif days == 1:
return "1 day"
elif days < 7:
return f"{days} days"
elif days < 30:
weeks = days // 7
return f"about {weeks} week{'s' if weeks != 1 else ''}"
elif days < 365:
months = days // 30
return f"about {months} month{'s' if months != 1 else ''}"
else:
years = days // 365
months = (days % 365) // 30
if months > 0:
return f"about {years} year{'s' if years != 1 else ''} and {months} month{'s' if months != 1 else ''}"
return f"about {years} year{'s' if years != 1 else ''}"