摘要

本文通过一个完整的教育学习平台案例,演示如何使用 OpenClaw 构建智能在线教育系统。文章涵盖课程管理、学习路径、智能推荐、学习分析等核心功能,帮助开发者掌握 OpenClaw 在教育科技场景的应用。通过详细的系统设计和代码实现,让读者了解教育学习平台的完整构建过程。🎓


1. 引言 - 教育学习平台概述

1.1 在线教育痛点

在线教育面临诸多挑战,传统平台难以满足个性化学习需求:

痛点 传统平台 OpenClaw方案
学习路径单一 固定课程顺序 智能学习路径
进度难以追踪 简单完成标记 多维度分析
缺乏互动 单向视频 智能问答
推荐不精准 热门推荐 个性化推荐
效果难评估 考试分数 综合评估

1.2 平台架构设计

用户服务层

分析引擎层

学习服务层

内容管理层

课程管理

资源管理

题库管理

学习路径

进度追踪

智能问答

作业考试

学习分析

能力评估

推荐引擎

用户管理

学习计划

成就系统

1.3 核心功能规划

功能模块 核心能力 技术实现
课程管理 课程内容管理 结构化存储
学习路径 个性化学习 知识图谱 + 推荐
智能问答 即时答疑 RAG + LLM
学习分析 多维分析 数据分析 + 可视化
能力评估 综合评估 模型评估

2. 课程管理模块

2.1 课程实体设计

from dataclasses import dataclass, field
from typing import Dict, List, Optional
from enum import Enum
import time

class CourseStatus(Enum):
    """课程状态"""
    DRAFT = "draft"
    PUBLISHED = "published"
    ARCHIVED = "archived"

class ResourceType(Enum):
    """资源类型"""
    VIDEO = "video"
    DOCUMENT = "document"
    QUIZ = "quiz"
    ASSIGNMENT = "assignment"
    LINK = "link"

@dataclass
class Resource:
    """学习资源"""
    id: str
    title: str
    type: ResourceType
    url: str
    duration: int = 0  # 视频时长(秒)
    description: str = ""
    metadata: Dict = field(default_factory=dict)

@dataclass
class Chapter:
    """章节"""
    id: str
    title: str
    description: str
    order: int
    resources: List[Resource] = field(default_factory=list)

@dataclass
class Course:
    """课程"""
    id: str
    title: str
    description: str
    instructor_id: str
    status: CourseStatus
    category: str
    tags: List[str] = field(default_factory=list)
    chapters: List[Chapter] = field(default_factory=list)
    prerequisites: List[str] = field(default_factory=list)  # 前置课程ID
    difficulty: str = "intermediate"  # beginner, intermediate, advanced
    estimated_hours: float = 0
    enrollment_count: int = 0
    rating: float = 0
    created_at: float = field(default_factory=time.time)
    updated_at: float = field(default_factory=time.time)

