Backend du bot (Node/TS)

Un Policy Router ( @Finary ) lit le profil élève via un token / code de login → récupère classe, niveau → résout les vector_store_ids prioritaires.

Appelle la Responses API avec l’outil file_search en priorité (tool_choice forcé), puis fallback vers le modèle général uniquement si la recherche n’a rien donné ou si le score est trop faible.

Routage & appel OpenAI (Node/TypeScript)


// router.ts
import type { Request, Response } from "express";
import OpenAI from "openai";
const client = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

// Helpers
type StudentContext = {
  classe: string;
  niveau: "fort" | "moyen" | "fragile";
  lang?: string;
  vectorStores: string[];          // Option A: 1 store (classe)
  metadataFilter?: Record<string, any>; // ex. { metadata: { niveau: "fort", lang: "fr" } }
};

async function fetchStudentContextByCode(code: string): Promise<StudentContext> {
  // appelez Odoo (controller maison) avec le code, ex.
  // GET https://odoo.mercollege/api/student_profile?code=XXXX
  // Ici, on simule :
  return {
    classe: "2nde2",
    niveau: "moyen",
    lang: "fr",
    vectorStores: [process.env.VS_2NDE2!], // Option A: un store par classe
    metadataFilter: { metadata: { niveau: "moyen", lang: "fr" } }
  };
}

function shouldFallbackToGeneral(searchResults: any): boolean {
  // À adapter: si 0 résultat OU scores < seuil
  const items = searchResults ?? [];
  if (!items.length) return true;
  // si la réponse inclut des scores, filtrez; sinon, laissez 0 résultat déclencher le fallback
  return false;
}

export async function chatHandler(req: Request, res: Response) {
  const { code, message } = req.body as { code: string; message: string };

  const student = await fetchStudentContextByCode(code);

  // 1) Appel PRIORITAIRE RAG (file_search forcé)
  const ragResp = await client.responses.create({
    model: "gpt-4o-mini", // ou gpt-5-mini si dispo & latence ok
    input: [
      {
        role: "system",
        content: [
          {
            type: "text",
            text: [
              "Rôle: Assistant pédagogique personnalisé du MerCollège Lycée.",
              "Procédure:",
              "1) Tu DOIS d'abord utiliser file_search sur les ressources de la CLASSE et du NIVEAU de l'élève.",
              "2) Si la recherche ne renvoie rien de pertinent, informe-le et complète avec ta connaissance générale.",
              "3) Adapter la difficulté et le ton au niveau (fort/moyen/fragile)."
            ].join("\n")
          }
        ]
      },
      { role: "user", content: [{ type: "text", text: message }] }
    ],
    tools: [
      {
        type: "file_search",
        vector_store_ids: student.vectorStores,
        // Option A: filtrez par niveau/langue
        filters: student.metadataFilter
      }
    ],
    tool_choice: { type: "file_search" }, // FORCE la recherche
    // utile en debug: inclure les résultats et citations
    include: ["output_text", "citations", "output[*].file_search_call.search_results"]
  });

  // 2) Décidez si fallback
  const results =
    ragResp.output?.find?.((o: any) => o.file_search_call)?.file_search_call?.search_results ?? [];
  const needFallback = shouldFallbackToGeneral(results);

  if (needFallback) {
    const generalResp = await client.responses.create({
      model: "gpt-4o-mini",
      input: [
        {
          role: "system",
          content: [
            {
              type: "text",
              text: [
                "Rôle: Assistant pédagogique personnalisé du MerCollège Lycée.",
                "Aucune ressource de classe/niveau pertinente trouvée.",
                "Tu peux répondre avec connaissance générale, mais précise clairement que tu n'as pas trouvé dans les ressources officielles."
              ].join("\n")
            }
          ]
        },
        { role: "user", content: [{ type: "text", text: message }] }
      ]
    });
    res.json({ text: generalResp.output_text, mode: "general" });
    return;
  }

  res.json({
    text: ragResp.output_text,
    sources: ragResp.citations, // pour afficher les sources à l’élève
    mode: "rag"
  });
}

