"""OpenAI GPT service — chat completions, JSON mode, structured generation.""" import json import logging import time _logger = logging.getLogger(__name__) try: import openai as _openai_mod except ImportError: _openai_mod = None class OpenAIService: """Wraps the OpenAI Python SDK with Odoo settings and logging.""" def __init__(self, env): self.env = env self._get_param = env["ir.config_parameter"].sudo().get_param self.enabled = self._get_param("encoach_ai.enabled", "True").lower() in ("1", "true", "yes") self.max_retries = int(self._get_param("encoach_ai.max_retries", "3")) api_key = self._get_param("encoach_ai.openai_api_key", "") if not api_key: import os api_key = os.environ.get("OPENAI_API_KEY", "") if _openai_mod and api_key: self.client = _openai_mod.OpenAI(api_key=api_key) else: self.client = None self.model = self._get_param("encoach_ai.openai_model", "gpt-4o") self.fast_model = self._get_param("encoach_ai.openai_fast_model", "gpt-3.5-turbo") def _log(self, action, model, usage, latency, status="success", error=None, inp=None, out=None): try: self.env["encoach.ai.log"].sudo().create({ "service": "openai", "action": action, "model_used": model, "prompt_tokens": getattr(usage, "prompt_tokens", 0) if usage else 0, "completion_tokens": getattr(usage, "completion_tokens", 0) if usage else 0, "total_tokens": getattr(usage, "total_tokens", 0) if usage else 0, "latency_ms": latency, "status": status, "error_message": error, "input_preview": (inp or "")[:500], "output_preview": (out or "")[:500], }) except Exception: _logger.warning("Failed to log AI call", exc_info=True) def _check_enabled(self): if not self.enabled: raise RuntimeError("AI is disabled — enable in Settings > AI Configuration") def _retry_with_backoff(self, fn, action, model): """Execute fn with exponential backoff retries.""" last_exc = None for attempt in range(self.max_retries): try: return fn() except Exception as exc: last_exc = exc err_str = str(exc).lower() is_rate_limit = "rate" in err_str or "429" in err_str is_server_error = "500" in err_str or "502" in err_str or "503" in err_str if not (is_rate_limit or is_server_error) or attempt == self.max_retries - 1: raise wait = min(2 ** attempt, 16) _logger.warning("AI retry %d/%d for %s (wait %ds): %s", attempt + 1, self.max_retries, action, wait, exc) time.sleep(wait) raise last_exc def chat(self, messages, *, model=None, temperature=0.7, max_tokens=2048, action="chat"): """Standard chat completion. Returns the assistant message content string.""" self._check_enabled() if not self.client: raise RuntimeError("OpenAI not configured — set API key in AI Settings") model = model or self.model t0 = time.time() try: def _call(): return self.client.chat.completions.create( model=model, messages=messages, temperature=temperature, max_tokens=max_tokens, ) resp = self._retry_with_backoff(_call, action, model) content = resp.choices[0].message.content self._log(action, model, resp.usage, int((time.time() - t0) * 1000), inp=json.dumps(messages[-1:])[:500], out=content[:500]) return content except Exception as exc: self._log(action, model, None, int((time.time() - t0) * 1000), status="error", error=str(exc)) raise def chat_json(self, messages, *, model=None, temperature=0.3, max_tokens=4096, action="chat_json"): """Chat completion with JSON response format. Returns parsed dict/list.""" self._check_enabled() if not self.client: raise RuntimeError("OpenAI not configured — set API key in AI Settings") model = model or self.model t0 = time.time() try: def _call(): return self.client.chat.completions.create( model=model, messages=messages, temperature=temperature, max_tokens=max_tokens, response_format={"type": "json_object"}, ) resp = self._retry_with_backoff(_call, action, model) raw = resp.choices[0].message.content self._log(action, model, resp.usage, int((time.time() - t0) * 1000), inp=json.dumps(messages[-1:])[:500], out=raw[:500]) return json.loads(raw) except Exception as exc: self._log(action, model, None, int((time.time() - t0) * 1000), status="error", error=str(exc)) raise def chat_fast(self, messages, **kwargs): """Use the fast/cheap model for classification, tagging, simple tasks.""" return self.chat(messages, model=self.fast_model, **kwargs) def grade_writing(self, rubric, task_text, response_text): """Grade a writing response using GPT with a rubric.""" messages = [ {"role": "system", "content": ( "You are an expert IELTS examiner. Grade the following response using the rubric provided. " "Return JSON: {\"scores\": {\"task_achievement\": float, \"coherence_cohesion\": float, " "\"lexical_resource\": float, \"grammatical_range\": float}, " "\"overall_band\": float, \"feedback\": string, \"suggestions\": [string]}" )}, {"role": "user", "content": f"## Rubric\n{rubric}\n\n## Task\n{task_text}\n\n## Student Response\n{response_text}"}, ] return self.chat_json(messages, action="grade_writing") def grade_speaking(self, rubric, transcript): """Grade a speaking transcript using GPT.""" messages = [ {"role": "system", "content": ( "You are an expert IELTS Speaking examiner. Grade the transcript. " "Return JSON: {\"scores\": {\"fluency_coherence\": float, \"lexical_resource\": float, " "\"grammatical_range\": float, \"pronunciation\": float}, " "\"overall_band\": float, \"feedback\": string, \"suggestions\": [string]}" )}, {"role": "user", "content": f"## Rubric\n{rubric}\n\n## Transcript\n{transcript}"}, ] return self.chat_json(messages, action="grade_speaking") def generate_content(self, content_type, brief, *, cefr_level="B2"): """Generate educational content (reading passage, grammar exercise, etc.).""" messages = [ {"role": "system", "content": ( f"You are an expert EFL content creator. Generate a {content_type} " f"at CEFR {cefr_level} level. Return well-structured JSON with the content, " "questions/exercises if applicable, answer keys, and metadata." )}, {"role": "user", "content": json.dumps(brief)}, ] return self.chat_json(messages, action=f"generate_{content_type}", max_tokens=4096) def explain_grade(self, score_data, student_context=""): """Explain a grade to a student in simple terms.""" messages = [ {"role": "system", "content": ( "You are a supportive English learning coach. Explain the grade to the student " "in an encouraging way. Highlight strengths, then areas for improvement with " "concrete tips. Keep it under 200 words." )}, {"role": "user", "content": f"Score data: {json.dumps(score_data)}\nContext: {student_context}"}, ] return self.chat(messages, model=self.fast_model, action="explain_grade") def search_answer(self, query, context=""): """Answer a natural language search query about the platform.""" messages = [ {"role": "system", "content": ( "You are an intelligent assistant for the EnCoach IELTS & English learning platform. " "Answer the query based on available context. Be concise and helpful. " "Return JSON: {\"answer\": string, \"suggestions\": [string], \"related_actions\": [{\"label\": string, \"action\": string}]}" )}, {"role": "user", "content": f"Query: {query}\nContext: {context}"}, ] return self.chat_json(messages, model=self.fast_model, action="search") def generate_insights(self, data_summary, insight_type="general"): """Generate AI insights from data.""" messages = [ {"role": "system", "content": ( f"You are a data analyst for an education platform. Generate {insight_type} insights. " "Return JSON: {\"insights\": [{\"title\": string, \"description\": string, " "\"severity\": \"info\"|\"warning\"|\"critical\", \"recommendation\": string}]}" )}, {"role": "user", "content": json.dumps(data_summary)}, ] return self.chat_json(messages, model=self.fast_model, action="insights") def generate_report_narrative(self, report_type, data): """Generate a human-readable narrative for a report.""" messages = [ {"role": "system", "content": ( f"Write a concise professional narrative summary for a {report_type} report. " "2-3 paragraphs. Highlight key trends, concerns, and recommendations." )}, {"role": "user", "content": json.dumps(data)}, ] return self.chat(messages, model=self.fast_model, action="report_narrative") def suggest_study_plan(self, student_profile): """Suggest a personalized study plan.""" messages = [ {"role": "system", "content": ( "You are an IELTS preparation expert coach. Create a personalized study suggestion. " "Return JSON: {\"suggestion\": string, \"focus_areas\": [string], " "\"daily_plan\": [{\"activity\": string, \"duration_min\": int, \"skill\": string}], " "\"motivation\": string}" )}, {"role": "user", "content": json.dumps(student_profile)}, ] return self.chat_json(messages, model=self.fast_model, action="study_suggest") def writing_help(self, task, draft, help_type="improve"): """Provide writing assistance.""" messages = [ {"role": "system", "content": ( f"You are a writing tutor. Help the student {help_type} their draft. " "Return JSON: {\"improved_text\": string, \"changes\": [{\"original\": string, " "\"revised\": string, \"reason\": string}], \"tips\": [string]}" )}, {"role": "user", "content": f"Task: {task}\n\nDraft:\n{draft}"}, ] return self.chat_json(messages, action="writing_help") def batch_optimize(self, items, optimization_type="schedule"): """Optimize a batch of items (schedule, grouping, etc.).""" messages = [ {"role": "system", "content": ( f"You are an optimization specialist. Optimize these items for {optimization_type}. " "Return JSON: {\"optimized\": [items with suggested changes], \"summary\": string, \"impact\": string}" )}, {"role": "user", "content": json.dumps(items)}, ] return self.chat_json(messages, action="batch_optimize") # ── RAG-enhanced methods ───────────────────────────────────────── def _get_vector_context(self, query, *, content_types=None, limit=5): """Retrieve relevant context from the vector store.""" try: from odoo.addons.encoach_vector.services.embedding_service import EmbeddingService svc = EmbeddingService(self.env) if content_types: results = [] for ct in content_types: results.extend(svc.search(query, content_type=ct, limit=limit)) results.sort(key=lambda r: r['similarity'], reverse=True) return results[:limit] return svc.search(query, limit=limit) except Exception: _logger.debug("Vector search unavailable, proceeding without RAG", exc_info=True) return [] def _format_context(self, vector_results): """Format vector search results as context for the LLM.""" if not vector_results: return "" parts = [] for r in vector_results: text = (r.get('text') or '')[:500] meta = r.get('metadata', {}) label = f"[{r['content_type']}#{r['content_id']}]" if meta: label += f" ({', '.join(f'{k}={v}' for k, v in meta.items())})" parts.append(f"{label}\n{text}") return "\n---\n".join(parts) def chat_with_context(self, messages, query, *, content_types=None, limit=5, **kwargs): """RAG-enhanced chat: search vectors, inject context, then call GPT.""" context_results = self._get_vector_context(query, content_types=content_types, limit=limit) if context_results: context_text = self._format_context(context_results) rag_msg = { "role": "system", "content": ( "The following relevant content was found in the knowledge base. " "Use it to provide accurate, contextual answers:\n\n" + context_text ), } messages = [messages[0], rag_msg] + messages[1:] kwargs.setdefault("action", "chat_rag") return self.chat(messages, **kwargs) def search_with_rag(self, query, context=""): """RAG-enhanced search: vector search + GPT synthesis.""" vector_results = self._get_vector_context(query, limit=8) context_text = self._format_context(vector_results) messages = [ {"role": "system", "content": ( "You are an intelligent assistant for the EnCoach IELTS & English learning platform. " "Answer the query based on the knowledge base content provided below. " "Be concise, accurate, and cite specific content when possible. " "Return JSON: {\"answer\": string, \"suggestions\": [string], " "\"related_actions\": [{\"label\": string, \"action\": string}], " "\"sources\": [{\"type\": string, \"id\": number}]}" )}, ] if context_text: messages.append({"role": "system", "content": f"Knowledge base:\n{context_text}"}) if context: messages.append({"role": "system", "content": f"Additional context: {context}"}) messages.append({"role": "user", "content": f"Query: {query}"}) return self.chat_json(messages, model=self.fast_model, action="search_rag") def generate_content_dedup(self, content_type, brief, *, cefr_level="B2"): """Generate content with dedup-awareness: checks for similar existing content.""" brief_text = json.dumps(brief) if isinstance(brief, dict) else str(brief) similar = self._get_vector_context(brief_text, content_types=[content_type], limit=3) messages = [ {"role": "system", "content": ( f"You are an expert EFL content creator. Generate a {content_type} " f"at CEFR {cefr_level} level. Return well-structured JSON with the content, " "questions/exercises if applicable, answer keys, and metadata." )}, ] if similar: context_text = self._format_context(similar) messages.append({"role": "system", "content": ( "IMPORTANT: The following similar content already exists. " "Make your output DISTINCT — different angles, examples, or approaches. " "Do NOT duplicate existing content:\n\n" + context_text )}) messages.append({"role": "user", "content": brief_text}) return self.chat_json(messages, action=f"generate_{content_type}_dedup", max_tokens=4096)