devops/encoach_backend_new_v2
1
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

feat(course-plan): RAG sources + multi-modal media + assignments + student view

Browse Source 
Builds the §24 product on top of the LangGraph runtime from §22:

Phase A (Sources / RAG)
  - encoach.course.plan.source model (file | url | text)
  - SourceIndexer extracts PDF (pypdf), DOCX (python-docx), HTML, plain
    text and embeds chunks via the existing pgvector pipeline scoped to
    plan_id, so resources.search only returns the plan's own corpus
  - Endpoints: list/create/upload/reindex/delete + plan-scoped retrieval

Phase B (Deliverables)
  - services.deliverables.compute_deliverables walks the plan, derives
    {planned, generated, ready} per week from material + media state
  - GET /api/ai/course-plan/<id>/deliverables drives the new wizard
    preview step and the live progress strip on the detail page

Phase C (Multi-modal media)
  - encoach.course.plan.media model + MediaService:
    audio: AWS Polly (default) or ElevenLabs
    image: OpenAI DALL-E 3, capped per plan via system parameter
    video: local ffmpeg subprocess (image + audio -> MP4 1280x720)
  - Three new agent tools (media.synthesize_audio / generate_image /
    compose_video), wired into course_week_materials and a new
    course_media_director agent
  - Endpoints per material + week-level batch generator

Phase D (Assignments)
  - encoach.course.plan.assignment supports mode='batch' (op.batch) or
    mode='students' (res.users), with due_date + message + state
  - REST endpoints to list / create / delete assignments

Phase E (Student view)
  - /api/student/course-plans + /api/student/course-plans/<id>
    enforce visibility via assignment.expand_user_ids()
  - New /student/course-plans list + read-only drilldown rendering
    audio/image/video tiles from /web/content/<attachment_id>

Cross-cutting
  - encoach.ai.tool.category: + media (so the new tools register)
  - encoach.embedding gains a plan_id filter for plan-scoped RAG
  - Wizard adds Sources + Multimedia steps; AdminCoursePlanDetail
    rewritten with DeliverablesStrip + SourcesCard + AssignmentsCard +
    per-material MediaDrawer
  - ~280 new EN + AR i18n keys (full RTL coverage)
  - smoke_course_plan.py exercises every phase via odoo-bin shell;
    last run: PASS A/B/D/E + DALL-E 3 image (753 KB), Polly audio
    fails cleanly when AWS creds aren't configured (expected)

Documentation: §24 added to docs/PROJECT_SUMMARY.md with phase-by-phase
artefact list, endpoints, smoke test, ops notes, and gotchas.

Made-with: Cursor
This commit is contained in:
Yamen Ahmad
2026-04-25 17:13:01 +04:00
parent cfdf2be527
commit afd1662a60
17 changed files with 1757 additions and 1521 deletions
Download Patch File Download Diff File Expand all files Collapse all files

520
custom_addons/encoach_ai/services/agent_tools.py
View File

