from __future__ import annotations import logging from .base_agent import BaseAgent, AgentReport, AgentDirective, SweepReport from ..tools.elearning_tools import ElearningTools logger = logging.getLogger(__name__) ELEARNING_TOOLS = [ {'name': 'get_courses', 'description': 'List eLearning courses', 'parameters': {'active': {'type': 'boolean', 'optional': True}, 'limit': {'type': 'integer', 'optional': True}}}, {'name': 'get_course_stats', 'description': 'Get detailed stats for a course', 'parameters': {'channel_id': {'type': 'integer'}}}, {'name': 'get_enrolled_users', 'description': 'Get users enrolled in a course', 'parameters': {'channel_id': {'type': 'integer'}, 'limit': {'type': 'integer', 'optional': True}}}, {'name': 'get_slide_completion', 'description': 'Get slide completion by user', 'parameters': {'channel_id': {'type': 'integer'}, 'min_completion': {'type': 'number', 'optional': True}}}, {'name': 'get_learning_summary', 'description': 'Get overall learning summary', 'parameters': {}}, {'name': 'flag_low_completion', 'description': 'Flag a course with low completion', 'parameters': {'channel_id': {'type': 'integer'}, 'reason': {'type': 'string'}}}, {'name': 'suggest_next_course', 'description': 'Suggest next course for a learner', 'parameters': {'partner_id': {'type': 'integer'}}}, {'name': 'post_chatter_note', 'description': 'Post a note on a record', 'parameters': {'model': {'type': 'string'}, 'record_id': {'type': 'integer'}, 'note': {'type': 'string'}}}, ] class ElearningAgent(BaseAgent): name = 'elearning_agent' domain = 'elearning' required_odoo_module = 'website_slides' system_prompt_file = 'elearning_system.txt' tools = ELEARNING_TOOLS def __init__(self, odoo, llm, peer_bus=None): super().__init__(odoo, llm, peer_bus) self._el = ElearningTools(odoo) self._gathered_data = {} self._actions_taken = [] self._escalations_list = [] async def _plan(self, directive: AgentDirective) -> dict: intent = (directive.intent or '').lower() return { 'fetch_summary': any(k in intent for k in ('summary', 'overview', 'learning')), 'fetch_courses': 'course' in intent, 'channel_id': directive.context.get('channel_id'), 'partner_id': directive.context.get('partner_id'), } async def _gather(self, ctx: dict) -> dict: plan = ctx.get('plan', {}) data: dict = {} data['summary'] = await self._el.get_learning_summary() if plan.get('fetch_courses') or plan.get('channel_id'): if plan.get('channel_id'): data['course_stats'] = await self._el.get_course_stats(channel_id=plan['channel_id']) else: data['courses'] = await self._el.get_courses(limit=20) self._gathered_data = data return data async def _reason(self, ctx: dict) -> dict: data = self._gathered_data analysis: dict = {'escalations': [], 'low_completion': []} summary = data.get('summary', {}) low = summary.get('low_completion_courses', []) analysis['low_completion'] = low if len(low) > 3: analysis['escalations'].append(f'{len(low)} courses have <30% completion rate.') self._escalations_list = analysis['escalations'] return analysis async def _act(self, ctx: dict) -> list: actions = [] analysis = ctx.get('analysis', {}) for course in analysis.get('low_completion', [])[:3]: try: await self._el.flag_low_completion( channel_id=course.get('id'), reason=f'Completion rate {course.get("completion_rate", 0):.1f}% is below 30% threshold', ) actions.append({'action': 'flag_low_completion', 'course_id': course.get('id'), 'success': True}) except Exception as exc: logger.warning('flag_low_completion failed: %s', exc) self._actions_taken = actions return actions async def _report(self, ctx: dict) -> AgentReport: data = self._gathered_data summary = data.get('summary', {}) parts = [] if summary: parts.append( f'eLearning: {summary.get("total_courses", 0)} courses, ' f'{summary.get("total_enrollments", 0)} enrollments, ' f'{summary.get("avg_completion", 0):.1f}% avg completion.' ) if not parts: parts.append('eLearning review complete.') return AgentReport(agent=self.name, summary=chr(10).join(parts), data=data, escalations=self._escalations_list, actions_taken=self._actions_taken) async def _dispatch_tool(self, name: str, args: dict): dispatch = { 'get_courses': self._el.get_courses, 'get_course_stats': self._el.get_course_stats, 'get_enrolled_users': self._el.get_enrolled_users, 'get_slide_completion': self._el.get_slide_completion, 'get_learning_summary': self._el.get_learning_summary, 'flag_low_completion': self._el.flag_low_completion, 'suggest_next_course': self._el.suggest_next_course, 'post_chatter_note': self._el.post_chatter_note, } if name not in dispatch: raise ValueError(f'Unknown tool: {name}') return await dispatch[name](**args) async def handle_peer_request(self, request: dict) -> dict: req_type = request.get('type', '') try: if req_type == 'learning_summary': return await self._el.get_learning_summary() if req_type == 'suggest_courses': return {'courses': await self._el.suggest_next_course(partner_id=request['partner_id'])} return {'error': f'Unknown type: {req_type}'} except Exception as exc: return {'error': str(exc)} async def sweep(self) -> SweepReport: findings = [] try: summary = await self._el.get_learning_summary() for course in summary.get('low_completion_courses', []): findings.append({'type': 'low_completion', 'course_id': course.get('id'), 'name': course.get('name'), 'completion': course.get('completion_rate', 0), 'severity': 'medium'}) except Exception as exc: return SweepReport(agent=self.name, findings=[], actions=[], error=str(exc)) return SweepReport(agent=self.name, findings=findings, actions=[], summary=f'eLearning sweep: {len(findings)} low-completion courses.')