LLM'de Kişiselleştirilmiş Öğretmen: Bilgi Temellendirme için RAG
Her öğrencinin hayali, 24 saat ulaşılabilir, seviyeye uyum sağlayabilen özel öğretmen ve herkesin öğrenme stili, artık bilim kurgu değil. THE Büyük Dil Modeli (LLM) tekniklerle birleştirilmiş Alma-Artırılmış Nesil (RAG) bunu mümkün kılıyorlar genel sohbet robotlarının ve statik SSS'lerin sınırlamalarını aşan kişiselleştirilmiş yapay zeka eğitmenleri oluşturmak.
Eğitim bağlamında Yüksek Lisans'ların temel sorunu, bilgi temellendirme: bir model GPT-4o veya Llama 3 gibi geniş bir genel bilgiye sahip ancak spesifik programı bilmiyor Bologna Üniversitesi Matematiksel Analiz dersinin içeriği, profesörün ders notları, geçmiş yıllara ait sınavlar veya birinci sınıf öğrencilerinin tipik kavram yanılgıları. Topraklama olmadan, Yapay zeka eğitmeni makul ancak pedagojik olarak yanlış veya bağlam dışı cevaplar verme riskiyle karşı karşıyadır.
Serinin bu makalesinde Eğitim Teknolojisi Mühendisliği eksiksiz bir yapay zeka öğretmeni oluşturacağız LLM ve RAG ile: eğitim belgelerinin indeksleme hattından pedagojik korkuluklara kadar Uyarlanabilir geri bildirime kadar modelin alıştırmalara doğrudan çözüm sunmasını engelleyen Öğrenci profiline göre. Hepsi Python ve TypeScript'te somut örneklerle.
Bu Makalede Neler Öğreneceksiniz?
- Yüksek Lisans ve RAG'lı bir yapay zeka öğretmeninin uçtan uca mimarisi
- Eğitim içeriği indeksleme hattı (PDF, video transkripti, sınav)
- Bilgi temeli: Yüksek Lisans'ın kurs materyaliyle nasıl sınırlandırılacağı
- Doğrudan tepkiyi değil, eleştirel düşünmeyi teşvik eden pedagojik korkuluklar
- Öğrenci profili ve uyarlanabilir geri bildirim özelleştirmesi
- Çok oturumlu konuşma belleği yönetimi
- Yanıtların kalitesinin RAG metrikleriyle değerlendirilmesi (sadıklık, alaka)
- FastAPI ve anlamsal önbelleğe alma ile ölçeklenebilir dağıtım
1. Neden Eğitim Öğretmenleri için RAG
Alana özgü bilgiye erişimi olmayan saf bir Yüksek Lisans, üç kritik sorunla karşı karşıyadır eğitim bağlamında: halüsinasyonlar (icat edilmiş ancak makul bilgi), bayat bilgi (eğitim tarihindeki bilgi firması) e müfredat bağlamının eksikliği (Öğrencinin daha önce ne okuduğunu bilmiyor, hangi kitabı kullanıyor, programın hangi bölümünü kapsıyor).
2024-2025 akademik araştırması, eğitimde uygulanan RAG sistemlerinin, halüsinasyonları saf LLM'lere kıyasla %80 oranında azaltır ve öğrenci memnuniyetini artırır Kurs materyaline sabitlenmiş cevaplar sayesinde %40 oranında. LPITutor sistemi (2025) İyi bir RAG boru hattıyla 7-17 milyar parametreli açık kaynak modeli gösterdi GPT-4o ile karşılaştırılabilir bir performans elde ederek şirket içi dağıtımı mümkün hale getirin bütçesi kısıtlı kurumlar için bile.
Anahtar kavram, bilgi temellendirme: Yüksek Lisans cevaplarını tutturun doğrulanmış ve bağlama özel belgelere (çalışma notları, ders kitapları, çözülmüş alıştırmalar). Öğrenci "sin(x)'in türevini nasıl hesaplarsınız?" diye sorduğunda öğretmen erişim sağlayamıyor. genel bilgisine dayanarak derste kullanılan tam tanımı notasyonla kurtarır Profesörün sözleri ve benimsenen kitaptan örnekler.
Üst Düzey Mimari
| Bileşen | Teknoloji | İşlev |
|---|---|---|
| Belge Alma | LangChain, PyMuPDF | PDF, slayt, transkript ayrıştırılıyor |
| Gömme Modeli | metin gömme-3-küçük, BGE-M3 | Metin öbeği vektörleştirmesi |
| Vektör Mağazası | pgvektör, Qdrant, Chroma | Anlamsal depolama ve alma |
| Yüksek Lisans | GPT-4o, Lama 3.1, Mistral | Pedagojik tepkinin üretilmesi |
| Hafıza | Redis, PostgreSQL | Konuşma oturumları |
| Korkuluk Katmanı | Özel yönlendirme, NeMo Guardrails | Pedagojik kontrol |
| Öğrenci Profili | PostgreSQL, Redis Önbelleği | Seviye, geçmiş, tercihler |
| API Katmanı | FastAPI, WebSocket | Akış arayüzü |
2. Eğitim Belgesi Dizinleme Hattı
İlk adım, öğretmenin bilgi tabanını oluşturmaktır. Öğretim materyallerinin özellikleri vardır spesifik ve genel belgeler: matematiksel formüller, kaynak kodu, şemalar, tablolar ve kesin bir teknik kelime dağarcığı. Parçalama stratejisi anlamsal tutarlılığı korumalıdır.
# 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. Erişim ve Bilgi Temellendirme
Geri alma RAG sisteminin kalbidir. Soruya en çok benzeyen parçaları almak yeterli değildir: eğitim bağlamında şunları da dikkate almalıyız: zorluk seviyesi öğrencinin, güncel konu programda ve belge türü (örneğin, öğrenci pratik yapmasını ister).
# 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. Pedagojik Korkuluklar: Öğretmen Cevapları Vermiyor
Korkulukları olmayan bir yapay zeka öğretmeninin en büyük riski, kopyalama için bir araç haline gelmesidir egzersizler. İyi bir pedagojik öğretmen doğrudan cevap vermez, öğrenciye rehberlik eder. Sokratik sorularla, aşamalı önerilerle çözüme doğru ilerleme (iskele) ve kavramsal hatalara ilişkin geri bildirim.
Üç katmanlı bir korkuluk sistemi uyguluyoruz: amaç sınıflandırması (Cevabı mı soruyorsunuz yoksa kavramsal bir şüpheniz mi var?), pedagojik politika (ne kadar iskele uygulanmalı?) e hızlı mühendislik (nasıl formüle edilir LLM'nin aktif öğrenmeyi teşvik etmeye tepkisi).
# 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. Çok Oturumlu Konuşma Belleği
Etkili bir mentor önceki konuşmaları hatırlar. Öğrenci zorluk yaşıyorsa Geçen hafta türevlerle ilgili olarak, öğretmen cevap verirken bunu aklında tutmalıdır İntegrallerle ilgili sorular. İki seviyeli bir hafıza uyguluyoruz: kısa süreli hafıza (mevcut görüşme, Redis) e uzun süreli hafıza (oturum geçmişi, LLM özetleriyle birlikte PostgreSQL).
# 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. FastAPI ile API Akışı
Yapay Zeka eğitmeninin kullanıcı deneyimi bununla birlikte muazzam bir şekilde gelişiyor akış Cevapların sayısı: öğrenci metnin sanki öğretmenmiş gibi yavaş yavaş belirdiğini görür gerçek zamanlı olarak yazıyordu. Sunucu Tarafından Gönderilen Olaylar (SSE) ile bir FastAPI uç noktası uyguluyoruz.
# 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. RAG kalite değerlendirmesi
Üretimdeki bir yapay zeka öğretmeni sürekli olarak izlenmelidir. Çerçeveyi kullanıyoruz RAGAS (RAG Değerlendirmesi) dört boyutu değerlendirmek için: sadakat (Cevap kurtarılan belgelere sadık mı?), cevap. alaka (Cevap soruyla alakalı mı?), bağlam hassasiyeti (kurtarılan belgeler konuyla alakalı mı?) e bağlam hatırlama (gerekli tüm belgeleri kurtardık mı?).
# 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,
}Kaçınılması Gereken Anti-Desenler
- Kiracı filtreleri olmayan RAG: Farklı kurslar veya kurumlar arasında asla belge paylaşmayın. Her zaman tenant_id ve course_id'ye göre filtreleyin.
- Parçalar çok büyük: 2000'den fazla jeton yığını alaka düzeyini azaltır. %10-15 örtüşme ile 512-768 token kullanın.
- Yüksek sıcaklık: 0,5'in üzerindeki sıcaklıklar halüsinasyonları artırır. Eğitim öğretmenleri için 0,2-0,4 kullanın.
- Korkuluk yok: Pedagojik korkulukları olmayan bir Yüksek Lisans, ödevlerin kopyalanması için bir sistem haline gelir. Korkuluklar isteğe bağlı değil esastır.
- Sonsuz Bellek: Tüm konuşma geçmişinin yüklenmesi bağlam penceresini aşar ve maliyetleri artırır. Sürgülü pencereler ve özetler kullanın.
- Derecelendirme yok: RAGAS veya benzeri ölçümler olmadan öğretmeninizin gerçekten iyi performans gösterip göstermediğini bilemezsiniz.
Sonuçlar ve Sonraki Adımlar
Yüksek Lisans ve RAG'a dayalı bir AI eğitmeninin eksiksiz mimarisini oluşturduk: eğitim materyallerinin indekslenmesinden uyarlanabilir erişime kadar teşvik ettikleri pedagojik korkuluklara kadar öğrencinin profilini dikkate alan aktif öğrenme, çoklu oturum hafızası ve kalite izleme.
Sonuç, yalnızca soruları yanıtlamakla kalmayıp aynı zamanda rehber Öğrencinin öğrenme süreci boyunca kendi seviyesine uyum sağlaması, Kavram yanılgılarını tespit etmek ve eleştirel düşünmeyi teşvik etmek, genel LLM bilgisine değil, belirli ders materyaline bağlı.
Serinin bir sonraki makalesinde, bir yapı oluşturmayı keşfedeceğiz. Oyunlaştırma Motoru durum makineleri ve angajman mekaniği ile Bu da öğrencilerin motivasyonunu ve platformdaki kalıcılığını artırır.
EdTech Mühendislik Serisi
- Ölçeklenebilir LMS Mimarisi: Çok Kiracılı Model
- Uyarlanabilir Öğrenme Algoritmaları: Teoriden Üretime
- Eğitim için Video Yayını: WebRTC vs HLS vs DASH
- Yapay Zeka Gözetleme Sistemleri: Bilgisayarlı Görme ile Öncelik Gizlilik
- LLM'de Kişiselleştirilmiş Öğretmen: Bilgi Temellendirme için RAG (bu makale)
- Oyunlaştırma Motoru: Mimari ve Durum Makinesi
- Öğrenme Analitiği: xAPI ve Kafka ile Veri Hattı
- Eğitim Teknolojisinde Gerçek Zamanlı İşbirliği: CRDT ve WebSocket
- Mobil Öncelikli Eğitim Teknolojisi: Çevrimdışı Öncelikli Mimari
- Çok Kiracılı İçerik Yönetimi: Sürüm Oluşturma ve SCORM