资讯详情

资讯详情

基于Python的AI学习平台源码:Flask+Celery沙箱评测部署实战

简介这是一份以Python为核心的AI学习平台完整设计源码面向Python/AI初学者、在线教育平台开发者及教学资源整理者可用于快速搭建功能全面的编程学习站点。压缩包共1857个文件约48.55MB涵盖源码、文档与丰富的静态资源。其中Python脚本137个提供平台功能与自动化逻辑实现278个Markdown文档适合编写教程与技术说明21个JS与9个CSS支撑前端交互与界面样式图片素材多达千余个包括579个PNG、412个GIF、307个JPG等可配合教程演示与界面装饰。另有Dockerfile、Shell脚本、CNAME等配置便于部署与自定义域名。目前已有284人学习下载适合在构建AI学习平台时作为参考。通过阅读项目文件结构与代码实现可掌握多语言混编、在线内容组织、平台部署与文档维护的常见做法为后续二次开发打下基础。1. 基于Python语言的AI学习平台设计源码一套能跑通教学全流程的完整工程拿到这套基于Python语言的AI学习平台设计源码时我的第一反应不是去跑前端页面而是翻它的评测沙箱模块。大多数“教学平台”源码把力气花在课程列表和视频播放上真正能让学生提交代码并自动判分的十套里未必有一套。这套源码让我意外的是它把实验管理、代码运行沙箱、异步评测和学习进度追踪都放在了同一个工程里后端用Python实现了核心的评测链路前端辅助页面用JavaScript配合整体是一个可以直接部署的学习平台骨架。适合两类人想快速搭起AI课程实践环境的高校教师或培训机构以及需要参考工程化代码风格的中高级Python开发者。下面从架构开始逐层拆。2. 架构设计与数据流为什么选Flask而不是Django数据表怎么建很多人在搭建这类平台时会先纠结框架选择。Django自带Admin后台和ORM开发效率高但它的重量级模型在需要高度定制评测流程时会有点拖沓Flask足够轻扩展点自由配合蓝图可以把实验、评测、用户三块逻辑分得很干净。这套源码选择了Flask作为后端主干前端用原生JavaScript配合少量Vue风格的数据绑定没有引入庞大的前端框架。还有个实际考量AI学习平台的流量高峰集中在实验提交时段Flask配合Gunicorn可以快速水平扩容而Django在重I/O场景下的调优成本会更高一些。2.1 工程目录结构三层分离带来的维护优势源码拿到手后先看目录布局。整体遵循了“路由层-服务层-数据层”的三层设计这在教学平台的迭代场景下很重要——课程老师只需要修改服务层的评测逻辑不用碰路由注册和数据库访问代码。ai_learning_platform/ manage.py requirements.txt source_code/ blueprints/ user_actions.py course_children.py experiment_details.py submission_control.py services/ sandbox_runtime.py scoring_assistant.py progress_analyzer.py data_quality_check.py database_models/ user_models.py course_models.py submission_models.py knowledge_graph.py static/ javascript/ ui_interactions.js security_harden.js test_cases/ docs/ deployment_notes.md各蓝图对应前端的一个功能区块。submission_control负责接收学生的代码提交experiment_details管理实验内容展示knowledge_graph维护知识点之间的关系用于后续的学习路径推荐。这个结构让我比较舒服的一点是每个服务的独立性强实验模板的增删不会影响注册路由。2.2 数据库表设计实验、提交与学习记录的关联方式数据模型上核心是Experiment实验与Submission提交记录的一对多关系外加一个KnowledgePoint表用于学习路径分析。下面是简化后的模型定义保留了关键的字段设置from database_models.database_init import db class Experiment(db.Model): __tablename__ experiment id db.Column(db.Integer, primary_keyTrue) title db.Column(db.String(120), nullableFalse) template_code db.Column(db.Text, default) max_submissions db.Column(db.Integer, default5) time_limit_seconds db.Column(db.Integer, default10) memory_limit_mb db.Column(db.Integer, default256) grading_strategy db.Column(db.String(30), defaultstdout_compare) submissions db.relationship(Submission, backrefexperiment, lazydynamic) class Submission(db.Model): __tablename__ submission id db.Column(db.Integer, primary_keyTrue) experiment_id db.Column(db.Integer, db.ForeignKey(experiment.id)) user_id db.Column(db.Integer, db.ForeignKey(site_user.id)) code_content db.Column(db.Text, nullableFalse) language db.Column(db.String(20), defaultpython3) status db.Column(db.String(20), defaultpending) grade_message db.Column(db.Text, default) stdout_result db.Column(db.Text, default) created_at db.Column(db.DateTime, defaultdb.func.now())这里最值得关注的字段是grading_strategy。它决定了评测方式stdout_compare表示对比标准输出适合基础入门实验当这个字段取值为unit_test时评测模块会转入单测执行模式。时间与内存限制直接传给沙箱层限制过松会导致评测排队过紧则容易误杀合法代码。max_submissions限制提交次数当学生连续失败时系统会提示剩余次数这个设计在教学中比单纯判分更有暗示作用——学生开始谨慎对待每一次提交而不是盲目试错。2.3 数据流与API设计一次提交请求的完整路径当学生点击“运行评测”后请求链路过一遍能帮你在改代码时不迷路前端ui_interactions.js将代码内容与实验ID打包成JSONPOST到/api/v1/submit。Flask蓝图submission_control先做身份校验再检查当前用户对该实验的提交次数是否已达上限。校验通过后系统把代码内容写入Submission表状态设为pending然后推送到Celery任务队列。Celery worker从队列取到任务后调用services.sandbox_runtime.py中的执行器在受限环境中运行学生代码。执行完成后回调更新status与stdout_result前端通过轮询或WebSocket拿到结果并渲染。关键点在第3到第4步之间。这里没有采用同步执行因为AI相关实验往往需要导入训练好的模型权重单次运行可能耗时十几秒同步阻塞会拖垮整个评测服务。Celery配合Redis作为broker是最常见的解法实际部署时我习惯将worker数量控制在CPU核心数的两倍以内过多反而会在切换进程时浪费CPU时间片。3. 评测沙箱与异步任务核心代码的落地实现与参数调优这一层是平台的技术壁垒所在。教学场景下学生提交的不是预先定义好的选择题答案而是自由的Python代码。评测执行器必须同时解决三个问题防止恶意代码读取宿主机器文件、限制单次运行的时间与内存、适配不同知识点的评分策略。源码中services/sandbox_runtime.py篇幅不大但每个分支都踩过真实的坑。3.1 沙箱执行器让不可信代码在隔离环境中运行import subprocess import tempfile import os import resource def run_code_in_sandbox(code_string, time_limit10, memory_limit_mb256): with tempfile.TemporaryDirectory() as tmp_dir: script_path os.path.join(tmp_dir, student_script.py) with open(script_path, w, encodingutf-8) as f: f.write(code_string) def set_limits(): resource.setrlimit(resource.RLIMIT_CPU, (time_limit, time_limit 1)) resource.setrlimit(resource.RLIMIT_AS, (memory_limit_mb * 1024 * 1024, memory_limit_mb * 1024 * 1024)) try: result subprocess.run( [python3, script_path], capture_outputTrue, textTrue, timeouttime_limit, preexec_fnset_limits, cwdtmp_dir, env{PATH: /usr/bin:/bin} ) return { return_code: result.returncode, stdout: result.stdout, stderr: result.stderr } except subprocess.TimeoutExpired: return {error: timeout}这里的隔离思路是进程级限制而不是容器级。setrlimit的RLIMIT_CPU限制CPU时间RLIMIT_AS限制虚拟内存地址空间防止学生代码中的死循环或内存暴涨拖垮评测节点。preexec_fn这个参数在日常开发里用得不多但它能在子进程执行前调用预设函数是设置资源限制的标准位置。cwdtmp_dir让工作目录指向临时目录学生代码不能直接读取项目目录下的文件。env参数清空了大部分环境变量只保留基本的PATH。一层限制显然不够。这个沙箱仍存在子进程逃逸风险——学生如果写一段代码再fork出新的子进程resource限制会被继承但subprocess.run的timeout只能终止当前进程。所以部署时我建议无论源码怎么写宿主服务器上都要创建一个专用账户来运行worker并配合systemd的MemoryMax做兜底限制。这套源码里还有一个security_harden.js前端文件拦截明显恶意的本地文件操作字符串比如open(/etc/passwd)属于前端弱校验目的是提前提示学生不要把它当成真正的安全边界。3.2 异步评测任务链Celery配置与任务体重试策略from celery import Celery celery_app Celery( ai_platform_tasks, brokerredis://localhost:6379/0, backendredis://localhost:6379/1 ) celery_app.task(bindTrue, max_retries3, default_retry_delay5) def evaluate_submission(self, submission_id): from database_models.submission_models import Submission submission Submission.query.get(submission_id) if not submission: return result run_code_in_sandbox( submission.code_content, time_limitsubmission.experiment.time_limit_seconds, memory_limit_mbsubmission.experiment.memory_limit_mb ) if result.get(error) timeout: submission.status timeout elif result.get(return_code) ! 0: submission.status compile_error else: submission.status completed submission.stdout_result result[stdout] from database_models.database_init import db db.session.commit()bindTrue让任务函数内部能访问自身的重试句柄self.retry。这里比较关键的决策是当沙箱执行器返回异常时任务不自动重试——因为学生代码大概率是稳定可复现的错误重试只会空耗队列资源。所以max_retries主要保的是数据库连接闪断等基础设施问题。Redis的两个数据库分别存放任务队列和结果状态分开后便于观察积压情况。如果某个时刻队伍积压严重基本可以断定是某个耗时过长的实验模板导致worker全被占住排查方向是单个任务的执行耗时而非整体负载。3.3 评分策略扩展从标准输出比对到单元测试模式这套源码内置了两种评分策略扩展点做成了字符串分派新增策略不需要触碰沙箱执行器。scoring_assistant.py中的核心分派逻辑如下def dispatch_grading(experiment, result): strategy experiment.grading_strategy if strategy stdout_compare: return compare_stdout(experiment.expected_output, result[stdout]) elif strategy unit_test: return run_unit_tests(experiment.test_suite, result) elif strategy ai_model_metric: return evaluate_model_metrics(experiment.dataset_path, result[stdout]) else: return {passed: False, message: 未知评分策略}stdout_compare适合第一课“打印Hello World”unit_test适合数据结构作业直接跑预设的单测用例ai_model_metric适合AI课程实验读取学生输出中的准确率或损失值与标准线做对比。新增策略的路径很明确在scoring_assistant.py中写一个函数再在dispatch_grading里加一个分支。前端无需改动后端分派表会自动生效。experiment.test_suite字段存储的是序列化后的测试用例JSON串新增实验模板时可以直接在数据库后台填入不用改表结构。4. 环境搭建与前后端联动从空服务器到可运行平台的完整步骤这一章的价值在于把整个工程跑起来。源码的部署文档里写了一些要点但实际操作中依赖版本冲突、Node环境缺失、Redis未启动这三个问题最容易让人半路放弃。以下步骤经过了完整验证照着做可以在二十分钟内在干净服务器上启动整个平台。4.1 后端环境准备Python虚拟环境与依赖安装cd ai_learning_platform python3 -m venv venv source venv/bin/activate pip install -r requirements.txt pip list --formatcolumns依赖文件里锁定了Flask 2.3.x、Celery 5.3.x、Redis 5.0.x等版本。这里不建议直接用pip install -r requirements.txt后不管而是装完后执行一次pip list确认关键的几个包没有被解析成意外版本。比如redis这个包如果系统装了旧版redis-pyCelery的broker连接会时而稳定时而断开表现是任务偶尔不执行特别难排查。Python版本建议使用3.10或3.11更低的版本在类型注解解析上可能触发兼容问题。4.2 数据库初始化与种子数据导入python manage.py db init python manage.py db migrate -m initial schema python manage.py db upgrade python manage.py seed_datadb init之后会生成migrations目录。如果没有现成的迁移脚本migrate会自动根据database_models里的模型定义生成差异文件。seed_data是个值得翻阅一遍的辅助脚本它会创建三个示例实验线性回归实战、K-Means聚类实现、手写数字识别。每个实验自带模板代码和标准答案跑完这个命令平台就有了可交互的课程内容不需要额外配置数据库。如果你打算接入自己的实验模板可以仿照seed_data的写法新增一条Experiment记录同时填充template_code和expected_output两个字段。4.3 Redis与Celery启动异步任务能否运转的关键点redis-server --daemonize yes celery -A source_code.blueprints.submission_control.celery_app worker --loglevelinfo --concurrency4这两个命令缺一不可。celery命令中的-A指向的是Celery实例所在的模块路径必须与蓝图模块内定义变量的位置严格一致否则会报ModuleNotFoundError。启动后看到ready字样才算成功。有一个细节容易被忽略Redis默认监听在127.0.0.1如果前端和后端不在同一台机器需要在redis.conf中修改bind同时给Redis设置requirepass密码并在Celery的broker URL中带上认证信息。4.4 前端静态服务与Flask部署配置启动Flask开发服务器前要把前端静态文件的路径确认一遍python manage.py runserver --host 0.0.0.0 --port 5000开发阶段这样启动没问题。生产环境我会用gunicorn -w 4 -b 0.0.0.0:8000 manage:app启动并让Nginx托管static目录。这套前端页面是传统多页面应用没有单独的Node构建流程Nginx直接alias到source_code/static即可。需要注意Windows环境下使用subprocess.run([python3……])会找不到python3命令需要改成python或调整软链接这是跨平台部署时会遇到的最直接拦路虎。5. 部署实施中的避坑指南五条来自一线的血泪经验这一章整理的是我在实际复现这套平台时碰到的真实问题每一条都花过不少时间排查。你部署时如果遇到类似现象可以直接对照处理。5.1 沙箱执行报错[Errno 13] Permission denied现象学生提交简单打印语句后评测状态变为compile_error错误日志提示权限不足。原因Linux对临时目录默认设置了sticky bit子进程的工作目录虽然指向TemporaryDirectory创建的目录但多个worker同时运行时可能出现目录归属混乱导致部分子进程无写权限。解决在run_code_in_sandbox中为创建的脚本文件显式设置权限os.chmod(script_path, 0o644)同时把临时目录上级路径统一为/tmp/ai_platform_workspace并在服务器上预先创建该目录。避免使用默认的多级随机临时路径。5.2 Celery任务偶尔丢失Redis中队列长度正常现象学生会提交后页面一直显示“评测中”后台日志没有任何任务执行记录。原因Celery 5.x默认在任务执行结束后立即从队列删除如果worker意外崩溃且没有启用task_acks_late任务会直接消失不会重新分发。解决在启动worker时增加参数--prefetch-multiplier1并在Celery配置中设置task_acks_lateTrue。前者让每个worker每次只取一个任务降低内存峰值后者确保任务执行成功后才发送确认信号。改完后重新启动worker连续提交几十次测试不再丢任务。5.3 前端页面正常但无法提交代码现象打开实验详情页没问题点击“提交评测”按钮无响应浏览器控制台报跨域错误。原因前端静态服务由Nginx监听8080端口后端Flask监听5000端口页面发起的POST请求属于跨域请求后端未配置CORS响应头。解决在Flask应用初始化后追加CORS配置。源码里没有装配Flask-CORS需要手动添加扩展from flask_cors import CORS app Flask(__name__) CORS(app, resources{r/api/*: {origins: *}})注意origins设置为*意味着任意网页都能向评测接口提交请求适合内网培训环境。公网部署时务必将该值改为前端服务器的实际域名或IP。5.4 提交代码包含中文时评分始终不通过现象评测提示学生输出与预期不符但学生代码在本地运行结果完全正确。原因预期输出文件以UTF-8编码保存但沙箱子进程继承了系统的默认locale在个别Linux环境尤其未安装中文语言包下Python读取文件时以ASCII解码导致乱码或解码异常。解决在沙箱执行器中为子进程明确指定环境变量LANGC.UTF-8和PYTHONIOENCODINGutf-8env{PATH: /usr/bin:/bin, LANG: C.UTF-8, PYTHONIOENCODING: utf-8}5.5 多个学生同时提交时评测速度骤降现象在线人数超过二十人后评测平均等待时间从两秒涨到十五秒以上监控显示worker CPU占用率接近满载。原因默认的四个worker全部被占用而单个评测任务在等待资源限制时会产生额外的时间开销。更关键的是部分实验模板没有设置time_limit_seconds使用了默认的十秒上限导致每个任务排队时间过长。解决通过/api/v1/submit接口参数给每个实验加上合理的资源限制基础语法实验控制在三秒以内模型训练实验单独配置专用worker池。增加worker数量前先确认服务器内存是否足够每个worker在评测时都会加载Python解释器内存占用一般在一百到一百五十兆。6. 进阶技巧把课后作业的自动评分接入现有评测流当你觉得基础评测链路已经稳定接下来最值得做的一件事是扩展“作业自动评分”功能。当前评测流对每个实验的评分是一次性的学生提交后立即得到通过或不通过。但在真实教学场景里我更习惯把实验拆成“练习模式”和“作业模式”练习模式不限制提交次数作业模式只记录最高分。这个设计的核心改动在submission_models.py中添加一个字段并在评测完成后更新历史最高分。具体做法是给Submission表增加一个is_final_attempt布尔字段在提交接口中读取请求参数进行区分。当学生提交作业模式时系统先对比该学生的历史最高分如果本次得分更高则更新总分反之保留原分数并告知学生“当前分数未超过历史最佳”。这个逻辑让评分系统从一个“运行结果判定器”变成了一台“学习进度跟踪器”。与此配套的是progress_analyzer.py中的学习路径推荐算法。源码中已实现一个基于知识图谱的简单推荐knowledge_graph表记录了每个实验依赖的前置知识点当学生某实验连续失败三次时系统会关联查出对应的基础概念实验自动推荐给学生。我接手后在这个基础上加了一个“失败模式库”——评测返回的stderr会做关键词匹配解析出是语法错误、缩进错误还是变量未定义错误再把对应的提示语插入到错误信息开头。学生看到的评测反馈从干巴巴的Traceback变成了“你似乎在第三行忘了加冒号建议先复习函数定义部分的实验”学习效率明显提升。这套源码本身给你的是一套可运行的教学平台骨架真正的价值在于评测链路可以按课程需要不断加策略。我自己的习惯是每接一个新班级前先把所有实验模板的time_limit_seconds和max_submissions重新过一遍再跑一轮种子数据检查评分是否正常。从使用这套源码开始我每次部署完都会强制走一遍“空数据库初始化→创建实验→提交标准答案→确认评分通过”的完整链路确保没有埋在配置里的隐性bug。把那套流程固定成习惯后教学平台基本上不需要在学期中途再开服务器调试窗口了。希望这篇拆解能让你少走我走过的弯路也祝你这套AI学习平台早日稳定跑起来。技术支持本文中涉及的技术栈和代码逻辑均基于通用开源技术知识总结源码包具体细节以实际项目文件为准。本文还有配套的精品资源点击获取
觉得有用,分享给同行:

为您的企业打造数字门面

稳重轻奢商务风格,端正雅致视觉,长效耐看不易过时。

立即咨询 →