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

feat(ai): LangGraph as core runtime + AI Agents/Tools console + full-demo seed
Some checks failed
CI / Frontend — lint + build + e2e (pull_request) Failing after 1m20s
Details
CI / Backend — Odoo HttpCase (pull_request) Failing after 1s
Details

Browse Source 
Core AI runtime
- New encoach.ai.agent + encoach.ai.tool models with M2M tool binding,
  graph topology (simple|plan_review_revise|rag|react), model + fallback,
  temperature, max_tokens, response_format, max_revisions, quality checks
  and system prompt fields.
- services/agent_runtime.py compiles a langgraph.StateGraph per agent
  and caches the build per (key, write_date). Emits a structured trace
  (output, tool_calls, retrieval_hits, revisions, quality_issues,
  ms, model_used, fallback_used) and auto-falls-back on rate-limit/5xx.
- services/agent_tools.py registers 11 tool handlers wrapping existing
  services: resources.search, rubric.fetch, outcomes.fetch,
  student.profile, quality.cefr_check, quality.ai_detect,
  quality.content_gate, course_plan.save (mutates),
  course_plan.save_materials (mutates), scoring.grade_writing,
  scoring.grade_speaking.
- 7 default agents seeded via data/agents_defaults.xml: course_planner,
  course_week_materials, exam_generator, exercise_generator, lms_tutor,
  writing_grader, speaking_grader.
- Feature flag encoach_ai.use_langgraph_runtime (default True).
- encoach_ai_course pipeline now routes through AgentRuntime when on,
  legacy SDK path kept as fallback.

Admin UI
- /admin/ai/prompts is now a tabbed Agents | Tools | Prompts console.
- AIAgentsPanel: card grid + config dialog (model/temp/graph/tools/
  system prompt) + built-in Test Runner showing live trace.
- AIToolsPanel: registry table with category badges, mutates flag,
  schema viewer, edit dialog.
- New /api/ai/agents* and /api/ai/tools* controller (list/get/update/
  test, list-tools, toggle-tool).
- Sidebar label nav.aiPrompts -> nav.aiAgents (AI Agents and Tools).
- EN + AR (RTL) translations for ~80 new keys.

Smart Wizard pages
- /admin/quick-setup hub + CourseWizard, CoursePlanWizard,
  RubricWizard, ExamStructureWizard step-by-step flows.
- /admin/course-plans list + detail pages.
- /teacher/quick-setup mirror.

Full demo seed + 8-role E2E
- seed_full_demo.py adds the 5 missing user_types (approver, corporate,
  mastercorporate, agent, developer), activates a 2-stage exam-approval
  workflow with one pending request, creates a GE1-aligned 12-week B1
  course plan with 6 detailed Week-1 materials (reading 400w, writing,
  listening 4-min script, speaking, grammar present simple vs continuous,
  vocabulary), and inserts sample ai.log + ai.feedback rows.
- reset_demo_passwords.py forces every demo login back to canonical
  passwords (admin123/teacher123/student123/approver123/corporate123/
  master123/agent123/dev123).
- e2e_full_scenario.py: 46/46 PASS read-only API smoke across all
  8 roles, including a live LangGraph round-trip on writing_grader.
- e2e_approval_chain.py: 6/6 PASS mutation E2E - approver approves
  stage 1, admin approves stage 2, linked encoach.exam.custom flips
  to status=published, verified via psql.

Docs
- docs/PROJECT_SUMMARY.md updated to 2026-04-25: new Latest events
  bullets, refreshed credentials table, full sections 22 (LangGraph
  runtime) and 23 (full demo seed + 8-role E2E).
- docs/ENCOACH_FULL_DEMO_QA_REPORT.md added with credentials,
  per-endpoint PASS/FAIL, mutation chain proof, LangGraph live output.
- backend/GE1 Course Outline_ Fall AY25-26.pdf vendored as the
  reference outline the GE1 plan/materials are aligned to.

Dependencies
- requirements.txt: langgraph>=0.2.0, langchain-core>=0.3.0.
- encoach_ai/__manifest__.py: external_dependencies updated.

Made-with: Cursor
This commit is contained in:
Yamen Ahmad
2026-04-25 03:13:55 +04:00
parent 1223074bde
commit e2aa8031ff
56 changed files with 9846 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
backend/custom_addons/encoach_ai/__manifest__.py
View File

@@ -17,12 +17,13 @@
"author": "EnCoach",
"depends": ["base", "encoach_core", "encoach_api"],
"external_dependencies": {
"python": ["openai", "boto3"],
"python": ["openai", "boto3", "langgraph", "langchain_core"],
},
"data": [
"security/ir.model.access.csv",
"views/ai_settings_views.xml",
"data/ai_defaults.xml",
"data/agents_defaults.xml",
],
"installable": True,
"application": True,

1
backend/custom_addons/encoach_ai/controllers/__init__.py
View File

@@ -3,3 +3,4 @@ from . import coach_controller
from . import media_controller
from . import prompt_controller
from . import feedback_controller
from . import agents_controller

244
backend/custom_addons/encoach_ai/controllers/agents_controller.py Normal file
View File

