简介这是一套基于Python的实时课程教学数据推荐与个性化智能问答系统源码主要面向教育技术开发者、课程平台运营者以及需要搭建智慧教学辅助工具的研究者。系统通过采集与分析课程教学数据实现学习资源个性化推荐、课程内容智能管理及自然语言问答可有效提升教学互动与资源利用效率。资源包为ZIP格式共54个文件大小约108KB其中包含40个Python源文件覆盖数据处理、推荐算法、问答服务、用户认证与后台管理等核心模块另有8个XML配置、1个iml项目文件、文本说明及SQLite数据库便于直接运行和二次开发。源码按课程管理、服务层、工具模块清晰分层并集成Celery异步任务、JWT认证、文本预处理等实用技术适合用来学习Django框架、推荐系统与智能问答的实现思路。目前已有344人学习下载可作为毕业设计、课设项目或教育产品原型的参考基础。1. 基于Python的实时课程教学数据推荐与个性化问答先把链路画清只单独做推荐或单独做问答都不难难的是让两者共享同一份实时教学数据。学生刚提交一道习题推荐系统需要立刻把他错题知识点附近的课程推到列表前面同一秒智能问答也要基于“这道题暴露出来的知识缺口”重新组织答案而不是给一段通用解释。要支撑这种联动源码不能只包含几个训练好的模型文件而需要一条从行为采集、特征引擎、向量召回、RAG检索到异步推理的完整管道。下面直接按这条管道拆解用Python代码把每一步落到可运行程度讨论范围也限定在在线系统真正需要的实现上而不是离线实验流程。2. 实时教学数据接入与特征工程行为日志怎么变成在线特征“实时”在不同系统里的定义差别很大。课程表更新可以容忍分钟级延迟但“学生刚看完视频点下一课”的推荐延迟超过两三秒就会被用户感知。因此第一阶段要把实时定义在秒级从课程视频播放进度、习题提交、课件翻阅这些行为事件产生到特征缓存更新完成端到端耗时不超过5秒。常见做法是行为数据不进数仓先进入消息队列由Python消费者边消费边更新特征缓存把“写过离线特征表”的思路改成“在线状态持续更新”。2.1 第一时间要收集哪些行为事件教学平台里最常见的事件类型分四类内容浏览、视频播放、习题作答、检索提问。每类事件都至少携带四个公共字段学生ID、课程ID、时间戳和事件类型码。下表是一个可落地的字段约定。字段示例说明user_idu_10234学生标识item_idc_lesson_045课程、视频或习题标识event_typevideo_play_doneclick、play_progress、submit_answer、ask_queryts1721024000000毫秒时间戳extraJSON字符串逗留时长、习题得分、知识点ID等扩展信息事件设计越简单越好。extra里建议只放原始字段不要在采集端做聚合聚合放到消费者里做否则事件类型会膨胀到无法管理。视频播放事件尤其需要记录“播放到第几秒”和“总时长”答题事件需要记录“得分”和“关联知识点ID”这两类字段是后来计算知识掌握度和课程相似度的最直接素材。2.2 用 Redis Stream 接收行为事件消费者组保证不丢不重如果团队已经有Kafka继续用Kafka没问题但新项目从零起推荐系统我更倾向先用Redis Stream。它没有额外依赖一个Redis实例就能做消息队列Python接入也简单。行为事件通过SDK写入Stream推荐服务用消费者组读取处理完再确认。import redis import json r redis.Redis(host127.0.0.1, port6379, decode_responsesTrue) stream_key behavior_stream group_name course_reco_group # 首次创建消费者组exists参数避免重复创建报错 try: r.xgroup_create(stream_key, group_name, id0-0, mkstreamTrue) except redis.ResponseError: pass def consume_behavior(): consumer_name python_consumer_1 while True: # 阻塞读最多500ms每次最多取20条 entries r.xreadgroup( group_name, consumer_name, {stream_key: }, count20, block500 ) for stream, messages in entries: for msg_id, payload in messages: event json.loads(payload[event]) update_online_feature(event) # 处理成功后ACK防止重启后重复消费 r.xack(stream_key, group_name, msg_id) def update_online_feature(event): # 具体特征更新逻辑见2.3节 pass这段代码把消费逻辑封装在循环里。xreadgroup的第一个表示只读新消息count20是批量大小block500让消费者在没有消息时最多阻塞半秒。xack负责向Redis确认消息已经处理这样消费者宕机重启后未确认消息会重新被其他消费者拿到。需要重点注意的是mkstreamTrue只在首次创建流时生效第二次调用幂等包住即可。2.3 从原始事件算特征时间衰减与滑动窗口原始事件进入消费者后下一步是计算在线特征。在线特征通常分两类计数特征和时间衰减特征。课程点击数、答题次数属于计数特征视频停留占比、连续学习天数属于强度特征。时间衰减特征比简单计数更适合教学场景因为它能表达“三个月前学过”和“昨天刚学过”的区别。import time import math FEATURE_KEY student_course_feature def update_online_feature(event): user_id event[user_id] item_id event[item_id] event_ts event[ts] / 1000 # 毫秒转秒 now time.time() # 用时间衰减因子平滑历史计数 last_ts float(r.hget(FEATURE_KEY, f{user_id}:{item_id}) or 0) if last_ts 0: decay math.exp(- (now - event_ts) / (7 * 86400)) count float(r.hget(FEATURE_KEY, f{user_id}:{item_id}:count) or 0) count count * decay 1 r.hset(FEATURE_KEY, f{user_id}:{item_id}:count, count) else: r.hset(FEATURE_KEY, f{user_id}:{item_id}:count, 1) # 同时更新最近一次学习时间 r.hset(FEATURE_KEY, f{user_id}:{item_id}:last_ts, event_ts)这里把计数和最近时间都放在同一个Hash里key为“学生:课程”value为计数字段和时间戳字段。衰减因子分母7 * 86400表示半衰期是7天也就是7天前的行为权重约为e^-1约等于0.37。如果想更灵敏改成3 * 86400如果想保留更长周期记忆改成14 * 86400。这个参数需要结合课程学习周期反复调不是越大越好。2.4 特征缓存的结构与 TTL 设计在线特征不要写进关系型数据库否则高并发查询下会拖垮服务。Redis Hash天然适合存储“学生课程”粒度的特征但要注意TTL不能直接作用到Hash里的单个字段只能对整个Hash设置TTL。更好的做法是额外维护一个用户维度TTL键。键名类型过期时间含义student_course_featureHash15天学生与课程的交互计数、时间student_embeddingHash7天学生实时兴趣向量推荐用knowledge_point_scoreHash24小时知识点的掌握分问答用推荐系统关注最近15天的行为就够了问答系统则更适合24小时粒度因为一道错题必须尽快反映到回答里。TTL不是随便设的学习行为有周期TTL设得太短会让周末不学习的学生周一回来冷启动设得太长又会放大过期行为的影响。生产环境里第15天没有新行为的用户建议直接走冷启动分支。3. 实时个性化推荐从召回、粗排到重排的 Python 实现推荐系统的实时性主要体现在候选集更新上。离线推荐可以花一晚上更新矩阵分解模型在线推荐则要在几十毫秒内返回结果。要做秒级更新召回层必须能增量读取特征缓存而不是每次重建全量索引。3.1 召回层为什么选向量检索而不是纯 SQL如果课程数量只有几百直接SQL过滤加排序就够了。但课程平台上内容动辄上万纯SQL无法表达语义相似和兴趣演化。传统协同过滤也有同样的问题矩阵分解需要定期重新训练新课程和新学生进入时拿不到向量。这里是实时的最大矛盾点刚发生的点击行为不应该等下一轮离线训练才生效。所以在线召回采用双通道第一个通道从Redis里读学生最近交互的课程ID然后查课程Embedding表做向量相似召回第二个通道从活跃课程池里按热度召回用于冷启动和探索。两个召回集合合并后再交给粗排做规则过滤和去重。3.2 Faiss 索引的在线召回代码课程Embedding可以用离线模型训练训练好之后加载进Faiss。Faiss是Python生态里最成熟的向量检索库支持CPU和GPU单机处理几十万条向量没问题。查询时只用在Faiss上查一次拿到候选集后再从Redis回填课程元数据。import faiss import numpy as np # 加载离线训练好的课程Embedding矩阵每行是一个课程的128维向量 course_emb np.load(course_emb.npy).astype(float32) course_ids [line.strip() for line in open(course_ids.txt)] # 使用内积索引向量训练时已经做了归一化内积等于余弦相似度 index faiss.IndexFlatIP(course_emb.shape[1]) index.add(course_emb) def recall_by_vector(query_emb, top_k20): # query_emb是学生当前兴趣向量形状(1, 128) scores, indices index.search(query_emb.reshape(1, -1), top_k) return [(course_ids[i], float(scores[0][j])) for j, i in enumerate(indices[0])]IndexFlatIP是暴力内积检索返回的indices是课程在矩阵里的行号scores是相似度分数。如果课程量超过10万考虑换成faiss.IndexIVFFlat它需要先训练聚类索引再添加向量。注意使用IndexFlatIP时训练前必须对向量做L2归一化否则内积分会被向量模长干扰导致长文本课程永远排前面。索引构建完需要保存到磁盘服务启动时直接加载避免每次重启都重新add。3.3 粗排与重排规则分数 模型分数合并召回返回20到50条候选集数量有限重排阶段可以稍微奢侈一点。常见策略是先把规则过滤做彻底再叠加一个轻量模型或分数公式。我一般用三部分分数加权实时行为分、课程难度匹配分、知识缺口覆盖分。def rerank_candidates(user_id, candidates, knowledge_gaps): results [] for course_id, vector_score in candidates: # 实时行为分学生在过去24小时对该课程的点击、停留时间 realtime_score get_realtime_score(user_id, course_id) # 难度匹配分当前学生掌握度与课程难度的负误差 mastery get_mastery(user_id) diff get_course_difficulty(course_id) diff_score 1 - abs(mastery - diff) / 5 # 知识缺口覆盖课程有多少知识点正好是学生薄弱点 course_kps get_course_knowledge_points(course_id) deficit_cover len(set(course_kps) set(knowledge_gaps)) / len(course_kps) final_score 0.5 * vector_score 0.3 * realtime_score 0.2 * diff_score deficit_cover * 0.2 results.append((course_id, final_score)) results.sort(keylambda x: x[1], reverseTrue) return results[:10]这个重排函数把四类信号合并成一个最终分数。权重系数0.5、0.3、0.2可以放进配置里不要硬编码。get_realtime_score要读Redis如果候选集有50条一次请求就会读50次。这里的优化空间很大可以改用Redis Pipeline一次取回50个score或者把score在事件消费者阶段就直接更新到候选课程列表中。3.4 冷启动与结果去重新学生没有行为历史召回向量检索拿不到query向量。冷启动干脆从简单开始先按课程的热度、平均评分、教学大纲顺序推一轮等学生积累了3个有效事件后再切换到个性化召回。另一个容易踩的坑是重复学生刚学完的课程不该又出现在推荐第一位。重排后还需要做一个规则去重把最近14天内学完的课程ID放进Redis Set去重时直接过滤。场景召回源重排逻辑新学生、无行为热度 大纲顺序不叠加实时行为分老学生、行为密集Faiss向量召回 热度召回实时行为分权重提升到0.4课程上新上新课程池给3小时临时加权4. 课程知识库上的个性化智能问答RAG 与学习状态注入推荐解决的是“学什么”问答解决的是“没学会时问谁”。通用大模型对知识点能给出高度概括的答案但它不知道这个学生刚刚在哪道题上卡住过也不知道课程老师指定的教材版本。要让问答个性化不能只把问题丢给大模型得把教学知识库和学生状态一起接进来。4.1 纯问答模型在课程场景的三个缺口第一个缺口是时效性。教材改版后新旧概念会混在一起通用模型无法区分课程当前版本。第二个缺口是上下文缺失。同一个问题学完基础模块和学完进阶模块的学生需要的回答深度完全不同。第三个缺口是归因要求。在学校场景里答案要能引用到具体课件页或题目编号而不是只有一段推理。这三个缺口都指向一个结论必须用检索增强生成把课程资料先分词入库再按需取用。4.2 课程资料切分与向量入库课程资料通常是PPT、Word、PDF混合格式。切分策略直接决定检索质量。按固定长度硬切会截断公式和代码块我一般用“先按标题结构分再按段落分单块不超过512个字符”的策略。import re from typing import List def split_course_document(text: str) - List[str]: # 先按章节标题切保留标题信息用于回答引用 sections re.split(r\n(?(第[一二三四五六七八九十百\d][章节])), text) chunks [] for sec in sections: if len(sec) 512: chunks.append(sec) else: # 长段落按句号切合并到接近512字符为止 sentences re.split(r(?[。]), sec) buf for sent in sentences: if len(buf) len(sent) 512: buf sent else: if buf: chunks.append(buf) buf sent if buf: chunks.append(buf) return chunks切分不是越短越好。512字符对中文来说大约是300到500个汉字既能容纳一个小节的核心概念又不会因为上下文过长稀释答案。切分后的每个chunk需要记录来源章节和课程ID写入向量库时带上这些字段作为metadata。向量化模型可以用开源中文Embedding模型也可以用通用text-embedding接口本质是让语义相近的教学内容在向量空间里靠在一起。4.3 检索后重排把“知识状态”塞进排序分数知识库检索出来3到5个候选段落接下来要让“知识状态”参与排序。学生刚错了一道题和这道题关联的课程片段应该优先被问答模型看到。def rerank_chunks(query, chunks, user_id): # 获取学生薄弱知识点 weak_points get_weak_points(user_id) # 列表: [勾股定理, 二次函数] re_scored [] for chunk in chunks: # chunk[knowledge_points] 是切分时存的知识点标签 overlap set(chunk[knowledge_points]) set(weak_points) weakness_bonus len(overlap) * 0.15 # 保持检索基础分weakness_bonus 让相关薄弱点片段排到前面 re_scored.append({ chunk: chunk, score: chunk[retrieval_score] weakness_bonus }) re_scored.sort(keylambda x: x[score], reverseTrue) return [item[chunk] for item in re_scored[:3]]排序分数里加的是“知识缺口覆盖度”的加成而不是直接过滤。如果学生薄弱点匹配到了两个不相关段落加成也会把它排上来所以这个系数只设0.15避免把基础语义相关性完全盖掉。get_weak_points的实时性很关键它的数据来源就是第2.3节里更新的答题事件一旦学生提交错误答案弱点列表通常需要30秒内更新完。4.4 低置信度时怎么办RAG系统一定会遇到检索结果和问题不匹配的情况。推荐做法是给问答加一个简单的置信度判断如果检索到的最相似片段得分低于阈值就明确告诉学生“当前课程资料里暂时没有找到直接答案”同时列出可能相关的章节名。这比大模型硬编答案更好至少不会给学生错误的安全感。检索最高分行为回答示例 0.75直接生成结合知识库内容引用课件章节0.50 - 0.75生成并附“相关章节”回答加上“参考第3.2节和课堂练习2” 0.50拒绝回答转人工“资料库未覆盖建议咨询授课教师”5. 源码工程落地目录结构、配置管理与 FastAPI 服务化实时推荐和问答的源码不能只在一台机器上跑通就结束还需要考虑多进程部署、接口并发和配置切换。很多Python项目死在“本地能跑”就是输在目录职责不清和配置依赖硬编码上。5.1 推荐的源码目录这样组织功能边界最清楚把数据管道、推荐服务、问答服务、公共组件分开是后续能继续加功能的前提。下面是一个常见且不冗余的目录结构。course_reco_qa/ ├── app/ │ ├── api/ # FastAPI 路由 │ │ ├── recommend.py │ │ └── qa.py │ ├── core/ # 配置、日志、Redis连接池 │ ├── features/ # 特征计算和更新 │ ├── recommender/ # 召回、粗排、重排 │ ├── qa/ # RAG 检索、知识库管理 │ └── models/ # Embedding、重排模型 ├── jobs/ │ ├── consume_behavior.py │ └── train_embedding.py ├── tests/ ├── docker-compose.yml ├── requirements.txt └── config.yamlfeatures目录只负责读写Redisrecommender和qa通过函数调用拿特征绝不在推荐函数里直接拼接Redis key。jobs目录放后台常驻脚本比如第2.2节的行为消费者。这样设计后推荐接口和问答接口各自依赖的函数边界非常明确后期做性能分析和单元测试都省力。5.2 配置管理环境变量与 YAML 合并源码里最忌讳出现Redis IP、模型路径、API Key这类硬编码。可以用一个轻量配置类在启动时读取环境变量和YAML文件并把敏感配置放到.env中。from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): redis_host: str 127.0.0.1 redis_port: int 6379 faiss_index_path: str models/course_emb.index embedding_model_path: str models/embedding.bin answer_confidence_threshold: float 0.5 model_config SettingsConfigDict( env_file.env, env_prefixCOURSE_RECO_ ) settings Settings()env_prefixCOURSE_RECO_表示环境变量COURSE_RECO_REDIS_HOST会覆盖YAML里的redis_host。这样做的好处是本地开发用配置文件测试环境用环境变量生产环境用密钥管理服务同一套源码不需要改任何文件。启动服务时所有组件都从settings取参数避免到处飘着Redisl连接字符串。5.3 用 FastAPI 把推荐和问答暴露成 HTTP 接口推荐和问答接口有明确区别推荐接口接收用户ID返回课程ID列表问答接口接收用户ID加问题返回答案和引用片段。下面把两个接口放在一个文件里示意。from fastapi import FastAPI from pydantic import BaseModel app FastAPI(titlecourse-reco-qa) class RecoRequest(BaseModel): user_id: str class RecoResponse(BaseModel): course_ids: list[str] reason: str class QARequest(BaseModel): user_id: str question: str class QAResponse(BaseModel): answer: str references: list[str] app.post(/v1/recommend, response_modelRecoResponse) def recommend(req: RecoRequest): course_ids, reason recommend_service.run(req.user_id) return RecoResponse(course_idscourse_ids, reasonreason) app.post(/v1/qa, response_modelQAResponse) def qa(req: QARequest): answer, refs qa_service.run(req.user_id, req.question) return QAResponse(answeranswer, referencesrefs)接口只做参数校验和转发业务逻辑全部放在recommend_service和qa_service。FastAPI的Pydantic模型负责处理请求和响应的数据格式如果前端传入的user_id不是字符串接口会在进入业务逻辑之前就报422错误。生产环境还需要给接口加超时控制推荐接口不能超过200毫秒问答接口根据模型不同可以放宽到2秒。5.4 用 docker compose 把依赖一起跑起来本地调试最省事的方式是用docker compose启动Redis、向量库和应用服务。这样一个docker compose up就能拉起全链路依赖。services: redis: image: redis:7 ports: [6379:6379] api: build: . environment: COURSE_RECO_REDIS_HOST: redis COURSE_RECO_FAISS_INDEX_PATH: /data/models/course_emb.index volumes: - ./models:/data/models ports: - 8000:8000 depends_on: - redis consumer: build: . command: python jobs/consume_behavior.py environment: COURSE_RECO_REDIS_HOST: redis depends_on: - redisapi和consumer共享同一个镜像但command不同。api启动Uvicorn进程consumer启动行为消费者。两者的Redis连接都指向compose网络内的redis服务。需要提醒的是向量索引文件比较大时不要打进镜像用volume挂载方式更容易替换。6. 实时推荐和问答联动预取、批量推理与可解释输出推荐和问答分开实现只是第一步真正提升体验的是让两部分共享在线特征和实时事件。学生在问答里问了一个问题这个行为本身应该影响推荐学生看了推荐里的课程问答检索时也应该优先考虑刚看的章节。6.1 共享特征缓存问答行为也进推荐特征在2.3节的行消费者中问答提问也需要被当成一种行为事件处理。学生问了一道关于“Python列表推导式”的问题消费者把事件写入特征缓存推荐服务下一次召回时自然会把与“列表推导式”相关的课程排到前面。实现上可以给事件类型加一个ask_query并在特征更新函数里把提问的文本转化成短期兴趣向量。def event_to_interest_vector(event): # 简单做法用高频关键词映射到课程标签再做加权 keyword extract_keyword(event.get(extra, )) label_vec label_to_vector(keyword) # 128维向量 return label_vec这类跨系统联动很有效但不要每个提问都更新一次长期向量。一个小时内同一个学生的提问应该合并成一次特征更新否则一个学生在问答中连续追问十次推荐结果会被单一知识点顶到全屏。6.2 批量推理与结果缓存在线推荐最怕的是每个学生请求都要实时跑一遍Faiss。学生行为不活跃时推荐结果可以缓存10分钟活跃时只对新增行为影响的候选集做增量重排。一个折中方案是把推荐结果写入Redis并设置短TTLTTL内直接返回缓存TTL外重新计算。def get_recommendations_with_cache(user_id): cache_key freco_result:{user_id} cached r.get(cache_key) if cached: return json.loads(cached) course_ids recompute(user_id) r.setex(cache_key, 300, json.dumps(course_ids)) return course_idssetex的过期时间设置为300秒。如果实时事件到达消费者会主动删除该用户对应的缓存键下一次请求就会强制重算。这种“过期加主动失效”的组合比单一缓存方式更可靠。6.3 给前端一个“为什么推荐/为什么这样答”的理由做实时推荐系统不能只看算法指标还要让用户和老师理解结果来源。返回的reason和references字段就是要回答“为什么”。推荐理由可以用最直接的事件拼接“因为你刚完成了二次函数单元测验且做错了解不等式题目因此推荐相关视频。”问答回答则同时返回引用的课件章节ID。{ course_ids: [c_lesson_045, c_lesson_032], reason: weak_point:linear_inequality, watched:quadratic_lesson_04 }这个解释结构有两个好处前端可以根据字段渲染标签运营人员可以快速判断推荐结果是否合理。调试实时服务时看到weak_point字段是否和最新错题一致就能确认特征链路是否在正常工作不需要再翻日志确认每一个数据环节。把可解释字段纳入接口设计会让整套源码在真实平台上的可维护性高出一个档次。本文还有配套的精品资源点击获取