""" 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()