Ce code router.ts définit un routeur Express (dans un backend Node.js / TypeScript) qui permet à un élève d’échanger avec un assistant pédagogique intelligent.

Il combine OpenAI (RAG – Retrieval-Augmented Generation) avec les profils d’élèves stockés dans Odoo, pour donner des réponses personnalisées selon la classe, le niveau et la langue de l’élève.

🔍 Vue d’ensemble

L’objectif du code est de :

Identifier le profil de l’élève (classe, niveau, langue) à partir d’un code.
Faire une recherche RAG (dans des vector stores OpenAI spécifiques à sa classe).
Générer une réponse adaptée à son niveau à partir de ces ressources.
Basculer sur un fallback général si aucune ressource pertinente n’est trouvée.

🧱 Structure générale du fichier

1️⃣ Importations et initialisation


import type { Request, Response } from "express";
import OpenAI from "openai";
const client = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

Express : framework web pour gérer les requêtes HTTP (req, res).
OpenAI : SDK officiel pour appeler les modèles (GPT-4o, GPT-5…).
La clé API est récupérée dans les variables d’environnement.

2️⃣ Définition du type StudentContext


type StudentContext = {
  classe: string;
  niveau: "fort" | "moyen" | "fragile";
  lang?: string;
  vectorStores: string[];
  metadataFilter?: Record<string, any>;
};

Ce type structure le profil de l’élève récupéré depuis Odoo :

classe : ex. "2nde2"
niveau : force académique
lang : langue principale
vectorStores : identifiants OpenAI des vector stores contenant les documents pédagogiques
metadataFilter : filtre sur les métadonnées du store (niveau/langue)

3️⃣ Fonction fetchStudentContextByCode


async function fetchStudentContextByCode(code: string): Promise<StudentContext> {
  // Ici, on simule un appel API Odoo
  return {
    classe: "2nde2",
    niveau: "moyen",
    lang: "fr",
    vectorStores: [process.env.VS_2NDE2!],
    metadataFilter: { metadata: { niveau: "moyen", lang: "fr" } }
  };
}

👉 Dans la vraie version :

On ferait un appel REST à Odoo :
GET https://odoo.mercollege/api/student_profile?code=XXXX
On récupère les infos du profil pour savoir dans quel vector store chercher les cours adaptés.

4️⃣ Fonction shouldFallbackToGeneral


function shouldFallbackToGeneral(searchResults: any): boolean {
  const items = searchResults ?? [];
  if (!items.length) return true;
  return false;
}

Vérifie si la recherche RAG a renvoyé des résultats pertinents.
Si aucun résultat → on utilisera le mode général (connaissance du modèle uniquement).

5️⃣ Fonction principale chatHandler

C’est le cœur du système, appelé lorsqu’un élève envoie un message.


