LLM을 통한 맞춤형 교사: 지식 기반 구축을 위한 RAG
모든 학생을 위한 개인 교사의 꿈, 하루 24시간 이용 가능, 수준에 적응할 수 있음 그리고 모든 사람의 학습 스타일은 더 이상 공상 과학 소설이 아닙니다. 그만큼 대형 언어 모델(LLM) 기술이 결합된 검색 증강 생성(RAG) 그들은 그것을 가능하게 만들고 있다 일반 챗봇과 정적인 FAQ의 한계를 극복하는 맞춤형 AI 튜터를 구축합니다.
교육적 맥락에서 LLM의 핵심 문제는 지식 접지: 모델 GPT-4o나 Llama 3처럼 광범위한 일반 지식을 갖고 있지만 구체적인 프로그램에 대해서는 알지 못합니다. 볼로냐 대학의 수학적 분석 과정, 교수의 강의 노트, 전년도 시험이나 1학년 학생들의 전형적인 오해. 접지하지 않고, AI 교사는 그럴듯하지만 교육학적으로 부정확하거나 맥락에 맞지 않는 답변을 제공할 위험이 있습니다.
이 시리즈 기사에서는 교육기술 엔지니어링 우리는 완전한 AI 교사를 구축할 것입니다 LLM 및 RAG 사용: 교육 문서의 색인 파이프라인부터 교육학적 가드레일까지 모델이 적응형 피드백까지 연습에 대한 솔루션을 직접 제공하는 것을 방지합니다. 학생 프로필을 기반으로 합니다. Python 및 TypeScript의 구체적인 예가 포함되어 있습니다.
이 기사에서 배울 내용
- LLM 및 RAG를 사용한 AI 교사의 엔드투엔드 아키텍처
- 교육 콘텐츠 인덱싱 파이프라인(PDF, 비디오 대본, 퀴즈)
- 지식 기반: LLM을 강의 자료로 제한하는 방법
- 직접적인 반응이 아닌 비판적 사고를 촉진하는 교육적 가드레일
- 학생 프로필 및 적응형 피드백 사용자 정의
- 다중 세션 대화형 메모리 관리
- RAG 지표(충실성, 관련성)를 사용한 응답 품질 평가
- FastAPI 및 의미론적 캐싱을 통한 확장 가능한 배포
1. 교육 교사를 위한 RAG 이유
도메인별 지식에 접근할 수 없는 순수 LLM은 세 가지 중요한 문제를 안고 있습니다. 교육적 맥락에서: 환각 (창안되었지만 그럴듯한 정보), 진부한 지식 (교육일 당시 지식회사) e 커리큘럼 맥락 부족 (학생이 이미 무엇을 공부했는지 알지 못함, 그는 어떤 책을 사용하는지, 프로그램의 어느 부분을 다루었는지).
2024~2025년 학술 연구에 따르면 교육에 적용된 RAG 시스템은 순수 LLM에 비해 환각이 80% 감소하고 학생 만족도가 높아집니다. 강좌 자료에 기반한 답변 덕분에 40% 증가했습니다. LPI튜터 시스템(2025) 우수한 RAG 파이프라인을 갖춘 70억~170억 개의 매개변수 오픈 소스 모델 시연 GPT-4o에 필적하는 성능을 달성하여 온프레미스 배포가 가능해집니다. 예산이 제한된 기관의 경우에도 마찬가지입니다.
핵심 컨셉은 지식 접지: LLM 답변 고정 검증된 상황별 문서(유인물, 교과서, 해결된 연습 문제) 학생이 "sin(x)의 도함수는 어떻게 계산하나요?"라고 질문하면 교사는 액세스하지 않습니다. 그의 일반적인 지식을 바탕으로 하지만 표기법을 사용하여 과정에서 사용된 정확한 정의를 복구합니다. 교수님의 말씀과 채택된 책의 사례입니다.
상위 수준 아키텍처
| 요소 | 기술 | 기능 |
|---|---|---|
| 문서 수집 | 랭체인, PyMuPDF | PDF, 슬라이드, 스크립트 구문 분석 |
| 임베딩 모델 | 텍스트 임베딩-3-소형, BGE-M3 | 텍스트 청크 벡터화 |
| 벡터 스토어 | pg벡터, Qdrant, 크로마 | 의미론적 저장 및 검색 |
| 법학대학원 | GPT-4o, 라마 3.1, 미스트랄 | 교육적 반응의 생성 |
| 메모리 | 레디스, 포스트그레SQL | 대화 세션 |
| 난간 레이어 | 맞춤형 프롬프트, NeMo Guardrails | 교육적 통제 |
| 학생 프로필 | PostgreSQL, Redis 캐시 | 레벨, 기록, 선호도 |
| API 레이어 | FastAPI, 웹소켓 | 스트리밍 인터페이스 |
2. 교육용 문서 인덱싱 파이프라인
첫 번째 단계는 교사의 지식 기반을 구축하는 것입니다. 교재에는 특성이 있습니다. 특정 문서와 일반 문서: 수학 공식, 소스 코드, 회로도, 표 그리고 정확한 기술 용어. 청킹 전략은 의미론적 일관성을 유지해야 합니다.
# 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. 검색 및 지식 접지
검색은 RAG 시스템의 핵심입니다. 질문과 가장 유사한 덩어리를 검색하는 것만으로는 충분하지 않습니다. 교육적인 맥락에서 우리는 다음 사항도 고려해야 합니다. 난이도 학생의, 현재 주제 프로그램과 문서 유형 (예: 학생이 공부할 때 해결 연습을 선호합니다. 연습을 요청합니다).
# 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. 교육학적 가드레일: 교사는 답을 주지 않습니다.
가드레일 없는 AI 튜터의 가장 큰 리스크는 복사의 도구가 된다는 점이다. 연습. 훌륭한 교육학 교사는 직접적인 답변을 제공하지 않고 학생을 지도합니다. 소크라테스식 질문, 점진적인 제안(스캐폴딩)을 통해 해결책을 향해 개념적 오류에 대한 피드백.
우리는 3단계 가드레일 시스템을 구현합니다. 의도 분류 (답을 묻는 건가요, 아니면 개념적인 의심이 있는 건가요?), 교육학 정책 (얼마나 비계를 적용할 것인가?) e 신속한 엔지니어링 (공식화하는 방법 적극적인 학습 촉진에 대한 LLM의 대응).
# 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. 다중 세션 대화 메모리
효과적인 멘토는 이전 대화를 기억합니다. 학생에게 어려움이 있는 경우 지난주에 파생상품에 관해서, 튜터는 답변할 때 이 점을 명심해야 합니다. 적분에 관한 질문입니다. 우리는 2단계 메모리를 구현합니다: 단기 기억 (현재 대화, Redis) e 장기 기억 (세션 기록, LLM 요약이 포함된 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를 사용한 API 스트리밍
AI 튜터의 사용자 경험은 이를 통해 엄청나게 향상됩니다. 스트리밍 답변: 학생은 마치 교사가 가르치는 것처럼 텍스트가 점차적으로 나타나는 것을 봅니다. 실시간으로 쓰고 있었어요. 우리는 SSE(Server-Sent Events)를 사용하여 FastAPI 엔드포인트를 구현합니다.
# 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 품질 평가
생산 중인 AI 교사를 지속적으로 모니터링해야 합니다. 우리는 프레임워크를 사용합니다. 라가스 (RAG 평가)를 통해 4가지 차원을 평가합니다. 충실 (답변은 회수된 문서에 충실한가?), 답변. 관련성 (답변이 질문과 관련이 있습니까?) 상황 정밀도 (복구된 문서가 관련이 있나요?) e 맥락 회상 (필요한 문서를 모두 복구했나요?)
# 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,
}피해야 할 안티패턴
- 테넌트 필터가 없는 RAG: 다른 코스나 기관 간에 문서를 공유하지 마십시오. 항상tenant_id 및course_id로 필터링하십시오.
- 청크가 너무 큼: 2000개 이상의 토큰 덩어리는 관련성을 희석시킵니다. 10~15% 중복되는 512~768 토큰을 사용하세요.
- 고온: 0.5보다 높은 온도는 환각을 증가시킵니다. 교육 교사의 경우 0.2-0.4를 사용하십시오.
- 난간 없음: 교육학적 가드레일이 없는 LLM은 숙제를 복사하는 시스템이 됩니다. 가드레일은 선택이 아닌 필수입니다.
- 무한한 기억: 모든 대화 기록을 로드하면 컨텍스트 창을 초과하고 비용이 증가합니다. 슬라이딩 윈도우와 요약을 사용하세요.
- 평가 없음: RAGAS 또는 유사한 측정항목이 없으면 교사가 실제로 잘 수행하고 있는지 알 수 없습니다.
결론 및 다음 단계
우리는 LLM 및 RAG를 기반으로 AI 교사의 전체 아키텍처를 구축했습니다. 교육 자료의 색인 파이프라인부터 적응형 검색까지 학생의 프로필을 고려하여 그들이 장려하는 교육적 가드레일까지 활성 학습, 최대 다중 세션 메모리 및 품질 모니터링.
그 결과 단순히 질문에 답하는 것이 아닌, 가이드 학습 과정을 통해 학생은 자신의 수준에 적응하고, 오해를 식별하고 비판적 성찰을 촉진하는 것, 모두 일반적인 LLM 지식이 아닌 특정 과정 자료에 기반을 두고 있습니다.
시리즈의 다음 기사에서는 게임화 엔진 상태 머신 및 참여 메커니즘 사용 이는 플랫폼에 대한 학생들의 동기와 지속성을 증가시킵니다.
EdTech 엔지니어링 시리즈
- 확장 가능한 LMS 아키텍처: 다중 테넌트 패턴
- 적응형 학습 알고리즘: 이론에서 생산까지
- 교육용 비디오 스트리밍: WebRTC, HLS, DASH
- AI 감독 시스템: 컴퓨터 비전을 통한 개인정보 보호 우선
- LLM을 통한 맞춤형 교사: 지식 기반 구축을 위한 RAG(이 기사)
- 게임화 엔진: 아키텍처 및 상태 머신
- 학습 분석: xAPI 및 Kafka를 사용한 데이터 파이프라인
- EdTech의 실시간 협업: CRDT 및 WebSocket
- 모바일 우선 교육 기술: 오프라인 우선 아키텍처
- 다중 테넌트 콘텐츠 관리: 버전 관리 및 SCORM