教育 Agent：个性化学习路径与自动批改

---

一、引言

1.1 钩子：传统教育的困境与AI的机遇

你是否曾经在课堂上感到无聊，因为老师讲的内容你早已掌握？或者相反，你是否曾经因为跟不上课程进度而感到沮丧，希望老师能放慢脚步，多给你一些解释？作为家长，你是否曾为孩子的学习成绩发愁，却不知道如何针对性地提供帮助？作为教师，你是否曾被堆积如山的作业批改压得喘不过气，感叹如果能有更多时间关注每个学生的个性化需求该有多好？

这些场景，相信很多人都不陌生。传统的教育模式，本质上是一种"标准化批量生产"模式：同样的教材、同样的进度、同样的评价标准，应用于一个个独特的个体。这种模式在工业时代为社会培养了大量标准化人才，但在知识经济和人工智能时代，它的局限性日益凸显。

一个令人震惊的数据是：据联合国教科文组织统计，全球有超过1.2亿儿童和青少年未能达到基本的阅读和数学水平，而与此同时，又有大量有天赋的学生因为教育内容不够有挑战性而失去学习兴趣。这就是我们面临的教育悖论——我们需要一种既能满足基础学习需求，又能激发每个学生最大潜能的教育方式。

好消息是，人工智能技术的快速发展正在为解决这一悖论提供新的可能。特别是近年来大型语言模型（LLMs）的突破，使得构建智能教育Agent成为现实。这些Agent可以像一位耐心、专业、不知疲倦的私人导师一样，为每个学生提供个性化的学习体验，同时还能自动完成繁琐的批改工作，让教师有更多时间关注教育中最核心的人文关怀部分。

1.2 定义问题/阐述背景：教育Agent究竟是什么？

在深入探讨之前，让我们先明确几个核心概念：

Agent（智能体）：在人工智能领域，Agent是指能够感知环境、做出决策并采取行动以实现特定目标的实体。一个好的Agent应该具备自主性（能够在没有人类干预的情况下运行）、反应性（能够感知并响应环境变化）、主动性（能够采取主动行动实现目标）和社交能力（能够与其他Agent或人类交互）。

教育Agent（Educational Agent）：顾名思义，教育Agent是专门为教育目的设计的智能体。它可以扮演多种角色：私人导师、学习伙伴、助教、评估专家等。教育Agent的核心价值在于能够提供个性化和自适应的教育体验。

个性化学习路径：指根据学生的学习风格、兴趣、能力水平、学习进度等个人特征，为其量身定制的学习计划和内容序列。传统教育中，教师很难为每个学生都设计这样的路径，但教育Agent可以通过数据分析和AI算法轻松实现这一点。

自动批改：指利用AI技术自动评估学生的作业、测验答案，并提供反馈的过程。这不仅包括客观题（如选择题）的批改，还包括主观题（如作文、简答题、编程题）的评估。

为什么教育Agent在今天变得如此重要？除了前面提到的传统教育模式的局限性，还有几个关键趋势推动着这一领域的发展：

1. 数据驱动决策的普及：教育领域正在积累越来越多的学生数据，教育Agent可以利用这些数据做出更智能的决策。
2. 大规模开放在线课程（MOOCs）的局限性暴露：虽然MOOCs实现了教育资源的广泛传播，但它们的完成率通常很低（研究显示不到10%），一个重要原因就是缺乏个性化指导和反馈。
3. 后疫情时代的教育新常态：COVID-19疫情加速了教育数字化转型，远程学习成为常态，这为教育Agent的应用提供了广阔的场景。
4. AI技术的成熟：特别是大语言模型的出现，使得教育Agent能够进行自然、流畅的对话，理解复杂的教育内容，提供高质量的反馈。

1.3 亮明观点/文章目标：你将从这篇文章中学到什么？

本文将带你深入了解教育Agent的世界，特别是如何构建能够提供个性化学习路径和自动批改功能的教育Agent。读完这篇文章，你将：

