Create agent_orchestrator.py

agent_orchestrator.py

@@ -0,0 +1,338 @@
+"""
+Agent Orchestrator - Coordinates multiple specialized agents (Free Version)
+Uses only free resources and local models
+"""
+
+from typing import Dict, List, Optional, Any
+import asyncio
+from dataclasses import dataclass
+from enum import Enum
+import json
+from datetime import datetime
+import random
+
+# Import free agents
+from agents.diagnostician import DiagnosticianAgent
+from agents.socratic_tutor import SocraticTutorAgent
+from agents.practice_generator import PracticeGeneratorAgent
+from agents.calculator_agent import CalculatorAgent
+from agents.knowledge_agent import KnowledgeAgent
+from agents.assessment_agent import AssessmentAgent
+
+
+class AgentRole(Enum):
+    """Enumeration of agent roles"""
+    DIAGNOSTICIAN = "diagnostician"
+    SOCRATIC_TUTOR = "socratic_tutor"
+    PRACTICE_GENERATOR = "practice_generator"
+    CALCULATOR = "calculator"
+    KNOWLEDGE_EXPERT = "knowledge_expert"
+    ASSESSOR = "assessor"
+
+
+@dataclass
+class AgentResponse:
+    """Container for agent responses"""
+    agent: str
+    response: str
+    confidence: float
+    metadata: Dict[str, Any]
+    timestamp: datetime
+
+
+class AgentOrchestrator:
+    """Orchestrates multiple specialized agents using free resources"""
+    
+    def __init__(self):
+        self.agents = {}
+        self.conversation_history = []
+        self.user_context = {}
+        self.system_prompts = self._load_system_prompts()
+        
+    def _load_system_prompts(self) -> Dict[str, str]:
+        """Load system prompts for different agent roles"""
+        return {
+            "diagnostician": """You are a learning diagnostician. Analyze the user's question to identify:
+1. What type of learning problem this is (conceptual, calculation, application, analysis)
+2. What knowledge gaps might exist
+3. What prerequisite knowledge is needed
+4. Specific areas where the user might be struggling
+
+Provide clear, actionable feedback.""",
+            
+            "socratic_tutor": """You are a Socratic tutor. Guide the user to discover answers themselves by asking questions.
+1. Start with a guiding question
+2. Provide hints if the user struggles
+3. Build up to the full concept
+4. Encourage critical thinking
+
+Do not give direct answers immediately.""",
+            
+            "calculator": """You are a financial calculator expert. For calculation requests:
+1. Identify the formula needed
+2. Show the formula clearly
+3. Walk through each step
+4. Provide the final answer
+5. Explain what the result means
+
+For non-calculation questions, suggest relevant calculations.""",
+            
+            "knowledge_expert": """You are a CFA knowledge expert. Provide accurate information about:
+1. Financial concepts and theories
+2. Formulas and their applications
+3. Real-world examples
+4. CFA exam relevance
+
+Cite sources from the knowledge base when available.""",
+            
+            "practice_generator": """You are a practice problem generator. Create relevant practice problems:
+1. Match the user's learning level
+2. Cover the requested topic
+3. Include multiple choice options
+4. Provide step-by-step solutions
+5. Explain why other options are wrong
+
+Start with easier problems and increase difficulty.""",
+            
+            "assessor": """You are a learning assessor. Evaluate the user's progress:
+1. Identify strengths and weaknesses
+2. Track learning patterns
+3. Provide personalized recommendations
+4. Estimate readiness for CFA exam
+
+Be encouraging but honest."""
+        }
+    
+    def initialize_agents(self):
+        """Initialize all specialized agents with free resources"""
+        print("🔄 Initializing AI Agents (Free Version)...")
+        
+        try:
+            self.agents = {
+                AgentRole.DIAGNOSTICIAN: DiagnosticianAgent(),
+                AgentRole.SOCRATIC_TUTOR: SocraticTutorAgent(),
+                AgentRole.PRACTICE_GENERATOR: PracticeGeneratorAgent(),
+                AgentRole.CALCULATOR: CalculatorAgent(),
+                AgentRole.KNOWLEDGE_EXPERT: KnowledgeAgent(),
+                AgentRole.ASSESSOR: AssessmentAgent()
+            }
+            
+            print("✅ All agents initialized successfully")
+            return True
+            
+        except Exception as e:
+            print(f"❌ Error initializing agents: {e}")
+            # Create fallback agents
+            self._create_fallback_agents()
+            return False
+    
+    def _create_fallback_agents(self):
+        """Create simple fallback agents if initialization fails"""
+        print("⚠️ Using fallback agents")
+        
+        class FallbackAgent:
+            def process(self, query, user_profile, conversation_history):
+                return {
+                    'response': f"I'm a fallback agent. I received: {query[:100]}...",
+                    'confidence': 0.5,
+                    'metadata': {'fallback': True}
+                }
+        
+        for role in AgentRole:
+            self.agents[role] = FallbackAgent()
+    
+    def process_query(self, 
+                     query: str, 
+                     user_profile: Dict[str, Any],
+                     selected_agents: List[str] = None) -> Dict[str, Any]:
+        """
+        Process a user query through the agent pipeline
+        
+        Args:
+            query: User's question or request
+            user_profile: User's learning profile
+            selected_agents: List of agent names to involve
+        
+        Returns:
+            Combined response from agents
+        """
+        if not self.agents:
+            self.initialize_agents()
+        
+        # Update user context
+        self.user_context.update(user_profile)
+        self.user_context['last_query'] = query
+        self.user_context['timestamp'] = datetime.now().isoformat()
+        
+        # Default to all agents if none selected
+        if not selected_agents:
+            selected_agents = [agent.value for agent in AgentRole]
+        
+        # Convert string agent names to AgentRole enum
+        agent_roles = []
+        for agent_name in selected_agents:
+            for role in AgentRole:
+                if role.value == agent_name.lower().replace(" ", "_"):
+                    agent_roles.append(role)
+                    break
+        
+        # If no valid agents found, use diagnostician and knowledge expert
+        if not agent_roles:
+            agent_roles = [AgentRole.DIAGNOSTICIAN, AgentRole.KNOWLEDGE_EXPERT]
+        
+        # Execute agents
+        responses = self._execute_agents(agent_roles, query, user_profile)
+        
+        # Synthesize responses
+        final_response = self._synthesize_responses(responses, query)
+        
+        # Update conversation history
+        self.conversation_history.append({
+            'query': query,
+            'responses': [r.__dict__ for r in responses],
+            'final_response': final_response,
+            'timestamp': datetime.now().isoformat()
+        })
+        
+        # Keep only last 50 conversations
+        if len(self.conversation_history) > 50:
+            self.conversation_history = self.conversation_history[-50:]
+        
+        return final_response
+    
+    def _execute_agents(self, 
+                       agent_roles: List[AgentRole],
+                       query: str,
+                       user_profile: Dict[str, Any]) -> List[AgentResponse]:
+        """
+        Execute selected agents and collect responses
+        """
+        responses = []
+        
+        for role in agent_roles:
+            agent = self.agents.get(role)
+            if agent:
+                try:
+                    # Process with agent
+                    result = agent.process(query, user_profile, self.conversation_history)
+                    
+                    agent_response = AgentResponse(
+                        agent=role.value.replace("_", " ").title(),
+                        response=result.get('response', 'No response generated'),
+                        confidence=result.get('confidence', 0.5),
+                        metadata=result.get('metadata', {}),
+                        timestamp=datetime.now()
+                    )
+                    responses.append(agent_response)
+                    
+                except Exception as e:
+                    print(f"❌ Error in agent {role.value}: {e}")
+                    # Create error response
+                    error_response = AgentResponse(
+                        agent=role.value.replace("_", " ").title(),
+                        response=f"This agent is currently unavailable. Error: {str(e)[:100]}",
+                        confidence=0.0,
+                        metadata={'error': True, 'message': str(e)},
+                        timestamp=datetime.now()
+                    )
+                    responses.append(error_response)
+        
+        return responses
+    
+    def _synthesize_responses(self, 
+                             responses: List[AgentResponse],
+                             original_query: str) -> Dict[str, Any]:
+        """
+        Synthesize multiple agent responses into a coherent answer
+        """
+        if not responses:
+            return {
+                'response': "I'm sorry, no agents were able to process your query. Please try rephrasing.",
+                'primary_agent': 'System',
+                'sources': [],
+                'confidence': 0.0,
+                'metadata': {'error': 'No responses'}
+            }
+        
+        # Filter out error responses
+        valid_responses = [r for r in responses if r.confidence > 0.1]
+        
+        if not valid_responses:
+            return {
+                'response': "The agents encountered errors. Please try a simpler query or check back later.",
+                'primary_agent': 'System',
+                'sources': [],
+                'confidence': 0.0
+            }
+        
+        # Sort by confidence
+        valid_responses.sort(key=lambda x: x.confidence, reverse=True)
+        
+        # Get primary response (highest confidence)
+        primary_response = valid_responses[0]
+        
+        # Build final response
+        response_parts = []
+        
+        # Add primary response
+        response_parts.append(f"## {primary_response.agent}\n\n{primary_response.response}")
+        
+        # Add other high-confidence responses
+        other_responses = []
+        for resp in valid_responses[1:]:
+            if resp.confidence > 0.6 and resp.agent != primary_response.agent:
+                other_responses.append(resp)
+        
+        if other_responses:
+            response_parts.append("\n\n---\n\n## Additional Insights")
+            for resp in other_responses[:2]:  # Limit to 2 additional responses
+                response_parts.append(f"\n\n### {resp.agent}\n\n{resp.response[:200]}...")
+        
+        final_response = "\n".join(response_parts)
+        
+        return {
+            'response': final_response,
+            'primary_agent': primary_response.agent,
+            'supporting_agents': [resp.agent for resp in valid_responses[1:]],
+            'confidence': primary_response.confidence,
+            'timestamp': datetime.now().isoformat(),
+            'metadata': {
+                'response_count': len(valid_responses),
+                'avg_confidence': sum(r.confidence for r in valid_responses) / len(valid_responses)
+            }
+        }
+    
+    def get_agent_status(self) -> Dict[str, Dict[str, Any]]:
+        """
+        Get status of all agents
+        """
+        status = {}
+        for role, agent in self.agents.items():
+            status[role.value] = {
+                'active': True,
+                'type': agent.__class__.__name__,
+                'initialized': True,
+                'capabilities': self._get_agent_capabilities(role)
+            }
+        return status
+    
+    def _get_agent_capabilities(self, role: AgentRole) -> List[str]:
+        """Get capabilities of an agent"""
+        capabilities = {
+            AgentRole.DIAGNOSTICIAN: ["Learning gap analysis", "Problem diagnosis", "Recommendation generation"],
+            AgentRole.SOCRATIC_TUTOR: ["Guided questioning", "Concept building", "Critical thinking"],
+            AgentRole.CALCULATOR: ["Financial calculations", "Formula explanation", "Step-by-step solutions"],
+            AgentRole.KNOWLEDGE_EXPERT: ["Concept explanation", "Knowledge retrieval", "Example provision"],
+            AgentRole.PRACTICE_GENERATOR: ["Problem creation", "Solution generation", "Difficulty scaling"],
+            AgentRole.ASSESSOR: ["Progress tracking", "Performance evaluation", "Study recommendations"]
+        }
+        return capabilities.get(role, ["General assistance"])
+    
+    def reset_conversation(self):
+        """Reset conversation history"""
+        self.conversation_history = []
+        print("🔄 Conversation history reset")
+
+
+# Singleton instance
+orchestrator = AgentOrchestrator()