Personalizovaný lektor s LLM: RAG pro ukotvení znalostí
Sen o soukromém učiteli pro každého studenta, dostupný 24 hodin denně, schopný se přizpůsobit dané úrovni a styl učení každého, už to není sci-fi. THE Velký jazykový model (LLM) v kombinaci s technikami Retrieval-Augmented Generation (RAG) oni to umožňují vytváření personalizovaných učitelů umělé inteligence, kteří překonávají omezení obecných chatbotů a statických častých dotazů.
Ústředním problémem LLM ve vzdělávacím kontextu je znalostní uzemnění: modelka jako GPT-4o nebo Llama 3 má rozsáhlé obecné znalosti, ale nezná konkrétní program z kurzu Matematická analýza na univerzitě v Bologni, poznámky profesora k přednáškám, otázky zkoušky z minulých let nebo typické mylné představy studentů prvních ročníků. Bez uzemnění, učitel umělé inteligence riskuje, že poskytne odpovědi, které jsou věrohodné, ale pedagogicky nesprávné nebo vytržené z kontextu.
V tomto článku seriálu EdTech Engineering vybudujeme kompletního tutora AI s LLM a RAG: od indexování vzdělávacích dokumentů až po pedagogické mantinely které brání modelu v přímém poskytování řešení cvičení, až po adaptivní zpětnou vazbu na základě profilu studenta. Vše s konkrétními příklady v Pythonu a TypeScriptu.
Co se dozvíte v tomto článku
- End-to-end architektura učitele AI s LLM a RAG
- Indexování vzdělávacího obsahu (PDF, přepis videa, kvíz)
- Ukotvení znalostí: jak omezit LLM na materiál kurzu
- Pedagogické mantinely k podpoře kritického myšlení, nikoli přímé reakce
- Profil studenta a adaptivní přizpůsobení zpětné vazby
- Správa konverzační paměti pro více relací
- Hodnocení kvality odpovědí pomocí RAG metrik (věrnost, relevance)
- Škálovatelné nasazení s FastAPI a sémantickým ukládáním do mezipaměti
1. proč RAG for Educational Tutors
Čistý LLM bez přístupu ke znalostem specifickým pro doménu trpí třemi kritickými problémy ve vzdělávacím kontextu: halucinace (vymyšlené, ale věrohodné informace), zatuchlé znalosti (znalostní firma v den školení) e nedostatek kurikulárního kontextu (neví, co student již studoval, jakou knihu používá, která část programu byla pokryta).
Akademický výzkum v letech 2024–2025 ukazuje, že systémy RAG aplikované na vzdělávání se snižují halucinace o 80 % ve srovnání s čistými LLM a zvyšují spokojenost studentů o 40 % díky odpovědím zakotveným v materiálu kurzu. Systém LPITutor (2025) předvedl 7-17 miliard parametrů open-source modelů s dobrým RAG potrubím dosáhnout výkonu srovnatelného s GPT-4o, díky čemuž je možné nasazení na místě i pro instituce s omezeným rozpočtem.
Klíčovým konceptem je znalostní uzemnění: ukotvení odpovědí LLM k ověřeným a kontextově specifickým dokumentům (práce, učebnice, řešená cvičení). Když se student zeptá: "Jak vypočítáte derivaci sin(x)?", učitel nemá přístup k jeho obecným znalostem, ale obnoví přesnou definici použitou v kurzu se zápisem profesora a příklady přijaté knihy.
Architektura na vysoké úrovni
| Komponent | Technologie | Funkce |
|---|---|---|
| Požití dokumentu | LangChain, PyMuPDF | Analýza PDF, snímek, přepis |
| Model vkládání | text-embedding-3-small, BGE-M3 | Vektorizace bloku textu |
| Vektorový obchod | pgvector, Qdrant, Chroma | Sémantické ukládání a vyhledávání |
| LLM | GPT-4o, Llama 3.1, Mistral | Generování pedagogické odezvy |
| Paměť | Redis, PostgreSQL | Konverzační sezení |
| Vrstva zábradlí | Vlastní výzvy, NeMo Guardrails | Pedagogická kontrola |
| Profil studenta | PostgreSQL, mezipaměť Redis | Úroveň, historie, preference |
| Vrstva API | FastAPI, WebSocket | Streamovací rozhraní |
2. Průběh indexování vzdělávacích dokumentů
Prvním krokem je vybudování znalostní báze lektora. Učební materiály mají vlastnosti specifické versus obecné dokumenty: matematické vzorce, zdrojový kód, schémata, tabulky a přesnou technickou slovní zásobu. Strategie chunkingu musí zachovat sémantickou koherenci.
# 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. Retrieval and Knowledge Grounding
Vyzvedávání je srdcem systému RAG. Nestačí načíst kousky nejvíce podobné otázce: ve vzdělávacím kontextu musíme také vzít v úvahu úroveň obtížnosti studenta, aktuální téma v programu a typ dokumentu (např. preferovat řešená cvičení, když student žádá o procvičení).
# 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. Pedagogické mantinely: Tutor neodpovídá
Největší riziko AI tutora bez mantinelů je, že se z něj stane nástroj pro kopírování cvičení. Dobrý pedagogický tutor nedává přímou odpověď, ale vede studenta k řešení přes sokratovské otázky, postupné návrhy (lešení) a zpětná vazba na koncepční chyby.
Realizujeme třípatrový systém zábradlí: klasifikace záměru (Ptáte se na odpověď nebo máte koncepční pochybnosti?), pedagogická politika (jaké množství lešení použít?) e rychlé inženýrství (jak formulovat reakce LLM na podporu aktivního učení).
# 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. Multi-Session konverzační paměť
Efektivní mentor si pamatuje předchozí rozhovory. Pokud měl student potíže s derivacemi minulý týden, tutor musí mít na paměti, když odpovídá na otázky ohledně integrálů. Implementujeme dvouúrovňovou paměť: krátkodobá paměť (aktuální konverzace, Redis) e dlouhodobá paměť (historie relace, PostgreSQL se souhrny 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 s FastAPI
Uživatelská zkušenost učitele umělé inteligence se s tím ohromně zlepšuje streamování odpovědí: student vidí, jak se text objevuje postupně, jako by vyučoval psal v reálném čase. Implementujeme koncový bod FastAPI s 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. Hodnocení kvality RAG
Učitel AI ve výrobě musí být neustále sledován. Používáme framework RAGAS (RAG Assessment) k vyhodnocení čtyř dimenzí: věrnost (je odpověď věrná nalezeným dokumentům?), odpověď. relevance (je odpověď relevantní k otázce?), kontextová přesnost (jsou získané dokumenty relevantní?) e připomenutí kontextu (obnovili jsme všechny potřebné dokumenty?).
# 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-vzory, kterým je třeba se vyhnout
- RAG bez filtrů nájemců: Nikdy nesdílejte dokumenty mezi různými kurzy nebo institucemi. Vždy filtrujte podle tenant_id a course_id.
- Příliš velké kusy: Kousky 2000+ tokenů snižují relevanci. Použijte 512-768 tokenů s 10-15% překrytím.
- Vysoká teplota: Teploty vyšší než 0,5 zvyšují halucinace. Pro pedagogy použijte 0,2-0,4.
- Bez mantinelů: LLM bez pedagogických mantinelů se stává systémem pro kopírování domácích úkolů. Zábradlí jsou nezbytná, nikoli volitelná.
- Nekonečná paměť: Načítání celé historie konverzace přesahuje kontextové okno a zvyšuje náklady. Použijte posuvná okna a souhrny.
- Žádné hodnocení: Bez RAGAS nebo podobných metrik nevíte, zda váš lektor skutečně funguje dobře.
Závěry a další kroky
Postavili jsme kompletní architekturu AI tutora založeného na LLM a RAG: od indexování vzdělávacích materiálů až po adaptivní vyhledávání který zohledňuje profil studenta, pedagogické mantinely, které prosazují aktivní učení, až multisekční paměť a sledování kvality.
Výsledkem je systém, který nejen odpovídá na otázky, ale také průvodce student se během procesu učení přizpůsobuje jeho úrovni, identifikace mylných představ a podpora kritické reflexe, vše zakotvené v konkrétním materiálu kurzu a nikoli v obecných znalostech LLM.
V dalším článku série prozkoumáme budovu a Gamification Engine se stavovými stroji a záběrovou mechanikou které zvyšují motivaci a vytrvalost studentů na platformě.
EdTech Engineering Series
- Škálovatelná architektura LMS: Vzor pro více nájemců
- Algoritmy adaptivního učení: Od teorie k produkci
- Streamování videa pro vzdělávání: WebRTC vs HLS vs DASH
- AI Proctoring Systems: Privacy-first with Computer Vision
- Personalizovaný lektor s LLM: RAG for Knowledge Grounding (tento článek)
- Gamification Engine: Architektura a státní stroj
- Learning Analytics: Data Pipeline s xAPI a Kafka
- Spolupráce v reálném čase v EdTech: CRDT a WebSocket
- Mobile-First EdTech: Offline-First Architecture
- Správa obsahu pro více nájemců: Správa verzí a SCORM