class CourseManager:
    """课程管理器"""
    
    def __init__(self):
        self.courses: Dict[str, Course] = {}
    
    def create_course(self, title: str, description: str, instructor_id: str,
                     category: str) -> Course:
        """创建课程"""
        course = Course(
            id=f"course_{int(time.time() * 1000)}",
            title=title,
            description=description,
            instructor_id=instructor_id,
            status=CourseStatus.DRAFT,
            category=category
        )
        
        self.courses[course.id] = course
        return course
    
    def add_chapter(self, course_id: str, title: str, description: str) -> Optional[Chapter]:
        """添加章节"""
        course = self.courses.get(course_id)
        if not course:
            return None
        
        order = len(course.chapters) + 1
        
        chapter = Chapter(
            id=f"chap_{int(time.time() * 1000)}",
            title=title,
            description=description,
            order=order
        )
        
        course.chapters.append(chapter)
        course.updated_at = time.time()
        
        return chapter
    
    def add_resource(self, course_id: str, chapter_id: str, resource: Resource) -> bool:
        """添加资源"""
        course = self.courses.get(course_id)
        if not course:
            return False
        
        for chapter in course.chapters:
            if chapter.id == chapter_id:
                chapter.resources.append(resource)
                course.updated_at = time.time()
                return True
        
        return False
    
    def publish_course(self, course_id: str) -> bool:
        """发布课程"""
        course = self.courses.get(course_id)
        if not course:
            return False
        
        # 验证课程完整性
        if not course.chapters:
            return False
        
        course.status = CourseStatus.PUBLISHED
        course.updated_at = time.time()
        
        return True
    
    def get_course(self, course_id: str) -> Optional[Course]:
        """获取课程"""
        return self.courses.get(course_id)
    
    def search_courses(self, query: str = None, category: str = None,
                      difficulty: str = None) -> List[Course]:
        """搜索课程"""
        results = list(self.courses.values())
        
        # 过滤已发布
        results = [c for c in results if c.status == CourseStatus.PUBLISHED]
        
        # 按条件过滤
        if category:
            results = [c for c in results if c.category == category]
        
        if difficulty:
            results = [c for c in results if c.difficulty == difficulty]
        
        if query:
            query_lower = query.lower()
            results = [
                c for c in results
                if query_lower in c.title.lower() or query_lower in c.description.lower()
            ]
        
        return results
    
    def get_course_structure(self, course_id: str) -> Dict:
        """获取课程结构"""
        course = self.courses.get(course_id)
        if not course:
            return {}
        
        total_resources = 0
        total_duration = 0
        
        for chapter in course.chapters:
            total_resources += len(chapter.resources)
            for resource in chapter.resources:
                total_duration += resource.duration
        
        return {
            "course_id": course.id,
            "title": course.title,
            "chapters": [
                {
                    "id": ch.id,
                    "title": ch.title,
                    "resource_count": len(ch.resources)
                }
                for ch in course.chapters
            ],
            "total_chapters": len(course.chapters),
            "total_resources": total_resources,
            "total_duration": total_duration
        }

# 使用示例
cm = CourseManager()

# 创建课程
course = cm.create_course(
    title="Python编程入门",
    description="从零开始学习Python编程",
    instructor_id="instructor_001",
    category="编程"
)

# 添加章节
chap1 = cm.add_chapter(course.id, "Python基础", "Python语言基础语法")
chap2 = cm.add_chapter(course.id, "数据结构", "Python数据结构详解")

# 添加资源
cm.add_resource(course.id, chap1.id, Resource(
    id="res_001",
    title="Python环境搭建",
    type=ResourceType.VIDEO,
    url="/videos/python_setup.mp4",
    duration=600
))

cm.add_resource(course.id, chap1.id, Resource(
    id="res_002",
    title="变量与数据类型",
    type=ResourceType.VIDEO,
    url="/videos/python_variables.mp4",
    duration=900
))

# 发布课程
cm.publish_course(course.id)

# 获取结构
structure = cm.get_course_structure(course.id)
print(f"课程结构: {structure}")

2.2 题库管理

from dataclasses import dataclass, field
from typing import Dict, List, Optional
from enum import Enum
import random

class QuestionType(Enum):
    """题目类型"""
    SINGLE_CHOICE = "single_choice"
    MULTIPLE_CHOICE = "multiple_choice"
    TRUE_FALSE = "true_false"
    FILL_BLANK = "fill_blank"
    SHORT_ANSWER = "short_answer"
    CODE = "code"

@dataclass
class Question:
    """题目"""
    id: str
    type: QuestionType
    content: str
    options: List[str] = field(default_factory=list)  # 选择题选项
    answer: str = ""  # 答案
    explanation: str = ""  # 解析
    difficulty: int = 1  # 1-5
    tags: List[str] = field(default_factory=list)
    points: int = 10

