Trong 3 năm xây dựng hệ thống gợi ý cho nền tảng edtech phục vụ 500K+ học sinh, tôi đã triển khai và vận hành nhiều phiên bản recommendation engine. Bài viết này là bản tổng hợp chi tiết về kiến trúc, code production-ready, và những bài học xương máu khi xây dựng 学生画像构建系统 (hệ thống xây dựng hồ sơ học sinh) cho nền tảng giáo dục AI.
Tại sao Student Profiling quan trọng trong EdTech
Mỗi học sinh có cách học khác nhau — người tiếp thu qua hình ảnh, người cần nhiều ví dụ thực tế, người học nhanh nhưng mau quên. Hệ thống gợi ý hiệu quả không chỉ đơn thuần lọc nội dung theo độ khó, mà phải hiểu sâu hành vi và nhu cầu của từng cá nhân.
Kiến trúc Tổng quan
┌─────────────────────────────────────────────────────────────────────┐
│ STUDENT PROFILING ARCHITECTURE │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────────┐ ┌─────────────────────────┐ │
│ │ Web/App │───▶│ API Gateway │───▶│ Student Profile Service │ │
│ │ Client │ │ (Rate Limit)│ │ - Behavior Tracker │ │
│ └──────────┘ └──────────────┘ │ - Learning Style Calc │ │
│ │ - Knowledge Graph │ │
│ └───────────┬─────────────┘ │
│ │ │
│ ┌────────────────────────┼───────────────┐ │
│ ▼ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌─────────┐ │
│ │ Vector DB │ │ Analytics │ │ Cache │ │
│ │ (Pinecone/ │ │ Pipeline │ │ (Redis) │ │
│ │ Milvus) │ │ │ │ │ │
│ └──────────────┘ └──────────────┘ └─────────┘ │
│ │
│ ┌────────────────────────────────────────────────────────────────┐ │
│ │ RECOMMENDATION ENGINE │ │
│ │ ┌─────────┐ ┌──────────┐ ┌───────────┐ ┌────────────────┐ │ │
│ │ │ Filter │─▶│ Rerank │─▶│ Context │─▶│ Final Output │ │ │
│ │ │ Layer │ │ Layer │ │ Aware │ │ + Logging │ │ │
│ │ └─────────┘ └──────────┘ └───────────┘ └────────────────┘ │ │
│ └────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘Triển khai Student Profile Service
1. Cấu trúc dữ liệu Student Profile
# models/student_profile.py
from pydantic import BaseModel, Field
from typing import List, Dict, Optional
from datetime import datetime
from enum import Enum
class LearningStyle(str, Enum):
VISUAL = "visual" # Học qua hình ảnh, biểu đồ
AUDITORY = "auditory" # Học qua âm thanh, ghi chép
KINESTHETIC = "kinesthetic" # Học qua thực hành, trải nghiệm
READING = "reading" # Học qua văn bản, sách giáo khoa
class MasteryLevel(str, Enum):
NOVICE = "novice"
BEGINNER = "beginner"
INTERMEDIATE = "intermediate"
ADVANCED = "advanced"
EXPERT = "expert"
class StudentProfile(BaseModel):
student_id: str = Field(..., description="Định danh học sinh")
# Demographics & Context
grade_level: int = Field(..., ge=1, le=12)
subjects: List[str] = Field(default_factory=list)
enrollment_date: datetime
# Learning Style Analysis (0.0 - 1.0)
learning_style_scores: Dict[LearningStyle, float] = Field(
default_factory=lambda: {style: 0.0 for style in LearningStyle}
)
dominant_learning_style: LearningStyle
# Knowledge Graph
knowledge_graph: Dict[str, Dict[str, float]] = Field(
default_factory=dict,
description="subject -> {concept: mastery_score}"
)
# Behavioral Metrics
engagement_metrics: Dict[str, float] = Field(default_factory=dict)
avg_session_duration: float = 0.0
sessions_per_week: int = 0
completion_rate: float = 0.0
# Performance Analytics
strength_areas: List[str] = Field(default_factory=list)
improvement_areas: List[str] = Field(default_factory=list)
recent_performance_trend: str # "improving", "stable", "declining"
# AI-Generated Insights
ai_insights: Optional[Dict[str, str]] = Field(default_factory=None)
# Timestamps
last_activity: datetime
profile_last_updated: datetime
class Config:
use_enum_values = True2. AI-Powered Profile Analysis với HolySheep
Điểm mấu chốt của hệ thống là khả năng phân tích hành vi học tập và sinh ra insights chính xác. Tôi sử dụng HolySheep AI vì chi phí chỉ $0.42/MTok (so với $8 của GPT-4.1), phù hợp cho việc xử lý hàng triệu hồ sơ học sinh mà không lo về chi phí.
# services/profile_analysis.py
import httpx
import json
from typing import List, Dict, Optional
from datetime import datetime
import asyncio
from dataclasses import dataclass
@dataclass
class AnalysisResult:
learning_style: str
strengths: List[str]
weaknesses: List[str]
recommended_content_types: List[str]
engagement_insights: str
suggested_difficulty_adjustment: float
confidence_score: float
class HolySheepProfileAnalyzer:
"""
Sử dụng HolySheep AI để phân tích hồ sơ học sinh
Chi phí: $0.42/MTok — tiết kiệm 85%+ so với OpenAI
Độ trễ trung bình: <50ms với cấu hình optimized
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.client = httpx.AsyncClient(
timeout=30.0,
limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
)
async def analyze_student_behavior(
self,
student_id: str,
interaction_history: List[Dict],
performance_data: Dict
) -> AnalysisResult:
"""
Phân tích toàn diện hành vi học sinh
"""
# Xây dựng prompt cho AI analysis
analysis_prompt = self._build_analysis_prompt(
interaction_history,
performance_data
)
# Gọi HolySheep API — chi phí cực thấp
response = await self._call_holysheep(analysis_prompt)
# Parse và trả về structured result
return self._parse_analysis_response(response)
def _build_analysis_prompt(
self,
interactions: List[Dict],
performance: Dict
) -> str:
"""
Xây dựng prompt chi tiết cho việc phân tích học sinh
"""
# Format interaction data
interaction_summary = self._summarize_interactions(interactions)
prompt = f"""Bạn là chuyên gia phân tích giáo dục. Phân tích hồ sơ học sinh sau:
Dữ liệu tương tác (30 ngày gần nhất):
{interaction_summary}
Dữ liệu hiệu suất:
- Điểm trung bình: {performance.get('avg_score', 'N/A')}
- Tỷ lệ hoàn thành bài tập: {performance.get('completion_rate', 'N/A')}%
- Số giờ học/tuần: {performance.get('hours_per_week', 'N/A')}
- Các môn học yếu: {', '.join(performance.get('weak_subjects', []))}
- Các môn học mạnh: {', '.join(performance.get('strong_subjects', []))}
Yêu cầu phân tích:
1. Xác định phong cách học tập chính (visual/auditory/kinesthetic/reading)
2. Liệt kê 3-5 điểm mạnh cần phát huy
3. Liệt kê 3-5 điểm cần cải thiện
4. Đề xuất loại nội dung phù hợp nhất (video/bài tập/sơ đồ/casestudy)
5. Mức độ điều chỉnh độ khó (-1.0 đến +1.0)
6. Insights về mức độ tương tác và động lực
Trả lời theo format JSON:
{{
"learning_style": "string",
"strengths": ["string"],
"weaknesses": ["string"],
"recommended_content_types": ["string"],
"engagement_insights": "string",
"difficulty_adjustment": number,
"confidence": number
}}"""
return prompt
async def _call_holysheep(self, prompt: str) -> Dict:
"""
Gọi HolySheep API với retry logic và error handling
"""
payload = {
"model": "deepseek-v3.2", # Model rẻ nhất, hiệu năng tốt
"messages": [
{"role": "system", "content": "Bạn là chuyên gia giáo dục AI."},
{"role": "user", "content": prompt}
],
"temperature": 0.3, # Low temperature cho analysis
"max_tokens": 500,
"stream": False
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# Retry với exponential backoff
for attempt in range(3):
try:
response = await self.client.post(
f"{self.BASE_URL}/chat/completions",
json=payload,
headers=headers
)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
# Rate limit - wait và retry
await asyncio.sleep(2 ** attempt)
continue
else:
raise Exception(f"API Error: {response.status_code}")
except httpx.TimeoutException:
if attempt == 2:
raise
await asyncio.sleep(1)
raise Exception("Failed after 3 attempts")
def _summarize_interactions(self, interactions: List[Dict]) -> str:
"""Tóm tắt dữ liệu tương tác để giảm token usage"""
if not interactions:
return "Không có dữ liệu tương tác"
content_types = {}
time_distribution = {"morning": 0, "afternoon": 0, "evening": 0}
avg_duration = 0
for interaction in interactions:
content_type = interaction.get("content_type", "unknown")
content_types[content_type] = content_types.get(content_type, 0) + 1
hour = interaction.get("hour", 12)
if 6 <= hour < 12:
time_distribution["morning"] += 1
elif 12 <= hour < 18:
time_distribution["afternoon"] += 1
else:
time_distribution["evening"] += 1
avg_duration += interaction.get("duration_seconds", 0)
avg_duration = avg_duration / len(interactions) if interactions else 0
summary = f"""
- Tổng tương tác: {len(interactions)}
- Phân bố nội dung: {content_types}
- Thời gian học: Sáng {time_distribution['morning']}, Chiều {time_distribution['afternoon']}, Tối {time_distribution['evening']}
- Thời lượng trung bình: {avg_duration:.0f} giây
"""
return summary
def _parse_analysis_response(self, response: Dict) -> AnalysisResult:
"""Parse AI response thành structured result"""
content = response["choices"][0]["message"]["content"]
# Extract JSON từ response
try:
# Try direct parse
data = json.loads(content)
except json.JSONDecodeError:
# Extract from markdown code block
import re
json_match = re.search(r'``(?:json)?\s*([\s\S]*?)``', content)
if json_match:
data = json.loads(json_match.group(1))
else:
# Try to find JSON pattern
data = json.loads(content.strip())
return AnalysisResult(
learning_style=data["learning_style"],
strengths=data["strengths"],
weaknesses=data["weaknesses"],
recommended_content_types=data["recommended_content_types"],
engagement_insights=data["engagement_insights"],
suggested_difficulty_adjustment=data["difficulty_adjustment"],
confidence_score=data["confidence"]
)
async def close(self):
await self.client.aclose()
Ví dụ sử dụng
async def main():
analyzer = HolySheepProfileAnalyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
interactions = [
{"content_type": "video", "duration_seconds": 1800, "hour": 19},
{"content_type": "quiz", "duration_seconds": 300, "hour": 19},
{"content_type": "video", "duration_seconds": 2100, "hour": 20},
]
performance = {
"avg_score": 7.5,
"completion_rate": 85,
"hours_per_week": 10,
"weak_subjects": ["Toán", "Vật lý"],
"strong_subjects": ["Ngữ văn", "Lịch sử"]
}
result = await analyzer.analyze_student_behavior(
student_id="STU001",
interaction_history=interactions,
performance_data=performance
)
print(f"Learning Style: {result.learning_style}")
print(f"Strengths: {result.strengths}")
print(f"Difficulty Adjustment: {result.suggested_difficulty_adjustment}")
await analyzer.close()
if __name__ == "__main__":
asyncio.run(main())3. Knowledge Graph Construction
# services/knowledge_graph.py
from typing import Dict, List, Set, Tuple
from collections import defaultdict
import numpy as np
class KnowledgeGraph:
"""
Xây dựng knowledge graph cho mỗi học sinh
Theo dõi mối quan hệ giữa các khái niệm và độ thành thạo
"""
def __init__(self):
# adjacency: concept -> {related_concept: weight}
self.adjacency: Dict[str, Dict[str, float]] = defaultdict(dict)
# mastery: concept -> mastery_score (0.0 - 1.0)
self.mastery: Dict[str, float] = {}
# prerequisites: concept -> set of prerequisite concepts
self.prerequisites: Dict[str, Set[str]] = defaultdict(set)
def add_concept(
self,
concept: str,
mastery_score: float,
prerequisites: List[str] = None
):
"""Thêm một khái niệm vào knowledge graph"""
self.mastery[concept] = max(0.0, min(1.0, mastery_score))
if prerequisites:
for prereq in prerequisites:
self.prerequisites[concept].add(prereq)
# Cập nhật adjacency với weight dựa trên mastery
if prereq in self.mastery:
weight = (self.mastery[prereq] + self.mastery[concept]) / 2
self.adjacency[concept][prereq] = weight
self.adjacency[prereq][concept] = weight
def update_mastery(self, concept: str, new_score: float):
"""Cập nhật độ thành thạo và propagate changes"""
old_score = self.mastery.get(concept, 0.0)
self.mastery[concept] = max(0.0, min(1.0, new_score))
# Propagate to dependent concepts
self._propagate_mastery_change(concept, old_score, new_score)
def _propagate_mastery_change(
self,
concept: str,
old_score: float,
new_score: float
):
"""Propagate thay đổi mastery đến các concepts liên quan"""
change = new_score - old_score
# Tìm tất cả concepts phụ thuộc vào concept này
dependents = [
c for c, prereqs in self.prerequisites.items()
if concept in prereqs
]
for dependent in dependents:
# Giảm impact theo khoảng cách
adjustment = change * 0.3 # Decay factor
new_mastery = self.mastery[dependent] + adjustment
self.mastery[dependent] = max(0.0, min(1.0, new_mastery))
def get_learning_path(
self,
target_concepts: List[str],
max_concepts: int = 10
) -> List[Tuple[str, float]]:
"""
Tính toán lộ trình học tập tối ưu
Ưu tiên các concepts có prerequisite chưa thành thạo
"""
learning_path = []
covered = set()
for target in target_concepts:
if len(learning_path) >= max_concepts:
break
# Tìm prerequisites chưa cover
path_segments = self._find_optimal_path(target, covered)
for concept, priority in path_segments:
if concept not in covered and len(learning_path) < max_concepts:
learning_path.append((concept, priority))
covered.add(concept)
# Sort theo priority (cao -> thấp)
learning_path.sort(key=lambda x: x[1], reverse=True)
return learning_path[:max_concepts]
def _find_optimal_path(
self,
concept: str,
covered: Set[str]
) -> List[Tuple[str, float]]:
"""Tìm đường đi tối ưu đến một concept"""
path = []
# Thêm prerequisites trước
for prereq in self.prerequisites.get(concept, []):
if prereq not in covered:
mastery = self.mastery.get(prereq, 0.0)
priority = (1.0 - mastery) * 0.8 # Ưu tiên concept chưa thành thạo
path.append((prereq, priority))
# Thêm concept chính
mastery = self.mastery.get(concept, 0.0)
priority = (1.0 - mastery) * 0.9
path.append((concept, priority))
return path
def get_gaps_analysis(self) -> Dict[str, List[str]]:
"""
Phân tích các lỗ hổng kiến thức
Trả về: {concept: [missing_prerequisites]}
"""
gaps = {}
for concept, prereqs in self.prerequisites.items():
missing = []
for prereq in prereqs:
mastery = self.mastery.get(prereq, 0.0)
if mastery < 0.6: # Ngưỡng "đã thành thạo"
missing.append(prereq)
if missing:
gaps[concept] = missing
return gaps
def to_vector(self, all_concepts: List[str]) -> np.ndarray:
"""
Chuyển knowledge graph thành vector để so sánh/học
"""
return np.array([
self.mastery.get(concept, 0.0)
for concept in all_concepts
])
def similarity(self, other: 'KnowledgeGraph') -> float:
"""Tính cosine similarity với một knowledge graph khác"""
all_concepts = set(self.mastery.keys()) | set(other.mastery.keys())
if not all_concepts:
return 0.0
v1 = self.to_vector(list(all_concepts))
v2 = other.to_vector(list(all_concepts))
# Cosine similarity
dot_product = np.dot(v1, v2)
norm_product = np.linalg.norm(v1) * np.linalg.norm(v2)
if norm_product == 0:
return 0.0
return dot_product / norm_product
Ví dụ sử dụng
def demo():
kg = KnowledgeGraph()
# Thêm concepts với prerequisites (chủ đề Toán)
kg.add_concept("Pythagorean Theorem", 0.7, prerequisites=["Triangle Basics", "Square Root"])
kg.add_concept("Triangle Basics", 0.9, prerequisites=[])
kg.add_concept("Square Root", 0.6, prerequisites=["Basic Arithmetic"])
kg.add_concept("Basic Arithmetic", 0.95, prerequisites=[])
# Get learning path
path = kg.get_learning_path(
target_concepts=["Pythagorean Theorem", "Trigonometry Basics"],
max_concepts=5
)
print("Learning Path:")
for concept, priority in path:
print(f" - {concept} (priority: {priority:.2f})")
# Analyze gaps
gaps = kg.get_gaps_analysis()
print(f"\nKnowledge Gaps: {gaps}")
# Similarity với student khác
kg2 = KnowledgeGraph()
kg2.add_concept("Pythagorean Theorem", 0.3)
kg2.add_concept("Triangle Basics", 0.5)
print(f"\nSimilarity with student 2: {kg.similarity(kg2):.2%}")
if __name__ == "__main__":
demo()Recommendation Engine Implementation
Hybrid Filtering với Collaborative và Content-based
# services/recommendation_engine.py
from typing import List, Dict, Optional, Tuple
import numpy as np
from dataclasses import dataclass
import redis.asyncio as redis
import json
from datetime import datetime
@dataclass
class RecommendedItem:
item_id: str
score: float
reasons: List[str]
content_type: str
difficulty: float
estimated_time: int # minutes
class RecommendationEngine:
"""
Hybrid Recommendation Engine kết hợp:
1. Content-based filtering (dựa trên knowledge graph)
2. Collaborative filtering (dựa trên similar students)
3. Context-aware re-ranking
"""
def __init__(
self,
redis_client: redis.Redis,
knowledge_graph_service,
analyzer: HolySheepProfileAnalyzer
):
self.redis = redis_client
self.kg_service = knowledge_graph_service
self.analyzer = analyzer
# Weights cho hybrid scoring
self.weights = {
"content": 0.4,
"collaborative": 0.3,
"context": 0.2,
"diversity": 0.1
}
async def get_recommendations(
self,
student_id: str,
num_items: int = 10,
exclude_viewed: bool = True
) -> List[RecommendedItem]:
"""
Lấy danh sách gợi ý cá nhân hóa cho học sinh
"""
# 1. Load student profile từ cache
profile = await self._load_profile(student_id)
if not profile:
# Fallback to cold start strategy
return await self._cold_start_recommendations(num_items)
# 2. Get candidate items từ các nguồn
candidates = await self._generate_candidates(profile, num_items * 3)
# 3. Calculate scores
scored_items = []
for item in candidates:
content_score = self._content_based_score(item, profile)
collab_score = await self._collaborative_score(item, student_id)
context_score = self._context_score(item, profile)
# Hybrid score
hybrid_score = (
self.weights["content"] * content_score +
self.weights["collaborative"] * collab_score +
self.weights["context"] * context_score
)
scored_items.append((item, hybrid_score))
# 4. Rerank với diversity
reranked = self._rerank_with_diversity(
scored_items,
profile,
num_items
)
# 5. Log recommendation
await self._log_recommendation(student_id, reranked)
return reranked
async def _load_profile(self, student_id: str) -> Optional[Dict]:
"""Load profile từ Redis cache"""
cache_key = f"profile:{student_id}"
data = await self.redis.get(cache_key)
if data:
return json.loads(data)
return None
async def _generate_candidates(
self,
profile: Dict,
num_candidates: int
) -> List[Dict]:
"""
Generate candidate items từ content catalog
Lọc theo:
- Subject alignment
- Difficulty range
- Content type preference
"""
# 1. Get learning path từ knowledge graph
target_concepts = profile.get("improvement_areas", [])
learning_path = self.kg_service.get_learning_path(
target_concepts,
max_concepts=10
)
# 2. Query content catalog
# (Implementation depends on your content DB)
candidates = []
# Filter: difficulty within range
current_level = profile.get("knowledge_level", 0.5)
difficulty_range = 0.2 # +/- 20%
for concept, _ in learning_path:
# Query contents matching this concept
contents = await self._query_content_by_concept(
concept,
difficulty_min=current_level - difficulty_range,
difficulty_max=current_level + difficulty_range,
limit=num_candidates // len(learning_path)
)
candidates.extend(contents)
return candidates
def _content_based_score(
self,
item: Dict,
profile: Dict
) -> float:
"""
Content-based scoring:
- Subject match
- Difficulty match
- Content type preference
"""
score = 0.0
# Subject match
if item["subject"] in profile.get("subjects", []):
score += 0.3
# Difficulty match
item_diff = item.get("difficulty", 0.5)
profile_level = profile.get("knowledge_level", 0.5)
difficulty_gap = abs(item_diff - profile_level)
score += 0.3 * (1 - difficulty_gap) # Closer = higher score
# Content type preference
preferred_types = profile.get("preferred_content_types", [])
if item["content_type"] in preferred_types:
score += 0.2
# Novelty bonus (recommend things not yet learned)
if item.get("mastery_required", 1.0) > 0.7:
score += 0.2
return min(score, 1.0)
async def _collaborative_score(
self,
item: Dict,
student_id: str
) -> float:
"""
Collaborative filtering score:
Dựa trên hành vi của similar students
"""
# Get similar students
similar_key = f"similar_students:{student_id}"
similar_ids = await self.redis.smembers(similar_key)
if not similar_ids:
return 0.5 # Default score
# Count how many similar students interacted with this item
positive_count = 0
total_count = 0
for sim_id in similar_ids:
interaction_key = f"interactions:{sim_id}"
interacted = await self.redis.sismember(
interaction_key,
item["item_id"]
)
if interacted:
total_count += 1
# Check if they rated it positively
rating_key = f"rating:{sim_id}:{item['item_id']}"
rating = await self.redis.get(rating_key)
if rating and float(rating) >= 4.0:
positive_count += 1
if total_count == 0:
return 0.5
return positive_count / total_count
def _context_score(
self,
item: Dict,
profile: Dict
) -> float:
"""
Context-aware scoring:
- Time of day
- Session duration
- Recent performance
"""
score = 1.0
# Time of day preference
current_hour = datetime.now().hour
item_time_pref = item.get("optimal_time", {})
if item_time_pref:
if current_hour in item_time_pref.get("morning", []):
score *= 1.2
elif current_hour in item_time_pref.get("evening", []):
score *= 0.9 # Penalize if preferred morning content
# Performance trend adjustment
trend = profile.get("recent_performance_trend", "stable")
if trend == "declining" and item.get("difficulty", 0.5) > 0.6:
score *= 0.7 # Reduce difficulty items for struggling students
elif trend == "improving" and item.get("difficulty", 0.5) > 0.8:
score *= 1.3 # Challenge students doing well
return min(score, 1.0)
def _rerank_with_diversity(
self,
items: List[Tuple[Dict, float]],
profile: Dict,
num_items: int
) -> List[RecommendedItem]:
"""
Rerank với MMR (Maximal Marginal Relevance)
Đảm bảo diversity trong recommendations
"""
if not items:
return []
# Sort by score
items.sort(key=lambda x: x[1], reverse=True)
selected = []
selected_content_types = set()
max_same_type = num_items // 2 # Max 50% cùng content type
for item, score in items:
content_type = item["content_type"]
# Check diversity constraint
same_type_count = sum(
1 for s in selected
if s.content_type == content_type
)
if same_type_count >= max_same_type:
continue
# MMR calculation
mmr_score = (
0.5 * score -
0.5 * self._max_similarity_to_selected(item, selected)
)
if len(selected) < num_items:
selected.append(RecommendedItem(
item_id=item["item_id"],
score=mmr_score,
reasons=self._generate_reasons(item, profile),
content_type=content_type,
difficulty=item.get("difficulty", 0.5),
estimated_time=item.get("estimated_time", 15)
))
return selected
def _max_similarity_to_selected(
self,
item: Dict,
selected: List[RecommendedItem]
) -> float:
"""Tính max similarity với items đã chọn"""
if not selected:
return 0.0
max_sim = 0.0
for sel in selected:
# Similarity based on subject và difficulty
subj_sim = 1.0 if item["subject"] == sel.content_type else 0.0
diff_sim = 1.0 - abs(
item.get("difficulty", 0.5) - sel.difficulty
)
sim = 0.7 * subj_sim + 0.3 * diff_sim
max_sim = max(max_sim, sim)
return max_sim
def _generate_reasons(
self,
item: Dict,
profile: Dict
) -> List[str]:
"""Generate human-readable reasons cho recommendation"""
reasons = []
if item["subject"] in profile.get("weak_subjects", []):
reasons.append("Cải thiện điểm yếu")
if item.get("difficulty", 0.5) < profile.get("knowledge_level", 0.5):
reasons.append("Ôn tập kiến thức cơ bản")