@@ -0,0 +1,244 @@
"""Admin endpoints for configuring and exercising AI agents.
Design notes:
* Read endpoints require a valid JWT but not admin. The "AI Agents"
tab needs to be reachable by anyone who can see ``/admin/ai/prompts``
today (analysts, teachers auditing prompt changes, etc.).
* Write endpoints — ``PATCH /api/ai/agents/<id>`` and
``POST /api/ai/agents/<id>/test`` — additionally require admin
privileges (``base.group_system``), matching the existing prompt
controller's policy.
* ``/test`` is deliberately synchronous and uncached: admins use it to
quickly verify a config change produces sane output. It caps the
LLM at 500 tokens to keep iteration cheap.
"""
from __future__ import annotations
import json
import logging
from odoo import http
from odoo.http import request
from odoo.addons.encoach_api.controllers.base import (
_error_response,
_get_json_body,
_json_response,
jwt_required,
)
_logger = logging.getLogger(__name__)
def _require_admin():
if not request.env.user.has_group("base.group_system"):
return _error_response("Admin privileges required", 403)
return None
class EncoachAIAgentsController(http.Controller):
# ------------------------------------------------------------------
# GET /api/ai/agents
# ------------------------------------------------------------------
@http.route("/api/ai/agents", type="http", auth="none", methods=["GET"], csrf=False)
@jwt_required
def list_agents(self, **kw):
try:
search = (kw.get("search") or "").strip()
domain = []
if search:
domain = [
"|", "|",
("key", "ilike", search),
("name", "ilike", search),
("description", "ilike", search),
]
Agent = request.env["encoach.ai.agent"].sudo()
records = Agent.search(domain, order="sequence, name")
items = [r.to_api_dict(include_prompt=False) for r in records]
return _json_response({
"items": items,
"data": items,
"total": len(items),
})
except Exception as exc:
_logger.exception("list agents failed")
return _error_response(str(exc), 500)
# ------------------------------------------------------------------
# GET /api/ai/agents/<id>
# ------------------------------------------------------------------
@http.route(
"/api/ai/agents/<int:agent_id>",
type="http", auth="none", methods=["GET"], csrf=False,
)
@jwt_required
def get_agent(self, agent_id, **kw):
try:
agent = request.env["encoach.ai.agent"].sudo().browse(int(agent_id))
if not agent.exists():
return _error_response("Agent not found", 404)
data = agent.to_api_dict(include_prompt=True)
data["tools"] = [t.to_api_dict() for t in agent.tool_ids]
return _json_response(data)
except Exception as exc:
_logger.exception("get agent failed")
return _error_response(str(exc), 500)
# ------------------------------------------------------------------
# PATCH /api/ai/agents/<id> (admin-only)
# ------------------------------------------------------------------
@http.route(
"/api/ai/agents/<int:agent_id>",
type="http", auth="none", methods=["PATCH", "PUT"], csrf=False,
)
@jwt_required
def update_agent(self, agent_id, **kw):
err = _require_admin()
if err is not None:
return err
try:
agent = request.env["encoach.ai.agent"].sudo().browse(int(agent_id))
if not agent.exists():
return _error_response("Agent not found", 404)
body = _get_json_body() or {}
vals: dict = {}
# Whitelist every settable field so callers can't flip `active` or
# rewrite `key` without knowing they're allowed to.
for f in (
"name", "description", "system_prompt", "prompt_key",
"model", "fallback_model", "response_format",
"graph_type", "quality_checks",
):
if f in body:
vals[f] = body[f] or ""
for f in ("temperature",):
if f in body:
try:
vals[f] = float(body[f])
except (TypeError, ValueError):
pass
for f in ("max_tokens", "max_revisions", "sequence"):
if f in body:
try:
vals[f] = int(body[f])
except (TypeError, ValueError):
pass
if "active" in body:
vals["active"] = bool(body["active"])
if "tool_keys" in body and isinstance(body["tool_keys"], list):
tool_ids = request.env["encoach.ai.tool"].sudo().search(
[("key", "in", [str(k) for k in body["tool_keys"]])]
).ids
vals["tool_ids"] = [(6, 0, tool_ids)]
with request.env.cr.savepoint():
agent.write(vals)
return _json_response(agent.to_api_dict(include_prompt=True))
except Exception as exc:
_logger.exception("update agent failed")
return _error_response(str(exc), 400)
# ------------------------------------------------------------------
# POST /api/ai/agents/<id>/test (admin-only)
# ------------------------------------------------------------------
@http.route(
"/api/ai/agents/<int:agent_id>/test",
type="http", auth="none", methods=["POST"], csrf=False,
)
@jwt_required
def test_agent(self, agent_id, **kw):
err = _require_admin()
if err is not None:
return err
try:
agent = request.env["encoach.ai.agent"].sudo().browse(int(agent_id))
if not agent.exists():
return _error_response("Agent not found", 404)
body = _get_json_body() or {}
variables = body.get("variables") or {}
payload = body.get("payload")
language = body.get("language") or request.env.user.lang or "en"
from odoo.addons.encoach_ai.services.agent_runtime import AgentRuntime
runtime = AgentRuntime(request.env, agent, language=language)
# Small-budget test: cap max_tokens so iteration stays cheap.
original_max = agent.max_tokens
if original_max > 800:
agent.sudo().write({"max_tokens": 800})
try:
final = runtime.invoke(variables=variables, payload=payload)
finally:
if agent.max_tokens != original_max:
agent.sudo().write({"max_tokens": original_max})
output = final.get("output")
return _json_response({
"error": final.get("error") or "",
"output": output,
"output_raw": (final.get("output_raw") or "")[:6000],
"tool_results": (final.get("tool_results") or [])[:20],
"retrieval_hits": len(final.get("retrieval") or []),
"revisions_used": final.get("revisions_used") or 0,
"quality_issues": final.get("quality_issues") or [],
"iterations": final.get("iterations") or 0,
})
except Exception as exc:
_logger.exception("test agent failed")
return _error_response(str(exc), 500)
# ------------------------------------------------------------------
# GET /api/ai/agents/tools
# ------------------------------------------------------------------
@http.route(
"/api/ai/agents/tools",
type="http", auth="none", methods=["GET"], csrf=False,
)
@jwt_required
def list_tools(self, **kw):
try:
tools = request.env["encoach.ai.tool"].sudo().search([], order="category, sequence, name")
items = [t.to_api_dict() for t in tools]
return _json_response({"items": items, "data": items, "total": len(items)})
except Exception as exc:
_logger.exception("list tools failed")
return _error_response(str(exc), 500)
# ------------------------------------------------------------------
# PATCH /api/ai/agents/tools/<id> (admin-only; currently toggle active)
# ------------------------------------------------------------------
@http.route(
"/api/ai/agents/tools/<int:tool_id>",
type="http", auth="none", methods=["PATCH", "PUT"], csrf=False,
)
@jwt_required
def update_tool(self, tool_id, **kw):
err = _require_admin()
if err is not None:
return err
try:
tool = request.env["encoach.ai.tool"].sudo().browse(int(tool_id))
if not tool.exists():
return _error_response("Tool not found", 404)
body = _get_json_body() or {}
vals: dict = {}
if "active" in body:
vals["active"] = bool(body["active"])
for f in ("name", "description", "category"):
if f in body:
vals[f] = body[f] or ""
if "schema" in body:
# Accept a parsed dict OR raw JSON string.
raw = body["schema"]
if isinstance(raw, (dict, list)):
vals["schema_json"] = json.dumps(raw)
else:
vals["schema_json"] = str(raw)
with request.env.cr.savepoint():
tool.write(vals)
return _json_response(tool.to_api_dict())
except Exception as exc:
_logger.exception("update tool failed")
return _error_response(str(exc), 400)

337
backend/custom_addons/encoach_ai/data/agents_defaults.xml Normal file
View File