@dataclass
class Quiz:
    """测验"""
    id: str
    title: str
    course_id: str
    chapter_id: Optional[str] = None
    questions: List[Question] = field(default_factory=list)
    time_limit: int = 30  # 分钟
    passing_score: int = 60
    attempts_allowed: int = 3

class QuestionBank:
    """题库管理"""
    
    def __init__(self):
        self.questions: Dict[str, Question] = {}
        self.quizzes: Dict[str, Quiz] = {}
    
    def add_question(self, question: Question):
        """添加题目"""
        self.questions[question.id] = question
    
    def create_quiz(self, title: str, course_id: str, question_ids: List[str],
                   time_limit: int = 30) -> Quiz:
        """创建测验"""
        questions = [self.questions[qid] for qid in question_ids if qid in self.questions]
        
        quiz = Quiz(
            id=f"quiz_{int(time.time() * 1000)}",
            title=title,
            course_id=course_id,
            questions=questions,
            time_limit=time_limit
        )
        
        self.quizzes[quiz.id] = quiz
        return quiz
    
    def generate_quiz(self, course_id: str, tags: List[str] = None,
                     difficulty_range: tuple = (1, 5), count: int = 10) -> Quiz:
        """自动生成测验"""
        # 筛选题目
        candidates = []
        
        for question in self.questions.values():
            # 难度筛选
            if not (difficulty_range[0] <= question.difficulty <= difficulty_range[1]):
                continue
            
            # 标签筛选
            if tags and not any(tag in question.tags for tag in tags):
                continue
            
            candidates.append(question)
        
        # 随机选择
        selected = random.sample(candidates, min(count, len(candidates)))
        
        # 创建测验
        quiz = Quiz(
            id=f"quiz_{int(time.time() * 1000)}",
            title=f"自动生成测验 - {time.strftime('%Y%m%d')}",
            course_id=course_id,
            questions=selected
        )
        
        self.quizzes[quiz.id] = quiz
        return quiz
    
    def get_questions_by_tags(self, tags: List[str]) -> List[Question]:
        """按标签获取题目"""
        return [
            q for q in self.questions.values()
            if any(tag in q.tags for tag in tags)
        ]
    
    def get_statistics(self) -> Dict:
        """获取题库统计"""
        type_counts = {}
        difficulty_counts = {}
        
        for question in self.questions.values():
            type_counts[question.type.value] = type_counts.get(question.type.value, 0) + 1
            difficulty_counts[question.difficulty] = difficulty_counts.get(question.difficulty, 0) + 1
        
        return {
            "total_questions": len(self.questions),
            "total_quizzes": len(self.quizzes),
            "by_type": type_counts,
            "by_difficulty": difficulty_counts
        }

# 使用示例
qb = QuestionBank()

# 添加题目
qb.add_question(Question(
    id="q_001",
    type=QuestionType.SINGLE_CHOICE,
    content="Python中用于定义函数的关键字是?",
    options=["function", "def", "func", "define"],
    answer="def",
    explanation="Python使用def关键字定义函数",
    difficulty=1,
    tags=["Python", "基础"]
))

qb.add_question(Question(
    id="q_002",
    type=QuestionType.SINGLE_CHOICE,
    content="以下哪个不是Python的数据类型?",
    options=["list", "dict", "array", "tuple"],
    answer="array",
    explanation="Python内置没有array类型,需要导入array模块",
    difficulty=2,
    tags=["Python", "数据类型"]
))

# 创建测验
quiz = qb.create_quiz(
    title="Python基础测验",
    course_id=course.id,
    question_ids=["q_001", "q_002"],
    time_limit=15
)

print(f"测验: {quiz.title}, 题目数: {len(quiz.questions)}")

3. 学习路径模块

3.1 知识图谱构建

from typing import Dict, List, Set, Optional
from dataclasses import dataclass, field

@dataclass
class KnowledgeNode:
    """知识节点"""
    id: str
    name: str
    description: str
    prerequisites: List[str] = field(default_factory=list)  # 前置知识ID
    related_courses: List[str] = field(default_factory=list)  # 相关课程ID
    difficulty: int = 1
    estimated_hours: float = 1.0

