odoo-ai/agent_service/agents/elearning_agent.py

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.')