export async function chatHandler(req: Request, res: Response) {
  const { code, message } = req.body as { code: string; message: string };

code → identifiant élève (pour retrouver son profil)
message → texte ou question de l’élève

Étape 1 – Récupération du profil


  const student = await fetchStudentContextByCode(code);

On récupère son contexte (classe, niveau, store).

Étape 2 – Requête RAG (avec file_search)


  const ragResp = await client.responses.create({
    model: "gpt-4o-mini",
    input: [
      {
        role: "system",
        content: [{
          type: "text",
          text: [
            "Rôle: Assistant pédagogique personnalisé du MerCollège Lycée.",
            "Procédure:",
            "1) Tu DOIS d'abord utiliser file_search sur les ressources de la CLASSE et du NIVEAU de l'élève.",
            "2) Si la recherche ne renvoie rien de pertinent, informe-le et complète avec ta connaissance générale.",
            "3) Adapter la difficulté et le ton au niveau (fort/moyen/fragile)."
          ].join("\n")
        }]
      },
      { role: "user", content: [{ type: "text", text: message }] }
    ],
    tools: [
      {
        type: "file_search",
        vector_store_ids: student.vectorStores,
        filters: student.metadataFilter
      }
    ],
    tool_choice: { type: "file_search" },
    include: ["output_text", "citations", "output[*].file_search_call.search_results"]
  });

📘 Ce qui se passe ici :

Le modèle est forcé à utiliser l’outil file_search (RAG).
Il cherche dans le(s) vector_store_id (ex. VS_2NDE2) les documents correspondant à la classe et au niveau.
Il reçoit :
- output_text → la réponse finale
- citations → les sources citées
- search_results → les résultats bruts de la recherche

Étape 3 – Vérification du besoin de fallback


  const results = ragResp.output?.find?.((o: any) => o.file_search_call)?.file_search_call?.search_results ?? [];
  const needFallback = shouldFallbackToGeneral(results);

S’il n’y a pas de résultats → on active le mode général.

Étape 4 – Fallback vers le modèle général (sans RAG)


  if (needFallback) {
    const generalResp = await client.responses.create({
      model: "gpt-4o-mini",
      input: [
        {
          role: "system",
          content: [{
            type: "text",
            text: [
              "Rôle: Assistant pédagogique personnalisé du MerCollège Lycée.",
              "Aucune ressource de classe/niveau pertinente trouvée.",
              "Tu peux répondre avec connaissance générale, mais précise clairement que tu n'as pas trouvé dans les ressources officielles."
            ].join("\n")
          }]
        },
        { role: "user", content: [{ type: "text", text: message }] }
      ]
    });
    res.json({ text: generalResp.output_text, mode: "general" });
    return;
  }

Le modèle répond avec sa propre connaissance, mais précise qu’il ne s’appuie pas sur les documents officiels.

Étape 5 – Réponse normale (mode RAG)


  res.json({
    text: ragResp.output_text,
    sources: ragResp.citations,
    mode: "rag"
  });
}

Le serveur renvoie la réponse à l’élève, avec :

Le texte généré
Les sources citées
Le mode utilisé (rag ou general)

⚙️ Résumé logique

Étape	Action	Source de données	Résultat
1	Récupère le profil élève	Odoo	Classe, niveau, langue
2	Recherche RAG	Vector Store OpenAI	Réponse documentée
3	Si rien trouvé	—	Passe au fallback général
4	Répond à l’élève	JSON	Texte + sources + mode

💡 Objectif final

Ce code crée un assistant éducatif dynamique :

qui personnalise ses réponses selon la classe et le niveau ;
qui utilise d’abord les ressources pédagogiques officielles ;
qui bascule automatiquement sur la connaissance générale si besoin ;
et qui peut s’interfacer avec Odoo pour la gestion des profils élèves.

Alternative en code Python ?

Bien sûr — voici l’équivalent en Python (avec FastAPI) du flux « RAG prioritaire + fallback général », calqué sur votre router.ts.

# app.py

from typing import Literal, Optional, Dict, Any, List

from fastapi import FastAPI, HTTPException

from pydantic import BaseModel

import os

from openai import OpenAI

app = FastAPI(title="MerCollège Assistant")

client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

# ---------- Modèles ----------

class ChatRequest(BaseModel):

code: str

message: str

class StudentContext(BaseModel):

classe: str

niveau: Literal["fort", "moyen", "fragile"]

lang: Optional[str] = None

vectorStores: List[str]

metadataFilter: Optional[Dict[str, Any]] = None

# ---------- Helpers ----------

async def fetch_student_context_by_code(code: str) -> StudentContext:

"""

En prod: appelez votre contrôleur Odoo, ex.:

GET https://odoo.mercollege/api/student_profile?code=XXXX

Ici on simule la même logique que dans votre TS.

"""

vs_2nde2 = os.environ.get("VS_2NDE2")

if not vs_2nde2:

raise HTTPException(status_code=500, detail="VS_2NDE2 non défini dans l'environnement.")

return StudentContext(

classe="2nde2",

niveau="moyen",

lang="fr",

vectorStores=[vs_2nde2],

metadataFilter={"metadata": {"niveau": "moyen", "lang": "fr"}}

)

def should_fallback_to_general(search_results: Any) -> bool:

"""

Adaptez selon votre logique: seuils, scores, filtrage.

Ici: fallback si aucun résultat.

"""

items = search_results or []

return len(items) == 0

# ---------- Endpoint ----------

@app.post("/chat")