class KnowledgeGraph:
    """知识图谱"""
    
    def __init__(self):
        self.nodes: Dict[str, KnowledgeNode] = {}
        self.edges: Dict[str, List[str]] = {}  # 依赖关系
    
    def add_node(self, node: KnowledgeNode):
        """添加知识节点"""
        self.nodes[node.id] = node
        
        # 构建边
        if node.id not in self.edges:
            self.edges[node.id] = []
        
        for prereq in node.prerequisites:
            if prereq not in self.edges:
                self.edges[prereq] = []
            self.edges[prereq].append(node.id)
    
    def get_learning_order(self, target_knowledge: str) -> List[str]:
        """获取学习顺序(拓扑排序)"""
        if target_knowledge not in self.nodes:
            return []
        
        # 收集所有前置知识
        visited = set()
        order = []
        
        def dfs(node_id: str):
            if node_id in visited:
                return
            visited.add(node_id)
            
            node = self.nodes.get(node_id)
            if node:
                for prereq in node.prerequisites:
                    dfs(prereq)
            
            order.append(node_id)
        
        dfs(target_knowledge)
        
        return order
    
    def get_next_knowledge(self, completed: List[str]) -> List[str]:
        """获取下一步可学习的知识"""
        completed_set = set(completed)
        available = []
        
        for node_id, node in self.nodes.items():
            if node_id in completed_set:
                continue
            
            # 检查前置是否都已完成
            if all(p in completed_set for p in node.prerequisites):
                available.append(node_id)
        
        return available
    
    def find_path(self, start: str, end: str) -> List[str]:
        """查找学习路径"""
        if start not in self.nodes or end not in self.nodes:
            return []
        
        # BFS查找最短路径
        from collections import deque
        
        queue = deque([(start, [start])])
        visited = {start}
        
        while queue:
            current, path = queue.popleft()
            
            if current == end:
                return path
            
            for neighbor in self.edges.get(current, []):
                if neighbor not in visited:
                    visited.add(neighbor)
                    queue.append((neighbor, path + [neighbor]))
        
        return []
    
    def get_knowledge_map(self) -> Dict:
        """获取知识图谱结构"""
        return {
            "nodes": [
                {
                    "id": node.id,
                    "name": node.name,
                    "difficulty": node.difficulty
                }
                for node in self.nodes.values()
            ],
            "edges": [
                {"source": source, "target": target}
                for source, targets in self.edges.items()
                for target in targets
            ]
        }

# 使用示例
kg = KnowledgeGraph()

# 构建知识图谱
kg.add_node(KnowledgeNode(
    id="python_basics",
    name="Python基础",
    description="Python语言基础语法",
    difficulty=1,
    estimated_hours=10
))

kg.add_node(KnowledgeNode(
    id="python_oop",
    name="Python面向对象",
    description="Python面向对象编程",
    prerequisites=["python_basics"],
    difficulty=2,
    estimated_hours=8
))

kg.add_node(KnowledgeNode(
    id="python_advanced",
    name="Python进阶",
    description="Python高级特性",
    prerequisites=["python_oop"],
    difficulty=3,
    estimated_hours=12
))

# 获取学习顺序
order = kg.get_learning_order("python_advanced")
print(f"学习顺序: {order}")

# 获取下一步
next_knowledge = kg.get_next_knowledge(["python_basics"])
print(f"下一步可学: {next_knowledge}")

3.2 个性化学习路径

from typing import Dict, List, Optional
from dataclasses import dataclass
from datetime import datetime

@dataclass
class LearningProfile:
    """学习画像"""
    user_id: str
    knowledge_level: Dict[str, int]  # 知识点 -> 掌握程度(1-5)
    learning_style: str  # visual, auditory, reading, kinesthetic
    preferred_duration: int  # 每次学习时长(分钟)
    goals: List[str]
    completed_courses: List[str]
    current_courses: List[str]