@@ -0,0 +1,337 @@
<?xml version="1.0" encoding="UTF-8"?>
<odoo noupdate="1">
<!--
Default AI agents + tools seeded on first install.
These are the *sensible defaults* the user asked for: every platform
pillar (course planning, weekly materials, exam generation, exercise
generation, LMS tutor, grading) gets a pre-configured LangGraph
agent so the system works out of the box. Admins edit the system
prompts, models, temperatures and tool bindings from
/admin/ai/prompts → Agents tab.
-->
<!-- ============================== TOOLS ============================== -->
<!-- Retrieval -->
<record id="ai_tool_resources_search" model="encoach.ai.tool">
<field name="key">resources.search</field>
<field name="name">Search resources</field>
<field name="category">retrieval</field>
<field name="description">Semantic search over the LMS resource library. Returns resource ids, titles and snippets. Use this BEFORE generating content so the agent reuses existing, approved materials instead of hallucinating.</field>
<field name="schema_json">{"type":"object","properties":{"query":{"type":"string","description":"Natural language search query"},"skill":{"type":"string","enum":["reading","writing","listening","speaking","grammar","vocabulary"]},"cefr_level":{"type":"string","enum":["pre_a1","a1","a2","b1","b2","c1","c2"]},"limit":{"type":"integer","default":5,"minimum":1,"maximum":20}},"required":["query"]}</field>
<field name="sequence">10</field>
</record>
<record id="ai_tool_rubric_fetch" model="encoach.ai.tool">
<field name="key">rubric.fetch</field>
<field name="name">Fetch rubric</field>
<field name="category">reference</field>
<field name="description">Return the grading rubric and criterion descriptors for a given rubric id or skill. Always call before grading so the LLM uses the approved rubric, not its own defaults.</field>
<field name="schema_json">{"type":"object","properties":{"rubric_id":{"type":"integer"},"skill":{"type":"string","enum":["reading","writing","listening","speaking"]}}}</field>
<field name="sequence">20</field>
</record>
<record id="ai_tool_outcomes_fetch" model="encoach.ai.tool">
<field name="key">outcomes.fetch</field>
<field name="name">Fetch course outcomes</field>
<field name="category">reference</field>
<field name="description">Return the registered learning outcomes for a course or CEFR level. Use it when generating course plans to stay aligned with the programme specification.</field>
<field name="schema_json">{"type":"object","properties":{"course_id":{"type":"integer"},"cefr_level":{"type":"string","enum":["pre_a1","a1","a2","b1","b2","c1","c2"]}}}</field>
<field name="sequence">30</field>
</record>
<record id="ai_tool_student_profile" model="encoach.ai.tool">
<field name="key">student.profile</field>
<field name="name">Get student gap profile</field>
<field name="category">reference</field>
<field name="description">Return the student's CEFR band, strengths and gaps so content can be personalised. Required input for personalised exercise generation and tutor follow-ups.</field>
<field name="schema_json">{"type":"object","properties":{"student_id":{"type":"integer"}},"required":["student_id"]}</field>
<field name="sequence">40</field>
</record>
<!-- Quality gates -->
<record id="ai_tool_quality_cefr" model="encoach.ai.tool">
<field name="key">quality.cefr_check</field>
<field name="name">CEFR readability check</field>
<field name="category">quality</field>
<field name="description">Check whether a text reads at the target CEFR level using Flesch-Kincaid. Returns ok=false with specific issues when the passage is too easy or too hard for the requested band.</field>
<field name="schema_json">{"type":"object","properties":{"text":{"type":"string"},"target_cefr":{"type":"string","enum":["a1","a2","b1","b2","c1","c2"]}},"required":["text","target_cefr"]}</field>
<field name="sequence">50</field>
</record>
<record id="ai_tool_quality_ai" model="encoach.ai.tool">
<field name="key">quality.ai_detect</field>
<field name="name">AI-content detection</field>
<field name="category">quality</field>
<field name="description">Probability the text was written by an AI (via GPTZero). Used during submission review — not usually during generation.</field>
<field name="schema_json">{"type":"object","properties":{"text":{"type":"string"}},"required":["text"]}</field>
<field name="sequence">60</field>
</record>
<record id="ai_tool_quality_gate" model="encoach.ai.tool">
<field name="key">quality.content_gate</field>
<field name="name">Unified content gate</field>
<field name="category">quality</field>
<field name="description">Run the project's combined content-source gate (CEFR + toxicity + length checks). Returns ok=false with the first failing rule.</field>
<field name="schema_json">{"type":"object","properties":{"text":{"type":"string"},"cefr_level":{"type":"string"}},"required":["text"]}</field>
<field name="sequence">70</field>
</record>
<!-- Persistence -->
<record id="ai_tool_course_plan_save" model="encoach.ai.tool">
<field name="key">course_plan.save</field>
<field name="name">Save course plan</field>
<field name="category">persistence</field>
<field name="mutates" eval="True"/>
<field name="description">Persist an AI-generated course plan header and its weekly rows. Only call once you're confident the JSON is valid and has been reviewed.</field>
<field name="schema_json">{"type":"object","properties":{"plan_vals":{"type":"object"},"weeks":{"type":"array","items":{"type":"object"}}},"required":["plan_vals"]}</field>
<field name="sequence">80</field>
</record>
<record id="ai_tool_course_plan_save_materials" model="encoach.ai.tool">
<field name="key">course_plan.save_materials</field>
<field name="name">Save weekly teaching materials</field>
<field name="category">persistence</field>
<field name="mutates" eval="True"/>
<field name="description">Persist the generated per-week teaching materials against an existing course plan and week.</field>
<field name="schema_json">{"type":"object","properties":{"plan_id":{"type":"integer"},"week_id":{"type":"integer"},"materials":{"type":"array","items":{"type":"object"}}},"required":["plan_id","week_id","materials"]}</field>
<field name="sequence">90</field>
</record>
<!-- Scoring -->
<record id="ai_tool_scoring_writing" model="encoach.ai.tool">
<field name="key">scoring.grade_writing</field>
<field name="name">Grade writing response</field>
<field name="category">scoring</field>
<field name="description">Grade a writing response against a rubric using the platform's standard writing examiner prompt.</field>
<field name="schema_json">{"type":"object","properties":{"rubric":{"type":"string"},"task":{"type":"string"},"response":{"type":"string"}},"required":["rubric","response"]}</field>
<field name="sequence">100</field>
</record>
<record id="ai_tool_scoring_speaking" model="encoach.ai.tool">
<field name="key">scoring.grade_speaking</field>
<field name="name">Grade speaking transcript</field>
<field name="category">scoring</field>
<field name="description">Grade a speaking transcript against a rubric using the platform's standard speaking examiner prompt.</field>
<field name="schema_json">{"type":"object","properties":{"rubric":{"type":"string"},"transcript":{"type":"string"}},"required":["rubric","transcript"]}</field>
<field name="sequence">110</field>
</record>
<!-- ============================== AGENTS ============================== -->
<!-- 1. Course planner -->
<record id="ai_agent_course_planner" model="encoach.ai.agent">
<field name="key">course_planner</field>
<field name="name">Course Planner</field>
<field name="description">Generates a full course plan (description, objectives, per-skill outcomes, grammar scope, assessment weights, week-by-week delivery) from a short brief. Used by the Smart Wizard and /api/ai/course-plan.</field>
<field name="model">gpt-4o</field>
<field name="fallback_model">gpt-4o-mini</field>
<field name="temperature">0.4</field>
<field name="max_tokens">4096</field>
<field name="response_format">json</field>
<field name="graph_type">plan_review_revise</field>
<field name="max_revisions">1</field>
<field name="quality_checks">quality.cefr_check</field>
<field name="sequence">10</field>
<field name="system_prompt">You are an expert English language curriculum designer. You produce structured, institution-grade course outlines suitable for a general foundation English programme.
Rules:
- Output MUST be a single valid JSON object matching the schema the user supplies.
- Use CEFR can-do statements when writing outcomes; cite the CEFR level in objectives.
- Distribute the weeks so grammar and skills build cumulatively, not randomly.
- Keep outcome codes short and stable (RLO1, WLO1, LLO1, SLO1, GLO1, VLO1) and reuse them in weeks[*].items[*].outcome_codes.
- Never wrap the JSON in prose, markdown, or code fences.</field>
<field name="tool_ids" eval="[(6, 0, [
ref('ai_tool_outcomes_fetch'),
ref('ai_tool_resources_search'),
ref('ai_tool_quality_cefr'),
ref('ai_tool_course_plan_save'),
])]"/>
</record>
<!-- 2. Week materials -->
<record id="ai_agent_course_week_materials" model="encoach.ai.agent">
<field name="key">course_week_materials</field>
<field name="name">Week Teaching Materials</field>
<field name="description">Given a course plan and a week number, produces classroom-ready materials (reading passage, listening script, speaking prompt, writing prompt, grammar mini-lesson, vocabulary list) aligned to the registered outcomes.</field>
<field name="model">gpt-4o</field>
<field name="fallback_model">gpt-4o-mini</field>
<field name="temperature">0.6</field>
<field name="max_tokens">6000</field>
<field name="response_format">json</field>
<field name="graph_type">plan_review_revise</field>
<field name="max_revisions">1</field>
<field name="quality_checks">quality.cefr_check</field>
<field name="sequence">20</field>
<field name="system_prompt">You are an expert EFL teacher creating ready-to-use classroom materials.
Rules:
- Every material MUST target only the outcome codes supplied for that week.
- Keep reading passages within the CEFR band's word-count window (A1~80, A2~150, B1~250, B2~400, C1~600, C2~800 words).
- Listening scripts must be natural dialogue/monologue, 3-4 minutes, with 4-6 comprehension questions.
- Speaking prompts include useful-language chunks the learner can recycle.
- Grammar lesson: one clear rule + 3 examples + 5 practice items with answer keys.
- Vocabulary: 8-12 entries with part of speech, CEFR-appropriate definition, and an example sentence in context.
- Output valid JSON only; no prose or markdown around it.</field>
<field name="tool_ids" eval="[(6, 0, [
ref('ai_tool_outcomes_fetch'),
ref('ai_tool_resources_search'),
ref('ai_tool_quality_cefr'),
ref('ai_tool_course_plan_save_materials'),
])]"/>
</record>
<!-- 3. Exam generator -->
<record id="ai_agent_exam_generator" model="encoach.ai.agent">
<field name="key">exam_generator</field>
<field name="name">Exam Generator</field>
<field name="description">Generates exam questions (MCQ, short answer, cloze, speaking prompts, writing tasks) matching a structure and blueprint.</field>
<field name="model">gpt-4o</field>
<field name="fallback_model">gpt-4o-mini</field>
<field name="temperature">0.5</field>
<field name="max_tokens">6000</field>
<field name="response_format">json</field>
<field name="graph_type">plan_review_revise</field>
<field name="max_revisions">1</field>
<field name="quality_checks">quality.cefr_check</field>
<field name="sequence">30</field>
<field name="system_prompt">You are a senior EFL / IELTS examiner generating authentic, validly constructed exam questions.
Rules:
- Follow the exam structure blueprint exactly: same number of sections, same question types, same scoring weights.
- Every MCQ has exactly one correct answer and three plausible distractors; avoid "all of the above".
- Reading/listening stems reference only content present in the accompanying passage/transcript.
- Never produce content outside the requested CEFR band.
- Output is a single JSON object; no explanations around it.</field>
<field name="tool_ids" eval="[(6, 0, [
ref('ai_tool_resources_search'),
ref('ai_tool_outcomes_fetch'),
ref('ai_tool_rubric_fetch'),
ref('ai_tool_quality_cefr'),
])]"/>
</record>
<!-- 4. Personalised exercise generator -->
<record id="ai_agent_exercise_generator" model="encoach.ai.agent">
<field name="key">exercise_generator</field>
<field name="name">Personalised Exercise Generator</field>
<field name="description">Produces targeted practice items (gap-fill, reordering, short response) using a learner's gap profile to focus on their weakest outcomes.</field>
<field name="model">gpt-4o-mini</field>
<field name="fallback_model">gpt-4o</field>
<field name="temperature">0.7</field>
<field name="max_tokens">3000</field>
<field name="response_format">json</field>
<field name="graph_type">react</field>
<field name="max_revisions">0</field>
<field name="quality_checks"></field>
<field name="sequence">40</field>
<field name="system_prompt">You are a remedial English tutor generating short, focused practice items for one learner.
Workflow:
1. Call `student.profile` with the student_id you are given. Read their CEFR band and gap_json.
2. Optionally call `resources.search` to find an anchor text at the right level.
3. Then produce 6-10 practice items that target the biggest gaps. Prefer item types the learner has been scoring low on.
4. Each item has: prompt, correct_answer, distractors (for MCQ), brief rationale, target_outcome_code.
5. Output a JSON object {"items": [...]}.
Never fabricate gap data — if student.profile fails, ask for a student_id in your final message and emit an empty items list.</field>
<field name="tool_ids" eval="[(6, 0, [
ref('ai_tool_student_profile'),
ref('ai_tool_resources_search'),
ref('ai_tool_outcomes_fetch'),
ref('ai_tool_quality_cefr'),
])]"/>
</record>
<!-- 5. LMS tutor / study assistant -->
<record id="ai_agent_lms_tutor" model="encoach.ai.agent">
<field name="key">lms_tutor</field>
<field name="name">LMS Tutor</field>
<field name="description">Chat assistant students talk to from inside a lesson. Can look up their profile, search the library, fetch outcomes, and answer questions about their course.</field>
<field name="model">gpt-4o-mini</field>
<field name="fallback_model">gpt-4o</field>
<field name="temperature">0.6</field>
<field name="max_tokens">1500</field>
<field name="response_format">text</field>
<field name="graph_type">react</field>
<field name="max_revisions">0</field>
<field name="quality_checks"></field>
<field name="sequence">50</field>
<field name="system_prompt">You are a friendly, encouraging English tutor inside the EnCoach LMS. You speak to learners directly.
Principles:
- Adapt vocabulary and sentence length to the learner's CEFR level.
- When the learner asks about their progress, call `student.profile` first.
- When they ask about a topic, prefer searching `resources.search` for approved materials before inventing examples.
- Be concrete: give one example, one practice question, one next step.
- Never invent scores or progress data; if a tool fails, tell the learner you'll flag the issue to their teacher.</field>
<field name="tool_ids" eval="[(6, 0, [
ref('ai_tool_resources_search'),
ref('ai_tool_student_profile'),
ref('ai_tool_outcomes_fetch'),
])]"/>
</record>
<!-- 6. Writing grader -->
<record id="ai_agent_writing_grader" model="encoach.ai.agent">
<field name="key">writing_grader</field>
<field name="name">Writing Grader</field>
<field name="description">Grades a writing submission against its rubric and produces band scores, feedback, and targeted suggestions.</field>
<field name="model">gpt-4o</field>
<field name="fallback_model">gpt-4o-mini</field>
<field name="temperature">0.2</field>
<field name="max_tokens">1800</field>
<field name="response_format">json</field>
<field name="graph_type">simple</field>
<field name="max_revisions">0</field>
<field name="quality_checks"></field>
<field name="sequence">60</field>
<field name="system_prompt">You are a calibrated IELTS / EFL writing examiner.
Rules:
- Score every criterion in the rubric exactly once.
- Use only band values the rubric advertises (0-9 for IELTS, 0-100 or A1-C2 for other rubrics — follow what the user sends).
- Feedback must quote one line of evidence from the student's text before each judgement.
- Suggestions must be specific ("replace X with Y") not generic ("improve grammar").
- Output JSON: {"scores": {"criterion_code": number}, "overall_band": number, "feedback": string, "suggestions": [string]}.</field>
<field name="tool_ids" eval="[(6, 0, [
ref('ai_tool_rubric_fetch'),
ref('ai_tool_scoring_writing'),
])]"/>
</record>
<!-- 7. Speaking evaluator -->
<record id="ai_agent_speaking_grader" model="encoach.ai.agent">
<field name="key">speaking_grader</field>
<field name="name">Speaking Evaluator</field>
<field name="description">Grades a speaking transcript against its rubric and produces band scores, feedback on fluency / pronunciation, and next-step drills.</field>
<field name="model">gpt-4o</field>
<field name="fallback_model">gpt-4o-mini</field>
<field name="temperature">0.2</field>
<field name="max_tokens">1800</field>
<field name="response_format">json</field>
<field name="graph_type">simple</field>
<field name="max_revisions">0</field>
<field name="quality_checks"></field>
<field name="sequence">70</field>
<field name="system_prompt">You are a calibrated IELTS Speaking examiner judging a transcript.
Rules:
- Only score what the transcript supports; pronunciation judgements must be flagged as indirect.
- Quote a line of the transcript before each criterion judgement.
- Suggestions prescribe one concrete drill (e.g. "practice minimal pairs /iː/ vs /ɪ/ for 2 weeks").
- Output JSON: {"scores": {"criterion_code": number}, "overall_band": number, "feedback": string, "suggestions": [string]}.</field>
<field name="tool_ids" eval="[(6, 0, [
ref('ai_tool_rubric_fetch'),
ref('ai_tool_scoring_speaking'),
])]"/>
</record>
<!-- Feature flag: pipelines consult this before routing through AgentRuntime.
Default "True" so the defaults-ship-working contract holds. -->
<record id="ai_default_use_langgraph" model="ir.config_parameter">
<field name="key">encoach_ai.use_langgraph_runtime</field>
<field name="value">True</field>
</record>
</odoo>

