File size: 11,629 Bytes
25732fb | 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 | """
Coordinator Agent - Orchestrates all other agents
"""
from agents.base_agent import BaseAgent, StudentState
from agents.teaching_agent import TeachingAgent
from agents.assessment_agent import AssessmentAgent
from agents.knowledge_agent import KnowledgeAgent
from agents.tutor_agent import TutorAgent
from agents.recommendation_agent import RecommendationAgent
from datetime import datetime
import json
class CoordinatorAgent(BaseAgent):
"""
Master agent that coordinates all other agents
"""
def __init__(self):
super().__init__("CA-001", "CoordinatorAgent")
# Initialize all sub-agents
self.teaching_agent = TeachingAgent()
self.assessment_agent = AssessmentAgent()
self.knowledge_agent = KnowledgeAgent()
self.tutor_agent = TutorAgent()
self.recommendation_agent = RecommendationAgent()
self.tasks_coordinated = 0
self.log("Coordinator initialized with 5 sub-agents")
def perceive(self, environment):
"""
Perceive global learning environment
"""
self.update_state("perceiving")
self.log(f"Coordinating agents for task: {environment.get('task')}")
self.task = environment.get('task')
self.user_id = environment.get('user_id')
self.context = environment.get('context', {})
return self
def decide(self):
"""
Decide which agents to activate
"""
self.update_state("deciding")
# Route to appropriate agents based on task
self.active_agents = []
if self.task == "generate_lesson":
self.active_agents = [self.teaching_agent, self.knowledge_agent]
self.workflow = "sequential" # Knowledge first, then teaching
elif self.task == "generate_quiz":
self.active_agents = [self.knowledge_agent, self.assessment_agent]
self.workflow = "sequential"
elif self.task == "evaluate_answer":
self.active_agents = [self.assessment_agent, self.knowledge_agent]
self.workflow = "sequential"
elif self.task == "provide_hint":
self.active_agents = [self.tutor_agent]
self.workflow = "single"
elif self.task == "recommend_topic":
self.active_agents = [self.knowledge_agent, self.recommendation_agent]
self.workflow = "sequential"
elif self.task == "full_learning_session":
# Complex workflow involving all agents
self.active_agents = [
self.knowledge_agent,
self.recommendation_agent,
self.teaching_agent,
self.assessment_agent,
self.tutor_agent
]
self.workflow = "complex"
else:
self.log(f"Unknown task: {self.task}", "warning")
self.active_agents = []
self.workflow = "none"
self.log(f"Activating {len(self.active_agents)} agents in {self.workflow} workflow")
return self
def act(self):
"""
Execute coordinated agent workflow
"""
self.update_state("acting")
self.log("Executing coordinated workflow...")
results = {}
try:
# Create standardized student state
student_state = StudentState(
user_id=self.user_id,
**self.context
)
# Fetch Topic object if ID is present
current_topic = None
if self.context.get('topic_id'):
from models import Topic
current_topic = Topic.query.get(self.context['topic_id'])
if self.workflow == "single":
# Single agent execution
agent = self.active_agents[0]
agent.perceive(student_state)
agent.decide()
# Handle specific agent signatures
if isinstance(agent, TeachingAgent):
results = agent.act(topic=current_topic)
elif isinstance(agent, AssessmentAgent):
results = agent.act(topic=current_topic)
elif isinstance(agent, TutorAgent):
results = agent.act(prompt=self.context.get('question'))
elif isinstance(agent, KnowledgeAgent):
# KnowledgeAgent act requires specific action type usually
# For general single execution, we assume a state update or status check
action = self.context.get('action_type', 'get_status')
results = agent.act(self.user_id, action_type=action, **self.context)
elif isinstance(agent, RecommendationAgent):
results = agent.act()
elif self.workflow == "sequential":
# Sequential agent execution
# Note: This generic loop is hard to maintain with diverse signatures.
# We will customize it based on the specific sequence defined in decide()
for i, agent in enumerate(self.active_agents):
self.log(f"Step {i+1}: Executing {agent.name}")
# Update state with results from previous steps
for key, val in results.items():
setattr(student_state, key, val)
agent.perceive(student_state)
agent.decide()
agent_result = {}
if isinstance(agent, TeachingAgent):
# Teaching agent needs a topic. If not in context, maybe from previous agent?
topic_to_teach = current_topic
if 'next_best' in results: # From RecommendationAgent
# Mocking topic object from result dict if needed, or fetching DB
from models import Topic
topic_to_teach = Topic.query.get(results['next_best']['topic_id'])
if topic_to_teach:
agent_result = agent.act(topic=topic_to_teach)
else:
self.log("TeachingAgent needs a topic but none provided", "error")
elif isinstance(agent, AssessmentAgent):
agent_result = agent.act(topic=current_topic)
elif isinstance(agent, KnowledgeAgent):
# Sequential usually implies updating state after something
if 'quiz_score' in self.context:
agent_result = agent.act(self.user_id, 'update_state',
topic_id=current_topic.id,
is_correct=self.context.get('is_correct'),
difficulty=current_topic.difficulty)
else:
agent_result = agent.act(self.user_id, 'get_weak_areas')
elif isinstance(agent, RecommendationAgent):
agent_result = agent.act()
results[agent.name] = agent_result
elif self.workflow == "complex":
results = self._execute_complex_workflow(student_state, current_topic)
self.tasks_coordinated += 1
self.log(f"Workflow completed successfully")
self.update_state("completed")
return {
"success": True,
"results": results,
"workflow": self.workflow
}
except Exception as e:
self.log(f"Workflow error: {str(e)}", "error")
return {"success": False, "error": str(e)}
def _execute_complex_workflow(self, student_state, current_topic):
"""
Execute complex multi-agent workflow
"""
results = {}
# Step 1: Knowledge Agent analyzes current state
self.log("Complex workflow - Step 1: Knowledge analysis")
self.knowledge_agent.perceive(student_state)
# No decide/act needed for just analysis in this flow, but let's get weak areas
knowledge_result = self.knowledge_agent.act(self.user_id, 'get_weak_areas')
results['knowledge_analysis'] = knowledge_result
# Step 2: Recommendation Agent suggests path
self.log("Complex workflow - Step 2: Path recommendation")
self.recommendation_agent.perceive(student_state)
self.recommendation_agent.decide()
recommendation_result = self.recommendation_agent.act()
results['recommendations'] = recommendation_result
# Step 3: Teaching Agent creates lesson
if recommendation_result.get('next_best'):
next_topic_data = recommendation_result['next_best']
from models import Topic
next_topic = Topic.query.get(next_topic_data['topic_id'])
self.log("Complex workflow - Step 3: Lesson generation")
# Update student state with specific topic context
student_state.topic_id = next_topic.id
student_state.knowledge_level = next_topic_data.get('current_knowledge', 0)
self.teaching_agent.perceive(student_state)
self.teaching_agent.decide()
teaching_result = self.teaching_agent.act(topic=next_topic)
results['lesson'] = teaching_result
# Step 4: Assessment Agent prepares quiz for this new topic
self.log("Complex workflow - Step 4: Assessment preparation")
self.assessment_agent.perceive(student_state)
self.assessment_agent.decide()
assessment_result = self.assessment_agent.act(topic=next_topic)
results['assessment'] = assessment_result
return results
def get_agent_status(self):
"""
Get status of all agents
"""
return {
"coordinator": self.get_statistics(),
"sub_agents": {
"teaching": self.teaching_agent.get_statistics(),
"assessment": self.assessment_agent.get_statistics(),
"knowledge": self.knowledge_agent.get_statistics(),
"tutor": self.tutor_agent.get_statistics(),
"recommendation": self.recommendation_agent.get_statistics()
},
"total_memory": sum([
len(self.teaching_agent.memory),
len(self.assessment_agent.memory),
len(self.knowledge_agent.memory),
len(self.tutor_agent.memory),
len(self.recommendation_agent.memory)
])
}
def get_statistics(self):
"""Return coordinator statistics"""
return {
"agent": self.name,
"tasks_coordinated": self.tasks_coordinated,
"state": self.state,
"active_sub_agents": len(self.active_agents) if hasattr(self, 'active_agents') else 0,
"memory_size": len(self.memory)
} |