@dataclass
class LearningPath:
    """学习路径"""
    id: str
    user_id: str
    goal: str
    milestones: List[Dict]
    current_position: int
    estimated_completion: datetime
    created_at: float

class LearningPathGenerator:
    """学习路径生成器"""
    
    def __init__(self, knowledge_graph: KnowledgeGraph, course_manager: CourseManager):
        self.kg = knowledge_graph
        self.cm = course_manager
    
    def generate_path(self, profile: LearningProfile, goal: str) -> LearningPath:
        """生成个性化学习路径"""
        # 获取目标知识的学习顺序
        knowledge_order = self.kg.get_learning_order(goal)
        
        # 过滤已掌握的知识
        unlearned = [
            k for k in knowledge_order
            if profile.knowledge_level.get(k, 0) < 3
        ]
        
        # 为每个知识点匹配课程
        milestones = []
        
        for knowledge_id in unlearned:
            node = self.kg.nodes.get(knowledge_id)
            if not node:
                continue
            
            # 查找相关课程
            courses = self._find_courses(knowledge_id, profile)
            
            milestone = {
                "knowledge_id": knowledge_id,
                "knowledge_name": node.name,
                "difficulty": node.difficulty,
                "estimated_hours": node.estimated_hours,
                "recommended_courses": courses,
                "status": "pending"
            }
            
            milestones.append(milestone)
        
        # 计算预计完成时间
        total_hours = sum(m["estimated_hours"] for m in milestones)
        sessions_per_week = 5
        hours_per_session = profile.preferred_duration / 60
        weeks_needed = total_hours / (sessions_per_week * hours_per_session)
        
        estimated_completion = datetime.now() + timedelta(weeks=weeks_needed)
        
        return LearningPath(
            id=f"path_{int(time.time() * 1000)}",
            user_id=profile.user_id,
            goal=goal,
            milestones=milestones,
            current_position=0,
            estimated_completion=estimated_completion
        )
    
    def _find_courses(self, knowledge_id: str, profile: LearningProfile) -> List[Dict]:
        """查找适合的课程"""
        node = self.kg.nodes.get(knowledge_id)
        if not node:
            return []
        
        # 获取相关课程
        course_ids = node.related_courses
        courses = []
        
        for cid in course_ids:
            course = self.cm.get_course(cid)
            if course:
                # 检查是否适合用户水平
                if self._is_suitable(course, profile):
                    courses.append({
                        "id": course.id,
                        "title": course.title,
                        "difficulty": course.difficulty,
                        "duration": course.estimated_hours
                    })
        
        return courses
    
    def _is_suitable(self, course: Course, profile: LearningProfile) -> bool:
        """检查课程是否适合用户"""
        # 检查前置课程
        for prereq in course.prerequisites:
            if prereq not in profile.completed_courses:
                return False
        
        return True
    
    def update_progress(self, path: LearningPath, milestone_index: int, status: str):
        """更新学习进度"""
        if 0 <= milestone_index < len(path.milestones):
            path.milestones[milestone_index]["status"] = status
            
            if status == "completed":
                path.current_position = milestone_index + 1
    
    def get_next_milestone(self, path: LearningPath) -> Optional[Dict]:
        """获取下一个里程碑"""
        if path.current_position < len(path.milestones):
            return path.milestones[path.current_position]
        return None

# 使用示例
from datetime import timedelta

lpg = LearningPathGenerator(kg, cm)

# 创建学习画像
profile = LearningProfile(
    user_id="user_001",
    knowledge_level={"python_basics": 4},
    learning_style="visual",
    preferred_duration=30,
    goals=["掌握Python高级编程"],
    completed_courses=["course_python_basics"],
    current_courses=[]
)

# 生成学习路径
path = lpg.generate_path(profile, "python_advanced")
print(f"学习路径: {len(path.milestones)} 个里程碑")
print(f"预计完成: {path.estimated_completion.strftime('%Y-%m-%d')}")

