Spersonalizowany nauczyciel z LLM: RAG dla uziemienia wiedzy
Marzenie o prywatnym nauczycielu dla każdego ucznia, dostępnym 24 godziny na dobę, potrafiącym dostosować się do poziomu i styl uczenia się wszystkich, nie jest to już fantastyka naukowa. TO Model dużego języka (LLM) w połączeniu z technikami Generacja wzmocniona odzyskiwaniem (RAG) oni to umożliwiają budowanie spersonalizowanych tutorów AI, którzy pokonują ograniczenia ogólnych chatbotów i statycznych często zadawanych pytań.
Głównym problemem LLM w kontekście edukacyjnym jest uziemienie wiedzy: modelka podobnie jak GPT-4o czy Llama 3 ma ogromną wiedzę ogólną, ale nie zna konkretnego programu z kursu Analizy Matematycznej na Uniwersytecie w Bolonii, notatki profesora z wykładów, pytania egzaminów z poprzednich lat czy typowych błędnych wyobrażeń studentów pierwszego roku. Bez uziemienia, nauczyciel sztucznej inteligencji ryzykuje udzieleniem odpowiedzi, które są wiarygodne, ale niepoprawne pedagogicznie lub wyrwane z kontekstu.
W tym artykule z serii Inżynieria EdTech zbudujemy kompletnego nauczyciela AI z LLM i RAG: od rurociągu indeksowania dokumentów edukacyjnych po poręcze pedagogiczne co uniemożliwia modelowi bezpośrednie dostarczanie rozwiązań ćwiczeń, aż do adaptacyjnego sprzężenia zwrotnego na podstawie profilu studenta. Wszystko z konkretnymi przykładami w Pythonie i TypeScript.
Czego dowiesz się w tym artykule
- Kompleksowa architektura nauczyciela AI z LLM i RAG
- Plan indeksowania treści edukacyjnych (PDF, transkrypcja wideo, quiz)
- Ugruntowanie wiedzy: jak ograniczyć LLM do materiału kursu
- Poręcze pedagogiczne promujące krytyczne myślenie, a nie bezpośrednią reakcję
- Profil ucznia i dostosowywanie adaptacyjnych informacji zwrotnych
- Wielosesyjne zarządzanie pamięcią konwersacyjną
- Ocena jakości odpowiedzi za pomocą wskaźników RAG (wierność, trafność)
- Skalowalne wdrożenie dzięki FastAPI i buforowaniu semantycznemu
1. dlaczego RAG dla tutorów edukacyjnych
Czysty LLM, bez dostępu do wiedzy specyficznej dla danej dziedziny, cierpi na trzy krytyczne problemy w kontekście edukacyjnym: halucynacje (informacja wymyślona, ale wiarygodna), nieaktualna wiedza (firma wiedza na dzień szkolenia) e brak kontekstu programowego (nie wie, czego student się już uczył, z jakiej książki korzysta, jaką część programu omawiał).
Badania akademickie przeprowadzone w latach 2024–2025 pokazują, że systemy RAG stosowane w edukacji ograniczają halucynacje o 80% w porównaniu do czystych LLM i zwiększają satysfakcję uczniów o 40% dzięki odpowiedziom zakotwiczonym w materiale kursu. System LPITutor (2025) zademonstrował modele open source o parametrach 7–17 miliardów z dobrym potokiem RAG osiągnąć wydajność porównywalną do GPT-4o, dzięki czemu wdrożenie lokalne jest możliwe nawet dla instytucji o ograniczonych budżetach.
Kluczową koncepcją jest uziemienie wiedzy: zakotwicz odpowiedzi LLM do zweryfikowanych i kontekstowych dokumentów (ulotki, podręczniki, rozwiązane ćwiczenia). Kiedy uczeń pyta: „Jak obliczyć pochodną sin(x)?”, nauczyciel nie uzyskuje dostępu do swojej wiedzy ogólnej, ale odtwarza dokładną definicję używaną w kursie, wraz z zapisem profesora i przykłady przyjętej książki.
Architektura na wysokim poziomie
| Część | Technologia | Funkcjonować |
|---|---|---|
| Przyjmowanie dokumentów | LangChain, PyMuPDF | Analizowanie pliku PDF, slajdu i transkrypcji |
| Model osadzania | osadzanie tekstu-3-małe, BGE-M3 | Wektoryzacja fragmentów tekstu |
| Sklep wektorowy | pgvector, Qdrant, Chroma | Semantyczne przechowywanie i wyszukiwanie |
| LLM | GPT-4o, Lama 3.1, Mistral | Generowanie reakcji pedagogicznej |
| Pamięć | Redisa i PostgreSQLa | Sesje konwersacyjne |
| Warstwa poręczy | Niestandardowe podpowiedzi, poręcze NeMo | Kontrola pedagogiczna |
| Profil studenta | PostgreSQL, pamięć podręczna Redis | Poziom, historia, preferencje |
| Warstwa API | FastAPI, WebSocket | Interfejs przesyłania strumieniowego |
2. Rurociąg indeksowania dokumentów edukacyjnych
Pierwszym krokiem jest zbudowanie bazy wiedzy tutora. Materiały dydaktyczne mają swoje cechy dokumenty szczegółowe a dokumenty ogólne: wzory matematyczne, kod źródłowy, schematy, tabele i precyzyjne słownictwo techniczne. Strategia fragmentowania musi zachować spójność semantyczną.
# 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. Odzyskiwanie i uziemianie wiedzy
Odzyskiwanie jest sercem systemu RAG. Nie wystarczy pobrać fragmenty najbardziej podobne do pytania: w kontekście edukacyjnym musimy również wziąć pod uwagę poziom trudności ucznia, aktualny temat w programie i rodzaj dokumentu (np. preferuj rozwiązane ćwiczenia, gdy uczeń prosi o ćwiczenia).
# 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. Poręcze pedagogiczne: Tutor nie daje odpowiedzi
Największym ryzykiem związanym z nauczycielem AI bez poręczy jest to, że stanie się on narzędziem do kopiowania ćwiczenia. Dobry korepetytor pedagogiczny nie daje bezpośredniej odpowiedzi, ale prowadzi ucznia ku rozwiązaniu poprzez pytania sokratejskie, stopniowe sugestie (rusztowanie) oraz informacje zwrotne dotyczące błędów koncepcyjnych.
Realizujemy trójpoziomowy system poręczy: klasyfikacja intencji (pytasz o odpowiedź czy masz wątpliwości koncepcyjne?), politykę pedagogiczną (jakie rusztowanie zastosować?) e szybka inżynieria (jak sformułować odpowiedź LLM na promowanie aktywnego uczenia się).
# 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. Wielosesyjna pamięć konwersacyjna
Skuteczny mentor zapamiętuje poprzednie rozmowy. Jeśli uczeń miał trudności z instrumentami pochodnymi w zeszłym tygodniu, prowadzący musi o tym pamiętać podczas udzielania odpowiedzi na pytania dotyczące całek. Implementujemy pamięć dwupoziomową: pamięć krótkotrwała (bieżąca rozmowa, Redis) e pamięć długotrwała (historia sesji, PostgreSQL z podsumowaniami 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. Przesyłanie strumieniowe API za pomocą FastAPI
Dzięki temu doświadczenie użytkownika nauczyciela AI znacznie się poprawia przesyłanie strumieniowe odpowiedzi: uczeń widzi tekst pojawiający się stopniowo, niczym nauczyciel pisał w czasie rzeczywistym. Implementujemy punkt końcowy FastAPI ze zdarzeniami wysyłanymi przez serwer (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. Ocena jakości RAG
Nauczyciel AI w produkcji musi być stale monitorowany. Korzystamy z frameworka RAGA (ocena RAG) w celu oceny czterech wymiarów: wierność (czy odpowiedź jest zgodna z odzyskanymi dokumentami?), odpowiedź. znaczenie (czy odpowiedź ma związek z pytaniem?), precyzja kontekstu (czy odzyskane dokumenty są istotne?) przypomnienie kontekstu (czy odzyskaliśmy wszystkie niezbędne 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,
}Anty-wzorce, których należy unikać
- RAG bez filtrów najemców: Nigdy nie udostępniaj dokumentów pomiędzy różnymi kursami lub instytucjami. Zawsze filtruj według identyfikatora najemcy i identyfikatora kursu.
- Kawałki są za duże: Kawałki ponad 2000 tokenów osłabiają znaczenie. Użyj tokenów 512-768 z nakładaniem się 10-15%.
- Wysoka temperatura: Temperatury wyższe niż 0,5 zwiększają halucynacje. W przypadku nauczycieli edukacyjnych użyj 0,2-0,4.
- Brak poręczy: LLM bez poręczy pedagogicznych staje się systemem do kopiowania prac domowych. Poręcze są niezbędne, a nie opcjonalne.
- Nieskończona pamięć: Załadowanie całej historii konwersacji wykracza poza okno kontekstowe i zwiększa koszty. Używaj przesuwanych okien i podsumowań.
- Brak oceny: Bez RAGAS i podobnych wskaźników nie wiesz, czy Twój nauczyciel rzeczywiście radzi sobie dobrze.
Wnioski i dalsze kroki
Zbudowaliśmy kompletną architekturę tutora AI w oparciu o LLM i RAG: od procesu indeksowania materiałów edukacyjnych po wyszukiwanie adaptacyjne uwzględniającego profil ucznia, po poręcze pedagogiczne, które promują aktywne uczenie się, aż do pamięci wielosesyjnej i monitorowania jakości.
W rezultacie powstał system, który nie tylko odpowiada na pytania, ale... przewodnik ucznia przez proces uczenia się, dostosowując się do jego poziomu, identyfikowanie błędnych przekonań i promowanie krytycznej refleksji, wszystko zakotwiczone w konkretnym materiale kursu, a nie w ogólnej wiedzy LLM.
W następnym artykule z tej serii zajmiemy się budowaniem Silnik grywalizacji z maszynami stanu i mechaniką zaangażowania które zwiększają motywację i wytrwałość uczniów na platformie.
Seria inżynieryjna EdTech
- Skalowalna architektura LMS: wzorzec wielu najemców
- Algorytmy uczenia się adaptacyjnego: od teorii do produkcji
- Strumieniowe przesyłanie wideo dla edukacji: WebRTC vs HLS vs DASH
- Systemy AI Proctoring: przede wszystkim prywatność dzięki wizji komputerowej
- Spersonalizowany nauczyciel z LLM: RAG dla ugruntowania wiedzy (ten artykuł)
- Silnik grywalizacji: architektura i maszyna stanu
- Learning Analytics: Potok danych z xAPI i Kafką
- Współpraca w czasie rzeczywistym w EdTech: CRDT i WebSocket
- EdTech zorientowany na urządzenia mobilne: architektura zorientowana na tryb offline
- Zarządzanie treścią dla wielu dzierżawców: wersjonowanie i SCORM