Gamification Engine: Architektura a státní stroj
80 % studentů říká, že jsou více motivovaní, když studují na platformách, které začlenit herní prvky. Trh s gamifikací v EdTech poroste na a CAGR nad 36 % do roku 2032. Ale rozdíl mezi gamifikací efektivní a povrchní není o přidávání náhodných odznaků a bodů: je to o solidnosti zzákladní architektura.
Špatně navržený gamifikační engine rychle hromadí technický dluh: logiku pevně zakódovaná progrese, závodní podmínky na body, když mnoho uživatelů dokončí činnosti ve stejnou dobu, žebříčky, které se stávají úzkými hrdly pro tisíce uživatelů, odznaky udělené několikrát kvůli chybám ve správě stavu. Tyto problémy nejsou teoretické: jsou to běžné patologie každé platformy, která přidává gamifikaci jako dodatečný nápad místo toho, aby to bylo navrženo jako prvotřídní funkce.
V tomto článku postavíme a Gamification Engine robustní a škálovatelné s explicitními stavovými automaty, sourcingem událostí pro sledovatelnost, Redis pro výsledkové tabulky v reálném čase a konfigurovatelný systém pravidel, který umožňuje tvůrcům obsahu definovat novou mechaniku, aniž byste se dotkli kódu.
Co se dozvíte v tomto článku
- Událostmi řízená architektura škálovatelného gamifikačního enginu
- Státní stroj pro řízení postupu studentů
- Systém odznaků s konfigurovatelnými pravidly (bez pevného kódování)
- Žebříček v reálném čase s Redis Sorted Sets
- Sledování pruhů a systém odměn za čas
- XP (zkušenostní body) s úrovněmi a křivkami progrese
- Anti-vzory: jak se vyhnout rasovým podmínkám a dvojímu přiřazení
- Push upozornění na úspěchy s WebSocket
1. Principy návrhu: vnitřní vs vnější gamifikace
Před psaním kódu je nezbytné pochopit rozdíl mezi motivací vnější (body, odznaky, hodnocení - externí motivátory) vnitřní (zvědavost, smysl pro kompetence, samostatnost). Úspěšná vzdělávací gamifikace využívá vnější motivátory jako odrazový můstek rozvíjet vnitřní motivaci, ne jako náhradu.
Nejúčinnější mechanismy gamifikace pro vzdělávání v roce 2025 jsou: pruh (sekvence po sobě jdoucích dnů studia), mistrovský postup (vizualizace pokroku směrem k mistrovství), sociální učení (pořadí mezi vrstevníky s podobnou úrovní), e smysluplné volby (alternativní cesty s různými odměnami). Nejméně efektivní (a potenciálně škodlivou) mechanikou jsou globální žebříčky kde noví studenti vždy považují špičkové umělce za nedosažitelné.
Komponenty Gamification Engine
| Komponent | Odpovědnost | Technologie |
|---|---|---|
| Event Bus | Sbírá výukové akce | Kafka/Redis Streams |
| Pravidlo Engine | Vyhodnoťte nastavitelná pravidla | Pravidla Python + JSON |
| Státní stroj | Řídí postup studentů | Vlastní Python FSM |
| XP kalkulačka | Spočítejte body zkušeností | Python + PostgreSQL |
| Správce odznaků | Udělení idempotentního odznaku | Mezipaměť PostgreSQL + Redis |
| Žebříčky | Žebříčky v reálném čase | Seřazené sady Redis |
| Streak Tracker | Sledujte studijní sekvence | Redis + PostgreSQL |
| Notifikační centrum | Push úspěch v reálném čase | WebSocket / SSE |
2. Státní stroj pro postup studentů
Státní automat je srdcem gamifikace. Každý student je v jednom stát což odráží jeho úroveň zapojení a progrese. K přechodům mezi stavy dochází v reakci na události (dokončení lekce, správná odpověď, po sobě jdoucí přihlášení). Udělejte tento stavový automat explicitním odstraňuje problém „implicitních stavů“, které způsobují těžko reprodukovatelné chyby.
# gamification/state_machine.py
from enum import Enum, auto
from dataclasses import dataclass, field
from typing import Dict, List, Callable, Optional
from datetime import datetime
class LearnerState(Enum):
"""Stati possibili di un learner nella piattaforma."""
ONBOARDING = auto() # Primo accesso, < 3 corsi completati
ACTIVE = auto() # Studente attivo, streak in corso
AT_RISK = auto() # Nessuna attivita da 7-13 giorni
DORMANT = auto() # Inattivo da 14+ giorni
MASTERY = auto() # Ha completato un corso con punteggio >= 90%
MENTOR = auto() # Ha aiutato almeno 5 altri studenti
class LearnerEvent(Enum):
"""Eventi che causano transizioni di stato."""
LESSON_COMPLETED = "lesson_completed"
QUIZ_PASSED = "quiz_passed"
QUIZ_FAILED = "quiz_failed"
COURSE_COMPLETED = "course_completed"
PEER_HELPED = "peer_helped"
DAILY_LOGIN = "daily_login"
STREAK_BROKEN = "streak_broken"
INACTIVE_7_DAYS = "inactive_7_days"
INACTIVE_14_DAYS = "inactive_14_days"
REACTIVATED = "reactivated"
@dataclass
class StateTransition:
from_state: LearnerState
event: LearnerEvent
to_state: LearnerState
conditions: List[Callable] = field(default_factory=list)
actions: List[Callable] = field(default_factory=list)
@dataclass
class LearnerContext:
student_id: str
current_state: LearnerState
xp: int = 0
streak_days: int = 0
courses_completed: int = 0
peers_helped: int = 0
last_activity: Optional[datetime] = None
metadata: Dict = field(default_factory=dict)
class LearnerStateMachine:
"""
State machine per la progressione del learner.
Tutte le transizioni sono esplicite e testabili.
"""
TRANSITIONS: List[StateTransition] = [
# ONBOARDING -> ACTIVE: ha completato la prima lezione
StateTransition(
from_state=LearnerState.ONBOARDING,
event=LearnerEvent.LESSON_COMPLETED,
to_state=LearnerState.ACTIVE,
),
# ACTIVE -> AT_RISK: 7 giorni di inattivita
StateTransition(
from_state=LearnerState.ACTIVE,
event=LearnerEvent.INACTIVE_7_DAYS,
to_state=LearnerState.AT_RISK,
),
# AT_RISK -> DORMANT: altri 7 giorni senza attivita
StateTransition(
from_state=LearnerState.AT_RISK,
event=LearnerEvent.INACTIVE_14_DAYS,
to_state=LearnerState.DORMANT,
),
# DORMANT/AT_RISK -> ACTIVE: torna ad essere attivo
StateTransition(
from_state=LearnerState.AT_RISK,
event=LearnerEvent.REACTIVATED,
to_state=LearnerState.ACTIVE,
),
StateTransition(
from_state=LearnerState.DORMANT,
event=LearnerEvent.REACTIVATED,
to_state=LearnerState.ACTIVE,
),
# ACTIVE -> MASTERY: completa corso con voto alto
StateTransition(
from_state=LearnerState.ACTIVE,
event=LearnerEvent.COURSE_COMPLETED,
to_state=LearnerState.MASTERY,
conditions=[lambda ctx: ctx.metadata.get("score", 0) >= 90],
),
# MASTERY -> MENTOR: aiuta abbastanza peer
StateTransition(
from_state=LearnerState.MASTERY,
event=LearnerEvent.PEER_HELPED,
to_state=LearnerState.MENTOR,
conditions=[lambda ctx: ctx.peers_helped >= 5],
),
]
def __init__(self, event_publisher):
self.event_publisher = event_publisher
# Indice per lookup rapido O(1)
self._transition_index: Dict[tuple, StateTransition] = {
(t.from_state, t.event): t
for t in self.TRANSITIONS
}
def process_event(
self,
context: LearnerContext,
event: LearnerEvent,
event_data: Dict = None,
) -> LearnerContext:
"""
Processa un evento e restituisce un NUOVO context (immutabile).
Non modifica mai il context in input.
"""
event_data = event_data or {}
key = (context.current_state, event)
transition = self._transition_index.get(key)
if not transition:
# Nessuna transizione definita: evento ignorato
return context
# Aggiorna il metadata del context con i dati dell'evento
updated_metadata = {**context.metadata, **event_data}
temp_context = LearnerContext(
**{
**context.__dict__,
"metadata": updated_metadata,
}
)
# Verifica tutte le condizioni
if not all(cond(temp_context) for cond in transition.conditions):
return context # Condizioni non soddisfatte, nessuna transizione
# Crea nuovo context con nuovo stato
new_context = LearnerContext(
**{
**temp_context.__dict__,
"current_state": transition.to_state,
"last_activity": datetime.utcnow(),
}
)
# Pubblica evento di transizione
self.event_publisher.publish({
"type": "state_transition",
"student_id": context.student_id,
"from_state": context.current_state.name,
"to_state": transition.to_state.name,
"trigger_event": event.value,
"timestamp": datetime.utcnow().isoformat(),
})
return new_context3. Systém XP a úrovní s křivkami progrese
Systém zkušenostních bodů (XP) musí vyvážit rychlost postupu: příliš rychle a studenti vyčerpají maximum za pár dní a budou se nudit, příliš pomalé a odradí je. Použijme jeden logaritmická křivka pro prvních několik úrovní (rychlý a uspokojivý postup) je to strmější k vysokým úrovním (významným cílům).
# gamification/xp_system.py
import math
from dataclasses import dataclass
from typing import Dict, List
from enum import Enum
class ActivityType(Enum):
LESSON_COMPLETED = "lesson"
QUIZ_PASSED = "quiz_pass"
QUIZ_PERFECT = "quiz_perfect" # 100% corretto
COURSE_COMPLETED = "course"
STREAK_BONUS = "streak"
PEER_REVIEW = "peer_review"
HELP_PEER = "help_peer"
FIRST_LOGIN_DAY = "first_login"
# XP base per ogni tipo di attivita
XP_BASE_REWARDS: Dict[str, int] = {
ActivityType.LESSON_COMPLETED.value: 50,
ActivityType.QUIZ_PASSED.value: 100,
ActivityType.QUIZ_PERFECT.value: 200,
ActivityType.COURSE_COMPLETED.value: 500,
ActivityType.STREAK_BONUS.value: 25,
ActivityType.PEER_REVIEW.value: 75,
ActivityType.HELP_PEER.value: 150,
ActivityType.FIRST_LOGIN_DAY.value: 10,
}
@dataclass
class XPGrant:
student_id: str
activity_type: str
base_xp: int
multiplier: float
final_xp: int
new_total_xp: int
new_level: int
level_up: bool
reason: str
class XPCalculator:
"""
Calcola XP con moltiplicatori basati su streak, difficolta e tempo.
Usa la formula logaritmica per i livelli.
"""
def calculate_level(self, total_xp: int) -> int:
"""
Livello = floor(log2(total_xp / 100)) + 1
Esempi: 100 XP = Lv1, 200 XP = Lv2, 400 XP = Lv3, 800 XP = Lv4...
"""
if total_xp < 100:
return 1
return min(int(math.log2(total_xp / 100)) + 1, 100)
def xp_for_level(self, level: int) -> int:
"""XP necessari per raggiungere il livello."""
if level <= 1:
return 0
return 100 * (2 ** (level - 1))
def calculate_multiplier(
self,
streak_days: int,
difficulty: int, # 1-5
time_of_day_bonus: bool = False,
) -> float:
"""
Calcola il moltiplicatore XP.
- Streak: +5% per ogni giorno consecutivo (max +50%)
- Difficolta: +10% per ogni livello sopra il baseline (2)
- Bonus orario: +20% per studio nelle ore ottimali (8-12, 14-18)
"""
streak_bonus = min(streak_days * 0.05, 0.5) # Max 50%
difficulty_bonus = max(0, (difficulty - 2) * 0.10) # Baseline difficolta = 2
time_bonus = 0.20 if time_of_day_bonus else 0.0
return 1.0 + streak_bonus + difficulty_bonus + time_bonus
def grant_xp(
self,
student_id: str,
activity: ActivityType,
current_xp: int,
streak_days: int,
difficulty: int = 2,
time_of_day_bonus: bool = False,
) -> XPGrant:
base_xp = XP_BASE_REWARDS.get(activity.value, 10)
multiplier = self.calculate_multiplier(streak_days, difficulty, time_of_day_bonus)
final_xp = int(base_xp * multiplier)
new_total = current_xp + final_xp
old_level = self.calculate_level(current_xp)
new_level = self.calculate_level(new_total)
return XPGrant(
student_id=student_id,
activity_type=activity.value,
base_xp=base_xp,
multiplier=multiplier,
final_xp=final_xp,
new_total_xp=new_total,
new_level=new_level,
level_up=new_level > old_level,
reason=f"{activity.value} x{multiplier:.2f} (streak:{streak_days}d, diff:{difficulty}/5)",
)4. Systém odznaků s idempotenci a konfigurovatelnými pravidly
Nejčastějším problémem v systémech odznaků je dvojí přiřazení: stejný odznak se uděluje vícekrát za podmínky závodu nebo opakování neúspěšných událostí. Řešení aidempotence: každá operace přiřazení odznaku lze provést vícekrát vždy se stejným výsledkem. Používáme a omezení UNIKÁTNÍ v PostgreSQL jako zajištění úrovně databáze.
# gamification/badge_manager.py
import json
from dataclasses import dataclass
from typing import List, Dict, Any, Optional
from datetime import datetime
from enum import Enum
class BadgeTier(Enum):
BRONZE = "bronze"
SILVER = "silver"
GOLD = "gold"
PLATINUM = "platinum"
LEGENDARY = "legendary"
@dataclass
class BadgeDefinition:
badge_id: str
name: str
description: str
tier: BadgeTier
icon_url: str
# Regola come JSON (no hardcoding!)
rule: Dict[str, Any]
# Esempi di regole configurabili
BADGE_CATALOG: List[BadgeDefinition] = [
BadgeDefinition(
badge_id="first_lesson",
name="Primo Passo",
description="Hai completato la tua prima lezione",
tier=BadgeTier.BRONZE,
icon_url="/badges/first-lesson.svg",
rule={"event": "lesson_completed", "count": 1},
),
BadgeDefinition(
badge_id="week_streak",
name="Settimana di Fuoco",
description="7 giorni di studio consecutivi",
tier=BadgeTier.SILVER,
icon_url="/badges/week-streak.svg",
rule={"metric": "streak_days", "gte": 7},
),
BadgeDefinition(
badge_id="perfect_quiz",
name="Perfezionista",
description="Quiz completato con 100% di risposte corrette",
tier=BadgeTier.GOLD,
icon_url="/badges/perfect-quiz.svg",
rule={"event": "quiz_completed", "score": 100},
),
BadgeDefinition(
badge_id="mentor",
name="Mentore della Comunita",
description="Hai aiutato 10 compagni di corso",
tier=BadgeTier.PLATINUM,
icon_url="/badges/mentor.svg",
rule={"metric": "peers_helped", "gte": 10},
),
]
class BadgeManager:
def __init__(self, db, redis_client):
self.db = db
self.redis = redis_client
self._catalog: Dict[str, BadgeDefinition] = {b.badge_id: b for b in BADGE_CATALOG}
async def evaluate_and_award(
self,
student_id: str,
event: str,
student_metrics: Dict[str, Any],
) -> List[str]:
"""
Valuta tutti i badge applicabili e li assegna (idempotente).
Ritorna la lista di badge_id appena assegnati.
"""
newly_awarded = []
for badge_id, badge_def in self._catalog.items():
if await self._is_already_awarded(student_id, badge_id):
continue
if self._evaluate_rule(badge_def.rule, event, student_metrics):
awarded = await self._award_badge_idempotent(student_id, badge_id)
if awarded:
newly_awarded.append(badge_id)
return newly_awarded
async def _award_badge_idempotent(self, student_id: str, badge_id: str) -> bool:
"""
Assegna il badge usando INSERT ... ON CONFLICT DO NOTHING.
Garantisce idempotenza a livello DB.
Ritorna True se il badge e stato effettivamente assegnato ora.
"""
result = await self.db.execute(
"""INSERT INTO student_badges (student_id, badge_id, awarded_at)
VALUES (:sid, :bid, :ts)
ON CONFLICT (student_id, badge_id) DO NOTHING
RETURNING badge_id""",
{"sid": student_id, "bid": badge_id, "ts": datetime.utcnow()},
)
rows = result.fetchall()
if rows:
# Invalida cache badge studente
await self.redis.delete(f"student:badges:{student_id}")
await self.db.commit()
return True
return False # Già esisteva
async def _is_already_awarded(self, student_id: str, badge_id: str) -> bool:
"""Controlla cache Redis prima di interrogare il DB."""
cache_key = f"student:badges:{student_id}"
cached = await self.redis.get(cache_key)
if cached:
badges = json.loads(cached)
return badge_id in badges
# Cache miss: interroga DB
result = await self.db.execute(
"SELECT badge_id FROM student_badges WHERE student_id = :sid",
{"sid": student_id},
)
badges = [row[0] for row in result.fetchall()]
await self.redis.setex(cache_key, 300, json.dumps(badges)) # TTL 5 minuti
return badge_id in badges
def _evaluate_rule(
self,
rule: Dict,
current_event: str,
metrics: Dict[str, Any],
) -> bool:
"""Valuta una regola badge configurabile."""
# Regola basata su evento
if "event" in rule:
if rule["event"] != current_event:
return False
if "count" in rule:
return metrics.get(f"event_count_{rule['event']}", 0) >= rule["count"]
if "score" in rule:
return metrics.get("last_score", 0) >= rule["score"]
return True
# Regola basata su metrica
if "metric" in rule:
metric_val = metrics.get(rule["metric"], 0)
if "gte" in rule:
return metric_val >= rule["gte"]
if "eq" in rule:
return metric_val == rule["eq"]
if "lte" in rule:
return metric_val <= rule["lte"]
return False5. Žebříček v reálném čase s Redis tříděnými sadami
Globální žebříčky jsou dvojsečná zbraň: motivují špičkové hráče ale demotivují ty na konci žebříčku. Moderním řešením je segmentovaný žebříček: každý student soutěží s 50 uživateli blíže jeho skóre, takže soutěž má vždy smysl. Nástrojem jsou řazené sady Redis se složitostí O(log N) pro všechny operace ideální pro hodnocení v reálném čase.
# gamification/leaderboard.py
import json
from typing import List, Dict, Optional
from datetime import datetime, date
import redis.asyncio as redis
@dataclass
class LeaderboardEntry:
rank: int
student_id: str
display_name: str
xp: int
level: int
avatar_url: str
is_current_user: bool = False
class LeaderboardManager:
"""
Leaderboard multi-scope con Redis Sorted Sets.
Scope: global, course-specific, weekly, peer-group.
"""
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
def _key(self, scope: str, period: str = "all-time") -> str:
"""Genera chiave Redis per uno scope e periodo."""
if period == "weekly":
week = date.today().isocalendar()[1]
year = date.today().year
return f"leaderboard:{scope}:{year}:w{week}"
return f"leaderboard:{scope}:{period}"
async def update_score(
self,
student_id: str,
xp_delta: int,
scopes: List[str],
) -> None:
"""
Aggiorna il punteggio in tutti gli scope rilevanti.
Usa pipeline Redis per operazioni atomiche.
"""
async with self.redis.pipeline(transaction=True) as pipe:
for scope in scopes:
for period in ["all-time", "weekly"]:
key = self._key(scope, period)
pipe.zincrby(key, xp_delta, student_id)
# Imposta TTL per leaderboard settimanale
if period == "weekly":
pipe.expire(key, 7 * 24 * 3600)
await pipe.execute()
async def get_leaderboard(
self,
scope: str,
page: int = 1,
page_size: int = 10,
period: str = "all-time",
) -> List[Dict]:
"""Recupera la leaderboard paginata per uno scope."""
key = self._key(scope, period)
start = (page - 1) * page_size
end = start + page_size - 1
# ZREVRANGE: ordine decrescente (punteggio più alto prima)
entries = await self.redis.zrevrange(key, start, end, withscores=True)
result = []
for rank_offset, (student_id_bytes, score) in enumerate(entries):
student_id = student_id_bytes.decode() if isinstance(student_id_bytes, bytes) else student_id_bytes
result.append({
"rank": start + rank_offset + 1,
"student_id": student_id,
"xp": int(score),
})
return result
async def get_peer_leaderboard(
self,
student_id: str,
scope: str,
window: int = 25,
period: str = "all-time",
) -> List[Dict]:
"""
Leaderboard contestualizzata: mostra i 25 utenti sopra e sotto il corrente studente.
"""
key = self._key(scope, period)
# Posizione attuale dello studente (rank 0-indexed dal basso)
rank_from_bottom = await self.redis.zrank(key, student_id)
if rank_from_bottom is None:
return []
total = await self.redis.zcard(key)
rank_from_top = total - rank_from_bottom - 1
# Finestra: window utenti sopra e sotto
start = max(0, rank_from_top - window)
end = min(total - 1, rank_from_top + window)
entries = await self.redis.zrevrange(key, start, end, withscores=True)
result = []
for i, (sid_bytes, score) in enumerate(entries):
sid = sid_bytes.decode() if isinstance(sid_bytes, bytes) else sid_bytes
result.append({
"rank": start + i + 1,
"student_id": sid,
"xp": int(score),
"is_current_user": sid == student_id,
})
return result
async def get_student_rank(
self,
student_id: str,
scope: str,
period: str = "all-time",
) -> Optional[int]:
"""Recupera il rank assoluto di uno studente."""
key = self._key(scope, period)
total = await self.redis.zcard(key)
rank_from_bottom = await self.redis.zrank(key, student_id)
if rank_from_bottom is None:
return None
return total - rank_from_bottom # Rank dal top (1 = primo posto)6. Sledování pruhů a dočasné odměny
Pruhy jsou jednou z nejvýkonnějších mechanismů zapojení v EdTech. Duolingo vybudovalo impérium na konceptu „nepřerušte sérii“. Implementace musí správně zpracovat časová pásma (student v Japonsku nesmí ztratit sérii, protože studoval ve 23:59 místního času) a „zamrzne“ (dny milosti pro neočekávané události).
# gamification/streak_tracker.py
from datetime import datetime, date, timedelta
from typing import Optional, Dict
import zoneinfo
import redis.asyncio as redis
@dataclass
class StreakStatus:
student_id: str
current_streak: int
longest_streak: int
last_activity_date: Optional[date]
freeze_remaining: int # Giorni di grazia disponibili
streak_at_risk: bool # True se non ha ancora fatto attivita oggi
class StreakTracker:
FREEZE_MAX = 2 # Massimo 2 giorni di grazia
def __init__(self, redis_client: redis.Redis, db):
self.redis = redis_client
self.db = db
def _today_for_user(self, timezone: str) -> date:
"""Data corrente nel fuso orario dello studente."""
tz = zoneinfo.ZoneInfo(timezone)
return datetime.now(tz).date()
async def record_activity(
self,
student_id: str,
timezone: str = "UTC",
) -> Dict:
"""
Registra attivita dello studente e aggiorna la streak.
Ritorna le modifiche alla streak.
"""
today = self._today_for_user(timezone)
status = await self.get_status(student_id, timezone)
if not status.last_activity_date:
# Prima attivita in assoluto
new_streak = 1
streak_extended = True
elif status.last_activity_date == today:
# Già registrata oggi: nessun cambio
return {"streak_changed": False, "current_streak": status.current_streak}
elif status.last_activity_date == today - timedelta(days=1):
# Ieri: estendi streak
new_streak = status.current_streak + 1
streak_extended = True
elif (today - status.last_activity_date).days <= 1 + status.freeze_remaining:
# Streak salvata da un freeze
days_missed = (today - status.last_activity_date).days - 1
new_freeze = max(0, status.freeze_remaining - days_missed)
new_streak = status.current_streak + 1
streak_extended = True
await self._update_freeze(student_id, new_freeze)
else:
# Streak rotta
new_streak = 1
streak_extended = False
# Salva aggiornamento
await self._save_streak(student_id, new_streak, status.longest_streak, today)
return {
"streak_changed": True,
"streak_extended": streak_extended,
"current_streak": new_streak,
"longest_streak": max(new_streak, status.longest_streak),
"streak_broken": not streak_extended and status.current_streak > 1,
"previous_streak": status.current_streak if not streak_extended else None,
}
async def get_status(self, student_id: str, timezone: str = "UTC") -> StreakStatus:
"""Recupera lo stato streak corrente dalla cache Redis."""
cache_key = f"streak:{student_id}"
cached = await self.redis.hgetall(cache_key)
if cached:
last_date_str = cached.get(b"last_date", b"").decode()
last_date = date.fromisoformat(last_date_str) if last_date_str else None
today = self._today_for_user(timezone)
return StreakStatus(
student_id=student_id,
current_streak=int(cached.get(b"current", b"0").decode()),
longest_streak=int(cached.get(b"longest", b"0").decode()),
last_activity_date=last_date,
freeze_remaining=int(cached.get(b"freeze", b"0").decode()),
streak_at_risk=last_date != today if last_date else True,
)
# Cache miss: leggi dal DB
row = await self.db.execute(
"""SELECT current_streak, longest_streak, last_activity_date, freeze_remaining
FROM student_streaks WHERE student_id = :sid""",
{"sid": student_id},
)
data = row.fetchone()
if not data:
return StreakStatus(student_id, 0, 0, None, self.FREEZE_MAX, True)
today = self._today_for_user(timezone)
return StreakStatus(
student_id=student_id,
current_streak=data[0],
longest_streak=data[1],
last_activity_date=data[2],
freeze_remaining=data[3],
streak_at_risk=data[2] != today if data[2] else True,
)
async def _save_streak(
self,
student_id: str,
current: int,
longest: int,
last_date: date,
) -> None:
new_longest = max(current, longest)
cache_key = f"streak:{student_id}"
# Aggiorna Redis
await self.redis.hset(cache_key, mapping={
"current": current,
"longest": new_longest,
"last_date": last_date.isoformat(),
})
await self.redis.expire(cache_key, 86400 * 2) # TTL 2 giorni
# Aggiorna DB
await self.db.execute(
"""INSERT INTO student_streaks (student_id, current_streak, longest_streak, last_activity_date)
VALUES (:sid, :cur, :lon, :dt)
ON CONFLICT (student_id) DO UPDATE
SET current_streak = :cur, longest_streak = :lon, last_activity_date = :dt""",
{"sid": student_id, "cur": current, "lon": new_longest, "dt": last_date},
)
await self.db.commit()
async def _update_freeze(self, student_id: str, new_freeze: int) -> None:
await self.redis.hset(f"streak:{student_id}", "freeze", new_freeze)
await self.db.execute(
"UPDATE student_streaks SET freeze_remaining = :fr WHERE student_id = :sid",
{"sid": student_id, "fr": new_freeze},
)
await self.db.commit()Anti-vzory, kterým je třeba se vyhnout
- Jedinečný globální žebříček: Demotivujte 80 % uživatelů, kteří nejsou v top 10. Použijte peer-to-peer segmenty.
- Odznaky bez významu: Odznaky za každou triviální akci nafouknou systém. Odznaky musí představovat skutečné úspěchy.
- XP bez omezení rychlosti: Boti mohou akumulovat nekonečné XP. Implementujte omezení rychlosti a detekci anomálií.
- Implicitní stav: Správa progrese pomocí booleovských příznaků rozptýlených v kódu způsobuje chyby, které se obtížně ladí. Vždy používejte explicitní stavový automat.
- Přiřazení odznaku bez idempotence: Bez omezení UNIQUE a INSERT ON CONFLICT může opakování udělení stejného odznaku vícekrát.
- Proužek bez časového pásma: Uživatel v Asii by neměl ztratit řadu kvůli problému s offsetem UTC.
Závěry a další kroky
Postavili jsme základ robustního gamifikačního enginu: explicitní stavové automaty pro postup, XP systém s logaritmickými křivkami, idempotentní odznaky s pravidly konfigurovatelné výsledkové tabulky peer-to-peer v reálném čase s Redis a sledování pruhů s podporou časového pásma. Každá komponenta je navržena tak, aby se škálovala pro miliony uživatelů bez snížení výkonu.
Klíčem k úspěchu byl přístup konfigurovatelné: pravidla odznaků jsou definovány jako data JSON, nikoli napevno kódovaný kód. To umožňuje správcům obsahu vytvářet nové mechanismy gamifikace bez zapojení vývojářů.
V příštím článku prozkoumáme Průběh učení Analytics s xAPI a Kafkou: jak sbírat data o chování studentů standardizované a přeměnit je v praktické poznatky pro učitele a administrátory.
EdTech Engineering Series
- Škálovatelná architektura LMS: Vzor pro více nájemců
- Algoritmy adaptivního učení: Od teorie k produkci
- Streamování videa pro vzdělávání: WebRTC vs HLS vs DASH
- AI Proctoring Systems: Privacy-first with Computer Vision
- Personalizovaný lektor s LLM: RAG pro ukotvení znalostí
- Gamification Engine: Architecture and State Machine (tento článek)
- Learning Analytics: Data Pipeline s xAPI a Kafka
- Spolupráce v reálném čase v EdTech: CRDT a WebSocket
- Mobile-First EdTech: Offline-First Architecture
- Správa obsahu pro více nájemců: Správa verzí a SCORM