4. 学习分析模块

4.1 学习行为追踪

from typing import Dict, List
from dataclasses import dataclass, field
from datetime import datetime
import time

@dataclass
class LearningSession:
    """学习会话"""
    id: str
    user_id: str
    course_id: str
    resource_id: str
    start_time: float
    end_time: float = 0
    duration: int = 0  # 秒
    progress: float = 0  # 0-100
    completed: bool = False
    notes: str = ""

@dataclass
class LearningRecord:
    """学习记录"""
    user_id: str
    course_id: str
    resource_progress: Dict[str, float] = field(default_factory=dict)
    total_time: int = 0
    last_access: float = 0
    completion_rate: float = 0

class LearningTracker:
    """学习追踪器"""
    
    def __init__(self):
        self.sessions: Dict[str, LearningSession] = {}
        self.records: Dict[str, LearningRecord] = {}
    
    def start_session(self, user_id: str, course_id: str, resource_id: str) -> LearningSession:
        """开始学习会话"""
        session = LearningSession(
            id=f"session_{int(time.time() * 1000)}",
            user_id=user_id,
            course_id=course_id,
            resource_id=resource_id,
            start_time=time.time()
        )
        
        self.sessions[session.id] = session
        return session
    
    def end_session(self, session_id: str, progress: float = 100, notes: str = ""):
        """结束学习会话"""
        session = self.sessions.get(session_id)
        if not session:
            return
        
        session.end_time = time.time()
        session.duration = int(session.end_time - session.start_time)
        session.progress = progress
        session.completed = progress >= 100
        session.notes = notes
        
        # 更新学习记录
        self._update_record(session)
    
    def _update_record(self, session: LearningSession):
        """更新学习记录"""
        record_key = f"{session.user_id}_{session.course_id}"
        
        if record_key not in self.records:
            self.records[record_key] = LearningRecord(
                user_id=session.user_id,
                course_id=session.course_id
            )
        
        record = self.records[record_key]
        record.resource_progress[session.resource_id] = session.progress
        record.total_time += session.duration
        record.last_access = time.time()
        
        # 计算完成率
        if record.resource_progress:
            record.completion_rate = sum(record.resource_progress.values()) / len(record.resource_progress)
    
    def get_user_stats(self, user_id: str) -> Dict:
        """获取用户学习统计"""
        user_records = [
            r for r in self.records.values()
            if r.user_id == user_id
        ]
        
        total_courses = len(user_records)
        total_time = sum(r.total_time for r in user_records)
        avg_completion = sum(r.completion_rate for r in user_records) / max(total_courses, 1)
        
        return {
            "total_courses": total_courses,
            "total_time_hours": total_time / 3600,
            "average_completion": avg_completion,
            "recent_activity": max(r.last_access for r in user_records) if user_records else 0
        }
    
    def get_course_stats(self, course_id: str) -> Dict:
        """获取课程学习统计"""
        course_records = [
            r for r in self.records.values()
            if r.course_id == course_id
        ]
        
        if not course_records:
            return {}
        
        total_learners = len(course_records)
        avg_completion = sum(r.completion_rate for r in course_records) / total_learners
        avg_time = sum(r.total_time for r in course_records) / total_learners
        
        return {
            "total_learners": total_learners,
            "average_completion": avg_completion,
            "average_time_hours": avg_time / 3600
        }
    
    def get_learning_heatmap(self, user_id: str, days: int = 30) -> Dict:
        """获取学习热力图数据"""
        user_sessions = [
            s for s in self.sessions.values()
            if s.user_id == user_id
        ]
        
        # 按日期统计
        daily_time = {}
        
        for session in user_sessions:
            date = datetime.fromtimestamp(session.start_time).strftime("%Y-%m-%d")
            daily_time[date] = daily_time.get(date, 0) + session.duration
        
        return daily_time

