-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathproactive_agent.py
More file actions
431 lines (349 loc) Β· 16.9 KB
/
proactive_agent.py
File metadata and controls
431 lines (349 loc) Β· 16.9 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
"""
Proactive conversation system for voice agents
Implements intelligent conversation initiation and topic suggestions
"""
import os
import time
import logging
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
from openai import OpenAI
from long_term_memory import LongTermMemory, MemoryEntry
class ProactivityLevel(Enum):
"""Proactivity levels for user control"""
OFF = 0 # No proactive interactions
LOW = 1 # Minimal, only high-confidence suggestions
NORMAL = 2 # Balanced proactive interactions
HIGH = 3 # More frequent suggestions and follow-ups
@dataclass
class ProactiveOpportunity:
"""Represents a potential proactive interaction"""
trigger_type: str # conversation_pause, topic_connection, follow_up, etc.
relevance_score: float # 0.0 - 1.0
usefulness_score: float # 0.0 - 1.0
suggested_response: str # The proactive message to send
related_memories: List[str] # IDs of memories that triggered this
confidence: float # Overall confidence in this suggestion
def should_execute(self, threshold: float = 0.7) -> bool:
"""Determine if this opportunity should be executed"""
return self.confidence >= threshold
class ConversationAnalyzer:
"""Analyzes conversation patterns to identify proactive opportunities"""
def __init__(self, openai_client: OpenAI):
self.client = openai_client
self.analysis_prompt = """You are a conversation analyzer for a proactive voice assistant. Analyze the recent conversation to identify opportunities for helpful, natural proactive interactions.
ANALYSIS CRITERIA:
- Look for natural conversation pauses or topic completions
- Identify topics that could benefit from follow-up questions
- Find connections to user's interests or past experiences
- Detect opportunities to be helpful without being intrusive
PROACTIVITY TYPES:
1. topic_connection: "That reminds me of when you mentioned X..."
2. follow_up: "How did that Y you were working on turn out?"
3. helpful_suggestion: "Since you're interested in X, you might like..."
4. engaging_question: "What's your experience with X been like?"
5. conversation_bridge: "Speaking of X, have you ever tried Y?"
IMPORTANT: Return ONLY a valid JSON object. No explanations, no markdown, just JSON.
If no good opportunities, return: {{"opportunities": []}}
Format:
{{
"opportunities": [
{{
"type": "helpful_suggestion",
"relevance": 0.85,
"usefulness": 0.75,
"suggestion": "Natural, conversational proactive message",
"reasoning": "Brief explanation"
}}
]
}}
Recent conversation context:
{conversation_context}
User memories that might be relevant:
{relevant_memories}
JSON object:"""
def analyze_conversation(self, conversation_context: str, relevant_memories: str) -> List[Dict[str, Any]]:
"""Analyze conversation for proactive opportunities"""
try:
# Generate analysis prompt
prompt = self.analysis_prompt.format(
conversation_context=conversation_context,
relevant_memories=relevant_memories
)
# Call LLM for analysis
response = self.client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
max_tokens=400,
temperature=0.3
)
# Parse response
content = response.choices[0].message.content.strip()
# Clean up response
if content.startswith('```json'):
content = content[7:]
if content.endswith('```'):
content = content[:-3]
content = content.strip()
import json
try:
analysis = json.loads(content)
return analysis.get('opportunities', [])
except json.JSONDecodeError as e:
# Try to extract JSON from response if wrapped in text
import re
json_match = re.search(r'\{.*\}', content, re.DOTALL)
if json_match:
try:
analysis = json.loads(json_match.group(0))
return analysis.get('opportunities', [])
except json.JSONDecodeError:
logging.warning(f"Failed to parse extracted JSON: {json_match.group(0)}")
return []
else:
logging.warning(f"Failed to parse conversation analysis: {content[:100]}...")
return []
except Exception as e:
logging.warning(f"Conversation analysis failed: {e}")
return []
class ProactiveAgent:
"""Manages proactive interactions for the voice agent"""
def __init__(self, openai_client: OpenAI, long_term_memory: LongTermMemory, user_id: str):
self.client = openai_client
self.memory = long_term_memory
self.user_id = user_id
self.analyzer = ConversationAnalyzer(openai_client)
# Proactivity settings
self.proactivity_level = ProactivityLevel.NORMAL
self.min_pause_duration = 3.0 # seconds
self.rate_limit_minutes = 10 # max 1 proactive per N minutes
self.confidence_threshold = 0.7
# State tracking
self.last_proactive_time = None
self.declined_topics = set() # Topics user has declined
self.conversation_start_time = datetime.now()
# Quiet hours (can be configured per user)
self.quiet_start_hour = 22 # 10 PM
self.quiet_end_hour = 7 # 7 AM
logging.info(f"ProactiveAgent initialized for user {user_id}")
def set_proactivity_level(self, level: ProactivityLevel):
"""Update proactivity level based on user preference"""
self.proactivity_level = level
logging.info(f"Proactivity level set to: {level.name}")
def is_quiet_hours(self) -> bool:
"""Check if current time is during quiet hours"""
current_hour = datetime.now().hour
if self.quiet_start_hour < self.quiet_end_hour:
return not (self.quiet_end_hour <= current_hour < self.quiet_start_hour)
else: # Quiet hours span midnight
return not (self.quiet_end_hour <= current_hour < self.quiet_start_hour)
def can_be_proactive(self) -> bool:
"""Check if conditions are right for proactive interaction"""
# Check if proactivity is enabled
if self.proactivity_level == ProactivityLevel.OFF:
return False
# Check quiet hours
if self.is_quiet_hours():
return False
# Check rate limiting
if self.last_proactive_time:
time_since_last = datetime.now() - self.last_proactive_time
if time_since_last.total_seconds() < (self.rate_limit_minutes * 60):
return False
return True
def generate_proactive_opportunity(self, conversation_context: str, pause_duration: float) -> Optional[ProactiveOpportunity]:
"""Generate a proactive interaction opportunity"""
if not self.can_be_proactive():
return None
# Only trigger after sufficient pause
if pause_duration < self.min_pause_duration:
return None
try:
# Get relevant memories for context
recent_context = conversation_context[-500:] # Last 500 chars
memories = self.memory.retrieve_memories(recent_context, k=5)
# Format memories for analysis
memory_context = ""
if memories:
memory_context = "Relevant user memories:\n"
for memory, score in memories:
memory_context += f"- {memory.text} ({memory.domain})\n"
else:
memory_context = "No specific user memories available."
# Analyze conversation for opportunities
opportunities = self.analyzer.analyze_conversation(
conversation_context,
memory_context
)
if not opportunities:
return None
# Select best opportunity
best_opportunity = None
highest_confidence = 0.0
for opp in opportunities:
# Calculate confidence score
relevance = opp.get('relevance', 0.0)
usefulness = opp.get('usefulness', 0.0)
# Apply proactivity level adjustments
if self.proactivity_level == ProactivityLevel.LOW:
confidence = (relevance * 0.4 + usefulness * 0.6) * 0.8 # More conservative
elif self.proactivity_level == ProactivityLevel.HIGH:
confidence = (relevance * 0.6 + usefulness * 0.4) * 1.1 # More aggressive
else: # NORMAL
confidence = (relevance * 0.5 + usefulness * 0.5)
# Boost confidence based on pause duration (longer pause = more likely they want engagement)
pause_boost = min(0.2, (pause_duration - self.min_pause_duration) / 10.0)
confidence += pause_boost
if confidence > highest_confidence and confidence >= self.confidence_threshold:
highest_confidence = confidence
best_opportunity = ProactiveOpportunity(
trigger_type=opp.get('type', 'general'),
relevance_score=relevance,
usefulness_score=usefulness,
suggested_response=opp.get('suggestion', ''),
related_memories=[m[0].id for m in memories[:2]], # Top 2 related memories
confidence=confidence
)
return best_opportunity
except Exception as e:
logging.warning(f"Failed to generate proactive opportunity: {e}")
return None
def execute_proactive_interaction(self, opportunity: ProactiveOpportunity) -> str:
"""Execute a proactive interaction and return the message"""
try:
# Update state
self.last_proactive_time = datetime.now()
# Log the proactive interaction
logging.info(f"Proactive interaction: {opportunity.trigger_type} (confidence: {opportunity.confidence:.2f})")
# Return the suggested response
return opportunity.suggested_response
except Exception as e:
logging.error(f"Failed to execute proactive interaction: {e}")
return ""
def handle_user_decline(self, topic: str):
"""Handle when user declines a proactive suggestion"""
self.declined_topics.add(topic.lower())
# Increase rate limit temporarily (30 min cooldown)
if self.last_proactive_time:
self.last_proactive_time = datetime.now() + timedelta(minutes=20) # Extra 20 min cooldown
logging.info(f"User declined topic: {topic}, applying cooldown")
def get_proactivity_stats(self) -> Dict[str, Any]:
"""Get statistics about proactive interactions"""
total_conversation_time = (datetime.now() - self.conversation_start_time).total_seconds() / 60
return {
'user_id': self.user_id,
'proactivity_level': self.proactivity_level.name,
'conversation_minutes': round(total_conversation_time, 1),
'last_proactive': self.last_proactive_time.isoformat() if self.last_proactive_time else None,
'declined_topics_count': len(self.declined_topics),
'can_be_proactive_now': self.can_be_proactive()
}
# Integration patterns for the voice agent
class ProactiveVoiceAgentMixin:
"""Mixin to add proactive capabilities to the voice agent"""
def init_proactive_agent(self):
"""Initialize proactive agent (call from voice agent __init__)"""
if hasattr(self, 'ltm_enabled') and self.ltm_enabled:
try:
self.proactive_agent = ProactiveAgent(
self.client,
self.long_term_memory,
self.user_id
)
self.proactive_enabled = True
logging.info("Proactive agent initialized")
except Exception as e:
logging.warning(f"Failed to initialize proactive agent: {e}")
self.proactive_agent = None
self.proactive_enabled = False
else:
self.proactive_agent = None
self.proactive_enabled = False
def check_for_proactive_opportunity(self, pause_duration: float = 5.0) -> Optional[str]:
"""Check if there's a proactive interaction opportunity"""
if not self.proactive_enabled or not self.proactive_agent:
return None
# Get conversation context
context = self.memory.get_conversation_context()
# Generate opportunity
opportunity = self.proactive_agent.generate_proactive_opportunity(
context,
pause_duration
)
if opportunity and opportunity.should_execute():
return self.proactive_agent.execute_proactive_interaction(opportunity)
return None
def set_proactivity_level(self, level: str):
"""Set proactivity level from string"""
if self.proactive_enabled and self.proactive_agent:
try:
level_enum = ProactivityLevel[level.upper()]
self.proactive_agent.set_proactivity_level(level_enum)
return f"Proactivity set to {level}"
except KeyError:
return f"Invalid proactivity level. Use: off, low, normal, high"
return "Proactive agent not available"
# Enhanced Voice Agent with Natural AI Voice and Proactive Features
class EnhancedVoiceAgent:
"""Complete voice agent with natural AI voice, proactive interactions, and memory"""
def __init__(self, user_id: str = "default_user"):
try:
# Import the required classes
from voice_agent import VoiceAgent
from openai_tts_upgrade import OpenAITTSMixin
# Create a combined class with natural AI voice
class AIVoiceAgent(OpenAITTSMixin, VoiceAgent):
pass
# Initialize the enhanced agent
self.agent = AIVoiceAgent(user_id=user_id)
self.initialized = True
print("π€ Enhanced Voice Agent with Natural AI Voice initialized!")
print("β
Features: Natural speech, Memory, Proactive interactions")
except ImportError as e:
print(f"β Failed to import required modules: {e}")
print("Make sure you have installed: pip install pygame")
self.agent = None
self.initialized = False
except Exception as e:
print(f"β Failed to initialize enhanced voice agent: {e}")
self.agent = None
self.initialized = False
def start_conversation(self):
"""Start a conversation with enhanced natural voice"""
if not self.initialized:
print("β Enhanced voice agent is not properly initialized")
return
print("\n" + "π€"*20)
print("π€ Enhanced AI Voice Agent")
print("Features: Natural Voice + Memory + Proactive Chat")
print("π€"*20 + "\n")
try:
self.agent.run_conversation()
except Exception as e:
print(f"β Error during conversation: {e}")
def get_status(self) -> str:
"""Get the current status of the voice agent"""
if not self.initialized:
return "β Not initialized"
try:
memory_summary = self.agent.get_memory_summary()
return f"β
Ready | {memory_summary}"
except:
return "β
Ready | Natural AI Voice Active"
# Example usage
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
# Test enhanced voice agent
from dotenv import load_dotenv
load_dotenv()
# Create and start the enhanced voice agent
enhanced_agent = EnhancedVoiceAgent()
if enhanced_agent.initialized:
print(f"Status: {enhanced_agent.get_status()}")
print("\nπ€ Ready to start conversation with natural AI voice!")
print("Run: enhanced_agent.start_conversation()")
else:
print("β Enhanced voice agent failed to initialize")
print("Fallback: You can still use the basic proactive agent features")