1. 理解教育Agent的核心概念和技术基础：我们会从基础讲起，确保即使你是AI教育领域的新手，也能跟上节奏。
2. 掌握教育Agent的设计原则和架构模式：你将了解一个好的教育Agent应该包含哪些组件，以及它们如何协同工作。
3. 获得实战经验：我们将通过一个完整的项目示例，手把手教你如何构建一个简单但功能完整的教育Agent，包括个性化学习路径规划和自动批改功能。
4. 了解前沿趋势和最佳实践：我们会探讨教育Agent领域的最新研究成果、常见陷阱和未来发展方向。

无论你是一名想要探索AI教育应用的软件工程师，一位希望改进教学方法的教育工作者，还是一位对教育科技充满热情的创业者，这篇文章都将为你提供有价值的见解和实用的指导。

让我们开始这段激动人心的旅程吧！

---

二、基础知识/背景铺垫

2.1 核心概念定义

在深入探讨如何构建教育Agent之前，让我们先夯实基础，明确一些关键概念。

2.1.1 智能体（Agent）的基本理论

智能体理论是人工智能的一个重要分支，它的核心思想是将AI系统建模为能够感知环境并做出决策的实体。让我们从数学角度对智能体进行形式化定义。

一个智能体可以被定义为一个函数：

$$A:P^{\ast }\to A$$

其中，$P^{\ast }$ 表示感知历史的集合（即智能体到目前为止接收到的所有感知输入序列），$A$ 表示智能体可以执行的动作集合。这个函数描述了智能体如何根据其过去的经验选择下一步行动。

更具体地说，在每个时间步 $t$：

1. 智能体接收环境的感知 $p_t\in P$
2. 智能体根据感知历史 $p_1,p_2,...,p_t$ 选择动作 $a_t\in A$
3. 环境根据动作 $a_t$ 转换到新的状态，并产生下一个感知 $p_{t+1}$

智能体的目标是最大化一个效用函数 $U$，该函数衡量了智能体在特定环境中的表现：

$$U:S^{\ast }\to \mathbb{R}$$

其中 $S^{\ast }$ 是环境状态序列的集合。

这种抽象框架非常强大，因为它可以用来建模几乎任何类型的AI系统，包括我们将要讨论的教育Agent。

2.1.2 个性化学习的理论基础

个性化学习并不是一个新概念，它的理论基础可以追溯到几个重要的教育心理学理论：

1. 多元智能理论（Multiple Intelligences Theory）：由霍华德·加德纳（Howard Gardner）提出，认为人类拥有多种不同类型的智能，如语言智能、逻辑-数学智能、空间智能、音乐智能、身体-动觉智能、人际智能、内省智能和自然观察智能。传统教育往往只重视前两种智能，而个性化学习则旨在承认和发展所有类型的智能。

2. 学习风格理论（Learning Styles Theory）：虽然该理论在学术界存在一些争议，但它仍然是个性化学习的重要理论基础之一。常见的学习风格分类包括：

   • 视觉型学习者：通过看图片、图表和文字学习效果最好
   • 听觉型学习者：通过听讲座、讨论和音频学习效果最好
   • 动觉型学习者：通过动手实践、运动和体验学习效果最好

3. 掌握学习理论（Mastery Learning Theory）：由本杰明·布鲁姆（Benjamin Bloom）提出，核心思想是"只要给予足够的时间和适当的教学，几乎所有学生都能掌握任何学习目标"。在掌握学习模式中，学生需要在进入下一主题前证明自己已经掌握了当前主题，这为自适应学习路径提供了理论依据。

4. 最近发展区理论（Zone of Proximal Development, ZPD）：由列夫·维果茨基（Lev Vygotsky）提出，指学习者独立解决问题的能力与在成人指导或与更有能力的同伴合作下解决问题的能力之间的差距。教育的关键是提供"脚手架"（scaffolding）帮助学生在最近发展区内学习，个性化学习正是要为每个学生找到合适的脚手架。

这些理论为教育Agent的设计提供了重要的指导原则。

2.1.3 自动批改的技术维度

自动批改是教育Agent的另一个核心功能。根据批改对象和技术难度，我们可以将自动批改分为几个层次：

