MCP Self-Learning Server

#!/usr/bin/env python3 """ Claudia Voice Assistant Learning Integration Plugin Provides learning capabilities to enhance Claudia's voice interactions """ import asyncio import json import logging from typing import Dict, List, Optional, Any, Callable from datetime import datetime import sys import os # Add the parent directory to the path to import the client library sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'lib')) from self_learning_client import SelfLearningClient, ClaudiaLearningIntegration class ClaudiaLearningPlugin: """ Learning plugin for Claudia Voice Assistant This plugin integrates with the MCP Self-Learning Server to: - Learn from voice interactions - Improve intent recognition - Optimize response generation - Predict user needs """ def __init__(self, claudia_instance, config: Dict[str, Any] = None): self.claudia = claudia_instance self.config = config or {} self.logger = logging.getLogger(__name__) # Initialize learning integration mcp_url = self.config.get('mcp_server_url', 'http://localhost:8765') self.learning_integration = ClaudiaLearningIntegration(mcp_url) # Learning settings self.learning_enabled = self.config.get('learning_enabled', True) self.auto_optimize = self.config.get('auto_optimize', True) self.prediction_enabled = self.config.get('prediction_enabled', True) # Statistics tracking self.stats = { 'interactions_learned': 0, 'predictions_made': 0, 'optimizations_applied': 0, 'session_start': datetime.now() } # Current session tracking self.current_session = None self.session_interactions = [] self.is_initialized = False async def initialize(self): """Initialize the learning plugin""" if self.is_initialized: return try: await self.learning_integration.start() self.is_initialized = True self.logger.info("Claudia learning plugin initialized successfully") # Start periodic optimization if enabled if self.auto_optimize: asyncio.create_task(self._periodic_optimization()) except Exception as e: self.logger.error(f"Failed to initialize learning plugin: {e}") raise async def shutdown(self): """Shutdown the learning plugin""" try: # Learn from current session if active if self.current_session and self.session_interactions: await self._finalize_session() await self.learning_integration.stop() self.is_initialized = False self.logger.info("Claudia learning plugin shutdown complete") except Exception as e: self.logger.error(f"Error during learning plugin shutdown: {e}") def start_session(self, session_id: str = None) -> str: """Start a new learning session""" if not session_id: session_id = f"claudia-{datetime.now().strftime('%Y%m%d-%H%M%S')}" self.current_session = session_id self.session_interactions = [] self.logger.debug(f"Started learning session: {session_id}") return session_id async def end_session(self, session_success: bool = True, user_satisfaction: float = None): """End the current learning session""" if not self.current_session: return try: # Analyze the complete session if self.session_interactions: await self._analyze_session(session_success, user_satisfaction) await self._finalize_session() except Exception as e: self.logger.error(f"Error ending learning session: {e}") finally: self.current_session = None self.session_interactions = [] async def learn_from_interaction(self, user_input: str, assistant_response: str, intent: str, confidence: float, success: bool = True, metadata: Dict[str, Any] = None) -> Dict[str, Any]: """ Learn from a voice interaction Args: user_input: What the user said assistant_response: How Claudia responded intent: Detected intent confidence: Confidence in intent detection success: Whether the interaction was successful metadata: Additional interaction metadata Returns: Learning results and recommendations """ if not self.learning_enabled or not self.is_initialized: return {'learned': False, 'reason': 'Learning disabled or not initialized'} try: # Learn from the interaction result = await self.learning_integration.log_voice_command( user_input, assistant_response, intent, confidence, success ) # Track the interaction in current session if self.current_session: interaction_data = { 'user_input': user_input, 'assistant_response': assistant_response, 'intent': intent, 'confidence': confidence, 'success': success, 'timestamp': datetime.now().isoformat(), 'metadata': metadata or {} } self.session_interactions.append(interaction_data) self.stats['interactions_learned'] += 1 # Get immediate insights if this was a significant interaction if confidence < 0.5 or not success: insights = await self.get_learning_insights() result['insights'] = insights return result except Exception as e: self.logger.error(f"Failed to learn from interaction: {e}") return {'learned': False, 'error': str(e)} async def predict_user_intent(self, partial_input: str, context: Dict[str, Any] = None) -> Dict[str, Any]: """ Predict user intent from partial input Args: partial_input: Partial user input (e.g., from ongoing speech recognition) context: Current conversation context Returns: Predicted intent and confidence """ if not self.prediction_enabled or not self.is_initialized: return {'predictions': [], 'confidence': 0.0} try: result = await self.learning_integration.predict_user_intent(partial_input) self.stats['predictions_made'] += 1 # Format predictions for Claudia formatted_predictions = self._format_predictions_for_claudia(result) return formatted_predictions except Exception as e: self.logger.error(f"Failed to predict user intent: {e}") return {'predictions': [], 'confidence': 0.0, 'error': str(e)} async def get_voice_optimizations(self) -> Dict[str, Any]: """Get optimization suggestions for voice interactions""" if not self.is_initialized: return {'optimizations': [], 'recommendations': []} try: result = await self.learning_integration.get_voice_optimizations() # Format optimizations for Claudia formatted_optimizations = self._format_optimizations_for_claudia(result) return formatted_optimizations except Exception as e: self.logger.error(f"Failed to get voice optimizations: {e}") return {'optimizations': [], 'recommendations': [], 'error': str(e)} async def get_learning_insights(self) -> Dict[str, Any]: """Get learning insights and statistics""" if not self.is_initialized: return {'insights': {}, 'stats': self.stats} try: # Get insights from the learning server status = await self.learning_integration.client.get_status() insights = await self.learning_integration.client.get_insights() # Combine with local stats combined_insights = { 'server_status': { 'running': status.running, 'learning_active': status.learning_status.value, 'patterns': status.patterns_count, 'knowledge': status.knowledge_count }, 'learning_insights': { 'top_patterns': insights.top_patterns[:5], # Top 5 patterns 'performance': insights.performance_metrics, 'recommendations': insights.recommendations }, 'plugin_stats': self.stats, 'session_info': { 'current_session': self.current_session, 'session_interactions': len(self.session_interactions) } } return combined_insights except Exception as e: self.logger.error(f"Failed to get learning insights: {e}") return {'insights': {}, 'stats': self.stats, 'error': str(e)} async def improve_intent_recognition(self, training_data: List[Dict[str, Any]]) -> Dict[str, Any]: """ Improve intent recognition using training data Args: training_data: List of training examples with 'input', 'intent', 'confidence' Returns: Training results and improvements """ if not self.learning_enabled or not self.is_initialized: return {'improved': False, 'reason': 'Learning disabled or not initialized'} try: results = [] for example in training_data: # Create interaction for learning interaction = { 'type': 'intent_training', 'input': example['input'], 'output': example['intent'], 'success': True, 'context': { 'confidence': example.get('confidence', 1.0), 'training_mode': True, 'expected_intent': example['intent'] } } result = await self.learning_integration.client.analyze_pattern(interaction) results.append(result) # Trigger learning cycle after batch training await self.learning_integration.client.trigger_learning() return { 'improved': True, 'examples_processed': len(training_data), 'results': results, 'message': 'Intent recognition training completed' } except Exception as e: self.logger.error(f"Failed to improve intent recognition: {e}") return {'improved': False, 'error': str(e)} async def get_conversation_suggestions(self, conversation_history: List[Dict[str, Any]], user_profile: Dict[str, Any] = None) -> Dict[str, Any]: """ Get suggestions for continuing the conversation Args: conversation_history: Recent conversation turns user_profile: Known information about the user Returns: Conversation suggestions and predictions """ if not self.prediction_enabled or not self.is_initialized: return {'suggestions': [], 'confidence': 0.0} try: # Build context from conversation history context = { 'conversation_history': conversation_history[-5:], # Last 5 turns 'user_profile': user_profile or {}, 'conversation_length': len(conversation_history), 'prediction_type': 'conversation_continuation' } # Get predictions predictions = await self.learning_integration.client.predict_next_action(context) # Format as conversation suggestions suggestions = self._format_conversation_suggestions(predictions) return suggestions except Exception as e: self.logger.error(f"Failed to get conversation suggestions: {e}") return {'suggestions': [], 'confidence': 0.0, 'error': str(e)} async def _analyze_session(self, session_success: bool, user_satisfaction: float = None): """Analyze the complete session for learning""" if not self.session_interactions: return try: # Create a session summary interaction session_summary = { 'type': 'claudia_session', 'input': f"Session with {len(self.session_interactions)} interactions", 'output': f"Session {'successful' if session_success else 'unsuccessful'}", 'success': session_success, 'context': { 'session_id': self.current_session, 'total_interactions': len(self.session_interactions), 'user_satisfaction': user_satisfaction, 'average_confidence': sum(i.get('confidence', 0) for i in self.session_interactions) / len(self.session_interactions), 'intents': list(set(i.get('intent') for i in self.session_interactions if i.get('intent'))) }, 'performance': { 'duration': (datetime.now() - self.stats['session_start']).total_seconds() * 1000 } } # Learn from the session await self.learning_integration.client.analyze_pattern(session_summary) self.logger.info(f"Analyzed session {self.current_session} with {len(self.session_interactions)} interactions") except Exception as e: self.logger.error(f"Failed to analyze session: {e}") async def _finalize_session(self): """Finalize the current session""" if self.session_interactions: # Export session data for potential future analysis session_data = { 'session_id': self.current_session, 'interactions': self.session_interactions, 'stats': { 'total_interactions': len(self.session_interactions), 'successful_interactions': sum(1 for i in self.session_interactions if i.get('success', True)), 'average_confidence': sum(i.get('confidence', 0) for i in self.session_interactions) / len(self.session_interactions), 'session_duration': (datetime.now() - self.stats['session_start']).total_seconds() } } self.logger.debug(f"Session {self.current_session} finalized with {len(self.session_interactions)} interactions") async def _periodic_optimization(self): """Periodic optimization task""" while self.is_initialized: try: await asyncio.sleep(300) # Every 5 minutes if self.auto_optimize: optimizations = await self.get_voice_optimizations() # Apply automatic optimizations if available if optimizations.get('optimizations'): auto_applicable = [opt for opt in optimizations['optimizations'] if opt.get('auto_applicable', False)] for optimization in auto_applicable: await self._apply_optimization(optimization) self.stats['optimizations_applied'] += 1 except asyncio.CancelledError: break except Exception as e: self.logger.error(f"Error in periodic optimization: {e}") def _format_predictions_for_claudia(self, predictions: Dict[str, Any]) -> Dict[str, Any]: """Format predictions for Claudia's usage""" formatted = { 'predictions': [], 'confidence': predictions.get('confidence', 0.0) } suggestions = predictions.get('suggestions', []) for suggestion in suggestions: formatted_suggestion = { 'intent': suggestion.get('action', 'unknown'), 'confidence': suggestion.get('confidence', 0.0), 'reasoning': suggestion.get('reasoning', ''), 'parameters': suggestion.get('parameters', {}) } formatted['predictions'].append(formatted_suggestion) return formatted def _format_optimizations_for_claudia(self, optimizations: Dict[str, Any]) -> Dict[str, Any]: """Format optimizations for Claudia's usage""" formatted = { 'optimizations': [], 'recommendations': [] } opts = optimizations.get('optimizations', []) for opt in opts: formatted_opt = { 'type': opt.get('type', 'general'), 'description': opt.get('description', ''), 'impact': opt.get('impact', 'low'), 'auto_applicable': opt.get('type') in ['response_timing', 'confidence_threshold'], 'parameters': opt.get('parameters', {}) } formatted['optimizations'].append(formatted_opt) # Add general recommendations if optimizations.get('recommendations'): formatted['recommendations'] = optimizations['recommendations'] return formatted def _format_conversation_suggestions(self, predictions: Dict[str, Any]) -> Dict[str, Any]: """Format predictions as conversation suggestions""" suggestions = { 'suggestions': [], 'confidence': predictions.get('confidence', 0.0) } for prediction in predictions.get('suggestions', []): suggestion = { 'type': 'conversation_move', 'action': prediction.get('action', 'continue'), 'text': prediction.get('suggested_response', ''), 'confidence': prediction.get('confidence', 0.0), 'intent': prediction.get('suggested_intent', 'unknown') } suggestions['suggestions'].append(suggestion) return suggestions async def _apply_optimization(self, optimization: Dict[str, Any]): """Apply an optimization to Claudia""" try: opt_type = optimization.get('type') params = optimization.get('parameters', {}) if opt_type == 'response_timing': # Adjust response timing if Claudia supports it if hasattr(self.claudia, 'set_response_timing'): self.claudia.set_response_timing(params.get('timing', 1.0)) self.logger.info("Applied response timing optimization") elif opt_type == 'confidence_threshold': # Adjust confidence threshold if Claudia supports it if hasattr(self.claudia, 'set_confidence_threshold'): self.claudia.set_confidence_threshold(params.get('threshold', 0.7)) self.logger.info("Applied confidence threshold optimization") # Add more optimization types as needed except Exception as e: self.logger.error(f"Failed to apply optimization {optimization.get('type')}: {e}") def get_stats(self) -> Dict[str, Any]: """Get plugin statistics""" current_time = datetime.now() uptime = (current_time - self.stats['session_start']).total_seconds() return { **self.stats, 'uptime_seconds': uptime, 'uptime_formatted': f"{int(uptime // 3600)}h {int((uptime % 3600) // 60)}m", 'learning_rate': self.stats['interactions_learned'] / max(uptime / 3600, 1), # Per hour 'prediction_rate': self.stats['predictions_made'] / max(uptime / 3600, 1), # Per hour 'is_initialized': self.is_initialized, 'current_session': self.current_session } # Integration helper functions for easy setup async def setup_claudia_learning(claudia_instance, config: Dict[str, Any] = None) -> ClaudiaLearningPlugin: """ Set up learning integration for a Claudia instance Args: claudia_instance: Instance of Claudia voice assistant config: Configuration dictionary Returns: Initialized learning plugin """ plugin = ClaudiaLearningPlugin(claudia_instance, config) await plugin.initialize() return plugin def create_learning_config(mcp_server_url: str = "http://localhost:8765", learning_enabled: bool = True, auto_optimize: bool = True, prediction_enabled: bool = True) -> Dict[str, Any]: """Create a learning configuration dictionary""" return { 'mcp_server_url': mcp_server_url, 'learning_enabled': learning_enabled, 'auto_optimize': auto_optimize, 'prediction_enabled': prediction_enabled } # Example usage in Claudia's main loop: """ # In Claudia's initialization: learning_config = create_learning_config() learning_plugin = await setup_claudia_learning(self, learning_config) # In Claudia's interaction handler: async def handle_user_input(self, user_input, intent, confidence): # Process the interaction normally response = await self.generate_response(user_input, intent) # Learn from the interaction learning_result = await self.learning_plugin.learn_from_interaction( user_input, response, intent, confidence, success=True ) # Get predictions for better responses if confidence < 0.7: predictions = await self.learning_plugin.predict_user_intent(user_input) if predictions['predictions'] and predictions['confidence'] > confidence: # Use the predicted intent instead better_intent = predictions['predictions'][0]['intent'] response = await self.generate_response(user_input, better_intent) return response """