Spaces:
Runtime error
Runtime error
| """ | |
| 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" | |
| 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() |