1. 客观题批改：这是最简单的类型，包括选择题、判断题、填空题等。只要有标准答案，就可以通过简单的字符串匹配或模式匹配实现自动批改。

2. 编程题批改：这类批改需要检查代码的正确性、效率、风格等。常用技术包括：

   • 单元测试：运行预定义的测试用例，检查代码输出是否符合预期
   • 代码静态分析：检查代码风格、潜在bug、安全漏洞等
   • 代码相似度检测：防止抄袭

3. 数学题批改：不仅要检查最终答案，还要检查解题过程。这需要理解数学表达式和推理步骤。

4. 自然语言主观题批改：这是最具挑战性的类型，包括作文、简答题、论述题等。它需要AI系统具备深度语言理解能力，能够评估内容质量、论点逻辑性、语言表达等多个维度。

现代大语言模型的出现，极大地提升了自动批改的能力，特别是在主观题批改方面。

2.2 相关工具/技术概览

构建一个教育Agent需要多种技术和工具的协同工作。让我们对主要的技术栈进行一个概览。

2.2.1 大型语言模型（LLMs）

大型语言模型是现代教育Agent的核心引擎，它们为Agent提供了语言理解和生成能力。以下是一些主流的LLMs：

模型名称	开发商	特点	教育应用场景
GPT-4/GPT-3.5	OpenAI	强大的理解和生成能力，支持多模态	对话式辅导、作文批改、内容生成
Claude 2	Anthropic	长文本处理能力强，安全对齐好	长文本反馈、复杂推理指导
PaLM 2	Google	多语言能力强，与Google生态集成好	多语言教育，STEM领域辅导
Llama 2	Meta	开源，可自定义部署	需要数据隐私保护的教育场景
CodeLlama	Meta	专门为代码设计的LLM	编程教育、代码批改

选择合适的LLM需要考虑多种因素，包括能力、成本、速度、隐私保护、定制化需求等。在我们后面的实战项目中，我们将使用OpenAI的API，但你可以根据需要替换为其他模型。

2.2.2 向量数据库（Vector Databases）

教育Agent需要处理和检索大量的教育内容，向量数据库是实现这一功能的关键技术。向量数据库可以存储文本、图像等内容的向量表示（embeddings），并支持语义相似度检索。

主流的向量数据库包括：

• Pinecone：托管式向量数据库，易于使用
• Weaviate：开源，支持多种数据类型和模块化
• Chroma：轻量级开源向量数据库，适合原型开发
• FAISS：Facebook AI开源的向量检索库，性能优异
• Milvus：开源企业级向量数据库

在教育场景中，向量数据库可以用于：

• 个性化内容推荐：找到与学生当前学习状态最匹配的学习材料
• 相似问题检索：找到与学生提问相似的已解答问题
• 知识图谱构建：组织和关联教育内容

2.2.3 学习分析（Learning Analytics）工具

学习分析是指测量、收集、分析和报告关于学习者及其环境的数据，以理解和优化学习。常用的学习分析工具和技术包括：

1. xAPI（Experience API）：也称为Tin Can API，是一种用于收集学习体验数据的标准，比传统的SCORM更灵活。
2. 学习记录存储（LRS, Learning Record Store）：存储xAPI数据的系统。
3. 仪表盘工具：如Tableau、Power BI、Grafana等，用于可视化学习数据。
4. 预测分析模型：使用机器学习预测学生的学习结果，如辍学风险、考试成绩等。

2.2.4 其他相关技术

除了上述核心技术，构建教育Agent还可能需要：

1. 知识图谱：用于表示和组织领域知识，如ConceptNet、Wikidata等。
2. 语音识别和合成：如Whisper（OpenAI）、ElevenLabs等，用于提供语音交互界面。
3. 教育内容标准：如Common Core（美国）、IB（国际文凭）等，确保生成的教育内容符合标准。
4. 开发框架：如LangChain、LlamaIndex等，这些框架提供了构建AI应用的组件和工具链，可以大大加速开发过程。