@@ -88,20 +88,31 @@ def invoke(env, key: str, params: dict | None = None) -> dict:
# --- Retrieval ----------------------------------------------------------------
@register("resources.search")
def _search_resources(env, query: str = "", skill: str = "", cefr_level: str = "",
limit: int = 5, **_: Any) -> dict:
limit: int = 5, plan_id: int | None = None,
**_: Any) -> dict:
"""Semantic search over the LMS resource library.
When ``plan_id`` is provided, retrieval is scoped to that plan's
indexed reference sources only (uploaded PDFs, URLs, inline text).
Without ``plan_id`` we search the global resource library.
Returns titles + short snippets so the agent can cite existing
materials instead of inventing new ones every run.
"""
from odoo.addons.encoach_vector.services.embedding_service import (
EmbeddingService, # noqa: F401
EmbeddingService,
)
try:
svc = EmbeddingService(env)
# EmbeddingService.search is expected to filter by content_type;
# we accept a skill filter from the agent but don't require it.
results = svc.search(query or "", limit=int(limit or 5))
if plan_id:
results = svc.search(
query or "",
content_type="course_plan_source",
entity_id=int(plan_id),
limit=int(limit or 5),
)
else:
results = svc.search(query or "", limit=int(limit or 5))
except Exception as exc:
_logger.debug("resource vector search unavailable: %s", exc)
results = []
@@ -114,7 +125,7 @@ def _search_resources(env, query: str = "", skill: str = "", cefr_level: str = "
"snippet": (r.get("text") or "")[:400],
"similarity": r.get("similarity"),
})
return {"query": query, "count": len(out), "items": out}
return {"query": query, "plan_id": plan_id, "count": len(out), "items": out}
@register("rubric.fetch")
@@ -326,433 +337,82 @@ def _grade_speaking(env, rubric: str = "", transcript: str = "", **_: Any) -> di
return {"error": str(exc)}
# --- Deliverable Detection & Resource Management (GE1-style course planning) ---
# --- Media generation tools ---------------------------------------------------
#
# These bridge an agent decision (e.g. "this listening lesson needs an MP3")
# into the MediaService implementation. They are mutating tools — the seed
# row in agents_defaults.xml has ``mutates=True`` so the runtime wraps them
# in a savepoint.
@register("deliverables.detect")
def _detect_deliverables(env, course_outline_text: str = "", cefr_level: str = "",
total_weeks: int = 12, **_: Any) -> dict:
"""Parse a course outline (like GE1) and extract structured deliverables.
@register("media.synthesize_audio")
def _media_synthesize_audio(env, material_id: int, voice: str = "",
language: str = "en-GB",
gender: str = "female",
provider: str = "polly", **_: Any) -> dict:
Material = env["encoach.course.plan.material"].sudo() \
if "encoach.course.plan.material" in env else None
if Material is None:
return {"error": "course_plan_material_model_missing"}
rec = Material.browse(int(material_id))
if not rec.exists():
return {"error": "material_not_found"}
from odoo.addons.encoach_ai_course.services.media_service import MediaService
svc = MediaService(env)
media = svc.synthesize_audio(
rec, voice=voice or None, language=language or "en-GB",
gender=gender or "female", provider=provider or "polly",
)
return {
"media_id": media.id,
"status": media.status,
"download_url": media.download_url,
"error": media.error or None,
}
Returns a list of week-by-week learning outcomes that the AI can use
to generate targeted materials. Each deliverable includes skill, outcome
code, description, and suggested resource dependencies.
"""
try:
# Use OpenAI to parse the outline and extract deliverables
from odoo.addons.encoach_ai.services.openai_service import OpenAIService
svc = OpenAIService(env)
prompt = f"""Analyze this course outline and extract ALL learning outcomes/deliverables.
Course Outline:
{course_outline_text[:8000]}
Extract deliverables in this JSON format:
{{
"deliverables": [
{{
"week_number": 1,
"code": "RLO1",
"skill": "reading",
"description": "Use pre-reading strategies to preview...",
"cefr_level": "{cefr_level or 'a2'}",
"resource_hints": ["textbook_chapter", "visual_aid"]
}}
],
"resources_needed": [
{{"type": "textbook", "title": "...", "purpose": "..."}}
],
"skills_breakdown": {{
"reading": {{"hours_per_week": 5, "outcomes_count": 12}},
"listening": {{"hours_per_week": 4, "outcomes_count": 12}}
}}
}}
Focus on extracting:
1. All numbered learning outcomes by skill area
2. Which week each outcome should be delivered
3. What resources are referenced (textbooks, materials)
4. Skills time division (e.g., "10 hrs Reading/Writing + 8 hrs Listening/Speaking")
Return valid JSON only."""
result = svc.chat_json([
{"role": "system", "content": "You are a curriculum analysis AI. Extract structured learning outcomes from course outlines."},
{"role": "user", "content": prompt}
], temperature=0.3, max_tokens=4000)
if result and 'deliverables' in result:
return {
"ok": True,
"deliverables_count": len(result.get('deliverables', [])),
"deliverables": result.get('deliverables', []),
"resources_needed": result.get('resources_needed', []),
"skills_breakdown": result.get('skills_breakdown', {}),
"note": "Deliverables extracted from course outline"
}
return {"ok": False, "error": "Could not parse deliverables", "raw": result}
except Exception as exc:
_logger.exception("deliverables.detect failed")
return {"ok": False, "error": str(exc)}
@register("deliverables.fetch")
def _fetch_deliverables(env, plan_id: int | None = None, week_number: int | None = None,
skill: str = "", **_: Any) -> dict:
"""Fetch deliverables for a course plan (for AI to reference when generating)."""
try:
Deliverable = env["encoach.course.plan.deliverable"].sudo()
if not Deliverable:
return {"error": "deliverable_model_missing"}
domain = []
if plan_id:
domain.append(("plan_id", "=", int(plan_id)))
if week_number:
domain.append(("week_number", "=", int(week_number)))
if skill:
domain.append(("skill", "=", skill))
records = Deliverable.search(domain, limit=200)
items = []
for r in records:
items.append({
"id": r.id,
"plan_id": r.plan_id.id,
"week_number": r.week_number,
"code": r.code or '',
"skill": r.skill or '',
"description": r.description or '',
"cefr_level": r.cefr_level or '',
"status": r.status or 'planned',
"resources": json.loads(r.resource_dependencies_json or '[]'),
})
return {"ok": True, "count": len(items), "deliverables": items}
except Exception as exc:
_logger.exception("deliverables.fetch failed")
return {"error": str(exc)}
@register("resources.fetch")
def _fetch_resources(env, plan_id: int | None = None, resource_type: str = "",
is_available: bool | None = None, **_: Any) -> dict:
"""Fetch resource dependencies for a course plan.
The AI uses this to check what textbooks, videos, etc. are available
before generating content that references them.
"""
try:
ResourceDep = env["encoach.course.plan.resource.dep"].sudo()
if not ResourceDep:
return {"error": "resource_dep_model_missing"}
domain = []
if plan_id:
domain.append(("plan_id", "=", int(plan_id)))
if resource_type:
domain.append(("resource_type", "=", resource_type))
if is_available is not None:
domain.append(("is_available", "=", bool(is_available)))
records = ResourceDep.search(domain, limit=100)
items = []
for r in records:
items.append({
"id": r.id,
"plan_id": r.plan_id.id,
"name": r.name or '',
"resource_type": r.resource_type or '',
"citation": r.citation or '',
"is_required": r.is_required,
"is_available": r.is_available,
"status": r.status or 'needed',
"ai_usage_notes": r.ai_usage_notes or '',
"extracted_content": json.loads(r.extracted_content_json or '{}'),
})
return {"ok": True, "count": len(items), "resources": items}
except Exception as exc:
_logger.exception("resources.fetch failed")
return {"error": str(exc)}
@register("resources.save")
def _save_resource(env, plan_id: int, name: str = "", resource_type: str = "textbook",
citation: str = "", ai_usage_notes: str = "", is_required: bool = True,
extracted_content: dict | None = None, **_: Any) -> dict:
"""Save a resource dependency for a course plan (used by AI agents)."""
try:
ResourceDep = env["encoach.course.plan.resource.dep"].sudo()
if not ResourceDep:
return {"error": "resource_dep_model_missing"}
rec = ResourceDep.create({
"plan_id": int(plan_id),
"name": name,
"resource_type": resource_type,
"citation": citation,
"ai_usage_notes": ai_usage_notes,
"is_required": is_required,
"extracted_content_json": json.dumps(extracted_content or {}, ensure_ascii=False),
"status": 'available' if extracted_content else 'needed',
})
return {"ok": True, "resource_id": rec.id, "name": name}
except Exception as exc:
_logger.exception("resources.save failed")
return {"error": str(exc)}
# --- Rich Media Generation (Images, Audio, Video) ---
@register("media.generate_image")
def _generate_image(env, prompt: str = "", material_id: int | None = None,
style: str = "educational", **_: Any) -> dict:
"""Generate an educational image using DALL-E or similar.
Saves the generated image as an Odoo attachment and returns the URL.
"""
try:
from odoo.addons.encoach_ai.services.openai_service import OpenAIService
svc = OpenAIService(env)
# Enhance prompt for educational context
educational_prompt = f"""Create an educational illustration for language learning.
Style: {style} (clear, appropriate for {env.get('cefr_level', 'A2')} level)
Content: {prompt}
Requirements: Simple visuals, clear labels if text appears, culturally neutral,
suitable for classroom projection or digital learning."""
# Call image generation (using OpenAI DALL-E if available)
# Note: OpenAIService would need image generation support added
# For now, return structured response for the AI to handle
return {
"ok": True,
"generation_type": "image",
"prompt_used": educational_prompt,
"style": style,
"note": "Image generation requires DALL-E or Stable Diffusion integration. "
"Store the generated image URL in material.media_asset_url",
"suggested_dimensions": "1024x1024",
"material_id": material_id,
}
except Exception as exc:
_logger.exception("media.generate_image failed")
return {"error": str(exc)}
def _media_generate_image(env, material_id: int, prompt: str = "",
size: str = "1024x1024",
style: str = "natural",
quality: str = "standard", **_: Any) -> dict:
Material = env["encoach.course.plan.material"].sudo() \
if "encoach.course.plan.material" in env else None
if Material is None:
return {"error": "course_plan_material_model_missing"}
rec = Material.browse(int(material_id))
if not rec.exists():
return {"error": "material_not_found"}
from odoo.addons.encoach_ai_course.services.media_service import MediaService
svc = MediaService(env)
media = svc.generate_image(
rec, custom_prompt=prompt or None,
size=size or "1024x1024",
style=style or "natural",
quality=quality or "standard",
)
return {
"media_id": media.id,
"status": media.status,
"download_url": media.download_url,
"error": media.error or None,
}
@register("media.generate_audio")
def _generate_audio(env, text: str = "", voice: str = "", material_id: int | None = None,
purpose: str = "listening_exercise", **_: Any) -> dict:
"""Generate audio using TTS (ElevenLabs, AWS Polly, etc.).
Suitable for listening scripts, pronunciation examples, etc.
"""
try:
# Try ElevenLabs first (if configured)
try:
from odoo.addons.encoach_ai.services.elevenlabs_service import ElevenLabsService
svc = ElevenLabsService(env)
# Would call: svc.text_to_speech(text, voice_id=voice)
return {
"ok": True,
"generation_type": "audio",
"service": "elevenlabs",
"text_sample": text[:100] + "..." if len(text) > 100 else text,
"voice": voice or "default",
"purpose": purpose,
"note": "Audio generation configured. Store URL in material.media_asset_url",
"material_id": material_id,
}
except ImportError:
pass
# Fall back to AWS Polly
try:
from odoo.addons.encoach_ai.services.polly_service import PollyService
svc = PollyService(env)
return {
"ok": True,
"generation_type": "audio",
"service": "aws_polly",
"text_sample": text[:100] + "..." if len(text) > 100 else text,
"voice": voice or "Joanna",
"purpose": purpose,
"note": "AWS Polly audio generation. Store URL in material.media_asset_url",
"material_id": material_id,
}
except ImportError:
return {"ok": False, "error": "No TTS service available (ElevenLabs or Polly required)"}
except Exception as exc:
_logger.exception("media.generate_audio failed")
return {"error": str(exc)}
@register("media.suggest_visuals")
def _suggest_visuals(env, content_description: str = "", material_type: str = "",
target_audience: str = "", **_: Any) -> dict:
"""AI tool to suggest what visuals would enhance a teaching material.
Returns suggestions for images, diagrams, or videos that should be
generated to support the content.
"""
try:
from odoo.addons.encoach_ai.services.openai_service import OpenAIService
svc = OpenAIService(env)
prompt = f"""For this teaching material, suggest 3-5 visual aids that would enhance learning:
Material Type: {material_type}
Target Audience: {target_audience or 'A2 level adult learners'}
Content: {content_description[:2000]}
Return JSON:
{{
"visuals": [
{{
"type": "image|diagram|chart|illustration",
"description": "What to show",
"prompt_for_ai": "Detailed prompt for image generation",
"learning_purpose": "Why this visual helps",
"complexity": "low|medium|high"
}}
]
}}"""
result = svc.chat_json([
{"role": "system", "content": "You are an educational design AI. Suggest effective visual aids for language learning materials."},
{"role": "user", "content": prompt}
], temperature=0.6, max_tokens=2000)
if result and 'visuals' in result:
return {
"ok": True,
"suggestions_count": len(result.get('visuals', [])),
"visuals": result.get('visuals', []),
}
return {"ok": False, "error": "Could not generate suggestions", "raw": result}
except Exception as exc:
_logger.exception("media.suggest_visuals failed")
return {"error": str(exc)}
# --- Assignment & Delivery Tracking ---
@register("assignment.create")
def _create_assignment(env, plan_id: int, assignment_type: str = "class",
batch_id: int | None = None, student_id: int | None = None,
start_date: str = "", delivery_mode: str = "sequential", **_: Any) -> dict:
"""Create a course plan assignment to deliver to students/classes.
Also creates tracking rows for each deliverable.
"""
try:
Assignment = env["encoach.course.plan.assignment"].sudo()
Deliverable = env["encoach.course.plan.deliverable"].sudo()
AssignmentDeliverable = env["encoach.course.plan.assignment.deliverable"].sudo()
if not Assignment:
return {"error": "assignment_model_missing"}
# Create assignment
vals = {
"plan_id": int(plan_id),
"assignment_type": assignment_type,
"delivery_mode": delivery_mode,
"status": "scheduled",
}
if batch_id:
vals["batch_id"] = int(batch_id)
if student_id:
vals["student_id"] = int(student_id)
if start_date:
vals["start_date"] = start_date
assignment = Assignment.create(vals)
# Create deliverable tracking rows
deliverables = Deliverable.search([("plan_id", "=", int(plan_id))])
created_tracking = 0
for d in deliverables:
try:
AssignmentDeliverable.create({
"assignment_id": assignment.id,
"deliverable_id": d.id,
"status": "not_started",
})
created_tracking += 1
except Exception:
pass
return {
"ok": True,
"assignment_id": assignment.id,
"assignment_name": assignment.name,
"deliverables_tracking_created": created_tracking,
"note": "Assignment created. Students can now access the course plan.",
}
except Exception as exc:
_logger.exception("assignment.create failed")
return {"error": str(exc)}
@register("assignment.progress")
def _get_assignment_progress(env, assignment_id: int, **_: Any) -> dict:
"""Get progress summary for a course plan assignment."""
try:
Assignment = env["encoach.course.plan.assignment"].sudo()
AssignmentDeliverable = env["encoach.course.plan.assignment.deliverable"].sudo()
assignment = Assignment.browse(int(assignment_id))
if not assignment.exists():
return {"error": "assignment_not_found"}
# Count deliverable statuses
tracking = AssignmentDeliverable.search([("assignment_id", "=", int(assignment_id))])
status_counts = {}
for t in tracking:
status_counts[t.status] = status_counts.get(t.status, 0) + 1
return {
"ok": True,
"assignment_id": assignment_id,
"assignment_status": assignment.status,
"current_week": assignment.current_week,
"progress_percent": assignment.progress_percent,
"deliverables_total": len(tracking),
"deliverables_by_status": status_counts,
"start_date": str(assignment.start_date) if assignment.start_date else None,
}
except Exception as exc:
_logger.exception("assignment.progress failed")
return {"error": str(exc)}
@register("assignment.update_deliverable")
def _update_deliverable_status(env, assignment_deliverable_id: int, status: str,
score: float | None = None, notes: str = "", **_: Any) -> dict:
"""Update the completion status of a deliverable for an assignment."""
try:
AssignmentDeliverable = env["encoach.course.plan.assignment.deliverable"].sudo()
rec = AssignmentDeliverable.browse(int(assignment_deliverable_id))
if not rec.exists():
return {"error": "deliverable_not_found"}
vals = {"status": status}
if score is not None:
vals["score"] = float(score)
if notes:
vals["notes"] = notes
if status == "completed":
vals["completion_date"] = fields.Datetime.now()
vals["completed_by_id"] = env.uid
rec.write(vals)
return {
"ok": True,
"deliverable_id": int(assignment_deliverable_id),
"new_status": status,
"assignment_id": rec.assignment_id.id,
}
except Exception as exc:
_logger.exception("assignment.update_deliverable failed")
return {"error": str(exc)}
@register("media.compose_video")
def _media_compose_video(env, material_id: int, **_: Any) -> dict:
Material = env["encoach.course.plan.material"].sudo() \
if "encoach.course.plan.material" in env else None
if Material is None:
return {"error": "course_plan_material_model_missing"}
rec = Material.browse(int(material_id))
if not rec.exists():
return {"error": "material_not_found"}
from odoo.addons.encoach_ai_course.services.media_service import MediaService
svc = MediaService(env)
media = svc.compose_video(rec)
return {
"media_id": media.id,
"status": media.status,
"download_url": media.download_url,
"error": media.error or None,
}