1
backend/custom_addons/encoach_ai/models/__init__.py
View File

@@ -2,4 +2,5 @@ from . import ai_settings
from . import ai_log
from . import ai_prompt
from . import ai_feedback
from . import ai_agent
from . import constants

357
backend/custom_addons/encoach_ai/models/ai_agent.py Normal file
View File

@@ -0,0 +1,357 @@
"""LangGraph-backed AI agents and the tools they can invoke.
Architecture
------------
The platform already has:
* ``encoach.ai.prompt`` — versioned prompt *templates* with rendering.
* ``encoach.ai.log`` — per-call telemetry.
* ``OpenAIService`` — thin wrapper around the OpenAI Python SDK.
This module adds the **agent layer** that sits on top of those primitives:
* :py:class:`EncoachAIAgent` — a named, configurable agent (e.g.
``course_planner``, ``exam_generator``). Each agent has a system prompt,
model choice, temperature budget, a graph topology (``simple``,
``plan_review_revise`` or ``rag``) and an M2M list of tools it is
allowed to call.
* :py:class:`EncoachAITool` — a catalogue row describing one callable
capability. The *implementation* lives in
``encoach_ai.services.agent_tools`` keyed by :py:attr:`key`; this model
only stores the metadata (JSON Schema, human description, whether it
mutates data, which audiences may use it). Admins toggle tools on or
off per agent through the UI without touching Python code.
The runtime itself (LangGraph state machine, tool-routing loop, log
emission) is in :py:mod:`encoach_ai.services.agent_runtime`. Keeping the
models here and the execution engine there makes the runtime easy to
swap / upgrade without touching the DB schema.
"""
from __future__ import annotations
import json
import logging
import re
from odoo import api, fields, models
from odoo.exceptions import UserError, ValidationError
_logger = logging.getLogger(__name__)
# Keys follow the same shape as prompt keys: ``lowercase.dotted.identifiers``.
_KEY_RE = re.compile(r"^[a-z][a-z0-9_]*(?:\.[a-z0-9_]+)*$")
GRAPH_TYPES = [
# Single LLM call, no tools. Lowest latency, used for deterministic tasks.
("simple", "Simple (single LLM call)"),
# Plan → self-critique → optionally revise once. Used for long-form
# generation (course plans, exam papers) where we want quality checks.
("plan_review_revise", "Plan → Review → Revise"),
# Retrieval-augmented: runs the ``search_resources`` tool first, injects
# the hits as context, then calls the LLM. Used for curriculum-aware
# generation that must cite existing materials.
("rag", "Retrieval-augmented (RAG)"),
# LLM decides which tools to call via OpenAI function-calling, in a
# loop. Used for LMS tutor / study assistant / grading workflows that
# may need to fetch rubrics, student profiles, etc.
("react", "ReAct (tool-calling loop)"),
]
TOOL_CATEGORIES = [
("retrieval", "Retrieval"),
("persistence", "Persistence"),
("quality", "Quality & gating"),
("scoring", "Scoring & grading"),
("reference", "Reference lookup"),
("other", "Other"),
]
# Models we're OK advertising in the admin UI. Keep small — the whole point
# of the agent layer is to encapsulate model choice.
MODEL_CHOICES = [
("gpt-4o", "GPT-4o (quality)"),
("gpt-4o-mini", "GPT-4o mini (cheap / fast)"),
("gpt-4.1", "GPT-4.1"),
("gpt-4.1-mini", "GPT-4.1 mini"),
("gpt-3.5-turbo", "GPT-3.5 turbo (legacy)"),
]
# =============================================================================
# Tool catalogue
# =============================================================================
class EncoachAITool(models.Model):
_name = "encoach.ai.tool"
_description = "AI Agent Tool"
_order = "category, sequence, name"
key = fields.Char(
required=True,
index=True,
help="Stable dotted identifier (e.g. 'resources.search'). The Python "
"handler is resolved from this key via the agent_tools registry.",
)
name = fields.Char(required=True, translate=True)
description = fields.Text(
required=True, translate=True,
help="Shown to the LLM when the tool is exposed as a callable. "
"Keep it concrete: explain *when* to use the tool, not *how* it works.",
)
category = fields.Selection(
TOOL_CATEGORIES, required=True, default="other", index=True,
)
schema_json = fields.Text(
required=True,
default="{}",
help="JSON Schema (Draft-07 subset) describing the tool's parameters. "
"Passed verbatim to OpenAI function-calling.",
)
mutates = fields.Boolean(
default=False,
help="If True, the tool writes to the database. Used by the runtime "
"to wrap calls in a savepoint so a failed LLM step can't leave half-"
"created records behind.",
)
sequence = fields.Integer(default=10)
active = fields.Boolean(default=True, index=True)
_sql_constraints = [
("tool_key_uniq", "unique(key)", "Tool key must be unique."),
]
# ------------------------------------------------------------------
@api.constrains("key")
def _check_key(self):
for rec in self:
if not _KEY_RE.match(rec.key or ""):
raise ValidationError(
f"Invalid tool key {rec.key!r}. Use lowercase dotted identifiers."
)
@api.constrains("schema_json")
def _check_schema(self):
for rec in self:
try:
parsed = json.loads(rec.schema_json or "{}")
except Exception as exc:
raise ValidationError(f"schema_json must be valid JSON: {exc}")
if not isinstance(parsed, dict):
raise ValidationError("schema_json must be a JSON object.")
# ------------------------------------------------------------------
def to_api_dict(self):
self.ensure_one()
try:
schema = json.loads(self.schema_json or "{}")
except Exception:
schema = {}
return {
"id": self.id,
"key": self.key,
"name": self.name,
"description": self.description,
"category": self.category,
"schema": schema,
"mutates": bool(self.mutates),
"active": bool(self.active),
}
def to_openai_tool(self):
"""Render this row in the shape OpenAI function-calling expects."""
self.ensure_one()
try:
params = json.loads(self.schema_json or "{}")
except Exception:
params = {}
# Ensure the JSON Schema is OpenAI-compatible (an object with
# ``properties``). Fall back to an empty object if the author
# forgot to wrap parameters.
if "type" not in params:
params = {"type": "object", "properties": params.get("properties", {})}
return {
"type": "function",
"function": {
"name": self.key.replace(".", "__"),
"description": self.description or self.name,
"parameters": params,
},
}
# =============================================================================
# Agent
# =============================================================================
class EncoachAIAgent(models.Model):
_name = "encoach.ai.agent"
_description = "AI Agent"
_order = "sequence, name"
key = fields.Char(
required=True,
index=True,
help="Stable dotted identifier the platform uses to resolve this "
"agent (e.g. 'course_planner'). Pipelines look the agent up by key "
"via AgentRuntime.from_key().",
)
name = fields.Char(required=True, translate=True)
description = fields.Text(translate=True)
sequence = fields.Integer(default=10)
active = fields.Boolean(default=True, index=True)
# ------------------------------------------------------------------
# Prompt & model wiring
# ------------------------------------------------------------------
system_prompt = fields.Text(
required=True,
translate=False,
help="System message sent to the LLM. Referenced variables can be "
"filled in by the caller via AgentRuntime.invoke(variables=...).",
)
prompt_key = fields.Char(
help="Optional: key of an encoach.ai.prompt record. When set, the "
"active version of that prompt *overrides* system_prompt at runtime. "
"Use this when you want prompt authors to iterate without touching "
"the agent config.",
)
model = fields.Selection(
MODEL_CHOICES, required=True, default="gpt-4o",
help="OpenAI chat model used for this agent's LLM calls.",
)
fallback_model = fields.Selection(
MODEL_CHOICES, default="gpt-4o-mini",
help="Model tried automatically if the primary model fails with a "
"5xx / rate-limit error. Leave blank to disable the fallback.",
)
temperature = fields.Float(
default=0.4,
help="0.0 = deterministic, 1.0 = very creative. 0.3-0.5 is a sane "
"default for structured-JSON generation.",
)
max_tokens = fields.Integer(default=4096)
response_format = fields.Selection(
[("text", "Text"), ("json", "JSON object")],
default="json",
help="`json` enables OpenAI's JSON mode and asks the LLM to return "
"parseable JSON. Use `text` for free-form output (tutor chat, "
"coach feedback).",
)
# ------------------------------------------------------------------
# Graph & tools
# ------------------------------------------------------------------
graph_type = fields.Selection(
GRAPH_TYPES, required=True, default="simple", index=True,
help="Which LangGraph topology to use when invoking this agent.",
)
max_revisions = fields.Integer(
default=1,
help="For `plan_review_revise`: cap on how many times the agent may "
"revise its own draft before emitting the final answer.",
)
quality_checks = fields.Char(
default="",
help="Comma-separated list of quality-gate tool keys to run after "
"generation (e.g. 'quality.cefr_check,quality.ai_detect'). Used by "
"the `plan_review_revise` topology.",
)
tool_ids = fields.Many2many(
"encoach.ai.tool",
"encoach_ai_agent_tool_rel",
"agent_id", "tool_id",
string="Tools",
help="Tools this agent is allowed to call. For `react` graphs, "
"these are exposed to the LLM as OpenAI function-calling tools; "
"for `rag` / `plan_review_revise`, only tools whose key matches a "
"pre-defined hook (e.g. `resources.search` for RAG, "
"`quality.*` for review) are executed.",
)
tool_count = fields.Integer(compute="_compute_tool_count", store=False)
_sql_constraints = [
("agent_key_uniq", "unique(key)", "Agent key must be unique."),
]
# ------------------------------------------------------------------
@api.depends("tool_ids")
def _compute_tool_count(self):
for rec in self:
rec.tool_count = len(rec.tool_ids)
@api.constrains("key")
def _check_key(self):
for rec in self:
if not _KEY_RE.match(rec.key or ""):
raise ValidationError(
f"Invalid agent key {rec.key!r}. "
"Use lowercase dotted identifiers (e.g. 'course_planner')."
)
@api.constrains("temperature")
def _check_temperature(self):
for rec in self:
if rec.temperature < 0.0 or rec.temperature > 2.0:
raise ValidationError("Temperature must be between 0.0 and 2.0")
# ------------------------------------------------------------------
# Lookup helpers
# ------------------------------------------------------------------
@api.model
def get_by_key(self, key):
"""Return the active agent for ``key`` or an empty recordset."""
return self.sudo().search(
[("key", "=", key), ("active", "=", True)], limit=1,
)
def resolved_system_prompt(self, variables=None):
"""Resolve the prompt to send: versioned prompt (if set) or inline.
``variables`` are passed to ``encoach.ai.prompt.render`` when the
agent is bound to a prompt key.
"""
self.ensure_one()
variables = variables or {}
if self.prompt_key:
prompt = self.env["encoach.ai.prompt"].sudo().get_active(self.prompt_key)
if prompt:
try:
return prompt.render(variables)
except UserError:
# Missing variables — fall back to the inline prompt so
# the caller still gets *something* and logs tell us
# exactly which variable was missing.
_logger.warning(
"Prompt %s v%s has unfilled variables; falling back "
"to inline system_prompt for agent %s",
prompt.key, prompt.version, self.key,
)
return self.system_prompt or ""
def to_api_dict(self, *, include_prompt=True):
self.ensure_one()
data = {
"id": self.id,
"key": self.key,
"name": self.name,
"description": self.description or "",
"model": self.model,
"fallback_model": self.fallback_model or "",
"temperature": self.temperature,
"max_tokens": self.max_tokens,
"response_format": self.response_format,
"graph_type": self.graph_type,
"max_revisions": self.max_revisions,
"quality_checks": [
k.strip() for k in (self.quality_checks or "").split(",") if k.strip()
],
"prompt_key": self.prompt_key or "",
"tool_count": len(self.tool_ids),
"tool_keys": self.tool_ids.mapped("key"),
"active": bool(self.active),
}
if include_prompt:
data["system_prompt"] = self.system_prompt or ""
return data

