Tutor personalizat cu LLM: RAG pentru fundamentarea cunoștințelor
Visul unui profesor privat pentru fiecare elev, disponibil 24 de ore pe zi, capabil să se adapteze la nivel si stilul de invatare al tuturor, nu mai este science fiction. THE Model de limbă mare (LLM) combinate cu tehnici Recuperare-Augmented Generation (RAG) ei fac posibil construirea de tutori AI personalizați care depășesc limitările chatbot-urilor generice și ale întrebărilor frecvente statice.
Problema centrală a LLM-urilor în context educațional este fundamentarea cunoștințelor: un model ca GPT-4o sau Llama 3 are cunoștințe generale vaste, dar nu cunoaște programul specific a cursului de analiză matematică de la Universitatea din Bologna, notițele de curs ale profesorului, întrebările de la examenele din anii precedenți sau concepțiile greșite tipice ale studenților din anul I. Fără împământare, tutorele AI riscă să dea răspunsuri plauzibile, dar incorecte din punct de vedere pedagogic sau în afara contextului.
În acest articol al seriei Inginerie EdTech vom construi un tutor AI complet cu LLM și RAG: de la conducta de indexare a documentelor de învățământ, la balustradele pedagogice care împiedică modelul să ofere direct soluții exercițiilor, până la feedback adaptiv pe baza profilului studentului. Toate cu exemple concrete în Python și TypeScript.
Ce veți învăța în acest articol
- Arhitectură end-to-end a unui tutor AI cu LLM și RAG
- Tubul de indexare a conținutului educațional (PDF, transcriere video, test)
- Întemeierea cunoștințelor: cum să limitați LLM la materialul de curs
- Balustrade pedagogice pentru a promova gândirea critică, nu răspunsul direct
- Profilul elevului și personalizarea feedback-ului adaptiv
- Gestionarea memoriei conversaționale cu mai multe sesiuni
- Evaluarea calității răspunsurilor cu metrici RAG (fidelitate, relevanță)
- Implementare scalabilă cu FastAPI și cache semantică
1. de ce RAG for Educational Tutors
Un LLM pur, fără acces la cunoștințe specifice domeniului, suferă de trei probleme critice în context educațional: halucinații (informații inventate, dar plauzibile), cunoștințe învechite (firma de cunoștințe la data formării) e lipsa contextului curricular (nu știe ce a studiat deja studentul, ce carte folosește, ce parte a programului a fost acoperită).
Cercetările academice 2024-2025 demonstrează că sistemele RAG aplicate educației se reduc halucinații cu 80% comparativ cu LLM-urile pure și cresc satisfacția studenților cu 40% datorită răspunsurilor ancorate la materialul de curs. Sistemul LPItutor (2025) a demonstrat modele open-source cu 7-17 miliarde de parametri cu o conductă RAG bună atingeți performanțe comparabile cu GPT-4o, făcând implementarea on-premise fezabilă chiar şi pentru instituţiile cu bugete limitate.
Conceptul cheie este fundamentarea cunoștințelor: ancorați răspunsurile LLM la documente verificate și specifice contextului (fișe, manuale, exerciții rezolvate). Când elevul întreabă „Cum calculezi derivata sin(x)?”, tutorele nu accesează la cunoștințele sale generice, dar recuperează definiția exactă folosită în curs, cu notație a profesorului şi exemplele cărţii adoptate.
Arhitectură la nivel înalt
| Componentă | Tehnologie | Funcţie |
|---|---|---|
| Ingestie de documente | LangChain, PyMuPDF | Analizarea PDF, diapozitiv, transcriere |
| Model de încorporare | text-embedding-3-mic, BGE-M3 | Vectorizarea fragmentelor de text |
| Magazin de vectori | pgvector, Qdrant, Chroma | Stocare și recuperare semantică |
| LLM | GPT-4o, Llama 3.1, Mistral | Generarea răspunsului pedagogic |
| Memorie | Redis, PostgreSQL | Sesiuni de conversație |
| Strat de balustradă | Solicitare personalizată, NeMo Guardrails | Controlul pedagogic |
| Profilul studentului | PostgreSQL, Redis Cache | Nivel, istoric, preferințe |
| Stratul API | FastAPI, WebSocket | Interfață de streaming |
2. Conducta de indexare a documentelor educaționale
Primul pas este construirea bazei de cunoștințe a tutorelui. Materialele didactice au caracteristici documente specifice versus generice: formule matematice, cod sursă, scheme, tabele și un vocabular tehnic precis. Strategia de fragmentare trebuie să păstreze coerența semantică.
# pipeline/document_ingestion.py
import hashlib
from pathlib import Path
from typing import List, Dict, Any
from dataclasses import dataclass, field
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyMuPDFLoader, DirectoryLoader
from langchain.schema import Document
@dataclass
class ChunkConfig:
chunk_size: int = 512
chunk_overlap: int = 64
separators: List[str] = field(default_factory=lambda: [
"\n## ", "\n### ", "\n\n", "\n", ". ", " "
])
@dataclass
class CourseMetadata:
course_id: str
tenant_id: str
document_type: str # 'lecture', 'textbook', 'exercise', 'exam'
topic: str
difficulty_level: int # 1-5
class CourseDocumentPipeline:
def __init__(
self,
vector_store,
embedding_model,
config: ChunkConfig = None
):
self.vector_store = vector_store
self.embedding_model = embedding_model
self.config = config or ChunkConfig()
self.splitter = RecursiveCharacterTextSplitter(
chunk_size=self.config.chunk_size,
chunk_overlap=self.config.chunk_overlap,
separators=self.config.separators,
)
def ingest_pdf(
self,
file_path: str,
metadata: CourseMetadata
) -> int:
"""Carica un PDF, lo divide in chunk e indicizza."""
loader = PyMuPDFLoader(file_path)
raw_docs = loader.load()
# Arricchisci i metadati di ogni documento
enriched_docs = [
Document(
page_content=doc.page_content,
metadata={
**doc.metadata,
"course_id": metadata.course_id,
"tenant_id": metadata.tenant_id,
"document_type": metadata.document_type,
"topic": metadata.topic,
"difficulty_level": metadata.difficulty_level,
"source_hash": self._hash_content(doc.page_content),
}
)
for doc in raw_docs
]
# Split in chunk semantici
chunks = self.splitter.split_documents(enriched_docs)
# De-duplicazione basata su hash del contenuto
unique_chunks = self._deduplicate(chunks)
# Batch insertion nel vector store
self.vector_store.add_documents(unique_chunks, batch_size=100)
return len(unique_chunks)
def ingest_video_transcript(
self,
transcript: str,
timestamps: List[Dict],
metadata: CourseMetadata
) -> int:
"""Indicizza trascrizioni di video lezioni con timestamp."""
# Dividi per blocchi temporali (ogni 2 minuti di lezione)
chunks = self._split_transcript_by_time(transcript, timestamps, window_seconds=120)
docs = [
Document(
page_content=chunk["text"],
metadata={
"course_id": metadata.course_id,
"tenant_id": metadata.tenant_id,
"document_type": "video_transcript",
"topic": metadata.topic,
"start_time": chunk["start"],
"end_time": chunk["end"],
"video_url": chunk.get("video_url", ""),
}
)
for chunk in chunks
]
self.vector_store.add_documents(docs)
return len(docs)
def _hash_content(self, content: str) -> str:
return hashlib.sha256(content.encode()).hexdigest()[:16]
def _deduplicate(self, chunks: List[Document]) -> List[Document]:
seen = set()
unique = []
for chunk in chunks:
h = self._hash_content(chunk.page_content)
if h not in seen:
seen.add(h)
unique.append(chunk)
return unique
def _split_transcript_by_time(
self,
transcript: str,
timestamps: List[Dict],
window_seconds: int
) -> List[Dict]:
"""Raggruppa le parole della trascrizione in finestre temporali."""
chunks = []
current_chunk_words = []
current_start = timestamps[0]["start"] if timestamps else 0
words = transcript.split()
for i, (word, ts) in enumerate(zip(words, timestamps)):
current_chunk_words.append(word)
if ts["start"] - current_start >= window_seconds:
chunks.append({
"text": " ".join(current_chunk_words),
"start": current_start,
"end": ts["start"],
})
current_chunk_words = []
current_start = ts["start"]
if current_chunk_words:
chunks.append({
"text": " ".join(current_chunk_words),
"start": current_start,
"end": timestamps[-1]["end"] if timestamps else 0,
})
return chunks3. Recuperare și fundamentare a cunoștințelor
Recuperarea este inima sistemului RAG. Nu este suficient să regăsiți bucățile cele mai asemănătoare cu întrebarea: în context educaţional trebuie să avem în vedere şi nivelul de dificultate al studentului, cel subiect actual în program și în tip de document (de exemplu, preferă exercițiile rezolvate când elevul cere să exerseze).
# rag/retriever.py
from typing import List, Optional
from dataclasses import dataclass
from enum import Enum
class QueryIntent(Enum):
CONCEPT_EXPLANATION = "concept"
EXERCISE_HELP = "exercise"
EXAM_PREPARATION = "exam"
DEFINITION = "definition"
COMPARISON = "comparison"
@dataclass
class StudentProfile:
student_id: str
course_id: str
difficulty_level: int # 1-5 (adattivo)
current_topic: str
mastered_topics: List[str]
weak_areas: List[str]
preferred_style: str # 'visual', 'text', 'example-first'
@dataclass
class RetrievalContext:
query: str
student: StudentProfile
intent: QueryIntent
top_k: int = 5
class AdaptiveRetriever:
def __init__(self, vector_store, intent_classifier):
self.vector_store = vector_store
self.intent_classifier = intent_classifier
def retrieve(self, context: RetrievalContext) -> List[dict]:
"""
Retrieval adattivo che considera il profilo studente.
"""
intent = context.intent or self.intent_classifier.classify(context.query)
# Costruisci filtri metadata basati sul profilo
metadata_filter = self._build_filter(context.student, intent)
# Hybrid search: semantico + keyword per termini tecnici
semantic_results = self.vector_store.similarity_search_with_score(
query=context.query,
k=context.top_k * 2,
filter=metadata_filter,
)
# Re-ranking: penalizza documenti troppo avanzati o già masterizzati
reranked = self._rerank(
results=semantic_results,
student=context.student,
intent=intent,
)
return reranked[:context.top_k]
def _build_filter(
self,
student: StudentProfile,
intent: QueryIntent
) -> dict:
base_filter = {
"course_id": student.course_id,
"difficulty_level": {"$lte": student.difficulty_level + 1},
}
if intent == QueryIntent.EXERCISE_HELP:
base_filter["document_type"] = {"$in": ["exercise", "exam"]}
elif intent == QueryIntent.CONCEPT_EXPLANATION:
base_filter["document_type"] = {"$in": ["lecture", "textbook"]}
elif intent == QueryIntent.EXAM_PREPARATION:
base_filter["document_type"] = {"$in": ["exam", "exercise", "summary"]}
return base_filter
def _rerank(
self,
results: List[tuple],
student: StudentProfile,
intent: QueryIntent,
) -> List[dict]:
scored = []
for doc, semantic_score in results:
score = semantic_score
# Boost se il documento e sul topic corrente
if doc.metadata.get("topic") == student.current_topic:
score *= 1.3
# Penalizza se il topic e già masterizzato (mostra contenuti avanzati)
if doc.metadata.get("topic") in student.mastered_topics:
score *= 0.7
# Boost per aree deboli dello studente
if doc.metadata.get("topic") in student.weak_areas:
score *= 1.5
scored.append({"document": doc, "score": score})
return sorted(scored, key=lambda x: x["score"], reverse=True)4. Paravane pedagogice: Tutorul nu dă răspunsurile
Cel mai mare risc al unui tutor AI fără balustrade este că acesta devine un instrument de copiere exercitiile. Un bun tutore pedagogic nu dă un răspuns direct, ci îl îndrumă pe elev spre rezolvare prin întrebări socratice, sugestii graduale (schele) și feedback asupra erorilor conceptuale.
Implementăm un sistem de balustradă pe trei niveluri: clasificarea intențiilor (cereți răspunsul sau aveți o îndoială conceptuală?), politica pedagogica (câtă schelă să aplici?) e inginerie promptă (cum se formulează răspunsul LLM la promovarea învățării active).
# guardrails/pedagogical_guardrail.py
from enum import Enum
from typing import Optional
from pydantic import BaseModel
class ScaffoldingLevel(Enum):
HINT = "hint" # Solo un indizio
GUIDED = "guided" # Domande socratiche
STEP_BY_STEP = "steps" # Breakdown del processo
EXAMPLE = "example" # Esempio analogo (non la soluzione)
SOLUTION = "solution" # Soluzione completa (solo per esercizi risolti)
class PedagogicalPolicy(BaseModel):
allow_direct_answer: bool = False
max_scaffolding_level: ScaffoldingLevel = ScaffoldingLevel.GUIDED
promote_reflection: bool = True
suggest_resources: bool = True
track_misconceptions: bool = True
SYSTEM_PROMPT_TEMPLATE = """Sei un tutor educativo AI specializzato nel corso "{course_name}".
PROFILO STUDENTE:
- Livello: {difficulty_level}/5
- Topic corrente: {current_topic}
- Aree di debolezza: {weak_areas}
CONTESTO DEL CORSO (recuperato dalla knowledge base):
{retrieved_context}
REGOLE PEDAGOGICHE FONDAMENTALI:
1. NON fornire mai la risposta diretta a un esercizio non ancora risolto
2. Usa domande socratiche per guidare la riflessione ("Cosa succede se...?", "perchè pensi che...?")
3. Identifica le misconcezioni dello studente e correggile con gentilezza
4. Adatta il linguaggio al livello {difficulty_level}/5:
- Livello 1-2: linguaggio semplice, molti esempi quotidiani
- Livello 3: bilanciato tra intuizione e rigore
- Livello 4-5: terminologia tecnica precisa, proofs formali
5. Suggerisci sempre il materiale specifico del corso dove approfondire
6. Se lo studente e bloccato dopo 3 tentativi, aumenta gradualmente il supporto
7. Celebra i progressi e normalizza gli errori come parte dell'apprendimento
RISPOSTA:"""
class PedagogicalGuardrail:
def __init__(self, llm_client, policy: PedagogicalPolicy = None):
self.llm = llm_client
self.policy = policy or PedagogicalPolicy()
async def generate_response(
self,
query: str,
student: "StudentProfile",
retrieved_docs: list,
conversation_history: list,
course_name: str,
) -> dict:
# Classifica se la domanda chiede direttamente una soluzione
is_homework_request = await self._detect_homework_request(query)
# Scegli il livello di scaffolding appropriato
scaffolding = self._choose_scaffolding(
student=student,
is_homework=is_homework_request,
attempt_count=self._count_attempts(conversation_history, query),
)
# Costruisci il contesto RAG
context = self._format_context(retrieved_docs)
# Costruisci il prompt
system_prompt = SYSTEM_PROMPT_TEMPLATE.format(
course_name=course_name,
difficulty_level=student.difficulty_level,
current_topic=student.current_topic,
weak_areas=", ".join(student.weak_areas),
retrieved_context=context,
)
# Aggiungi istruzioni di scaffolding
scaffolding_instruction = self._get_scaffolding_instruction(scaffolding)
full_system = f"{system_prompt}\n\nMODALITA RISPOSTA: {scaffolding_instruction}"
response = await self.llm.chat(
system=full_system,
messages=conversation_history + [{"role": "user", "content": query}],
temperature=0.3, # Bassa temperatura per risposte più accurate e coerenti
max_tokens=1024,
)
return {
"content": response.content,
"scaffolding_used": scaffolding.value,
"sources": [doc["document"].metadata for doc in retrieved_docs],
}
def _choose_scaffolding(
self,
student,
is_homework: bool,
attempt_count: int,
) -> ScaffoldingLevel:
if not is_homework:
return ScaffoldingLevel.GUIDED
if attempt_count == 0:
return ScaffoldingLevel.HINT
elif attempt_count == 1:
return ScaffoldingLevel.GUIDED
elif attempt_count == 2:
return ScaffoldingLevel.STEP_BY_STEP
elif attempt_count >= 3:
return ScaffoldingLevel.EXAMPLE
else:
return ScaffoldingLevel.SOLUTION if self.policy.allow_direct_answer else ScaffoldingLevel.EXAMPLE
def _get_scaffolding_instruction(self, level: ScaffoldingLevel) -> str:
instructions = {
ScaffoldingLevel.HINT: "Fornisci solo un breve indizio (1-2 frasi) che metta lo studente sulla giusta strada. Non procedere oltre.",
ScaffoldingLevel.GUIDED: "Usa domande socratiche. Non dare la risposta, ma guida lo studente con 2-3 domande che stimolino la riflessione.",
ScaffoldingLevel.STEP_BY_STEP: "Scomponi il problema in passi. Descrivi i passi da seguire senza eseguirli tu. Chiedi allo studente di provare ogni passo.",
ScaffoldingLevel.EXAMPLE: "Mostra un esempio ANALOGO ma non identico al problema. Spiega l'esempio, poi chiedi allo studente di applicare lo stesso ragionamento.",
ScaffoldingLevel.SOLUTION: "Fornisci la soluzione completa con spiegazione dettagliata di ogni passaggio.",
}
return instructions.get(level, instructions[ScaffoldingLevel.GUIDED])
async def _detect_homework_request(self, query: str) -> bool:
"""Classifica se la domanda chiede la risposta a un esercizio."""
keywords = ["risolvi", "calcola", "trova", "dimostra", "soluzione", "risposta",
"solve", "calculate", "find", "answer", "result", "quanto fa"]
query_lower = query.lower()
return any(kw in query_lower for kw in keywords)
def _count_attempts(self, history: list, current_query: str) -> int:
"""Conta quante volte lo studente ha chiesto aiuto sullo stesso tema."""
similar_attempts = sum(
1 for msg in history
if msg["role"] == "user" and self._is_similar_query(msg["content"], current_query)
)
return similar_attempts
def _is_similar_query(self, q1: str, q2: str) -> bool:
words1 = set(q1.lower().split())
words2 = set(q2.lower().split())
overlap = len(words1 & words2) / max(len(words1 | words2), 1)
return overlap > 0.5
def _format_context(self, docs: list) -> str:
sections = []
for i, item in enumerate(docs, 1):
doc = item["document"]
source = doc.metadata.get("document_type", "documento")
topic = doc.metadata.get("topic", "")
sections.append(f"[Fonte {i} - {source} su '{topic}']\n{doc.page_content}")
return "\n\n---\n\n".join(sections)5. Memoria conversațională cu mai multe sesiuni
Un mentor eficient își amintește conversațiile anterioare. Dacă elevul avea dificultăţi cu derivate săptămâna trecută, tutorele trebuie să țină cont de acest lucru atunci când răspunde la întrebări despre integrale. Implementăm o memorie pe două niveluri: memorie pe termen scurt (conversație curentă, Redis) e memorie pe termen lung (istoricul sesiunii, PostgreSQL cu rezumate LLM).
# memory/session_manager.py
import json
from datetime import datetime, timedelta
from typing import List, Optional
import redis.asyncio as redis
from sqlalchemy.ext.asyncio import AsyncSession
class TutorMemoryManager:
SHORT_TERM_TTL = 3600 # 1 ora per sessione attiva
MAX_SHORT_TERM_MESSAGES = 20 # Finestra conversazione
def __init__(self, redis_client: redis.Redis, db_session: AsyncSession, llm_client):
self.redis = redis_client
self.db = db_session
self.llm = llm_client
async def get_conversation_history(
self,
student_id: str,
session_id: str
) -> List[dict]:
"""Recupera storia conversazione dalla cache Redis."""
key = f"tutor:session:{student_id}:{session_id}"
raw = await self.redis.get(key)
if raw:
return json.loads(raw)
# Se non in cache, prova a recuperare dall'ultimo riassunto
summary = await self._get_session_summary(student_id)
if summary:
return [{"role": "system", "content": f"Riassunto sessioni precedenti: {summary}"}]
return []
async def save_message(
self,
student_id: str,
session_id: str,
role: str,
content: str,
) -> None:
key = f"tutor:session:{student_id}:{session_id}"
history = await self.get_conversation_history(student_id, session_id)
# Rimuovi il messaggio di sistema con il riassunto se presente
history = [m for m in history if m.get("role") != "system"]
history.append({"role": role, "content": content, "timestamp": datetime.utcnow().isoformat()})
# Mantieni solo gli ultimi N messaggi (finestra scorrevole)
if len(history) > self.MAX_SHORT_TERM_MESSAGES:
await self._archive_old_messages(student_id, history[:-self.MAX_SHORT_TERM_MESSAGES])
history = history[-self.MAX_SHORT_TERM_MESSAGES:]
await self.redis.setex(key, self.SHORT_TERM_TTL, json.dumps(history))
async def end_session(self, student_id: str, session_id: str) -> None:
"""Chiudi sessione: genera riassunto e aggiorna profilo studente."""
history = await self.get_conversation_history(student_id, session_id)
if len(history) < 3:
return # Sessione troppo breve per riassumere
summary = await self._generate_session_summary(history)
misconceptions = await self._extract_misconceptions(history)
# Salva in PostgreSQL
await self.db.execute(
"""INSERT INTO tutor_sessions
(student_id, session_id, summary, misconceptions, created_at)
VALUES (:sid, :sess, :summary, :misc, :ts)""",
{
"sid": student_id,
"sess": session_id,
"summary": summary,
"misc": json.dumps(misconceptions),
"ts": datetime.utcnow(),
},
)
await self.db.commit()
# Aggiorna il profilo studente con le nuove misconcezioni
if misconceptions:
await self._update_student_weak_areas(student_id, misconceptions)
# Elimina dalla cache
key = f"tutor:session:{student_id}:{session_id}"
await self.redis.delete(key)
async def _generate_session_summary(self, history: List[dict]) -> str:
messages_text = "\n".join(
f"{m['role'].upper()}: {m['content']}"
for m in history if m.get("role") in ("user", "assistant")
)
prompt = f"""Riassumi in 3-4 frasi questa sessione di tutoring educativo.
Includi: argomenti discussi, difficolta incontrate, progressi dello studente.
Sessione:
{messages_text}
Riassunto conciso:"""
response = await self.llm.complete(prompt, max_tokens=200)
return response.text
async def _extract_misconceptions(self, history: List[dict]) -> List[str]:
"""Estrai le misconcezioni rilevate durante la sessione."""
# Implementazione semplificata basata su keyword
misconceptions = []
for msg in history:
if msg.get("role") == "assistant" and "misconcezione" in msg.get("content", "").lower():
misconceptions.append(msg["content"][:100])
return misconceptions
async def _get_session_summary(self, student_id: str) -> Optional[str]:
result = await self.db.execute(
"""SELECT summary FROM tutor_sessions
WHERE student_id = :sid
ORDER BY created_at DESC LIMIT 3""",
{"sid": student_id},
)
rows = result.fetchall()
if rows:
return " | ".join(row[0] for row in rows)
return None
async def _update_student_weak_areas(self, student_id: str, misconceptions: List[str]) -> None:
await self.db.execute(
"""UPDATE student_profiles
SET weak_areas = weak_areas || :misc::jsonb
WHERE student_id = :sid""",
{"sid": student_id, "misc": json.dumps(misconceptions)},
)
await self.db.commit()6. API Streaming cu FastAPI
Experiența utilizatorului unui tutor AI se îmbunătățește enorm cu aceasta streaming de răspunsuri: elevul vede textul apărând treptat, de parcă tutorele scria în timp real. Implementăm un punct final FastAPI cu Server-Sent Events (SSE).
# api/tutor_endpoint.py
from fastapi import FastAPI, Depends, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from typing import AsyncGenerator
import json
import uuid
app = FastAPI(title="EdTech AI Tutor API")
class TutorRequest(BaseModel):
student_id: str
query: str
session_id: str = None
course_id: str
@app.post("/api/tutor/stream")
async def tutor_stream(
request: TutorRequest,
retriever: AdaptiveRetriever = Depends(get_retriever),
guardrail: PedagogicalGuardrail = Depends(get_guardrail),
memory: TutorMemoryManager = Depends(get_memory),
):
session_id = request.session_id or str(uuid.uuid4())
async def generate() -> AsyncGenerator[str, None]:
try:
# 1. Carica profilo studente
student = await get_student_profile(request.student_id, request.course_id)
# 2. Recupera storico conversazione
history = await memory.get_conversation_history(request.student_id, session_id)
# 3. Salva il messaggio utente
await memory.save_message(request.student_id, session_id, "user", request.query)
# 4. Retrieval adattivo
context = RetrievalContext(
query=request.query,
student=student,
intent=None, # classificato automaticamente
)
docs = retriever.retrieve(context)
# 5. Genera risposta con guardrail pedagogici (streaming)
full_response = ""
async for chunk in guardrail.generate_response_stream(
query=request.query,
student=student,
retrieved_docs=docs,
conversation_history=history,
course_name=await get_course_name(request.course_id),
):
full_response += chunk
yield f"data: {json.dumps({'chunk': chunk, 'session_id': session_id})}\n\n"
# 6. Salva risposta in memoria
await memory.save_message(request.student_id, session_id, "assistant", full_response)
# 7. Invia metadata finali
yield f"data: {json.dumps({'done': True, 'session_id': session_id})}\n\n"
except Exception as e:
yield f"data: {json.dumps({'error': str(e)})}\n\n"
return StreamingResponse(
generate(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no",
},
)7. Evaluarea calității RAG
Un tutor AI în producție trebuie monitorizat continuu. Folosim cadrul RAGAS (Evaluare RAG) pentru a evalua patru dimensiuni: fidelitate (răspunsul este fidel documentelor recuperate?), răspuns. relevanţă (răspunsul este relevant pentru întrebare?), precizia contextului (sunt relevante documentele recuperate?) e amintirea contextului (am recuperat toate documentele necesare?).
# evaluation/rag_evaluator.py
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
)
from datasets import Dataset
from typing import List, Dict
import pandas as pd
class TutorRAGEvaluator:
def __init__(self, llm_client, embedding_model):
self.llm = llm_client
self.embeddings = embedding_model
self.metrics = [
faithfulness,
answer_relevancy,
context_precision,
context_recall,
]
def evaluate_batch(
self,
test_cases: List[Dict],
ground_truths: List[str],
) -> pd.DataFrame:
"""
Valuta un batch di interazioni tutor.
test_cases: list di {question, answer, contexts}
ground_truths: risposte attese (da esperti didattici)
"""
dataset = Dataset.from_dict({
"question": [tc["question"] for tc in test_cases],
"answer": [tc["answer"] for tc in test_cases],
"contexts": [tc["contexts"] for tc in test_cases],
"ground_truth": ground_truths,
})
results = evaluate(
dataset=dataset,
metrics=self.metrics,
llm=self.llm,
embeddings=self.embeddings,
)
return results.to_pandas()
def evaluate_pedagogical_quality(self, responses: List[Dict]) -> Dict:
"""
Valuta la qualità pedagogica delle risposte:
- Tasso di risposte dirette (dovrebbero essere basse per esercizi)
- Uso di domande socratiche
- Presenza di suggerimenti di risorse
"""
direct_answer_count = 0
socratic_question_count = 0
resource_suggestion_count = 0
for resp in responses:
content = resp.get("content", "").lower()
if resp.get("scaffolding_used") == "solution":
direct_answer_count += 1
if "?" in content:
socratic_question_count += 1
if any(kw in content for kw in ["vedi capitolo", "consulta", "approfondisci", "leggi"]):
resource_suggestion_count += 1
total = len(responses)
return {
"direct_answer_rate": direct_answer_count / total if total else 0,
"socratic_rate": socratic_question_count / total if total else 0,
"resource_suggestion_rate": resource_suggestion_count / total if total else 0,
"total_evaluated": total,
}Anti-modele de evitat
- RAG fără filtre pentru chiriași: Nu partajați niciodată documente între diferite cursuri sau instituții. Filtrați întotdeauna după tenant_id și course_id.
- Bucăți prea mari: Bucățile de peste 2000 de jetoane diluează relevanța. Utilizați 512-768 jetoane cu 10-15% suprapunere.
- Temperatura ridicata: Temperaturile mai mari de 0,5 cresc halucinațiile. Pentru tutorii educaționali folosiți 0,2-0,4.
- Fara balustrade: Un LLM fără balustrade pedagogice devine un sistem de copiere a temelor. Parapetele sunt esențiale, nu opționale.
- Memorie infinită: Încărcarea întregului istoric conversațional depășește fereastra de context și crește costurile. Utilizați ferestre glisante și rezumate.
- Fără evaluare: Fără RAGAS sau valori similare, nu știi dacă profesorul tău realizează performanțe bune.
Concluzii și pașii următori
Am construit arhitectura completă a unui tutor AI bazat pe LLM și RAG: de la conducta de indexare a materialelor educaționale, la recuperarea adaptivă care ține cont de profilul studentului, de balustradele pedagogice pe care le promovează învățare activă, până la memorie cu mai multe sesiuni și monitorizare a calității.
Rezultatul este un sistem care nu răspunde doar la întrebări, ci ghid elevul prin procesul de învățare, adaptându-se la nivelul său, identificarea concepțiilor greșite și promovarea reflecției critice, toate ancorat în materialul de curs specific și nu în cunoștințele generice LLM.
În următorul articol din serie, vom explora construirea unui Motor de gamification cu mașini de stat și mecanici de angajare care cresc motivația și persistența elevilor pe platformă.
Seria EdTech Engineering
- Arhitectură LMS scalabilă: model multi-chiriași
- Algoritmi de învățare adaptivă: de la teorie la producție
- Streaming video pentru educație: WebRTC vs HLS vs DASH
- Sisteme de supraveghere AI: confidențialitate-în primul rând cu computer Vision
- Tutor personalizat cu LLM: RAG for Knowledge Grounding (acest articol)
- Motor de gamification: Arhitectură și mașină de stat
- Learning Analytics: Data Pipeline cu xAPI și Kafka
- Colaborare în timp real în EdTech: CRDT și WebSocket
- Mobile-First EdTech: Offline-First Architecture
- Managementul conținutului cu mai mulți chiriași: Versiune și SCORM