在接下来的实战项目中，我们将使用LangChain，因为它提供了丰富的组件来处理对话、记忆、检索等任务，非常适合构建教育Agent。

---

三、核心内容/实战演练

在这一部分，我们将进入激动人心的实战环节。我们将从零开始构建一个名为"EduAgent"的教育Agent原型，它具备两个核心功能：个性化学习路径规划和自动批改。

3.1 项目介绍与环境准备

3.1.1 项目概述

我们的EduAgent将是一个基于Web的应用程序，它可以：

1. 评估学生水平：通过简短的对话或测验，了解学生的当前知识水平和学习风格。
2. 制定个性化学习路径：根据学生的水平和目标，生成定制化的学习计划。
3. 提供互动学习体验：作为虚拟导师，回答学生问题，解释概念，提供例子。
4. 自动批改作业：评估学生的答案（包括客观题和主观题），提供详细反馈。
5. 调整学习路径：根据学生的学习进度和表现，动态调整学习计划。

为了使项目既实用又易于理解，我们将专注于一个特定的学科领域——Python编程入门。当然，你可以轻松将其扩展到其他学科。

3.1.2 技术栈选择

对于这个项目，我们将使用以下技术栈：

• 后端：Python + FastAPI（轻量级高性能Web框架）
• 前端：React + TypeScript（现代UI框架）
• LLM集成：OpenAI API + LangChain
• 向量存储：Chroma（轻量级，便于开发）
• 数据库：SQLite（轻量级，无需额外配置）
• 其他工具：Pydantic（数据验证）、Uvicorn（ASGI服务器）

这种组合在保持项目简单的同时，也提供了足够的灵活性和可扩展性。

3.1.3 环境搭建

让我们开始搭建开发环境。首先，确保你已经安装了Python 3.9或更高版本。

步骤1：创建项目目录和虚拟环境

mkdir eduagent
cd eduagent
python -m venv venv

激活虚拟环境：

• Windows: venv\Scripts\activate
• macOS/Linux: source venv/bin/activate

步骤2：安装依赖

创建一个requirements.txt文件：

fastapi==0.104.1
uvicorn[standard]==0.24.0
python-multipart==0.0.6
sqlalchemy==2.0.23
pydantic==2.5.0
pydantic-settings==2.1.0
langchain==0.0.335
langchain-openai==0.0.2
chromadb==0.4.18
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-dotenv==1.0.0

安装依赖：

pip install -r requirements.txt

步骤3：设置OpenAI API密钥

创建一个.env文件：

OPENAI_API_KEY=your-api-key-here
DATABASE_URL=sqlite:///./eduagent.db
SECRET_KEY=your-secret-key-here-change-in-production
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30

将your-api-key-here替换为你的OpenAI API密钥。如果你还没有，可以在OpenAI官网注册并获取。

同时，生成一个安全的SECRET_KEY用于JWT令牌：

python -c "import secrets; print(secrets.token_hex(32))"

将输出的字符串填入.env文件。

步骤4：创建基本项目结构

让我们创建项目的基本目录结构：

eduagent/
├── backend/
│   ├── __init__.py
│   ├── main.py
│   ├── config.py
│   ├── database.py
│   ├── models.py
│   ├── schemas.py
│   ├── crud.py
│   ├── auth.py
│   ├── agent/
│   │   ├── __init__.py
│   │   ├── tutor.py
│   │   ├── learning_path.py
│   │   └── grader.py
│   └── vectorstore/
│       ├── __init__.py
│       └── content.py
├── frontend/ (我们将稍后创建)
├── .env
└── requirements.txt

好了，我们的环境已经准备就绪。接下来，让我们开始设计和实现系统的各个组件。

3.2 系统设计

3.2.1 系统架构设计

让我们首先从宏观角度看一下EduAgent的系统架构：

这个架构图展示了EduAgent的主要组件及其交互关系。让我们简要解释各层的职责：