4
backend/custom_addons/encoach_ai/security/ir.model.access.csv
View File

@@ -5,3 +5,7 @@ access_ai_prompt_admin,encoach.ai.prompt admin,model_encoach_ai_prompt,base.grou
access_ai_prompt_user,encoach.ai.prompt user,model_encoach_ai_prompt,base.group_user,1,0,0,0
access_ai_feedback_admin,encoach.ai.feedback admin,model_encoach_ai_feedback,base.group_system,1,1,1,1
access_ai_feedback_user,encoach.ai.feedback user,model_encoach_ai_feedback,base.group_user,1,1,1,0
access_ai_agent_admin,encoach.ai.agent admin,model_encoach_ai_agent,base.group_system,1,1,1,1
access_ai_agent_user,encoach.ai.agent user,model_encoach_ai_agent,base.group_user,1,0,0,0
access_ai_tool_admin,encoach.ai.tool admin,model_encoach_ai_tool,base.group_system,1,1,1,1
access_ai_tool_user,encoach.ai.tool user,model_encoach_ai_tool,base.group_user,1,0,0,0
1 id name model_id:id group_id:id perm_read perm_write perm_create perm_unlink
5 access_ai_prompt_user encoach.ai.prompt user model_encoach_ai_prompt base.group_user 1 0 0 0
6 access_ai_feedback_admin encoach.ai.feedback admin model_encoach_ai_feedback base.group_system 1 1 1 1
7 access_ai_feedback_user encoach.ai.feedback user model_encoach_ai_feedback base.group_user 1 1 1 0
8 access_ai_agent_admin encoach.ai.agent admin model_encoach_ai_agent base.group_system 1 1 1 1
9 access_ai_agent_user encoach.ai.agent user model_encoach_ai_agent base.group_user 1 0 0 0
10 access_ai_tool_admin encoach.ai.tool admin model_encoach_ai_tool base.group_system 1 1 1 1
11 access_ai_tool_user encoach.ai.tool user model_encoach_ai_tool base.group_user 1 0 0 0