# 使用示例
tracker = LearningTracker()

# 开始学习
session = tracker.start_session("user_001", course.id, "res_001")

# 模拟学习过程
time.sleep(2)

# 结束学习
tracker.end_session(session.id, progress=80, notes="学习了环境搭建")

# 获取统计
stats = tracker.get_user_stats("user_001")
print(f"学习统计: {stats}")

4.2 学习效果评估

from typing import Dict, List
from dataclasses import dataclass
import math

@dataclass
class AssessmentResult:
    """评估结果"""
    user_id: str
    quiz_id: str
    score: float
    correct_count: int
    total_count: int
    time_spent: int
    weak_areas: List[str]
    strong_areas: List[str]

class LearningAssessment:
    """学习效果评估"""
    
    def __init__(self, question_bank: QuestionBank):
        self.qb = question_bank
        self.results: Dict[str, List[AssessmentResult]] = {}
    
    def evaluate_quiz(self, user_id: str, quiz_id: str, answers: Dict[str, str],
                     time_spent: int) -> AssessmentResult:
        """评估测验结果"""
        quiz = self.qb.quizzes.get(quiz_id)
        if not quiz:
            return None
        
        correct_count = 0
        tag_performance = {}
        
        for question in quiz.questions:
            user_answer = answers.get(question.id, "")
            is_correct = user_answer == question.answer
            
            if is_correct:
                correct_count += 1
            
            # 统计标签表现
            for tag in question.tags:
                if tag not in tag_performance:
                    tag_performance[tag] = {"correct": 0, "total": 0}
                
                tag_performance[tag]["total"] += 1
                if is_correct:
                    tag_performance[tag]["correct"] += 1
        
        # 计算分数
        score = correct_count / len(quiz.questions) * 100
        
        # 识别强弱项
        weak_areas = []
        strong_areas = []
        
        for tag, perf in tag_performance.items():
            accuracy = perf["correct"] / perf["total"]
            if accuracy < 0.6:
                weak_areas.append(tag)
            elif accuracy >= 0.8:
                strong_areas.append(tag)
        
        result = AssessmentResult(
            user_id=user_id,
            quiz_id=quiz_id,
            score=score,
            correct_count=correct_count,
            total_count=len(quiz.questions),
            time_spent=time_spent,
            weak_areas=weak_areas,
            strong_areas=strong_areas
        )
        
        # 保存结果
        if user_id not in self.results:
            self.results[user_id] = []
        self.results[user_id].append(result)
        
        return result
    
    def get_learning_progress(self, user_id: str, course_id: str) -> Dict:
        """获取学习进度"""
        user_results = self.results.get(user_id, [])
        
        # 筛选课程相关测验
        course_quizzes = [
            r for r in user_results
            if self._is_course_quiz(r.quiz_id, course_id)
        ]
        
        if not course_quizzes:
            return {"progress": 0, "mastery_level": "未开始"}
        
        avg_score = sum(r.score for r in course_quizzes) / len(course_quizzes)
        
        # 计算掌握程度
        if avg_score >= 90:
            mastery = "精通"
        elif avg_score >= 75:
            mastery = "熟练"
        elif avg_score >= 60:
            mastery = "掌握"
        else:
            mastery = "学习中"
        
        return {
            "quiz_count": len(course_quizzes),
            "average_score": avg_score,
            "mastery_level": mastery,
            "recent_score": course_quizzes[-1].score if course_quizzes else 0
        }
    
    def _is_course_quiz(self, quiz_id: str, course_id: str) -> bool:
        """检查测验是否属于课程"""
        quiz = self.qb.quizzes.get(quiz_id)
        return quiz and quiz.course_id == course_id
    
    def generate_study_plan(self, user_id: str, weak_areas: List[str]) -> List[Dict]:
        """生成学习建议"""
        recommendations = []
        
        for area in weak_areas:
            # 查找相关题目
            questions = self.qb.get_questions_by_tags([area])
            
            if questions:
                recommendations.append({
                    "area": area,
                    "recommended_practice": len(questions),
                    "priority": "high"
                })
        
        return sorted(recommendations, key=lambda x: x["priority"], reverse=True)