async def chat_handler(payload: ChatRequest):

# 1) Contexte élève (Odoo)

student = await fetch_student_context_by_code(payload.code)

# 2) Appel PRIORITAIRE RAG (file_search forcé)

try:

rag_resp = client.responses.create(

model="gpt-4o-mini", # ou "gpt-5-mini" si disponible/low-latency

input=[

{

"role": "system",

"content": [

{

"type": "text",

"text": "\n".join([

"Rôle: Assistant pédagogique personnalisé du MerCollège Lycée.",

"Procédure:",

"1) Tu DOIS d'abord utiliser file_search sur les ressources de la CLASSE et du NIVEAU de l'élève.",

"2) Si la recherche ne renvoie rien de pertinent, informe-le et complète avec ta connaissance générale.",

"3) Adapter la difficulté et le ton au niveau (fort/moyen/fragile).",

])

}

]

},

{"role": "user", "content": [{"type": "text", "text": payload.message}]}

],

tools=[

{

"type": "file_search",

"vector_store_ids": student.vectorStores,

# Option A: filtrer par niveau/langue

"filters": student.metadataFilter

}

],

tool_choice={"type": "file_search"}, # force la recherche

include=["output_text", "citations", "output[*].file_search_call.search_results"]

)

except Exception as e:

raise HTTPException(status_code=500, detail=f"Erreur RAG: {e}")

# 3) Décider le fallback

try:

# On parcourt rag_resp.output pour trouver l'appel file_search

output = getattr(rag_resp, "output", None)

search_results = []

if isinstance(output, list):

for o in output:

call = getattr(o, "file_search_call", None) or o.get("file_search_call") if isinstance(o, dict) else None

if call and ("search_results" in call):

search_results = call["search_results"]

break

need_fallback = should_fallback_to_general(search_results)

except Exception:

# En cas de structure inattendue, on sécurise en fallback

need_fallback = True

if need_fallback:

# 4) Réponse "générale" (sans RAG)

try:

general_resp = client.responses.create(

model="gpt-4o-mini",

input=[

{

"role": "system",

"content": [

{

"type": "text",

"text": "\n".join([

"Rôle: Assistant pédagogique personnalisé du MerCollège Lycée.",

"Aucune ressource de classe/niveau pertinente trouvée.",

"Tu peux répondre avec connaissance générale, mais précise clairement que tu n'as pas trouvé dans les ressources officielles."

])

}

]

},

{"role": "user", "content": [{"type": "text", "text": payload.message}]}

]

)

return {

"text": getattr(general_resp, "output_text", None),

"mode": "general"

}

except Exception as e:

raise HTTPException(status_code=500, detail=f"Erreur fallback général: {e}")

# 5) Réponse RAG avec sources

return {

"text": getattr(rag_resp, "output_text", None),

"sources": getattr(rag_resp, "citations", None),

"mode": "rag"

}

# ---------- Démarrage ----------

# uvicorn app:app --reload --host 0.0.0.0 --port 8000

Points clés

FastAPI pour un endpoint simple /chat.
OpenAI Python SDK (openai v=client OpenAI) et la même logique file_search que votre code TS.
Récupération du profil élève (simulée) → vector_store_ids + filters (niveau/langue).
tool_choice={"type": "file_search"} pour forcer le RAG en premier.
Fallback contrôlé par should_fallback_to_general.
Retour JSON avec mode: "rag" ou mode: "general" et éventuellement sources.

Si vous voulez, je peux aussi vous fournir une version Flask ou ajouter un timeout/latence (pour du temps réel audio) et une télémétrie simple (logs des hits RAG vs fallback).

Autre version de code : en Flash

D'abord Voici un comparatif pragmatique entre un backend en Node.js (Express), Python (FastAPI) et Python (Flask) pour votre bot RAG + temps réel audio.

Tableau 1 — Avantages / Inconvénients