2
backend/custom_addons/encoach_ai/services/__init__.py
View File

@@ -7,3 +7,5 @@ from .elai_service import ElaiService
from .coach_service import CoachService
from . import cefr_mapper # canonical CEFR / band / theta mapper (P0.9)
from . import question_validator # schema + quality gate for AI-generated questions (P1.6/P1.1)
from . import agent_tools # registry of tool handlers used by AgentRuntime
from .agent_runtime import AgentRuntime # LangGraph-backed core agent runtime

500
backend/custom_addons/encoach_ai/services/agent_runtime.py Normal file
View File

@@ -0,0 +1,500 @@
"""LangGraph-based agent runtime.
This is the core AI engine for EnCoach — every pipeline that used to call
``OpenAIService`` directly can instead go through an
:py:class:`AgentRuntime` loaded from a named :py:class:`encoach.ai.agent`
row. That buys us:
* A single place to reason about retries, logging, tool execution and
self-review, instead of the same boilerplate inside every pipeline.
* Admins editing prompts, model choice, temperature or enabled tools in
the UI without redeploys.
* A consistent shape (``invoke(variables, payload)``) across course
planning, exam generation, LMS tutor, grading, etc.
Graph topologies
----------------
Each agent picks one of four graphs. They're all built on the same
:py:class:`AgentState` TypedDict so upgrading an agent from one topology
to another only means flipping a selection field.
``simple``
``START → llm → END``. The workhorse — used for deterministic
JSON generation (course plan header, exam question batches).
``plan_review_revise``
``START → llm → review → [revise → llm]? → END``. ``review`` runs
every configured quality tool (``quality.cefr_check`` etc.); if any
returns ``ok=False`` we ask the LLM to revise once, capped by
``max_revisions`` on the agent. Keeps structured outputs (reading
passages, listening scripts) inside their CEFR band.
``rag``
``START → retrieve → llm → END``. Runs ``resources.search`` before
the LLM and injects the hits as extra system context. Used for
curriculum-aware generation that must cite real library material.
``react``
Classic tool-calling loop. The LLM is given the OpenAI-format tool
list and can call tools in a loop until it emits a final answer.
Used for the LMS tutor / study assistant.
We depend on ``langgraph`` for the orchestration (state machine,
conditional edges) but still call OpenAI through the existing
:py:class:`OpenAIService` so the API key wiring, retry behaviour and
``encoach.ai.log`` rows keep working unchanged.
"""
from __future__ import annotations
import json
import logging
import time
from typing import Any, TypedDict
from odoo.tools import config as odoo_config # noqa: F401 (future use)
from . import agent_tools
from .openai_service import OpenAIService
_logger = logging.getLogger(__name__)
# =============================================================================
# State
# =============================================================================
class AgentState(TypedDict, total=False):
"""Shared state passed between every node of the graph."""
messages: list[dict] # chat history sent to the LLM
output: Any # parsed final answer (dict or string)
output_raw: str # the raw LLM text (for logging)
tool_calls: list[dict] # pending tool_calls emitted by the LLM
tool_results: list[dict] # results of executed tools, appended
quality_issues: list[str] # issues collected by the review node
revisions_used: int # revision counter (capped by max_revisions)
variables: dict # caller-supplied variables (for prompt rendering)
retrieval: list[dict] # hits from the retrieval node (RAG)
iterations: int # guard against runaway ReAct loops
error: str # populated on fatal failure
# =============================================================================
# AgentRuntime
# =============================================================================
class AgentRuntime:
"""Wraps an :py:class:`encoach.ai.agent` row with a compiled LangGraph."""
MAX_REACT_ITERATIONS = 6 # hard cap on tool-calling loops
# ------------------------------------------------------------------
# Factories
# ------------------------------------------------------------------
def __init__(self, env, agent, *, language: str | None = None):
self.env = env
self.agent = agent
self.language = language
self.ai = OpenAIService(env, language=language)
self._graph = None # lazily compiled
@classmethod
def from_key(cls, env, key: str, *, language: str | None = None):
"""Factory: load the active agent with ``key`` and build its runtime.
Returns ``None`` if no agent is configured for that key — callers
can decide whether to fall back to their legacy direct-SDK path.
"""
agent = env["encoach.ai.agent"].sudo().get_by_key(key)
if not agent:
return None
return cls(env, agent, language=language)
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def invoke(self, variables: dict | None = None, payload: Any = None,
*, extra_system: str = "") -> AgentState:
"""Run the agent's graph end-to-end and return the terminal state.
``variables`` are substituted into the system prompt (if the agent
is bound to a prompt_key). ``payload`` becomes the first user
message; pass a dict to have it rendered as JSON, or a string to
pass it through verbatim. ``extra_system`` lets callers tack on
a context block without touching the agent's stored prompt.
"""
t0 = time.time()
graph = self._compile()
initial = self._initial_state(variables or {}, payload, extra_system)
try:
# LangGraph is sync here — we're already inside an Odoo
# request worker so sticking with sync keeps the control
# flow simple.
final: AgentState = graph.invoke(initial)
except Exception as exc:
_logger.exception("agent %s crashed", self.agent.key)
final = {**initial, "error": str(exc)}
latency_ms = int((time.time() - t0) * 1000)
self._log(final, latency_ms)
return final
# ------------------------------------------------------------------
# Graph construction
# ------------------------------------------------------------------
def _compile(self):
if self._graph is not None:
return self._graph
try:
from langgraph.graph import StateGraph, START, END
except ImportError as exc:
raise RuntimeError(
"LangGraph is not installed. "
"Add `langgraph>=0.2.0` to backend/requirements.txt and pip install."
) from exc
g = StateGraph(AgentState)
if self.agent.graph_type == "simple":
g.add_node("llm", self._node_llm)
g.add_edge(START, "llm")
g.add_edge("llm", END)
elif self.agent.graph_type == "rag":
g.add_node("retrieve", self._node_retrieve)
g.add_node("llm", self._node_llm)
g.add_edge(START, "retrieve")
g.add_edge("retrieve", "llm")
g.add_edge("llm", END)
elif self.agent.graph_type == "plan_review_revise":
g.add_node("llm", self._node_llm)
g.add_node("review", self._node_review)
g.add_edge(START, "llm")
g.add_edge("llm", "review")
g.add_conditional_edges(
"review",
self._route_after_review,
{"revise": "llm", "done": END},
)
elif self.agent.graph_type == "react":
g.add_node("llm", self._node_llm_tools)
g.add_node("tools", self._node_tools)
g.add_edge(START, "llm")
g.add_conditional_edges(
"llm",
self._route_after_llm_tools,
{"tools": "tools", "done": END},
)
g.add_edge("tools", "llm")
else:
raise ValueError(f"Unknown graph_type: {self.agent.graph_type}")
self._graph = g.compile()
return self._graph
# ------------------------------------------------------------------
# Initial state
# ------------------------------------------------------------------
def _initial_state(self, variables: dict, payload: Any, extra_system: str) -> AgentState:
system_prompt = self.agent.resolved_system_prompt(variables)
messages: list[dict] = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
if extra_system:
messages.append({"role": "system", "content": extra_system})
if payload is not None:
user_content = (
payload if isinstance(payload, str)
else json.dumps(payload, ensure_ascii=False)
)
messages.append({"role": "user", "content": user_content})
return {
"messages": messages,
"output": None,
"output_raw": "",
"tool_calls": [],
"tool_results": [],
"quality_issues": [],
"revisions_used": 0,
"variables": variables,
"retrieval": [],
"iterations": 0,
"error": "",
}
# ------------------------------------------------------------------
# Nodes
# ------------------------------------------------------------------
def _node_llm(self, state: AgentState) -> AgentState:
"""Plain LLM call respecting the agent's model + response_format."""
messages = list(state.get("messages") or [])
model = self.agent.model
action = f"agent.{self.agent.key}"
try:
if self.agent.response_format == "json":
content = self.ai.chat_json(
messages,
model=model,
temperature=self.agent.temperature,
max_tokens=self.agent.max_tokens,
action=action,
)
raw = json.dumps(content, ensure_ascii=False)
else:
raw = self.ai.chat(
messages,
model=model,
temperature=self.agent.temperature,
max_tokens=self.agent.max_tokens,
action=action,
)
content = raw
except Exception as exc:
# Try the fallback model exactly once before giving up.
if self.agent.fallback_model and model != self.agent.fallback_model:
_logger.warning(
"agent %s primary model %s failed (%s); retrying with %s",
self.agent.key, model, exc, self.agent.fallback_model,
)
try:
if self.agent.response_format == "json":
content = self.ai.chat_json(
messages, model=self.agent.fallback_model,
temperature=self.agent.temperature,
max_tokens=self.agent.max_tokens,
action=f"{action}.fallback",
)
raw = json.dumps(content, ensure_ascii=False)
else:
raw = self.ai.chat(
messages, model=self.agent.fallback_model,
temperature=self.agent.temperature,
max_tokens=self.agent.max_tokens,
action=f"{action}.fallback",
)
content = raw
except Exception as exc2:
return {**state, "error": str(exc2)}
else:
return {**state, "error": str(exc)}
new_messages = messages + [{"role": "assistant", "content": raw}]
return {
**state,
"messages": new_messages,
"output": content,
"output_raw": raw,
}
def _node_retrieve(self, state: AgentState) -> AgentState:
"""RAG node: call resources.search with the user's payload as query."""
variables = state.get("variables") or {}
query = ""
# The last user message is our best default query.
for m in reversed(state.get("messages") or []):
if m.get("role") == "user":
query = m.get("content") or ""
break
query = variables.get("query") or query
hits = agent_tools.invoke(self.env, "resources.search", {
"query": query[:1000],
"limit": 5,
})
items = hits.get("items") or []
context = self._format_retrieval(items)
messages = list(state.get("messages") or [])
if context:
# Insert the context block *after* the system prompt(s).
last_sys = -1
for i, m in enumerate(messages):
if m.get("role") == "system":
last_sys = i
insert_at = last_sys + 1 if last_sys >= 0 else 0
messages.insert(insert_at, {
"role": "system",
"content": (
"Relevant content from the library (use it when accurate, "
"cite ids; do not fabricate):\n\n" + context
),
})
return {**state, "messages": messages, "retrieval": items}
def _node_review(self, state: AgentState) -> AgentState:
"""Run every configured quality tool against the LLM's output."""
text = state.get("output_raw") or ""
if isinstance(state.get("output"), dict):
# Flatten the dict to text so the quality tools see something
# meaningful (most just want prose).
text = json.dumps(state["output"], ensure_ascii=False)
issues: list[str] = []
checks = [
k.strip() for k in (self.agent.quality_checks or "").split(",")
if k.strip()
]
variables = state.get("variables") or {}
target_cefr = (
variables.get("cefr_level")
or variables.get("target_cefr")
or "b1"
)
for key in checks:
res = agent_tools.invoke(self.env, key, {
"text": text,
"target_cefr": target_cefr,
"cefr_level": target_cefr,
})
if res.get("ok") is False:
issues.extend(res.get("issues") or [res.get("error") or key])
return {**state, "quality_issues": issues}
def _route_after_review(self, state: AgentState) -> str:
issues = state.get("quality_issues") or []
if not issues:
return "done"
if (state.get("revisions_used") or 0) >= max(0, self.agent.max_revisions):
return "done"
# Queue up a revision: add a system message with the critique and
# bump the counter. We return via "revise" which loops back to
# the LLM node.
critique = (
"Your previous draft was rejected for the following reasons:\n- "
+ "\n- ".join(issues)
+ "\n\nProduce an improved version that addresses every issue. "
"Keep the same JSON schema if one was requested."
)
messages = list(state.get("messages") or []) + [
{"role": "system", "content": critique}
]
state["messages"] = messages
state["revisions_used"] = (state.get("revisions_used") or 0) + 1
return "revise"
# ReAct / tool-calling -------------------------------------------------
def _node_llm_tools(self, state: AgentState) -> AgentState:
"""ReAct step: ask the LLM, exposing all enabled tools."""
if state.get("iterations", 0) >= self.MAX_REACT_ITERATIONS:
return {**state, "error": "react_iteration_limit_exceeded"}
client = self.ai.client
if client is None:
return {**state, "error": "openai_not_configured"}
tools = [t.to_openai_tool() for t in self.agent.tool_ids]
try:
resp = client.chat.completions.create(
model=self.agent.model,
messages=state.get("messages") or [],
temperature=self.agent.temperature,
max_tokens=self.agent.max_tokens,
tools=tools or None,
tool_choice="auto" if tools else None,
timeout=self.ai.request_timeout,
)
except Exception as exc:
return {**state, "error": str(exc)}
choice = resp.choices[0].message
assistant_msg: dict[str, Any] = {
"role": "assistant",
"content": choice.content or "",
}
tool_calls = []
if getattr(choice, "tool_calls", None):
# Preserve the OpenAI-shaped tool_calls list on the message so
# the next round references them by id.
assistant_msg["tool_calls"] = [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments,
},
}
for tc in choice.tool_calls
]
for tc in choice.tool_calls:
try:
args = json.loads(tc.function.arguments or "{}")
except Exception:
args = {}
tool_calls.append({
"id": tc.id,
"name": tc.function.name,
"args": args,
})
new_messages = list(state.get("messages") or []) + [assistant_msg]
return {
**state,
"messages": new_messages,
"tool_calls": tool_calls,
"output": choice.content or state.get("output"),
"output_raw": choice.content or state.get("output_raw") or "",
"iterations": state.get("iterations", 0) + 1,
}
def _route_after_llm_tools(self, state: AgentState) -> str:
if state.get("error"):
return "done"
if state.get("tool_calls"):
return "tools"
return "done"
def _node_tools(self, state: AgentState) -> AgentState:
"""Execute every queued tool_call and append results to the chat."""
allowed = {t.key: t for t in self.agent.tool_ids}
messages = list(state.get("messages") or [])
results: list[dict] = list(state.get("tool_results") or [])
for call in state.get("tool_calls") or []:
# Tools are stored with dotted keys but OpenAI flattens dots to
# double-underscores (because function names must match [A-Za-z0-9_]).
key = (call.get("name") or "").replace("__", ".")
if key not in allowed:
result = {"error": f"tool_not_allowed:{key}"}
else:
result = agent_tools.invoke(self.env, key, call.get("args") or {})
results.append({"tool": key, "args": call.get("args"), "result": result})
messages.append({
"role": "tool",
"tool_call_id": call.get("id"),
"name": call.get("name") or key,
"content": json.dumps(result, ensure_ascii=False, default=str)[:6000],
})
return {
**state,
"messages": messages,
"tool_calls": [],
"tool_results": results,
}
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
@staticmethod
def _format_retrieval(items: list[dict]) -> str:
parts = []
for r in items or []:
label = f"[{r.get('type','?')}#{r.get('id','?')}]"
title = r.get("title") or ""
snippet = (r.get("snippet") or "")[:400]
parts.append(f"{label} {title}\n{snippet}")
return "\n---\n".join(parts)
def _log(self, final: AgentState, latency_ms: int):
try:
self.env["encoach.ai.log"].sudo().create({
"service": "openai",
"action": f"agent.{self.agent.key}",
"model_used": self.agent.model,
"latency_ms": latency_ms,
"status": "error" if final.get("error") else "success",
"error_message": final.get("error") or "",
"input_preview": json.dumps(final.get("variables") or {})[:500],
"output_preview": (final.get("output_raw") or "")[:500],
})
except Exception:
_logger.warning("agent %s log write failed", self.agent.key, exc_info=True)