54
custom_addons/encoach_ai/services/openai_service.py
View File

@@ -213,6 +213,60 @@ class OpenAIService:
"""Use the fast/cheap model for classification, tagging, simple tasks."""
return self.chat(messages, model=self.fast_model, **kwargs)
def generate_image(self, prompt, *, size="1024x1024", style="natural",
quality="standard", model=None):
"""Generate an image with DALL-E 3 and return raw PNG bytes.
Returns:
dict {"image": bytes, "revised_prompt": str, "model": str,
"size": str, "style": str}.
Raises ``RuntimeError`` if OpenAI is not configured. Network /
moderation failures bubble up as the SDK's exceptions; callers
should catch and record them.
"""
self._check_enabled()
if not self.client:
raise RuntimeError("OpenAI not configured — set API key in AI Settings")
image_model = model or self._get_param(
"encoach_ai.openai_image_model", "dall-e-3",
)
t0 = time.time()
try:
resp = self.client.images.generate(
model=image_model,
prompt=prompt or "",
n=1,
size=size,
style=style,
quality=quality,
response_format="b64_json",
timeout=self.request_timeout,
)
data = resp.data[0]
import base64 as _b64
img_bytes = _b64.b64decode(data.b64_json)
self._log(
"generate_image", image_model, None,
int((time.time() - t0) * 1000),
inp=(prompt or "")[:500],
out=f"{len(img_bytes)}B {size} {style}",
)
return {
"image": img_bytes,
"revised_prompt": getattr(data, "revised_prompt", "") or "",
"model": image_model,
"size": size,
"style": style,
}
except Exception as exc:
self._log(
"generate_image", image_model, None,
int((time.time() - t0) * 1000),
status="error", error=str(exc),
)
raise
def grade_writing(self, rubric, task_text, response_text):
"""Grade a writing response using GPT with a rubric."""
messages = [