Critère	Node.js (Express)	Python (FastAPI)	Python (Flask)
Modèle d’exécution	Événementiel non bloquant, excellent pour I/O intensifs et streaming	Async natif (async/await) performant, proche de Node pour I/O	Synchrone par défaut (plugins async possibles), plus simple mais moins performant en I/O pur
Latence & streaming	Très bon en SSE/WebSocket/RTC; faible overhead	Très bon en SSE/WebSocket via Uvicorn/ASGI	Correct en SSE; WebSocket possible via extensions, plus de friction
Sérialisation & schémas	TypeScript optionnel (fort atout)	Pydantic “first-class” (validation, schémas OpenAPI auto)	Non natif (ajouts via Marshmallow/Flask-Pydantic)
DX (expérience dev)	Écosystème JS massif; TS recommandé	API claire, doc auto (Swagger/Redoc), validations strictes	Minimaliste, très flexible; vous assemblez vous-même les briques
Écosystème data/IA	Moins d’outils science des données	Énorme (NumPy/Pandas/ML/LLM/RAG libs), intégration facile	Identique à FastAPI côté libs, mais moins “batteries-included” pour l’API
Intégration temps réel audio	Excellent (WebRTC, WebSocket, EventSource)	Bon (ASGI, websockets) mais un peu plus de plumbing	Faisable, moins fluide qu’Express/FastAPI
Déploiement & empreinte	Démarrage très rapide; images Docker légères	Uvicorn/Gunicorn + workers; très solide en prod	Gunicorn + workers; plus de tuning nécessaire pour scaler
Tests & typage	TS + Jest/Vitest très matures	Pytest + mypy/pyright; Pydantic aide énormément	Pytest + mypy/pyright; plus manuel
Courbe d’apprentissage	Facile si JS/TS; middleware simple	Facile si Python; conventions modernes	Ultra simple; laisse beaucoup de choix (risque d’hétérogénéité)
Perf CPU-bound	Comme tous runtimes, limité sans offload/worker	Idem; facile de déporter en tâches Celery/ProcessPool	Idem; généralement couplé à Celery/RQ
RAG/LLM toolings	SDK OpenAI OK; moins d’offres RAG “py-centric”	Plus de libs RAG (langchain, llama-index, haystack, etc.)	Même écosystème que FastAPI

Tableau 2 — Quand choisir quoi ?

Contexte	Choix recommandé	Pourquoi
Assistants conversationnels en temps réel (audio/vidéo), WebRTC, SSE massifs	Node.js (Express/WS)	Event loop et outillage temps réel excellents, latence basse, intégration front JS
API RAG, intégrations Odoo, data/analytics, ETL, scoring ML	Python (FastAPI)	Pydantic + écosystème data/IA Python, doc auto, validations strictes
POC rapide, microservice très simple, outillage minimal	Flask	Démarrage ultra rapide, code concis; bascule possible vers FastAPI plus tard
Equipe majoritairement JS/TS	Node.js	Cohérence de stack, réutilisation de compétences
Equipe majoritairement Python/data	FastAPI	Moins de friction, meilleure productivité sur data & IA
SSE/WebSocket modestes + forte validation de schémas	FastAPI	ASGI + Pydantic + OpenAPI out-of-the-box

Points d’attention concrets

RAG + faible latence audio (<500 ms)
- Node/Express: avantage pour WebSocket/SSE et orchestration front web.
- FastAPI: performe très bien si Uvicorn + HTTP/2 + keep-alive et vous streamez tôt (chunked).
Validation/qualité des payloads
- FastAPI gagne (Pydantic, types stricts, erreurs lisibles, OpenAPI auto).
- En Node, TypeScript + Zod/Yup égalent cette rigueur.
Observabilité & prod
- Node: PM2 ou containers; métriques Prometheus faciles.
- FastAPI: Uvicorn/Gunicorn workers, prometheus-fastapi-instrumentator; très standard.
- Flask: ajoutez manuellement vos middlewares/metrics.
Évolutivité “data-centric”
- Python (FastAPI) s’intègre mieux avec vos besoins RAG, ETL, stats, vérifs santé (HRV), etc.
Dette technique
- Flask “laisse faire” → architecture peut vite diverger.
- FastAPI impose une structure saine (routes, schémas, deps).
Coût/temps de dev
- Si vous avez déjà un front JS et de l’audio temps réel, Node réduit les frictions.
- Si vous multipliez les jobs data/RAG, FastAPI accélère la delivery.