# 使用示例
assessment = LearningAssessment(qb)

# 评估测验
result = assessment.evaluate_quiz(
    user_id="user_001",
    quiz_id=quiz.id,
    answers={"q_001": "def", "q_002": "array"},
    time_spent=300
)

print(f"测验结果: {result.score:.1f}分")
print(f"弱项: {result.weak_areas}")
print(f"强项: {result.strong_areas}")

5. 智能问答模块

5.1 课程问答助手

from typing import Dict, List, Optional
from dataclasses import dataclass

@dataclass
class QAResponse:
    """问答响应"""
    question: str
    answer: str
    sources: List[str]
    confidence: float
    related_questions: List[str]

class CourseQAAssistant:
    """课程问答助手"""
    
    def __init__(self, course_manager: CourseManager):
        self.cm = course_manager
        self.knowledge_base: Dict[str, List[str]] = {}
    
    def index_course(self, course_id: str):
        """索引课程内容"""
        course = self.cm.get_course(course_id)
        if not course:
            return
        
        documents = []
        
        # 提取课程内容
        documents.append(f"课程名称: {course.title}")
        documents.append(f"课程描述: {course.description}")
        
        for chapter in course.chapters:
            documents.append(f"章节: {chapter.title} - {chapter.description}")
            
            for resource in chapter.resources:
                documents.append(f"资源: {resource.title} - {resource.description}")
        
        self.knowledge_base[course_id] = documents
    
    def answer(self, course_id: str, question: str) -> QAResponse:
        """回答问题"""
        # 使用OpenClaw的RAG能力
        # 简化实现
        
        documents = self.knowledge_base.get(course_id, [])
        
        if not documents:
            return QAResponse(
                question=question,
                answer="抱歉,我没有找到相关信息。",
                sources=[],
                confidence=0,
                related_questions=[]
            )
        
        # 构建上下文
        context = "\n".join(documents[:5])
        
        # 生成回答(实际应调用LLM)
        answer = f"根据课程内容,{question}的答案是..."
        
        # 生成相关问题
        related = [
            f"关于{question}还有哪些内容?",
            f"如何深入学习{question}?"
        ]
        
        return QAResponse(
            question=question,
            answer=answer,
            sources=[course_id],
            confidence=0.8,
            related_questions=related
        )
    
    def get_study_hints(self, course_id: str, question_id: str) -> str:
        """获取学习提示"""
        # 根据题目提供提示,而不是直接答案
        return "提示:请回顾课程第X章的内容..."

# 使用示例
qa = CourseQAAssistant(cm)
qa.index_course(course.id)

# 提问
response = qa.answer(course.id, "Python中如何定义函数?")
print(f"回答: {response.answer}")
print(f"置信度: {response.confidence}")

6. 最佳实践

6.1 平台设计原则

原则 说明 实践
个性化 因材施教 学习路径 + 推荐
互动性 即时反馈 智能问答 + 评估
可视化 进度透明 图表 + 报告
激励性 持续学习 成就系统

6.2 常见问题

问题 原因 解决方案
学习动力不足 缺乏激励 成就系统
答疑不及时 人工成本高 智能问答
效果难评估 指标单一 多维评估

7. 总结

本文通过完整的教育学习平台案例,展示了 OpenClaw 在教育科技场景的应用:

模块 核心功能 技术要点
课程管理 内容管理 结构化存储
学习路径 个性化学习 知识图谱
学习分析 多维分析 数据追踪
智能问答 即时答疑 RAG + LLM

参考资料


Logo

中国智能体开发者社区,聚焦智能体与大模型开发,提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动,促进经验交流与协作,助力开发者快速构建创新智能应用。

更多推荐