326
backend/custom_addons/encoach_ai/services/agent_tools.py Normal file
View File

@@ -0,0 +1,326 @@
"""Python implementations of the tools the agent runtime can invoke.
Every :py:class:`encoach.ai.tool` row in the DB points to one of the
handler functions registered here via :py:func:`register`. The DB row
holds the metadata (description, JSON Schema, admin toggle); the real
logic is Python so it can import Odoo models, run transactions and
reuse existing services (vector search, quality gates, etc.).
Tools follow a strict contract:
* Signature: ``handler(env, **params) -> dict``.
* The returned dict must be JSON-serialisable. It becomes the tool
message the LLM sees next, so keys should be descriptive.
* Tools that write to the DB must set ``mutates=True`` on their catalogue
row so the runtime wraps the call in a savepoint.
* Tools never raise: catch exceptions and return ``{"error": str(exc)}``
so the agent can reason about failures and the top-level call keeps
going instead of aborting the whole graph.
Adding a new tool
-----------------
1. Write a handler below, decorated with ``@register("namespace.name")``.
2. Add a seed row in ``data/agents_defaults.xml`` with the same key.
3. Bind it to one or more agents in the same seed file.
"""
from __future__ import annotations
import json
import logging
from typing import Any, Callable
_logger = logging.getLogger(__name__)
# Registry: tool key → handler callable.
_REGISTRY: dict[str, Callable[..., dict]] = {}
def register(key: str):
"""Decorator to register a tool handler under ``key``."""
def decorator(func: Callable[..., dict]) -> Callable[..., dict]:
if key in _REGISTRY:
_logger.warning("Agent tool %r is being re-registered", key)
_REGISTRY[key] = func
return func
return decorator
def get_handler(key: str) -> Callable[..., dict] | None:
return _REGISTRY.get(key)
def list_keys() -> list[str]:
return sorted(_REGISTRY.keys())
def invoke(env, key: str, params: dict | None = None) -> dict:
"""Resolve and invoke the handler for ``key`` with ``params``.
All tool failures are normalised to ``{"error": "..."}`` so callers
(LangGraph nodes) don't need to care about exceptions. A missing
handler returns ``{"error": "unknown_tool"}``.
"""
handler = _REGISTRY.get(key)
if not handler:
return {"error": f"unknown_tool:{key}"}
try:
return handler(env, **(params or {}))
except TypeError as exc:
# Bad arguments from the LLM — make the complaint readable.
return {"error": f"bad_arguments: {exc}"}
except Exception as exc:
_logger.exception("agent tool %s failed", key)
return {"error": str(exc)}
# =============================================================================
# Built-in tools
# =============================================================================
#
# Each handler is deliberately small: they're thin adapters over services
# we already have. The LLM gets a compact JSON payload to reason over.
# --- Retrieval ----------------------------------------------------------------
@register("resources.search")
def _search_resources(env, query: str = "", skill: str = "", cefr_level: str = "",
limit: int = 5, **_: Any) -> dict:
"""Semantic search over the LMS 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
)
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))
except Exception as exc:
_logger.debug("resource vector search unavailable: %s", exc)
results = []
out = []
for r in results or []:
out.append({
"id": r.get("content_id") or r.get("id"),
"type": r.get("content_type") or "",
"title": (r.get("metadata") or {}).get("title", ""),
"snippet": (r.get("text") or "")[:400],
"similarity": r.get("similarity"),
})
return {"query": query, "count": len(out), "items": out}
@register("rubric.fetch")
def _fetch_rubric(env, rubric_id: int | None = None, skill: str = "", **_: Any) -> dict:
"""Return rubric criteria for a given id, or the newest rubric for a skill."""
Rubric = env["encoach.rubric"].sudo() if "encoach.rubric" in env else None
if Rubric is None:
return {"error": "rubric_model_missing"}
rec = None
if rubric_id:
rec = Rubric.browse(int(rubric_id))
if not rec.exists():
rec = None
if rec is None and skill:
rec = Rubric.search(
[("skill", "=", skill)], order="create_date desc", limit=1,
)
if not rec:
return {"error": "rubric_not_found"}
criteria = []
for crit in getattr(rec, "criterion_ids", rec.browse([])):
criteria.append({
"code": getattr(crit, "code", "") or "",
"name": crit.name or "",
"weight": getattr(crit, "weight", 0) or 0,
"descriptors": getattr(crit, "descriptors", "") or "",
})
return {
"id": rec.id,
"name": rec.name,
"skill": getattr(rec, "skill", "") or "",
"criteria": criteria,
}
@register("outcomes.fetch")
def _fetch_course_outcomes(env, course_id: int | None = None,
cefr_level: str = "", **_: Any) -> dict:
"""Return learning outcomes for a course (or CEFR level)."""
LO = env["encoach.learning.objective"].sudo() \
if "encoach.learning.objective" in env else None
if LO is None:
return {"error": "learning_objective_model_missing"}
domain = []
if course_id:
domain.append(("course_id", "=", int(course_id)))
if cefr_level:
domain.append(("cefr_level", "=", cefr_level))
records = LO.search(domain, limit=200)
return {
"count": len(records),
"items": [{
"id": r.id,
"code": getattr(r, "code", "") or "",
"skill": getattr(r, "skill", "") or "",
"cefr_level": getattr(r, "cefr_level", "") or "",
"description": r.name or getattr(r, "description", "") or "",
} for r in records],
}
@register("student.profile")
def _fetch_student_profile(env, student_id: int, **_: Any) -> dict:
"""Return a compact gap-profile the agent can use to personalise content."""
SP = env["encoach.student.profile"].sudo() \
if "encoach.student.profile" in env else None
if SP is None:
return {"error": "student_profile_model_missing"}
rec = SP.search([("student_id", "=", int(student_id))], limit=1)
if not rec:
return {"error": "profile_not_found"}
return {
"student_id": int(student_id),
"cefr_level": getattr(rec, "cefr_level", "") or "",
"strengths_json": getattr(rec, "strengths_json", "") or "",
"gaps_json": getattr(rec, "gaps_json", "") or "",
}
# --- Quality gates ------------------------------------------------------------
@register("quality.cefr_check")
def _cefr_check(env, text: str = "", target_cefr: str = "b1", **_: Any) -> dict:
"""Grade the readability of ``text`` against a target CEFR band."""
try:
import textstat # noqa: F401
fk = textstat.flesch_kincaid_grade(text or "")
fre = textstat.flesch_reading_ease(text or "")
except Exception:
fk, fre = None, None
# Rough mapping — deliberately conservative; the LLM uses it as a hint.
band_map = {"a1": (1, 3), "a2": (3, 5), "b1": (5, 7),
"b2": (7, 9), "c1": (9, 12), "c2": (12, 20)}
ok = True
issues = []
if fk is not None:
lo, hi = band_map.get((target_cefr or "b1").lower(), (5, 7))
if fk < lo - 0.5:
ok = False
issues.append(f"Text reads below {target_cefr.upper()} (FK={fk:.1f})")
elif fk > hi + 0.5:
ok = False
issues.append(f"Text reads above {target_cefr.upper()} (FK={fk:.1f})")
return {
"ok": ok,
"target_cefr": target_cefr,
"flesch_kincaid": fk,
"flesch_reading_ease": fre,
"issues": issues,
}
@register("quality.ai_detect")
def _ai_detect(env, text: str = "", **_: Any) -> dict:
"""Return AI-detection probability if GPTZero is configured, else neutral."""
try:
# Try to reuse whatever GPTZero wrapper the platform already has.
from odoo.addons.encoach_ai.services.gptzero_service import (
GPTZeroService, # type: ignore
)
svc = GPTZeroService(env)
return svc.score(text)
except Exception as exc:
_logger.debug("gptzero unavailable: %s", exc)
return {"ok": True, "ai_probability": None, "note": "detector_unavailable"}
@register("quality.content_gate")
def _content_gate(env, text: str = "", cefr_level: str = "b1", **_: Any) -> dict:
"""Run the project's unified content-source gate (if installed)."""
try:
from odoo.addons.encoach_quality_gate.services.content_source_gate import (
ContentSourceGate, # type: ignore
)
gate = ContentSourceGate(env)
return gate.check(text, cefr_level=cefr_level)
except Exception as exc:
_logger.debug("content_gate unavailable: %s", exc)
return {"ok": True, "note": "gate_unavailable"}
# --- Persistence --------------------------------------------------------------
@register("course_plan.save")
def _save_course_plan(env, plan_vals: dict, weeks: list | None = None, **_: Any) -> dict:
"""Persist an AI-generated course plan. Used by the course_planner agent.
``plan_vals`` is the dict the LLM produced (after the runtime's JSON
normalisation). ``weeks`` is the list of per-week rows. The handler is
idempotent-friendly: it always creates new rows (agents are expected
to decide whether to reuse an existing plan before calling this).
"""
Plan = env["encoach.course.plan"].sudo() if "encoach.course.plan" in env else None
Week = env["encoach.course.plan.week"].sudo() \
if "encoach.course.plan.week" in env else None
if Plan is None or Week is None:
return {"error": "course_plan_models_missing"}
plan = Plan.create(plan_vals or {})
created_weeks = 0
for w in weeks or []:
try:
Week.create({**w, "plan_id": plan.id})
created_weeks += 1
except Exception as exc:
_logger.warning("agent tool course_plan.save: bad week row %r: %s", w, exc)
return {"plan_id": plan.id, "weeks_created": created_weeks}
@register("course_plan.save_materials")
def _save_week_materials(env, plan_id: int, week_id: int,
materials: list | None = None, **_: Any) -> dict:
Material = env["encoach.course.plan.material"].sudo() \
if "encoach.course.plan.material" in env else None
if Material is None:
return {"error": "material_model_missing"}
created = 0
for m in materials or []:
try:
Material.create({
**m,
"plan_id": int(plan_id),
"week_id": int(week_id),
"body_json": json.dumps(m.get("body") or {}, ensure_ascii=False)
if "body" in m else m.get("body_json", ""),
})
created += 1
except Exception as exc:
_logger.warning("agent tool save_materials: bad row %r: %s", m, exc)
return {"plan_id": plan_id, "week_id": week_id, "materials_created": created}
# --- Scoring (best-effort wrappers over existing services) --------------------
@register("scoring.grade_writing")
def _grade_writing(env, rubric: str = "", task: str = "", response: str = "",
**_: Any) -> dict:
from odoo.addons.encoach_ai.services.openai_service import OpenAIService
svc = OpenAIService(env)
try:
return svc.grade_writing(rubric, task, response)
except Exception as exc:
return {"error": str(exc)}
@register("scoring.grade_speaking")
def _grade_speaking(env, rubric: str = "", transcript: str = "", **_: Any) -> dict:
from odoo.addons.encoach_ai.services.openai_service import OpenAIService
svc = OpenAIService(env)
try:
return svc.grade_speaking(rubric, transcript)
except Exception as exc:
return {"error": str(exc)}