Recommandation pour votre cas (MerCollège, Odoo, RAG, temps réel)

Architecture bi-modale (souvent gagnante) :
1. Service “temps réel” en Node.js (Express) pour WebSocket/SSE/Audio Realtime.
2. Service “RAG + data” en FastAPI pour la recherche, la validation stricte, l’intégration Odoo et les analytics (HRV, séries temporelles, etc.).
  → Communication interne via HTTP/JSON ou gRPC; traçage distribué (OpenTelemetry).
Si vous préférez une seule stack :
- Tout Node si votre priorité absolue = latence audio + front web.
- Tout FastAPI si votre priorité = rigueur data/RAG + outillage Python (et la latence reste <500 ms avec streaming et tuning).

Checklist de décision rapide

Besoin WebRTC/WebSocket lourd ➜ Node
Besoin Pydantic/validations/Docs auto ➜ FastAPI
POC express/ultra-léger ➜ Flask
Équipe JS majoritaire ➜ Node
Équipe data/IA majoritaire ➜ FastAPI
Observabilité “clé en main” ➜ FastAPI (léger avantage), sinon égalité

Les grandes histoires ont une personnalité. Envisagez de raconter une belle histoire qui donne de la personnalité. Écrire une histoire avec de la personnalité pour des clients potentiels aidera à établir un lien relationnel. Cela se traduit par de petites spécificités comme le choix des mots ou des phrases. Écrivez de votre point de vue, pas de l'expérience de quelqu'un d'autre.

Les grandes histoires sont pour tout le monde, même lorsqu'elles ne sont écrites que pour une seule personne. Si vous essayez d'écrire en pensant à un public large et général, votre histoire sonnera fausse et manquera d'émotion. Personne ne sera intéressé. Ecrire pour une personne en particulier signifie que si c'est authentique pour l'un, c'est authentique pour le reste.

CODE Voici une version Flask avec timeout, mesure de latence (pour l’audio temps réel) et télémétrie simple (compteurs RAG vs fallback + endpoint /metrics style Prometheus + logs JSON).

# app.py

from typing import Literal, Optional, Dict, Any, List

from flask import Flask, request, jsonify

import os, time, json, logging

from openai import OpenAI

from dataclasses import dataclass, asdict

from threading import Lock

# ---------- Config & Clients ----------

OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")

if not OPENAI_API_KEY:

raise RuntimeError("OPENAI_API_KEY manquant dans l'environnement")

client = OpenAI(api_key=OPENAI_API_KEY)

# Option: client avec timeout (secondes) pour appels OpenAI

OPENAI_TIMEOUT_S = float(os.environ.get("OPENAI_TIMEOUT_S", "6.0"))

client_tmo = client.with_options(timeout=OPENAI_TIMEOUT_S)

app = Flask(__name__)

# Logs JSON lignes pour ingestion facile (Datadog, Loki, etc.)

handler = logging.StreamHandler()

handler.setFormatter(logging.Formatter('%(message)s'))

app.logger.setLevel(logging.INFO)

app.logger.addHandler(handler)

# ---------- Télémétrie en mémoire ----------

@dataclass

class Meters:

rag_hits: int = 0

fallback_hits: int = 0

errors: int = 0

last_latency_ms: float = 0.0

meters = Meters()

meters_lock = Lock()

# ---------- Types ----------

@dataclass

class StudentContext:

classe: str

niveau: Literal["fort", "moyen", "fragile"]

lang: Optional[str]

vectorStores: List[str]

metadataFilter: Optional[Dict[str, Any]]

# ---------- Helpers ----------

def jlog(event: str, **fields):

payload = {"event": event, **fields}

app.logger.info(json.dumps(payload, ensure_ascii=False))

def fetch_student_context_by_code(code: str) -> StudentContext:

"""

En prod: appelez votre contrôleur Odoo, ex.:

GET https://odoo.mercollege/api/student_profile?code=XXXX

Ici on simule comme dans votre version TS.

"""

vs_2nde2 = os.environ.get("VS_2NDE2")

