ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

Python AI 作业辅助源码实战:从模型接入到批处理流水线

Python AI 作业辅助源码实战:从模型接入到批处理流水线 简介这是一份面向学生与AI初学者的作业辅助工具源码基于Python开发旨在帮助使用者更高效地完成学习任务并理解AI算法的实现过程。资源包共56个文件约23.66MB以34张PNG图片、10个Python源码文件、8个GZIP压缩包和4个文本文件为主图片多用于展示算法运行结果与可视化图表源码则覆盖核心算法实现压缩包用于存放数据集等较大文件。目录按模块划分涵盖深度学习、智能优化算法与搜索算法三大方向其中深度学习部分包含BP神经网络与CNN的完整实现及MNIST数据智能优化算法涵盖PSO、ACO、GA等经典算法搜索算法则提供BFS、DFS与Astar的实现与测试用例便于对照学习与二次开发。目前已有375人学习下载适合希望借助现成代码快速上手AI算法实践、完成课程作业或开展教学演示的读者参考使用。1. 从一份 Python AI 作业辅助源码说起它到底能帮你省下哪几个小时如果你手里正躺着一份「基于 Python 语言的 AI 作业辅助设计源码」大概率不是想研究它写得多优雅而是想知道这东西能不能跑起来、能不能接上大模型、能不能真的把批改和答疑的活儿接过去。我最初接触这类项目是因为带的一批学生每周要交几十份编程作业人工看一遍要三四个小时重复错误还老漏。后来我把这套思路拆开重做了一遍用 Python 做调度层用大模型做语义判断用规则做兜底最后落成一个能本地跑、能换模型、能记录每次判断依据的小系统。它解决的不是「AI 替你写作业」而是「AI 帮你把作业里的结构化信息抽出来、把重复判断做掉、把需要人看的留给人」。适合谁适合要批量处理文本类作业的老师、助教也适合想拿一个真实场景练手 Python AI 工程化的开发者。源码本身只是起点真正值钱的是它背后那条「输入清洗 → 提示词约束 → 模型调用 → 结果校验 → 人工复核」的链路。下面我按这条链路把能复现的部分一段段拆开讲。2. 先想清楚 AI 作业辅助的边界哪些判断交给模型哪些必须留给规则2.1 作业辅助系统的三层结构解析层、判断层、复核层一份能用的 AI 作业辅助源码不管它包装成 Web 还是命令行内部基本都逃不出三层。第一层是解析层负责把作业文件txt、md、docx、pdf、甚至截图 OCR 后的文本读成纯文本并按题号切成一段段。第二层是判断层把每段文本连同评分标准一起送进大模型让它输出结构化结果比如得分点命中情况、错误类型、建议评语。第三层是复核层把模型输出里置信度低、格式异常、或者命中敏感规则的部分标出来交回给人。很多人一上来就把所有东西塞给模型结果就是同一份作业两次跑出两个分数或者模型开始自由发挥写小作文。我的做法是解析层尽量用确定性代码判断层用模型但强约束输出格式复核层用规则做二次过滤。这样即使模型抽风系统也不会整体崩掉。源码里如果只有一次模型调用、没有复核层那它更适合当 demo不适合直接上生产。2.2 为什么不能全交给大模型三个真实翻车场景第一个场景学生答案里写了「我不确定但我觉得应该是 A」模型直接判对因为它读出了「A」。第二个场景评分标准里写「答出任意两点得满分」模型把「任意两点」理解成「必须两点」多扣了分。第三个场景同一道题换了个问法模型给出的错误类型从「概念混淆」变成「计算失误」导致统计报表完全没法用。这三个坑的共同原因是模型在做语义理解时会把「不确定」「任意」「换问法」这些人类觉得理所当然的弹性处理成硬性逻辑。所以我的原则是凡是能用规则判的比如关键词是否出现、代码是否能跑、格式是否合规绝不交给模型凡是需要理解语义的比如论述是否切题、代码逻辑是否等价才交给模型并且要求它给出判断依据。源码里如果有一份rules.yaml或类似的规则配置文件说明作者想过这件事如果没有你得自己补。2.3 选型本地模型还是 APIPython 版本和依赖怎么定选型这块没有标准答案但有几个硬指标。如果你处理的是学生作业数据不能出本地那就走本地部署常见做法是用 Ollama 或类似工具拉一个 7B 到 14B 的模型Python 侧用requests或官方 SDK 调本地端口。如果你处理的是公开数据集或者脱敏后的文本用 API 更省事延迟低、效果好但要注意调用频率和费用。Python 版本我一般锁 3.10 或 3.11太新的版本有些库还没跟上太老的版本类型提示写起来难受。依赖方面最小集合是requests调模型、pydantic约束输出结构、python-docx和pdfplumber读文件、pandas做统计。如果你要用异步并发加httpx和asyncio。不要一上来就上 LangChain这类作业辅助场景链路短自己写调度反而更可控。源码里如果依赖了几十个包先看它是不是把简单问题复杂化了。3. 把源码跑起来环境配置、模型接入和最小可运行命令3.1 环境准备Python 安装、虚拟环境和依赖清单不管你用 Windows 还是 Linux第一步都是把 Python 环境弄干净。我见过太多人因为系统里同时装了三个 Python 版本最后pip install装到了错误的位置。下面这套命令在 Windows 和 Linux 上都能用区别只是激活虚拟环境的命令不同。# 确认当前 Python 版本建议 3.10 或 3.11 python --version # 在项目根目录创建虚拟环境 python -m venv .venv # Windows 激活 .venv\Scripts\activate # Linux / macOS 激活 source .venv/bin/activate # 升级 pip 并安装依赖 python -m pip install --upgrade pip pip install requests pydantic python-docx pdfplumber pandas这段命令的逻辑是先确认版本再建隔离环境最后装最小依赖。参数说明--upgrade pip是为了避免老版本 pip 解析依赖时出玄学问题pydantic后面会用来定义模型输出的结构不装它你就得自己写一堆if来校验字段。如果你用的是 VSCode记得在右下角把解释器切到.venv里的那个否则终端跑通了、编辑器里还是报红。3.2 接入大模型本地端口和 API 两种方式的配置写法模型接入这块我一般把配置抽到一个config.py里方便切换。下面是一个同时支持本地和 API 的写法用环境变量控制走哪条路。# config.py import os # 模式选择local 或 api MODE os.getenv(AI_MODE, local) # 本地模型配置 LOCAL_BASE_URL os.getenv(LOCAL_BASE_URL, http://127.0.0.1:11434) LOCAL_MODEL os.getenv(LOCAL_MODEL, qwen2.5:7b) # API 配置 API_BASE_URL os.getenv(API_BASE_URL, https://api.example.com/v1) API_KEY os.getenv(API_KEY, ) API_MODEL os.getenv(API_MODEL, gpt-4o-mini) # 通用参数 TEMPERATURE 0.1 # 作业判断要稳定温度调低 MAX_TOKENS 1024 TIMEOUT 60逻辑说明MODE决定后面调用哪个函数TEMPERATURE设成 0.1 是因为作业判断需要可复现温度高了同一份作业两次跑出不同结果复核层会疯掉。参数说明LOCAL_BASE_URL默认指向本地常见端口你换成自己部署的地址即可API_KEY不要硬编码在代码里用环境变量注入。如果你的源码里直接把 key 写在.py文件里第一件事就是把它挪出去。3.3 最小可运行示例读一份作业、调一次模型、打印结构化结果下面这段代码是整个系统的最小闭环读文本、拼提示词、调模型、解析 JSON 输出。你可以直接拿它当起点再往里面加规则和复核。# minimal_run.py import json import requests from config import ( MODE, LOCAL_BASE_URL, LOCAL_MODEL, API_BASE_URL, API_KEY, API_MODEL, TEMPERATURE, MAX_TOKENS, TIMEOUT ) PROMPT_TEMPLATE 你是一个作业辅助判断助手。请根据评分标准判断学生答案。 评分标准{rubric} 学生答案{answer} 请严格输出 JSON格式如下 {{hit_points: [命中的得分点], miss_points: [遗漏的得分点], error_type: 错误类型或none, comment: 简短评语}} 不要输出 JSON 以外的任何内容。 def call_model(prompt: str) - str: if MODE local: url f{LOCAL_BASE_URL}/api/generate payload { model: LOCAL_MODEL, prompt: prompt, stream: False, options: {temperature: TEMPERATURE, num_predict: MAX_TOKENS} } resp requests.post(url, jsonpayload, timeoutTIMEOUT) return resp.json().get(response, ) else: url f{API_BASE_URL}/chat/completions headers {Authorization: fBearer {API_KEY}} payload { model: API_MODEL, messages: [{role: user, content: prompt}], temperature: TEMPERATURE, max_tokens: MAX_TOKENS } resp requests.post(url, jsonpayload, headersheaders, timeoutTIMEOUT) return resp.json()[choices][0][message][content] def judge(rubric: str, answer: str) - dict: prompt PROMPT_TEMPLATE.format(rubricrubric, answeranswer) raw call_model(prompt) # 模型有时会包一层 markdown 代码块这里做一次清洗 raw raw.strip().removeprefix(json).removeprefix().removesuffix().strip() try: return json.loads(raw) except json.JSONDecodeError: return {hit_points: [], miss_points: [], error_type: parse_error, comment: raw[:200]} if __name__ __main__: rubric 答出变量作用域得 5 分答出闭包得 5 分。 answer 变量作用域决定了变量在哪里能被访问闭包是函数记住外部变量的能力。 result judge(rubric, answer) print(json.dumps(result, ensure_asciiFalse, indent2))逻辑说明PROMPT_TEMPLATE里把评分标准和学生答案分开传入避免模型混淆call_model根据MODE走本地或 APIjudge里做了一次 markdown 清洗因为很多模型习惯把 JSON 包在代码块里不清洗直接json.loads会翻车。参数说明num_predict是本地模型的输出长度限制max_tokens是 API 侧的限制两者含义类似但字段名不同。如果你跑出来parse_error先看模型原始输出再决定是改提示词还是加清洗规则。4. 提示词和输出结构怎么定让模型稳定吐 JSON 的四个约束4.1 提示词模板把评分标准、学生答案、输出格式分开写提示词写得好不好直接决定系统能不能用。我的经验是不要把所有东西揉成一段话而是用明确的分段标签。上面代码里的模板已经做了这件事但还可以更细。比如评分标准里如果有多个得分点可以编号传入学生答案如果很长可以截断到关键段落再传。常见做法是加一句「如果学生答案中没有明确提到某个得分点不要推测直接算遗漏」这一句能挡掉很多模型自作主张的加分。另外输出格式的约束要放在提示词最后并且用「不要输出 JSON 以外的任何内容」这种强指令。有些模型会加一句「好的以下是结果」你不加这句约束解析层就要多写一堆清洗逻辑。如果你的源码里提示词只有一句话那基本可以判断它没怎么调过。4.2 用 Pydantic 做输出校验字段缺失和类型错误的兜底光靠提示词约束还不够模型偶尔会漏字段或者把列表写成字符串。这时候用 Pydantic 做一层校验能把问题挡在业务逻辑之前。# schema.py from pydantic import BaseModel, Field, field_validator from typing import List class JudgeResult(BaseModel): hit_points: List[str] Field(default_factorylist) miss_points: List[str] Field(default_factorylist) error_type: str none comment: str field_validator(hit_points, miss_points, modebefore) classmethod def ensure_list(cls, v): if isinstance(v, str): return [v] if v else [] return v or [] field_validator(error_type, comment, modebefore) classmethod def ensure_str(cls, v): return str(v) if v is not None else 逻辑说明ensure_list把模型可能返回的字符串转成列表ensure_str把None转成空字符串。参数说明modebefore表示在 Pydantic 默认校验之前执行这样即使模型返回null也不会直接抛异常。加上这层之后你的主流程只需要处理JudgeResult对象不用再写if hit_points in result这种防御代码。4.3 温度、最大长度、重试次数三个必须调的参数温度前面说了作业判断场景建议 0.1 到 0.3再高就不稳定。最大长度要根据你的评分标准长度来定一般 512 到 1024 够用太长模型容易在末尾加废话。重试次数我一般设 2 次第一次解析失败后把原始输出和错误信息拼成新提示词再问一次让模型自己修正格式。如果两次都失败直接标记为parse_error交给人工不要无限重试。这三个参数在源码里如果写死在函数内部建议你抽出来放到配置文件。因为不同科目、不同题型的作业对稳定性和长度的要求不一样写死了后面改起来很痛苦。5. 避坑与排查跑这份源码时最容易翻车的五个地方5.1 现象模型返回空字符串日志里没有任何报错原因通常是本地模型端口没通或者 API key 失效但请求库没抛异常。解决在call_model里加一行日志打印resp.status_code和resp.text[:200]先确认请求到底有没有发出去。如果是本地模型用curl直接打一下端口看服务是否在跑。5.2 现象同一份作业两次跑得分点命中数量不一样原因是温度设高了或者提示词里没有强约束「不要推测」。解决把温度降到 0.1并在提示词里加一句「只根据学生答案中明确出现的内容判断不要根据常识补充」。如果还不行检查模型本身是不是随机性太强换一个更稳定的模型。5.3 现象JSON 解析失败报Expecting value: line 1 column 1原因是模型输出里带了 markdown 代码块、前缀说明或者多余换行。解决在json.loads之前做清洗去掉json、、以及首尾空白。如果清洗后还失败把原始输出存到文件里看模型到底吐了什么。常见的是模型在 JSON 后面又加了一段解释这时候可以用正则截取第一个{到最后一个}之间的内容。5.4 现象读 docx 或 pdf 时中文乱码或段落丢失原因是python-docx读表格和文本框里的内容方式不同pdfplumber对扫描版 PDF 无能为力。解决docx 要同时遍历paragraphs和tablespdf 先用pdfplumber试如果提取出来是空的说明是扫描版需要走 OCR。源码里如果只写了docx.Document(path).paragraphs表格里的答案就全丢了。5.5 现象并发调用时 API 返回 429或者本地模型卡死原因是请求频率超过了限制或者本地模型同时处理多个请求时显存不够。解决加一个简单的信号量控制并发数API 侧一般 3 到 5 个并发就够本地模型建议串行。如果必须并发用asyncio.Semaphore限制同时进行的请求数并在每次请求后加一个小延迟。6. 进阶把单次判断变成可复用的批处理流水线6.1 批处理脚本遍历文件夹、并发调用、结果落表单次判断跑通之后下一步就是批处理。我一般会写一个batch.py遍历作业文件夹按学生和题号分组并发调用判断函数最后把结果写进 CSV 或 Excel。下面是一个简化版用concurrent.futures控制并发。# batch.py import os import json import pandas as pd from concurrent.futures import ThreadPoolExecutor, as_completed from minimal_run import judge def process_one(file_path: str, rubric: str) - dict: with open(file_path, r, encodingutf-8) as f: answer f.read() result judge(rubric, answer) result[file] os.path.basename(file_path) return result def run_batch(folder: str, rubric: str, max_workers: int 3): files [os.path.join(folder, f) for f in os.listdir(folder) if f.endswith(.txt)] rows [] with ThreadPoolExecutor(max_workersmax_workers) as executor: futures {executor.submit(process_one, fp, rubric): fp for fp in files} for future in as_completed(futures): try: rows.append(future.result()) except Exception as e: rows.append({file: futures[future], error_type: exception, comment: str(e)}) df pd.DataFrame(rows) df.to_csv(batch_result.csv, indexFalse, encodingutf-8-sig) return df if __name__ __main__: rubric 答出变量作用域得 5 分答出闭包得 5 分。 run_batch(./homework, rubric, max_workers3)逻辑说明ThreadPoolExecutor控制并发数as_completed按完成顺序收集结果异常也被捕获成一行记录不会因为单个文件失败中断整批。参数说明max_workers建议设 3 到 5太高会触发 API 限流或本地显存溢出encodingutf-8-sig是为了 Excel 打开 CSV 时中文不乱码。6.2 人工复核队列把低置信度和解析失败单独捞出来批处理跑完不要直接出分。我的习惯是加一列need_review把error_type是parse_error、exception或者hit_points和miss_points都为空的行标出来单独导一份review.csv。复核的时候只看这些行工作量能降一大半。这一步在源码里如果没做你自己加一个过滤条件就行成本很低但收益很高。6.3 我踩过的一个坑别在提示词里写「请尽量准确」最后说一个血泪经验。我早期在提示词里写过「请尽量准确判断」结果模型开始过度推理把学生没写的得分点也补上了理由是「他应该知道」。后来我把这类模糊指令全删了改成「只根据学生答案中明确出现的内容判断没有出现的一律算遗漏」。这一改准确率反而上去了。模型不需要你鼓励它努力它需要你告诉它边界在哪。希望帮到你。本文还有配套的精品资源点击获取
返回列表