1. 用户界面层：负责与用户交互，包括Web界面和移动界面。
2. API网关层：处理所有API请求，负责身份认证和路由。
3. 应用服务层：实现核心业务逻辑，如用户管理、学习管理、评估管理等。
4. Agent核心层：包含教育Agent的核心功能组件，这是我们项目的重点。
5. 基础设施层：提供数据存储、LLM访问等基础服务。

3.2.2 核心概念模型

在开始编码之前，让我们先设计系统的核心概念模型。使用实体关系图（ER图）来可视化这些概念及其关系：

这个ER图定义了我们系统中的主要实体及其关系。特别值得注意的是：

1. USER（用户）：存储用户基本信息和学习档案。learning_profile字段是一个JSON字段，用于存储用户的学习风格、偏好、历史表现等信息。

2. LEARNING_PATH（学习路径）：代表一个用户在特定学科上的个性化学习计划。

3. LEARNING_OBJECTIVE（学习目标）：学习路径中的具体目标，有顺序、状态和掌握程度。

4. ASSESSMENT（评估）：用于测量学生对学习目标的掌握程度，可以是预评估、形成性评估或总结性评估。

5. LEARNING_SESSION（学习会话）和MESSAGE（消息）：记录用户与Agent的交互历史。

这个数据模型为我们实现个性化学习路径和自动批改功能提供了基础。

3.2.3 系统接口设计

接下来，让我们设计系统的API接口。我们将使用RESTful API设计原则。

认证接口：

• POST /api/auth/register - 用户注册
• POST /api/auth/login - 用户登录
• GET /api/auth/me - 获取当前用户信息

用户接口：

• GET /api/users/profile - 获取用户学习档案
• PUT /api/users/profile - 更新用户学习档案

学习路径接口：

• POST /api/learning-paths - 创建新的学习路径
• GET /api/learning-paths - 获取用户的所有学习路径
• GET /api/learning-paths/{path_id} - 获取特定学习路径详情
• PUT /api/learning-paths/{path_id} - 更新学习路径

学习目标接口：

• GET /api/learning-paths/{path_id}/objectives - 获取学习路径的所有目标
• GET /api/learning-objectives/{objective_id} - 获取特定学习目标
• POST /api/learning-objectives/{objective_id}/start - 开始学习目标

评估接口：

• POST /api/assessments - 创建新评估
• GET /api/assessments/{assessment_id} - 获取评估详情
• POST /api/assessments/{assessment_id}/submit - 提交评估答案
• GET /api/assessments/{assessment_id}/feedback - 获取评估反馈

学习会话接口：

• POST /api/learning-sessions - 创建新学习会话
• POST /api/learning-sessions/{session_id}/messages - 发送消息
• GET /api/learning-sessions/{session_id}/messages - 获取会话消息

在后面的代码实现中，我们将详细实现这些接口。

3.3 核心实现

现在，让我们开始编写代码。我们将从基础设施开始，逐步构建到Agent核心功能。

3.3.1 基础设置

首先，让我们实现一些基础模块，包括配置、数据库连接等。

config.py - 管理应用配置：

from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    openai_api_key: str
    database_url: str = "sqlite:///./eduagent.db"
    secret_key: str
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30
    
    class Config:
        env_file = ".env"

@lru_cache()
def get_settings():
    return Settings()

database.py - 设置数据库连接：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

from .config import get_settings

settings = get_settings()