if not vs_2nde2:

raise RuntimeError("VS_2NDE2 non défini")

return StudentContext(

classe="2nde2",

niveau="moyen",

lang="fr",

vectorStores=[vs_2nde2],

metadataFilter={"metadata": {"niveau": "moyen", "lang": "fr"}},

)

def should_fallback_to_general(search_results: Any) -> bool:

items = search_results or []

return len(items) == 0 # ajustez: seuils de score, etc.

def extract_search_results(rag_resp) -> list:

"""

Extrait les search_results de la réponse Responses API.

La structure peut évoluer; on sécurise.

"""

out = getattr(rag_resp, "output", None)

if isinstance(out, list):

for o in out:

call = getattr(o, "file_search_call", None) or (o.get("file_search_call") if isinstance(o, dict) else None)

if call and isinstance(call, dict) and "search_results" in call:

return call["search_results"] or []

return []

# ---------- Routes ----------

@app.route("/chat", methods=["POST"])

def chat_handler():

t0 = time.monotonic()

req_json = request.get_json(silent=True) or {}

code = req_json.get("code")

message = req_json.get("message")

if not code or not message:

return jsonify({"error": "Paramètres requis: code, message"}), 400

try:

student = fetch_student_context_by_code(code)

except Exception as e:

with meters_lock:

meters.errors += 1

jlog("student_context_error", code=code, error=str(e))

return jsonify({"error": f"Profil élève indisponible: {e}"}), 500

# ---- Appel RAG prioritaire ----

try:

rag_resp = client_tmo.responses.create(

model=os.environ.get("OPENAI_MODEL_RAG", "gpt-4o-mini"),

input=[

{

"role": "system",

"content": [{

"type": "text",

"text": "\n".join([

"Rôle: Assistant pédagogique personnalisé du MerCollège Lycée.",

"Procédure:",

"1) Tu DOIS d'abord utiliser file_search sur les ressources de la CLASSE et du NIVEAU de l'élève.",

"2) Si la recherche ne renvoie rien de pertinent, informe-le et complète avec ta connaissance générale.",

"3) Adapte la difficulté et le ton au niveau (fort/moyen/fragile).",

])

}]

},

{"role": "user", "content": [{"type": "text", "text": message}]}

],

tools=[{

"type": "file_search",

"vector_store_ids": student.vectorStores,

"filters": student.metadataFilter

}],

tool_choice={"type": "file_search"},

include=["output_text", "citations", "output[*].file_search_call.search_results"]

)

except Exception as e:

with meters_lock:

meters.errors += 1

jlog("rag_call_error", code=code, error=str(e))

return jsonify({"error": f"Erreur RAG: {e}"}), 502

search_results = extract_search_results(rag_resp)

need_fallback = should_fallback_to_general(search_results)

if need_fallback:

# ---- Fallback général ----

try:

general_resp = client_tmo.responses.create(

model=os.environ.get("OPENAI_MODEL_GENERAL", "gpt-4o-mini"),

input=[

{

"role": "system",

"content": [{

"type": "text",

"text": "\n".join([

"Rôle: Assistant pédagogique personnalisé du MerCollège Lycée.",

"Aucune ressource de classe/niveau pertinente trouvée.",

"Tu peux répondre avec connaissance générale, mais précise clairement que tu n'as pas trouvé dans les ressources officielles."

])

}]

},

{"role": "user", "content": [{"type": "text", "text": message}]}

]

)

latency_ms = (time.monotonic() - t0) * 1000.0

with meters_lock:

meters.fallback_hits += 1

meters.last_latency_ms = latency_ms

jlog("chat_fallback",

code=code, classe=student.classe, niveau=student.niveau,

latency_ms=latency_ms, results_count=len(search_results))

return jsonify({

"text": getattr(general_resp, "output_text", None),

"mode": "general",

"latency_ms": round(latency_ms, 1)

})

except Exception as e:

with meters_lock:

meters.errors += 1

jlog("fallback_error", code=code, error=str(e))

return jsonify({"error": f"Erreur fallback: {e}"}), 502

# ---- Succès RAG ----

