from __future__ import annotations import logging from .base_agent import BaseAgent, AgentReport, SweepReport from ..tools.elearning_tools import ElearningTools logger = logging.getLogger(__name__) ELEARNING_TOOLS = [ # ── Read ────────────────────────────────────────────────────────────── {'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 rates by user', 'parameters': {'channel_id': {'type': 'integer'}, 'min_completion': {'type': 'number', 'optional': True}}}, {'name': 'get_learning_summary', 'description': 'Get overall eLearning summary', 'parameters': {}}, # ── Create / Write ───────────────────────────────────────────────────── {'name': 'create_course', 'description': 'Create a new eLearning course (slide.channel). Returns the new course id.', 'parameters': {'name': {'type': 'string'}, 'description': {'type': 'string', 'optional': True}, 'enroll_policy': {'type': 'string', 'optional': True}, 'website_published': {'type': 'boolean', 'optional': True}}}, {'name': 'update_course', 'description': 'Update name, description, or enroll_policy on an existing course', 'parameters': {'channel_id': {'type': 'integer'}, 'name': {'type': 'string', 'optional': True}, 'description': {'type': 'string', 'optional': True}, 'enroll_policy': {'type': 'string', 'optional': True}}}, {'name': 'publish_course', 'description': 'Publish a course to the website (sets website_published=True)', 'parameters': {'channel_id': {'type': 'integer'}}}, {'name': 'add_section', 'description': 'Add a section (category) to a course. Sections group slides into topics.', 'parameters': {'channel_id': {'type': 'integer'}, 'name': {'type': 'string'}, 'sequence': {'type': 'integer', 'optional': True}}}, {'name': 'create_slide', 'description': ( 'Create a slide/lesson inside a course. ' 'slide_type: document | video | infographic | webpage | quiz. ' 'Use html_content for webpage slides, url for video slides.' ), 'parameters': {'channel_id': {'type': 'integer'}, 'name': {'type': 'string'}, 'slide_type': {'type': 'string', 'optional': True}, 'sequence': {'type': 'integer', 'optional': True}, 'description': {'type': 'string', 'optional': True}, 'html_content': {'type': 'string', 'optional': True}, 'url': {'type': 'string', 'optional': True}}}, {'name': 'enroll_user', 'description': 'Enroll a partner (by partner_id) in a course', 'parameters': {'channel_id': {'type': 'integer'}, 'partner_id': {'type': 'integer'}}}, # ── Notify ───────────────────────────────────────────────────────────── {'name': 'flag_low_completion', 'description': 'Post a chatter flag on a course with low completion rate', 'parameters': {'channel_id': {'type': 'integer'}, 'reason': {'type': 'string'}}}, {'name': 'suggest_next_course', 'description': 'Suggest the next course for a learner based on completed courses', 'parameters': {'partner_id': {'type': 'integer'}}}, {'name': 'post_chatter_note', 'description': 'Post a note on any Odoo record', 'parameters': {'model': {'type': 'string'}, 'record_id': {'type': 'integer'}, 'note': {'type': 'string'}}}, ] _SYSTEM_PROMPT = """\ You are the ActiveBlue eLearning Agent with full access to Odoo 18 eLearning (website_slides module). When asked to build a course, follow this exact workflow: 1. Call create_course with the course name and a short description. Record the 'id' returned — this is the channel_id for all subsequent calls. 2. Call add_section for each major topic or module (sequence: 10, 20, 30 ...). 3. Call create_slide for each lesson within a section. Place slides after their section using sequence numbers (section at 10 → slides at 11, 12, 13 ...). Slide types: - 'document' — PDF / document lesson - 'webpage' — rich HTML lesson (provide html_content with the lesson text) - 'video' — video lesson (provide url) - 'quiz' — assessment / knowledge check 4. Call publish_course if the task requests a live/published course. 5. Call enroll_user for any specific learners mentioned in the task. Always use the actual IDs returned by tool calls — never invent IDs. After all tools are done, summarise what was created. """ class ElearningAgent(BaseAgent): name = 'elearning_agent' domain = 'elearning' required_odoo_module = 'website_slides' system_prompt_file = '' tools = ELEARNING_TOOLS def __init__(self, odoo, llm, peer_bus=None): super().__init__(odoo, llm, peer_bus) self._el = ElearningTools(odoo) self._actions_taken: list = [] # ── BaseAgent lifecycle ───────────────────────────────────────────────── async def _plan(self) -> dict: task = (self._directive.task or '').lower() create_kws = ('create', 'build', 'make', 'new course', 'set up course', 'add course') enroll_kws = ('enroll', 'register', 'sign up') return { 'intent': ( 'create' if any(k in task for k in create_kws) else 'enroll' if any(k in task for k in enroll_kws) else 'read' ), 'channel_id': self._directive.params.get('channel_id'), 'partner_id': self._directive.params.get('partner_id'), } async def _gather(self, plan: dict) -> None: self._gathered['intent'] = plan['intent'] channel_id = plan.get('channel_id') if plan['intent'] == 'read' and channel_id: self._gathered['course_stats'] = await self._el.get_course_stats(channel_id=channel_id) else: self._gathered['summary'] = await self._el.get_learning_summary() async def _reason(self) -> dict: if self._gathered.get('intent') == 'create': messages = [ {'role': 'system', 'content': _SYSTEM_PROMPT}, {'role': 'user', 'content': self._directive.task}, ] llm_output = await self._loop(messages, tools=ELEARNING_TOOLS, max_iter=20) return {'intent': 'create', 'llm_output': llm_output} summary = self._gathered.get('summary', {}) low = summary.get('low_completion_courses', []) return {'intent': 'read', 'low_completion': low} async def _act(self, reasoning: dict) -> None: if reasoning.get('intent') != 'read': return for course in reasoning.get('low_completion', [])[:3]: try: await self._el.flag_low_completion( channel_id=course['id'], reason=f'Completion rate {course.get("completion_rate", 0):.1f}% is below 30% threshold', ) self._actions_taken.append({'action': 'flag_low_completion', 'course_id': course['id']}) except Exception as exc: logger.warning('flag_low_completion failed course_id=%s: %s', course.get('id'), exc) async def _report(self) -> AgentReport: parts = [] summary = self._gathered.get('summary', {}) course_stats = self._gathered.get('course_stats', {}) 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 course_stats: ch = course_stats.get('channel', {}) parts.append(f'Course "{ch.get("name", "?")}": {course_stats.get("slide_count", 0)} slides.') creates = [a for a in self._actions_taken if 'create' in a.get('action', '')] if creates: names = ', '.join(a.get('name', f'id={a.get("id")}') for a in creates) parts.append(f'Created: {names}.') if not parts: parts.append('eLearning operation complete.') return AgentReport( directive_id=self._directive.directive_id, agent=self.name, status='complete', summary='\n'.join(parts), actions_taken=self._actions_taken, data={k: v for k, v in self._gathered.items() if not k.startswith('_')}, ) # ── Tool dispatch (_run_tool in BaseAgent calls _tool_) ────────── async def _tool_get_courses(self, active: bool = True, limit: int = 50) -> list: return await self._el.get_courses(active=active, limit=limit) async def _tool_get_course_stats(self, channel_id: int) -> dict: return await self._el.get_course_stats(channel_id=channel_id) async def _tool_get_enrolled_users(self, channel_id: int, limit: int = 100) -> list: return await self._el.get_enrolled_users(channel_id=channel_id, limit=limit) async def _tool_get_slide_completion(self, channel_id: int, min_completion: float = 0.0) -> list: return await self._el.get_slide_completion(channel_id=channel_id, min_completion=min_completion) async def _tool_get_learning_summary(self) -> dict: return await self._el.get_learning_summary() async def _tool_create_course(self, name: str, description: str = '', enroll_policy: str = 'public', website_published: bool = False) -> dict: result = await self._el.create_course(name=name, description=description, enroll_policy=enroll_policy, website_published=website_published) if result.get('success'): self._actions_taken.append({'action': 'create_course', 'id': result['id'], 'name': name}) return result async def _tool_update_course(self, channel_id: int, name: str = None, description: str = None, enroll_policy: str = None) -> dict: result = await self._el.update_course(channel_id=channel_id, name=name, description=description, enroll_policy=enroll_policy) if result.get('success'): self._actions_taken.append({'action': 'update_course', 'id': channel_id}) return result async def _tool_publish_course(self, channel_id: int) -> dict: result = await self._el.publish_course(channel_id=channel_id) if result.get('success'): self._actions_taken.append({'action': 'publish_course', 'id': channel_id}) return result async def _tool_add_section(self, channel_id: int, name: str, sequence: int = 0) -> dict: result = await self._el.add_section(channel_id=channel_id, name=name, sequence=sequence) if result.get('success'): self._actions_taken.append({'action': 'add_section', 'id': result['id'], 'name': name}) return result async def _tool_create_slide(self, channel_id: int, name: str, slide_type: str = 'document', sequence: int = 0, description: str = '', html_content: str = '', url: str = '') -> dict: result = await self._el.create_slide(channel_id=channel_id, name=name, slide_type=slide_type, sequence=sequence, description=description, html_content=html_content, url=url) if result.get('success'): self._actions_taken.append({'action': 'create_slide', 'id': result['id'], 'name': name}) return result async def _tool_enroll_user(self, channel_id: int, partner_id: int) -> dict: result = await self._el.enroll_user(channel_id=channel_id, partner_id=partner_id) if result.get('success'): self._actions_taken.append({'action': 'enroll_user', 'channel_id': channel_id, 'partner_id': partner_id}) return result async def _tool_flag_low_completion(self, channel_id: int, reason: str) -> dict: success = await self._el.flag_low_completion(channel_id=channel_id, reason=reason) return {'success': success} async def _tool_suggest_next_course(self, partner_id: int) -> list: return await self._el.suggest_next_course(partner_id=partner_id) async def _tool_post_chatter_note(self, model: str, record_id: int, note: str) -> dict: success = await self._el.post_chatter_note(model=model, record_id=record_id, note=note) return {'success': success} # ── Peer bus ──────────────────────────────────────────────────────────── async def handle_peer_request(self, request_type: str, params: dict, directive_id: str) -> dict: try: if request_type == 'learning_summary': return {'success': True, **await self._el.get_learning_summary()} if request_type == 'suggest_courses': partner_id = params.get('partner_id') if not partner_id: return {'success': False, 'error': 'partner_id required'} courses = await self._el.suggest_next_course(partner_id=partner_id) return {'success': True, 'courses': courses} return {'success': False, 'error': f'Unknown request type: {request_type}'} except Exception as exc: return {'success': False, 'error': str(exc)} # ── Sweep ─────────────────────────────────────────────────────────────── async def sweep(self) -> SweepReport: findings = [] try: summary = await self._el.get_learning_summary() for course in summary.get('low_completion_courses', []): findings.append({ 'issue': '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=[{'issue': 'sweep_failed', 'detail': str(exc), 'severity': 'low'}], ) return SweepReport(agent=self.name, findings=findings)