关注

OpenClaw 实战案例:教育学习平台构建

摘要

本文通过一个完整的教育学习平台案例,演示如何使用 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

参考资料


转载自 CSDN-专业IT技术社区

原文链接:https://blog.csdn.net/sinat_41617212/article/details/162704114

文章来源crawl

评论

赞0

评论列表

微信小程序
QQ小程序

关于作者

点赞数:0
关注数:0
粉丝:0
文章:0
关注标签:0
加入于:--