latency_ms = (time.monotonic() - t0) * 1000.0

with meters_lock:

meters.rag_hits += 1

meters.last_latency_ms = latency_ms

jlog("chat_rag",

code=code, classe=student.classe, niveau=student.niveau,

latency_ms=latency_ms, results_count=len(search_results))

return jsonify({

"text": getattr(rag_resp, "output_text", None),

"sources": getattr(rag_resp, "citations", None),

"mode": "rag",

"latency_ms": round(latency_ms, 1)

})

@app.route("/metrics", methods=["GET"])

def metrics():

"""

Exposition simple (format Prometheus) + JSON brut pour debug rapide.

"""

with meters_lock:

m = asdict(meters)

lines = [

"# HELP mercollege_rag_hits Total réponses RAG",

"# TYPE mercollege_rag_hits counter",

f"mercollege_rag_hits {m['rag_hits']}",

"# HELP mercollege_fallback_hits Total réponses fallback",

"# TYPE mercollege_fallback_hits counter",

f"mercollege_fallback_hits {m['fallback_hits']}",

"# HELP mercollege_errors Total erreurs",

"# TYPE mercollege_errors counter",

f"mercollege_errors {m['errors']}",

"# HELP mercollege_last_latency_ms Dernière latence en ms",

"# TYPE mercollege_last_latency_ms gauge",

f"mercollege_last_latency_ms {m['last_latency_ms']}",

"",

"# JSON mirror (debug)",

f"# {json.dumps(m)}"

]

return ("\n".join(lines), 200, {"Content-Type": "text/plain; version=0.0.4"})

# ---------- Lancement ----------

# Dév: python app.py

# Prod: gunicorn -w 2 -k gthread -b 0.0.0.0:8000 --threads 8 --timeout 30 app:app

if __name__ == "__main__":

app.run(host="0.0.0.0", port=int(os.environ.get("PORT", "8000")), debug=True)

Variables d’environnement utiles


OPENAI_API_KEY=xxxx
VS_2NDE2=vs_XXXXXXXXXXXX            # Vector Store de la classe
OPENAI_TIMEOUT_S=6.0               # Timeout API OpenAI (secondes)
OPENAI_MODEL_RAG=gpt-4o-mini       # ou gpt-5-mini si dispo
OPENAI_MODEL_GENERAL=gpt-4o-mini
PORT=8000

Conseils “temps réel audio” (latence < ~500 ms)

Pré-échauffez (warm-up) le process au boot avec un appel court.
Timeout agressif (ex. OPENAI_TIMEOUT_S=3.0) pour détecter tôt des lenteurs et basculer vite en fallback.
Chunking/streaming côté client** : commencez à parler dès les premiers tokens si vous utilisez l’API Realtime Audio.
Caches: gardez le contexte system stable, mutualisez les vector stores par classe, et filtrez finement (niveau, lang).
Monitoring: suivez latency_ms par mode (RAG vs fallback) et établissez un SLO (ex. P95 < 450 ms).

Bot adaptatif classe et niveau > Routage & appel OpenAI (Node/TypeScript)

Backend du bot (Node/TS)

Routage & appel OpenAI (Node/TypeScript)

🔍 Vue d’ensemble

🧱 Structure générale du fichier

1️⃣ Importations et initialisation

2️⃣ Définition du type StudentContext

3️⃣ Fonction fetchStudentContextByCode

4️⃣ Fonction shouldFallbackToGeneral

5️⃣ Fonction principale chatHandler

Étape 1 – Récupération du profil

Étape 2 – Requête RAG (avec file_search)

Étape 3 – Vérification du besoin de fallback

Étape 4 – Fallback vers le modèle général (sans RAG)

Étape 5 – Réponse normale (mode RAG)

⚙️ Résumé logique

💡 Objectif final

Points clés

Tableau 1 — Avantages / Inconvénients

Tableau 2 — Quand choisir quoi ?

Points d’attention concrets

Recommandation pour votre cas (MerCollège, Odoo, RAG, temps réel)

Checklist de décision rapide

Variables d’environnement utiles

Conseils “temps réel audio” (latence < ~500 ms)