简介这是一套面向教育技术方向的Python课程教学辅助系统源码适合计算机专业学生、教师及教育管理人员参考学习用于实现课程内容个性化推荐与智能问答。系统以Python为核心结合数据挖掘、推荐算法与自然语言处理可实时分析学习行为并调整推荐策略同时支持对用户提问进行语义理解与解答。压缩包共54个文件约108KB其中40个py文件承载推荐、搜索、文本预处理、JWT鉴权与Celery异步任务等核心逻辑8个xml为IDE与数据源配置另含sqlite3数据库、txt说明及gitignore等辅助文件目录按登录、课程、短信、工具等模块划分结构清晰。目前已有344人学习下载。读者可从中获取完整的推荐与问答实现思路、数据库与接口设计范例以及可运行的工程骨架便于二次开发或课程设计参考。1. 从课堂沉默到实时推荐这套系统到底在解决什么上周旁听了一节 Python 数据分析课老师在讲台上问「有多少人听懂了 Pandas 的 groupby」聊天框里只有三个人回复。课后翻后台日志才发现其实有二十多个学生在同一时间点反复回看那段录播——他们不是没听懂是不敢在课堂上说没听懂。这个场景几乎每个做在线教育平台的人都遇到过教学数据在实时产生但反馈链路是断的。基于 Python 的实时课程教学数据内容推荐与个性化智能问答系统要解决的就是这个断层——把学生看视频的进度、暂停点、答题正确率、提问记录这些实时数据接进来一边做内容推荐一边用问答系统兜住那些「不好意思问出口」的疑惑。这套系统适合谁如果你正在做在线教育平台的后端开发或者手头有一个课程管理系统想加上推荐和问答能力又或者你在做毕业设计需要一套能跑通的 Python 推荐加问答方案那接下来的内容就是冲着你来的。它不要求你从零训练大模型也不要求你有海量用户数据核心思路是用轻量级方案把实时数据流、推荐逻辑和问答检索串成一条可落地的链路。热搜词里「Python」「推荐系统」「智能问答系统」「源码」这几个词反复出现说明大家真正关心的是有没有一套能看懂、能改、能跑起来的代码结构而不是又一篇讲协同过滤公式的论文。我见过太多团队在这件事上翻车推荐模块用离线批处理跑学生都下课了才算出「你可能喜欢」问答模块直接调一个通用大模型回答跟课程内容毫无关系。这套系统的设计出发点就是反着来——推荐要实时问答要贴着课程知识库走。下面从数据管道怎么搭、推荐逻辑怎么选、问答怎么接、坑在哪里一步步拆开讲。2. 实时数据管道与特征工程从埋点到可用向量2.1 为什么不用离线批处理做推荐离线批处理推荐在课程场景下有个致命问题课程内容的消费节奏太快。一节 45 分钟的课学生的兴趣点可能在 10 分钟内切换三四次——前 10 分钟在听概念中间 20 分钟在看代码演示最后 15 分钟在做练习。如果推荐系统每小时才更新一次用户画像那它推荐的内容大概率是过时的。实时推荐的核心不是「快」而是「跟得上学生的当前状态」。常见做法是用消息队列把前端埋点数据实时推到处理端。我一般会选 Redis 的 Stream 结构做轻量级消息队列原因是它足够简单不需要额外部署 Kafka 集群对于中小规模课程平台完全够用。数据流是这样的前端在视频播放器里埋点记录 play、pause、seek、answer、ask 五类事件通过 HTTP 接口打到后端后端写入 Redis Stream消费者进程实时读取并更新用户特征向量。这里有个选型细节值得说清楚为什么不用数据库直接写因为推荐系统需要的是「最近 N 分钟内的行为序列」而不是全量历史。Redis Stream 天然支持按时间范围读取配合 XADD 和 XREAD 命令可以很方便地拿到滑动窗口内的行为数据。数据库适合做持久化存储和离线分析但实时特征计算走 Redis 更顺手。2.2 埋点数据采集与 Redis Stream 写入先看数据采集端的代码。前端埋点通过一个统一的接口上报后端接收到之后做基本校验再写入 Stream。import redis import json import time from flask import Flask, request, jsonify app Flask(__name__) r redis.Redis(hostlocalhost, port6379, db0, decode_responsesTrue) STREAM_KEY course:events VALID_EVENTS {play, pause, seek, answer, ask} app.route(/track, methods[POST]) def track(): data request.get_json() # 必填字段校验缺一个就丢弃避免脏数据污染特征 required [user_id, course_id, event_type, timestamp] if not all(k in data for k in required): return jsonify({error: missing fields}), 400 if data[event_type] not in VALID_EVENTS: return jsonify({error: invalid event}), 400 # 补充服务端时间戳防止客户端时间被篡改 data[server_ts] int(time.time() * 1000) # 写入 Redis Streammaxlen 控制内存占用保留最近 10 万条 r.xadd(STREAM_KEY, data, maxlen100000) return jsonify({status: ok}) if __name__ __main__: app.run(port5000)这段代码的逻辑很直接校验必填字段和事件类型补充服务端时间戳然后写入 Redis Stream。maxlen100000这个参数需要根据你的课程平台规模调整——如果同时在线人数在几百人级别10 万条大约能覆盖最近几小时的数据如果上千人同时在线建议调到 50 万。注意server_ts字段很多团队只信任客户端时间结果学生把手机时间改一下就能刷推荐这个坑后面还会细说。2.3 用户特征向量的实时更新数据进了 Stream 之后需要一个消费者进程持续读取并更新用户特征。特征向量的设计决定了推荐质量的上限。我一般会把用户特征分成三组短期兴趣最近 5 分钟行为、中期兴趣最近 30 分钟行为、长期偏好最近 7 天统计。短期和中期用 Redis 的 Sorted Set 存储长期用 Hash 存储。import redis import json import time r redis.Redis(hostlocalhost, port6379, db0, decode_responsesTrue) STREAM_KEY course:events GROUP_NAME feature_updater CONSUMER_NAME worker-1 # 创建消费者组从最新消息开始消费 try: r.xgroup_create(STREAM_KEY, GROUP_NAME, id$, mkstreamTrue) except redis.exceptions.ResponseError: pass # 组已存在 def update_features(user_id, event): now int(time.time()) # 短期兴趣最近 5 分钟内的课程 ID 及权重 short_key ffeat:short:{user_id} r.zadd(short_key, {event[course_id]: now}) r.zremrangebyscore(short_key, 0, now - 300) # 移除 5 分钟前的 # 中期兴趣最近 30 分钟 mid_key ffeat:mid:{user_id} r.zadd(mid_key, {event[course_id]: now}) r.zremrangebyscore(mid_key, 0, now - 1800) # 长期偏好按事件类型累加计数 long_key ffeat:long:{user_id} r.hincrby(long_key, f{event[event_type]}:{event[course_id]}, 1) while True: # 每次读 10 条阻塞 2 秒 messages r.xreadgroup(GROUP_NAME, CONSUMER_NAME, {STREAM_KEY: }, count10, block2000) if not messages: continue for stream, msg_list in messages: for msg_id, event in msg_list: update_features(event[user_id], event) # 处理完确认防止重复消费 r.xack(STREAM_KEY, GROUP_NAME, msg_id)这段代码的核心逻辑是用 Sorted Set 的 score 存时间戳通过zremrangebyscore自动淘汰过期数据保证短期和中期特征始终是滑动窗口内的。hincrby做长期计数简单但有效。参数方面count10和block2000需要根据消息量调整——消息多的时候增大 count消息少的时候 block 设长一点减少空轮询。xack这行不能省否则消息会一直留在 pending 列表里时间长了 Redis 内存会被撑爆。提示消费者组的id$表示只消费新消息如果你需要从头回放历史数据做测试改成id0。3. 推荐引擎的选型与实现协同过滤还是内容匹配3.1 课程推荐场景下两种路线的取舍推荐系统在课程场景下有个特殊之处课程数量通常不多几百到几千门但用户行为稀疏度极高。一个学生一学期可能只选五六门课跟电商平台上千次点击完全不是一个量级。这意味着传统的协同过滤在课程推荐上容易遇到冷启动和数据稀疏问题——新学生没有历史行为新课程没有交互记录。我的做法是混合推荐用内容匹配兜底冷启动用协同过滤做个性化排序。内容匹配基于课程标签和用户画像的余弦相似度协同过滤用 ItemCF 计算课程之间的关联度。两者加权融合权重根据用户行为数量动态调整——行为少于 10 条时内容匹配占 0.8行为超过 50 条时协同过滤占 0.7。热搜词里「推荐系统」出现频率很高但很多人一上来就搞深度学习模型结果数据量不够模型效果还不如规则。我的血泪经验是在课程推荐场景下ItemCF 加内容匹配的组合效果往往比神经协同过滤更稳定而且可解释性强——你能明确告诉老师「这门课被推荐是因为选了 A 课的学生也选了它」。3.2 ItemCF 相似度计算与增量更新ItemCF 的核心是计算课程之间的相似度矩阵。全量计算用 pandas 做就行但实时推荐需要增量更新——新产生的选课行为要能快速反映到相似度上。import pandas as pd import numpy as np from collections import defaultdict def build_item_similarity(interactions): interactions: list of (user_id, course_id, rating) 返回课程相似度字典 {course_a: {course_b: score}} # 构建用户-课程评分矩阵 df pd.DataFrame(interactions, columns[user_id, course_id, rating]) pivot df.pivot_table(indexuser_id, columnscourse_id, valuesrating, fill_value0) # 计算课程之间的余弦相似度 item_matrix pivot.T.values # 行是课程列是用户 norms np.linalg.norm(item_matrix, axis1, keepdimsTrue) norms[norms 0] 1 # 防止除零 normalized item_matrix / norms sim_matrix normalized normalized.T # 只保留每个课程最相似的前 20 个减少存储和计算 courses pivot.columns.tolist() sim_dict {} for i, course in enumerate(courses): sim_scores sim_matrix[i] top_indices np.argsort(sim_scores)[::-1][1:21] # 排除自身 sim_dict[course] { courses[j]: round(float(sim_scores[j]), 4) for j in top_indices if sim_scores[j] 0.01 } return sim_dict def incremental_update(sim_dict, new_interactions, decay0.95): 增量更新对已有相似度做衰减再叠加新行为的贡献 decay 参数控制历史相似度的遗忘速度 # 先对所有相似度做衰减 for course in sim_dict: for other in sim_dict[course]: sim_dict[course][other] * decay # 新行为产生的共现关系简单累加 co_occur defaultdict(lambda: defaultdict(float)) user_courses defaultdict(set) for uid, cid, rating in new_interactions: user_courses[uid].add(cid) for uid, courses in user_courses.items(): courses list(courses) for i in range(len(courses)): for j in range(i 1, len(courses)): co_occur[courses[i]][courses[j]] 1.0 co_occur[courses[j]][courses[i]] 1.0 # 合并到相似度字典 for course, others in co_occur.items(): if course not in sim_dict: sim_dict[course] {} for other, score in others.items(): old sim_dict[course].get(other, 0) sim_dict[course][other] round(old score * 0.1, 4) return sim_dictbuild_item_similarity用矩阵乘法一次性算出所有课程对的余弦相似度然后只保留 top 20这个截断很关键——不截断的话相似度矩阵会非常稀疏且存储成本高。incremental_update里的decay0.95是遗忘因子每来一批新数据就把历史相似度乘 0.95这样推荐结果会逐渐偏向近期行为。score * 0.1是新增行为的权重调大这个值会让推荐更敏感调小则更稳定。我一般会在测试环境用 0.05 到 0.2 之间做 A/B 测试。3.3 融合排序与实时推荐接口有了内容匹配分数和协同过滤分数之后需要融合成一个最终排序。融合策略用加权求和权重根据用户行为数量动态调整。def recommend(user_id, sim_dict, user_features, course_tags, top_n10): 混合推荐内容匹配 ItemCF user_features: 用户近期交互过的课程列表 course_tags: {course_id: [tag1, tag2, ...]} recent_courses user_features.get(recent, []) behavior_count user_features.get(count, 0) # 动态权重行为少时偏内容匹配行为多时偏协同过滤 if behavior_count 10: w_content, w_cf 0.8, 0.2 elif behavior_count 50: w_content, w_cf 0.5, 0.5 else: w_content, w_cf 0.3, 0.7 scores defaultdict(float) # 内容匹配分数用户近期课程标签与候选课程标签的 Jaccard 相似度 user_tags set() for cid in recent_courses: user_tags.update(course_tags.get(cid, [])) for cid, tags in course_tags.items(): if cid in recent_courses: continue # 已交互过的不再推荐 tag_set set(tags) if not tag_set or not user_tags: continue jaccard len(user_tags tag_set) / len(user_tags | tag_set) scores[cid] w_content * jaccard # 协同过滤分数基于 ItemCF 相似度累加 for cid in recent_courses: for similar_cid, sim_score in sim_dict.get(cid, {}).items(): if similar_cid not in recent_courses: scores[similar_cid] w_cf * sim_score # 排序取 top N ranked sorted(scores.items(), keylambda x: x[1], reverseTrue) return [cid for cid, _ in ranked[:top_n]]这段代码里内容匹配用的是 Jaccard 相似度而不是余弦相似度原因是课程标签是离散的集合Jaccard 更直观。协同过滤部分直接累加相似度分数没有做归一化因为 ItemCF 的相似度本身已经在 0 到 1 之间。top_n10是返回给前端的推荐数量实际接口里还会加一层业务过滤——比如过滤掉已下架的课程、学生已修过的必修课等。注意recent_courses建议取最近 20 条交互记录太多会稀释兴趣信号太少则覆盖不足。这个值我在不同项目里试过 10 到 3020 是比较稳的中间值。4. 智能问答系统的搭建检索增强而非纯生成4.1 为什么课程问答不能直接调通用大模型通用大模型在课程问答场景下有三个硬伤第一它不知道你这门课讲到哪了可能把下学期才讲的内容提前说出来第二它可能编造课程里根本没提过的概念学生信了反而更迷糊第三响应延迟不可控直播课场景下学生等三秒没答案就关掉了。我的方案是检索增强生成RAG的轻量版先用课程知识库做语义检索找到最相关的几个知识点再把知识点和问题一起送给模型做总结。这样既保证了答案贴着课程内容走又控制了生成范围。知识库的构建不需要多复杂把课程字幕、讲义、FAQ 文档切片存入向量数据库就行。向量化用 sentence-transformers 的轻量模型在 CPU 上也能跑。热搜词里「智能问答系统」和「源码」经常一起出现说明大家想要的是能直接跑起来的代码而不是概念介绍。下面从知识库构建到问答接口把关键代码过一遍。4.2 课程知识库的切片与向量化知识库的质量决定了问答的上限。切片策略上我一般按语义段落切每段 200 到 500 字重叠 50 字。课程字幕按时间戳切每 30 秒一段。讲义按标题层级切每个小节一段。from sentence_transformers import SentenceTransformer import numpy as np import faiss import json # 用轻量模型CPU 上单条编码约 20ms model SentenceTransformer(paraphrase-multilingual-MiniLM-L12-v2) def build_knowledge_base(documents): documents: list of {course_id: str, content: str, source: str} 返回 faiss 索引和对应的元数据列表 texts [doc[content] for doc in documents] # 批量编码batch_size 根据内存调整 embeddings model.encode(texts, batch_size32, show_progress_barTrue, normalize_embeddingsTrue) # 用内积索引因为向量已归一化内积等价于余弦相似度 dim embeddings.shape[1] index faiss.IndexFlatIP(dim) index.add(embeddings.astype(float32)) # 元数据单独存检索后按索引取回 metadata [{course_id: d[course_id], content: d[content], source: d[source]} for d in documents] return index, metadata def search_knowledge(index, metadata, query, top_k5): 检索最相关的 top_k 个知识点 query_vec model.encode([query], normalize_embeddingsTrue) scores, indices index.search(query_vec.astype(float32), top_k) results [] for score, idx in zip(scores[0], indices[0]): if idx -1: continue item metadata[idx].copy() item[score] float(score) results.append(item) return resultsparaphrase-multilingual-MiniLM-L12-v2这个模型的好处是支持中文且体积小编码 384 维向量在普通笔记本上就能跑。normalize_embeddingsTrue让向量归一化这样用内积索引就等价于余弦相似度省去了额外计算。IndexFlatIP是精确检索数据量在几万条以内完全够用如果知识库超过 10 万条可以换成IndexIVFFlat做近似检索但需要额外训练索引。4.3 检索结果与生成模型的拼接策略检索到相关知识点后需要把它们和用户问题拼成一个 prompt 送给生成模型。拼接策略直接影响回答质量。我的做法是按相似度排序取前 3 条每条截断到 300 字总 prompt 控制在 1500 字以内。def build_prompt(question, retrieved_docs, course_name): 拼接检索结果和问题构造生成 prompt context_parts [] for i, doc in enumerate(retrieved_docs[:3], 1): content doc[content][:300] # 截断防止 prompt 过长 context_parts.append(f[片段{i}] {content}) context \n.join(context_parts) prompt f你是一门名为《{course_name}》的课程助教。 请仅根据以下课程资料回答问题如果资料中没有相关内容直接说这个问题在课程资料中没有找到答案。 课程资料 {context} 学生问题{question} 请用简洁的中文回答不超过 200 字。 return prompt def answer_question(question, index, metadata, course_name, llm_client): 完整的问答流程检索 生成 docs search_knowledge(index, metadata, question, top_k5) if not docs or docs[0][score] 0.3: return 这个问题在课程资料中没有找到答案建议课后向老师提问。 prompt build_prompt(question, docs, course_name) # 调用生成模型temperature 设低一点保证稳定性 response llm_client.generate(prompt, temperature0.3, max_tokens300) return responsescore 0.3这个阈值是兜底逻辑——如果检索到的最相关片段相似度太低说明知识库里没有相关内容这时候直接告诉学生「没找到」比让模型硬编一个答案要好。temperature0.3是为了让回答稳定课程问答不需要创造性准确比有趣重要。max_tokens300限制回答长度避免模型长篇大论。提示如果用的是本地部署的开源模型建议把max_tokens设小一点生成速度会快很多。直播课场景下学生能接受的等待时间大约是 2 秒。5. 避坑与排查那些让系统翻车的细节5.1 时间戳被篡改导致推荐结果漂移现象部分学生的推荐内容突然全部变成同一门课而且持续好几天不变。原因客户端上报的timestamp字段被篡改或者时区设置错误导致 Redis Sorted Set 里的 score 异常。比如学生手机时区设成了 UTC-12上报的时间戳比服务端时间早了 12 小时zremrangebyscore清理时把这些记录当成了「未来数据」一直保留在窗口里。解决所有特征计算必须用服务端时间戳server_ts客户端时间只做参考。在写入 Stream 之前就把server_ts补上后续所有窗口计算都用这个字段。另外在消费者端加一个校验如果server_ts与当前时间差超过 5 分钟直接丢弃这条消息。5.2 冷启动用户拿到重复推荐现象新注册的学生第一次打开推荐页看到的十门课里有六门是同一系列的。原因内容匹配阶段新用户没有历史行为user_tags为空Jaccard 相似度全部为 0导致所有候选课程得分相同排序退化成按课程 ID 排序。如果课程 ID 是连续分配的同一系列的课程 ID 相邻就全被推出来了。解决冷启动阶段不能只靠内容匹配。我的做法是加一个「热门课程兜底」当用户行为少于 3 条时推荐结果中至少包含 3 门全平台热门课程剩余位置用内容匹配填充。热门课程按最近 7 天的选课人数排序每天更新一次。5.3 问答系统检索到无关片段后强行生成现象学生问「Pandas 的 merge 怎么用」系统回答了一段关于「Python 列表推导式」的内容还煞有介事地编了个例子。原因知识库里没有 merge 相关的片段但检索返回了 top 5其中第一条相似度只有 0.15。生成模型看到有上下文就硬着头皮编了一个答案。解决在answer_question里加相似度阈值判断docs[0][score] 0.3时直接返回「没找到」。这个阈值需要根据你的向量模型调整——不同模型的相似度分布不一样建议用一批测试问题跑一下看正确匹配和错误匹配的分数分布取一个能分开两者的值。我用的这个模型0.3 是比较稳的阈值。5.4 Redis Stream 消息积压导致内存暴涨现象服务运行几天后 Redis 内存占用从几百 MB 涨到几 GB最后触发 OOM 被系统杀掉。原因消费者进程处理速度跟不上生产速度或者消费者挂了但没被发现消息一直堆积在 Stream 里。maxlen100000只限制了大致的条数但如果单条消息体积大比如带了完整的用户行为上下文10 万条也能占好几个 GB。解决第一给 Stream 设置更严格的maxlen比如 50000并且用approximateTrue让 Redis 更积极地裁剪。第二加监控定期用XLEN查 Stream 长度超过阈值就告警。第三消费者端加xack确认机制处理失败的消息不要一直留在 pending 列表里设置一个重试上限超过就丢弃并记录日志。5.5 推荐接口响应时间随用户行为增长而线性上升现象系统上线初期推荐接口响应时间 50ms三个月后变成 500ms学生反馈推荐页加载明显变慢。原因recommend函数里遍历了用户所有历史交互记录来计算内容匹配分数而recent_courses没有做长度限制有些活跃学生的历史记录积累到了几百条。解决在特征更新阶段就限制recent_courses的长度只保留最近 20 条。Redis Sorted Set 用zremrangebyrank按排名裁剪保留 score 最高的 20 个。另外sim_dict的遍历也要限制——只遍历recent_courses里的课程而不是所有课程。这两个限制加上之后响应时间稳定在 80ms 以内。6. 把问答和推荐串起来一个可验证的联调技巧单独测推荐和单独测问答都容易难的是验证两者串起来之后的行为是否符合预期。我一般会写一个模拟脚本模拟一个学生从进入课程到提问的完整流程观察推荐结果和问答回答是否合理。import requests import time import random BASE_URL http://localhost:5000 def simulate_student(user_id, course_id): 模拟一个学生的完整学习流程 events [ {event_type: play, course_id: course_id}, {event_type: pause, course_id: course_id}, {event_type: seek, course_id: course_id}, {event_type: answer, course_id: course_id}, {event_type: ask, course_id: course_id}, ] for event in events: payload { user_id: user_id, course_id: event[course_id], event_type: event[event_type], timestamp: int(time.time() * 1000) } requests.post(f{BASE_URL}/track, jsonpayload) time.sleep(0.5) # 模拟真实操作间隔 # 等特征更新 time.sleep(2) # 请求推荐 rec_resp requests.get(f{BASE_URL}/recommend, params{user_id: user_id, top_n: 5}) recommendations rec_resp.json()[courses] print(f用户 {user_id} 的推荐结果: {recommendations}) # 模拟提问 question 这节课讲的 groupby 和 pivot_table 有什么区别 qa_resp requests.post(f{BASE_URL}/ask, json{user_id: user_id, course_id: course_id, question: question}) answer qa_resp.json()[answer] print(f问答结果: {answer}) return recommendations, answer # 跑三个不同行为量的用户做对比 for uid in [user_new, user_mid, user_active]: simulate_student(uid, course_python_101)这个脚本的价值在于它能帮你快速发现「推荐结果和问答回答是否匹配当前课程」。比如模拟脚本里问的是 groupby 和 pivot_table 的区别如果问答系统返回的是「列表和元组的区别」说明知识库切片或者检索出了问题。推荐结果也一样——如果推荐的全是跟 Python 无关的课程说明特征更新或者相似度计算有 bug。联调时我还会加一个「一致性检查」推荐结果里的课程其标签应该和用户近期交互课程的标签有重叠。如果完全不重叠要么是权重设置有问题要么是标签体系没对齐。这个检查用几行代码就能做def check_consistency(recommendations, recent_courses, course_tags): 检查推荐结果与用户近期兴趣的一致性 user_tags set() for cid in recent_courses: user_tags.update(course_tags.get(cid, [])) overlap_count 0 for cid in recommendations: rec_tags set(course_tags.get(cid, [])) if user_tags rec_tags: overlap_count 1 ratio overlap_count / len(recommendations) if recommendations else 0 print(f标签重叠率: {ratio:.2%}) # 低于 30% 说明推荐可能跑偏了 return ratio 0.3这个检查在每次修改推荐权重或者更新知识库之后跑一遍能快速判断改动是否引入了退化。我自己的习惯是任何涉及推荐逻辑的改动上线前必须跑通模拟脚本加一致性检查两个都过了才合并代码。这套流程帮我拦住了好几次「权重调参调出负优化」的事故。希望帮到你。本文还有配套的精品资源点击获取