engine = create_engine(
    settings.database_url,
    connect_args={"check_same_thread": False}  # 仅SQLite需要
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

# 依赖项
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

models.py - 定义数据库模型：

import uuid
from datetime import datetime
from sqlalchemy import Column, String, Text, Integer, Float, DateTime, ForeignKey, JSON
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
from sqlalchemy.ext.mutable import MutableDict

from .database import Base

# 使用兼容的UUID类型
class UUIDColumn(Column):
    def __init__(self, *args, **kwargs):
        super().__init__(UUID(as_uuid=True), *args, **kwargs)

class User(Base):
    __tablename__ = "users"
    
    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    learning_profile = Column(MutableDict.as_mutable(JSON), default=dict)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    learning_paths = relationship("LearningPath", back_populates="user")
    assessments = relationship("Assessment", back_populates="user")
    learning_sessions = relationship("LearningSession", back_populates="user")
    messages = relationship("Message", back_populates="user")

class LearningPath(Base):
    __tablename__ = "learning_paths"
    
    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    user_id = Column(String, ForeignKey("users.id"))
    subject = Column(String, index=True)
    objectives = Column(MutableDict.as_mutable(JSON), default=list)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    user = relationship("User", back_populates="learning_paths")
    learning_objectives = relationship("LearningObjective", back_populates="learning_path")

class LearningObjective(Base):
    __tablename__ = "learning_objectives"
    
    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    learning_path_id = Column(String, ForeignKey("learning_paths.id"))
    title = Column(String)
    description = Column(Text)
    order = Column(Integer)
    status = Column(String, default="not_started")  # not_started, in_progress, completed
    mastery_level = Column(Float, default=0.0)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    learning_path = relationship("LearningPath", back_populates="learning_objectives")
    learning_contents = relationship("LearningContent", back_populates="learning_objective")
    assessments = relationship("Assessment", back_populates="learning_objective")
    learning_sessions = relationship("LearningSession", back_populates="learning_objective")

class LearningContent(Base):
    __tablename__ = "learning_contents"
    
    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    learning_objective_id = Column(String, ForeignKey("learning_objectives.id"))
    title = Column(String)
    content_type = Column(String)  # text, video, link, etc.
    content_url = Column(String, nullable=True)
    content_text = Column(Text, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    learning_objective = relationship("LearningObjective", back_populates="learning_contents")

class Assessment(Base):
    __tablename__ = "assessments"
    
    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    user_id = Column(String, ForeignKey("users.id"))
    learning_objective_id = Column(String, ForeignKey("learning_objectives.id"), nullable=True)
    assessment_type = Column(String)  # pre, formative, summative
    questions = Column(MutableDict.as_mutable(JSON), default=list)
    answers = Column(MutableDict.as_mutable(JSON), nullable=True)
    score = Column(Float, nullable=True)
    feedback = Column(Text, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    completed_at = Column(DateTime, nullable=True)
    
    user = relationship("User", back_populates="assessments")
    learning_objective = relationship("LearningObjective", back_populates="assessments")
    learning_sessions = relationship("LearningSession", back_populates="assessment")

class LearningSession(Base):
    __tablename__ = "learning_sessions"
    
    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    user_id = Column(String, ForeignKey("users.id"))
    learning_objective_id = Column(String, ForeignKey("learning_objectives.id"), nullable=True)
    assessment_id = Column(String, ForeignKey("assessments.id"), nullable=True)
    started_at = Column(DateTime, default=datetime.utcnow)
    ended_at = Column(DateTime, nullable=True)
    
    user = relationship("User", back_populates="learning_sessions")
    learning_objective = relationship("LearningObjective", back_populates="learning_sessions")
    assessment = relationship("Assessment", back_populates="learning_sessions")
    messages = relationship("Message", back_populates="learning_session")

class Message(Base):
    __tablename__ = "messages"
    
    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    learning_session_id = Column(String, ForeignKey("learning_sessions.id"))
    user_id = Column(String, ForeignKey("users.id"), nullable=True)  # Null means it's from the agent
    role = Column(String)  # user, assistant, system
    content = Column(Text)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    learning_session = relationship("LearningSession", back_populates="messages")
    user = relationship("User", back_populates="messages")

schemas.py - 定义Pydantic模型用于API请求和响应：

import uuid
from datetime import datetime
from typing import List, Optional, Dict, Any
from pydantic import BaseModel, Field, EmailStr

# 用户相关模型
class UserBase(BaseModel):
    username: str
    email: EmailStr

class UserCreate(UserBase):
    password: str

class UserLogin(BaseModel):
    username: str
    password: str

class User(UserBase):
    id: str
    learning_profile: Dict[str, Any] = Field(default_factory=dict)
    created_at: datetime
    updated_at: datetime
    
    class Config:
        from_attributes = True

class UserUpdate(BaseModel):
    learning_profile: Optional[Dict[str, Any]] = None

# 令牌相关模型
class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: Optional[str] = None

# 学习路径相关模型
class LearningPathBase(BaseModel):
    subject: str

class LearningPathCreate(LearningPathBase):
    initial_assessment: Optional[bool] = True

class LearningPath(LearningPathBase):
    id: str
    user_id: str
    objectives: List[Dict[str, Any]]
    created_at: datetime
    updated_at: datetime
    
    class Config:
        from_attributes = True

# 学习目标相关模型
class LearningObjectiveBase(BaseModel):
    title: str
    description: Optional[str] = None

class LearningObjectiveCreate(LearningObjectiveBase):
    order: int

class LearningObjective(LearningObjectiveBase):
    id: str
    learning_path_id: str
    order: int
    status: str
    mastery_level: float
    created_at: datetime
    updated_at: datetime
    
    class Config:
        from_attributes = True

# 评估相关模型
class Question(BaseModel):
    id: str
    type: str  # multiple_choice, short_answer, essay, coding
    question: str
    options: Optional[List[str]] = None
    correct_answer: Optional[str] = None
    points: float = 1.0

class AssessmentBase(BaseModel):
    assessment_type: str  # pre, formative, summative
    learning_objective_id: Optional[str] = None

class AssessmentCreate(AssessmentBase):
    questions: Optional[List[Question]] = None

class AssessmentAnswer(BaseModel):
    question_id: str
    answer: str

class AssessmentSubmit(BaseModel):
    answers: List[AssessmentAnswer]

class Assessment(AssessmentBase):
    id: str
    user_id: str
    questions: List[Dict[str, Any]]
    answers: Optional[List[Dict[str, Any]]] = None
    score: Optional[float] = None
    feedback: Optional[str] = None
    created_at: datetime
    completed_at: Optional[datetime] = None
    
    class Config:
        from_attributes = True

# 学习会话相关模型
class LearningSessionBase(BaseModel):
    learning_objective_id: Optional[str] = None
    assessment_id: Optional[str] = None

class LearningSessionCreate(LearningSessionBase):
    pass

class LearningSession(LearningSessionBase):
    id: str
    user_id: str
    started_at: datetime
    ended_at: Optional[datetime] = None
    
    class Config:
        from_attributes = True

# 消息相关模型
class MessageBase(BaseModel):
    content: str

class MessageCreate(MessageBase):
    pass

class Message(MessageBase):
    id: str
    learning_session_id: str
    user_id: Optional[str] = None
    role: str
    created_at: datetime
    
    class Config:
        from_attributes = True

auth.py - 实现身份认证：

from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session

from .config import get_settings
from .database import get_db
from .models import User
from .schemas import TokenData

settings = get_settings()

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/auth/login")

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def get_user(db: Session, username: str) -> Optional[User]:
    return db.query(User).filter(User.username == username).first()

def authenticate_user(db: Session, username: str, password: str) -> Optional[User]:
    user = get_user(db, username)
    if not user:
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.secret_key, algorithm=settings.algorithm)
    return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = get_user(db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

好的，我们已经搭建了基础结构。接下来，让我们实现CRUD操作和API端点，然后是Agent的核心功能。

crud.py - 实现数据库操作：

from typing import List, Optional, Dict, Any
from datetime import datetime
from sqlalchemy.orm import Session

from .models import (
    User, LearningPath, LearningObjective, LearningContent,
    Assessment, LearningSession, Message
)
from .schemas import (
    UserCreate, LearningPathCreate, LearningObjectiveCreate,
    AssessmentCreate, AssessmentSubmit, LearningSessionCreate, MessageCreate
)
from .auth import get_password_hash

# 用户操作
def create_user(db: Session, user: UserCreate) -> User:
    hashed_password = get_password_hash(user.password)
    db_user = User(
        username=user.username,
        email=user.email,
        hashed_password=hashed_password,
        learning_profile={}
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def update_user_profile(db: Session, user_id: str, profile: Dict[str, Any]) -> User:
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user:
        db_user.learning_profile = profile
        db_user.updated_at = datetime.utcnow()
        db.commit()
        db.refresh(db_user)
    return db_user

# 学习路径操作
def create_learning_path(db: Session, user_id: str, path: LearningPathCreate) -> LearningPath:
    db_path = LearningPath(
        user_id=user_id,
        subject=path.subject,
        objectives=[]
    )
    db.add(db_path)
    db.commit()
    db.refresh(db_path)
    return db_path

def get_learning_paths(db: Session, user_id: str) -> List[LearningPath]:
    return db.query(LearningPath).filter(LearningPath.user_id == user_id).all()

def get_learning_path(db: Session, path_id: str) -> Optional[LearningPath]:
    return db.query(LearningPath).filter(LearningPath.id == path_id).first()

def update_learning_path_objectives(db: Session, path_id: str, objectives: List[Dict[str, Any]]) -> LearningPath:
    db_path = db.query(LearningPath).filter(LearningPath.id == path_id).first()
    if db_path:
        db_path.objectives = objectives
        db_path.updated_at = datetime.utcnow()
        db.commit()
        db.refresh(db_path)
    return db_path

# 学习目标操作
def create_learning_objective(db: Session, path_id: str, objective: LearningObjectiveCreate) -> LearningObjective:
    db_objective = LearningObjective(
        learning_path_id=path_id,
        title=objective.title,
        description=objective.description,
        order=objective.order
    )
    db.add(db_objective)
    db.commit()
    db.refresh(db_objective)
    return db_objective

def get_learning_objectives(db: Session, path_id: str) -> List[LearningObjective]:
    return db.query(LearningObjective).filter(LearningObjective.learning_path_id == path_id).order_by(LearningObjective.order).all()

def get_learning_objective(db: Session, objective_id: str) -> Optional[LearningObjective]:
    return db.query(LearningObjective).filter(LearningObjective.id == objective_id).first()

def update_learning_objective_status(db: Session, objective_id: str, status: str, mastery_level: Optional[float] = None) -> LearningObjective:
    db_objective = db.query(LearningObjective).filter(LearningObjective.id == objective_id).first()
    if db_objective:
        db_objective.status = status
        if mastery_level is not None:
            db_objective.mastery_level = mastery_level
        db_objective.updated_at = datetime.utcnow()
        db.commit()
        db.refresh(db_objective)
    return db_objective

# 评估操作
def create_assessment(db: Session, user_id: str, assessment: AssessmentCreate) -> Assessment:
    db_assessment = Assessment(
        user_id=user_id,
        learning_objective_id=assessment.learning_objective_id,
        assessment_type=assessment.assessment_type,
        questions=[q.model_dump() for q in assessment.questions] if assessment.questions else []
    )
    db.add(db_assessment)
    db.commit()
    db.refresh(db_assessment)
    return db_assessment

def get_assessment(db: Session, assessment_id: str) -> Optional[Assessment]:
    return db.query(Assessment).filter(Assessment.id == assessment_id).first()

def submit_assessment(db: Session, assessment_id: str, submission: AssessmentSubmit) -> Assessment:
    db_assessment = db.query(Assessment).filter(Assessment.id == assessment_id).first()
    if db_assessment:
        db_assessment.answers = [ans.model_dump() for ans in submission.answers]
        db_assessment.completed_at = datetime.utcnow()
        db.commit()
        db.refresh(db_assessment)
    return db_assessment

def update_assessment_result(db: Session, assessment_id: str, score: float, feedback: str) -> Assessment:
    db_assessment = db.query(Assessment).filter(Assessment.id == assessment_id).first()
    if db_assessment:
        db_assessment.score = score
        db_assessment.feedback = feedback
        db.commit()
        db.refresh(db_assessment)
    return db_assessment

# 学习会话操作
def create_learning_session(db: Session, user_id: str, session: LearningSessionCreate) -> LearningSession:
    db_session = LearningSession(
        user_id=user_id,
        learning_objective_id=session.learning_objective_id,
        assessment_id=session.assessment_id
    )
    db.add(db_session)
    db.commit