242 lines
12 KiB
Python
242 lines
12 KiB
Python
"""初中数学作业批改工作流状态定义 - 支持多学生多图片批改"""
|
||
from typing import List, Optional, Literal
|
||
from pydantic import BaseModel, Field
|
||
from utils.file.file import File
|
||
|
||
|
||
# === 基础数据结构 ===
|
||
|
||
class StudentHomework(BaseModel):
|
||
"""学生作业信息"""
|
||
student_id: int = Field(..., description="学生ID")
|
||
student_name: str = Field(default="", description="学生姓名")
|
||
homework_images: List[str] = Field(default=[], description="该学生的作业图片URL列表")
|
||
|
||
|
||
class ImageInfo(BaseModel):
|
||
"""图片信息"""
|
||
width: int = Field(..., description="图片宽度（像素）")
|
||
height: int = Field(..., description="图片高度（像素）")
|
||
dpi: int = Field(default=72, description="图片DPI")
|
||
|
||
|
||
class MarkPosition(BaseModel):
|
||
"""批改标记位置"""
|
||
x: int = Field(..., description="标记中心X坐标（绝对像素）")
|
||
y: int = Field(..., description="标记中心Y坐标（绝对像素）")
|
||
|
||
|
||
class QuestionItem(BaseModel):
|
||
"""题目项（一体化识别结果）"""
|
||
question_id: str = Field(..., description="题号（如10、10.1、10.2）")
|
||
parent_id: str = Field(default="", description="母题题号（子题时填写）")
|
||
is_sub_question: bool = Field(default=False, description="是否为子题")
|
||
question_text: str = Field(default="", description="题目内容")
|
||
student_answer: str = Field(default="", description="学生答案内容")
|
||
answer_bbox: List[int] = Field(..., description="学生答案区域的边界框 [x1, y1, x2, y2]（用于定位标注）")
|
||
mark_position: MarkPosition = Field(..., description="批改标记应该标注的位置坐标")
|
||
full_score: int = Field(default=10, description="满分")
|
||
|
||
|
||
class CorrectAnswer(BaseModel):
|
||
"""标准答案项（从Word文档解析）"""
|
||
question_id: str = Field(..., description="题号（如10、10.1、10.2）")
|
||
parent_id: str = Field(default="", description="母题题号（子题时填写）")
|
||
is_sub_question: bool = Field(default=False, description="是否为子题")
|
||
question_text: str = Field(default="", description="题目内容/题干")
|
||
correct_answer: str = Field(..., description="标准答案")
|
||
full_score: int = Field(default=10, description="满分")
|
||
answer_analysis: str = Field(default="", description="答案解析/解题步骤")
|
||
|
||
|
||
class CorrectionResult(BaseModel):
|
||
"""批改结果"""
|
||
question_id: str = Field(..., description="题号")
|
||
parent_id: str = Field(default="", description="母题题号")
|
||
status: Literal["correct", "incorrect", "partial"] = Field(..., description="批改状态")
|
||
score: int = Field(default=0, description="得分")
|
||
full_score: int = Field(default=10, description="满分")
|
||
comment: str = Field(default="", description="批改评语")
|
||
|
||
|
||
class Annotation(BaseModel):
|
||
"""完整批注"""
|
||
question_id: str = Field(..., description="题号")
|
||
parent_id: str = Field(default="", description="母题题号")
|
||
status: Literal["correct", "incorrect", "partial"] = Field(..., description="批改状态")
|
||
question_text: str = Field(default="", description="题目内容")
|
||
student_answer: str = Field(default="", description="学生答案")
|
||
answer_bbox: List[int] = Field(default=[], description="答案区域边界框")
|
||
comment: str = Field(default="", description="批改评语")
|
||
mark_position: MarkPosition = Field(..., description="批改标记位置")
|
||
score: int = Field(default=0, description="得分")
|
||
full_score: int = Field(default=10, description="满分")
|
||
|
||
|
||
class SingleImageResult(BaseModel):
|
||
"""单张图片的批改结果"""
|
||
image_index: int = Field(..., description="图片索引（从0开始）")
|
||
image_info: ImageInfo = Field(..., description="图片信息")
|
||
image_url: str = Field(default="", description="处理后的图片URL")
|
||
annotations: List[Annotation] = Field(default=[], description="该图片的批注列表")
|
||
|
||
|
||
class StudentResult(BaseModel):
|
||
"""单个学生的批改结果"""
|
||
student_id: int = Field(..., description="学生ID")
|
||
student_name: str = Field(default="", description="学生姓名")
|
||
total_images: int = Field(..., description="该学生的总图片数")
|
||
image_results: List[SingleImageResult] = Field(default=[], description="各图片的批改结果")
|
||
overall_comment: str = Field(default="", description="该学生的整体评价")
|
||
total_score: int = Field(default=0, description="该学生的总分")
|
||
full_score: int = Field(default=100, description="该学生的满分")
|
||
grade: str = Field(default="", description="该学生的等级")
|
||
|
||
|
||
# === 全局状态 ===
|
||
class GlobalState(BaseModel):
|
||
"""工作流全局状态"""
|
||
# 输入参数
|
||
student_homework: List[StudentHomework] = Field(default=[], description="学生作业列表")
|
||
answer_doc_url: str = Field(default="", description="正确答案Word文件的URL（.docx格式）")
|
||
comment_max_length: int = Field(default=100, description="评语最大字数")
|
||
max_concurrent: int = Field(default=10, description="并行批改的最大数量")
|
||
grade_standards: dict = Field(default={}, description="评价等级标准")
|
||
# 中间状态
|
||
correct_answers: List[CorrectAnswer] = Field(default=[], description="从Word解析的标准答案列表")
|
||
# 最终结果
|
||
student_results: List[StudentResult] = Field(default=[], description="各学生的批改结果列表")
|
||
|
||
|
||
# === 图输入输出 ===
|
||
class GraphInput(BaseModel):
|
||
"""工作流输入"""
|
||
student_homework: List[StudentHomework] = Field(..., description="学生作业列表，每个学生包含ID和作业图片")
|
||
answer_doc_url: str = Field(default="", description="正确答案Word文件的URL（.docx格式，可选）")
|
||
comment_max_length: int = Field(default=100, description="评语最大字数，默认100字")
|
||
max_concurrent: int = Field(default=10, description="并行批改的最大数量，默认10")
|
||
grade_standards: dict = Field(
|
||
default={
|
||
"A+": {"min_percentage": 95, "description": "答案全部正确，步骤完整规范，逻辑严谨；书写/格式整洁，无错别字、无遗漏；完成度100%，态度认真，质量上乘"},
|
||
"A": {"min_percentage": 90, "description": "答案完全正确，无任何错误；步骤合理、格式规范，无原则性问题；完成度100%，满足全部要求"},
|
||
"B": {"min_percentage": 80, "description": "存在少量非关键性错误，或步骤略有缺失；整体思路基本正确，仅细节、格式、计算等小问题；完成大部分内容，整体合格但不够严谨"},
|
||
"C": {"min_percentage": 70, "description": "错误较多，部分核心题目作答错误；步骤不完整、逻辑不够清晰；完成度一般，有明显应付、漏答情况"},
|
||
"D": {"min_percentage": 0, "description": "大面积错误，核心知识点未掌握；大量空白、敷衍、抄袭；未达到基本完成要求"}
|
||
},
|
||
description="评价等级标准，包含各等级的最低得分率百分比和描述"
|
||
)
|
||
|
||
|
||
class GraphOutput(BaseModel):
|
||
"""工作流输出"""
|
||
student_results: List[StudentResult] = Field(..., description="各学生的批改结果列表")
|
||
|
||
|
||
# === 文档答案解析节点 ===
|
||
class DocExtractInput(BaseModel):
|
||
"""文档答案解析节点输入"""
|
||
answer_doc_url: str = Field(default="", description="正确答案Word文件的URL（.docx格式，可选）")
|
||
|
||
|
||
class DocExtractOutput(BaseModel):
|
||
"""文档答案解析节点输出"""
|
||
correct_answers: List[CorrectAnswer] = Field(default=[], description="解析的标准答案列表")
|
||
|
||
|
||
# === 子图状态定义（单图片处理流程） ===
|
||
class SubgraphState(BaseModel):
|
||
"""子图全局状态（处理单张图片）"""
|
||
# 输入
|
||
homework_image: File = Field(..., description="当前处理的作业图片")
|
||
correct_answers: List[CorrectAnswer] = Field(default=[], description="标准答案列表")
|
||
image_index: int = Field(default=0, description="图片索引")
|
||
comment_max_length: int = Field(default=100, description="评语最大字数")
|
||
# 中间状态
|
||
image_info: ImageInfo = Field(default_factory=lambda: ImageInfo(width=0, height=0, dpi=72), description="图片信息")
|
||
image_url: str = Field(default="", description="处理后的图片URL")
|
||
question_items: List[QuestionItem] = Field(default=[], description="识别的题目项列表")
|
||
correction_results: List[CorrectionResult] = Field(default=[], description="批改结果列表")
|
||
annotations: List[Annotation] = Field(default=[], description="最终批注列表")
|
||
|
||
|
||
class SubgraphInput(BaseModel):
|
||
"""子图输入"""
|
||
homework_image: File = Field(..., description="当前处理的作业图片")
|
||
correct_answers: List[CorrectAnswer] = Field(default=[], description="标准答案列表")
|
||
image_index: int = Field(default=0, description="图片索引")
|
||
comment_max_length: int = Field(default=100, description="评语最大字数")
|
||
|
||
|
||
class SubgraphOutput(BaseModel):
|
||
"""子图输出"""
|
||
image_result: SingleImageResult = Field(..., description="单张图片的批改结果")
|
||
|
||
|
||
# === 子图节点输入输出 ===
|
||
|
||
# 1. 图像预处理节点
|
||
class ImagePreprocessInput(BaseModel):
|
||
"""图像预处理节点输入"""
|
||
homework_image: File = Field(..., description="上传的作业图片")
|
||
|
||
|
||
class ImagePreprocessOutput(BaseModel):
|
||
"""图像预处理节点输出"""
|
||
image_info: ImageInfo = Field(..., description="图片信息")
|
||
image_url: str = Field(..., description="处理后的图片URL")
|
||
|
||
|
||
# 2. 一体化识别批改节点（合并识别+批改）
|
||
class RecognizeAndCorrectInput(BaseModel):
|
||
"""一体化识别批改节点输入"""
|
||
image_url: str = Field(..., description="图片URL")
|
||
image_info: ImageInfo = Field(..., description="图片信息")
|
||
correct_answers: List[CorrectAnswer] = Field(default=[], description="标准答案列表")
|
||
comment_max_length: int = Field(default=100, description="评语最大字数")
|
||
|
||
|
||
class RecognizeAndCorrectOutput(BaseModel):
|
||
"""一体化识别批改节点输出"""
|
||
question_items: List[QuestionItem] = Field(default=[], description="识别的题目项列表")
|
||
correction_results: List[CorrectionResult] = Field(default=[], description="批改结果列表")
|
||
|
||
|
||
# 3. 批改判断节点
|
||
class CorrectionJudgeInput(BaseModel):
|
||
"""批改判断节点输入"""
|
||
question_items: List[QuestionItem] = Field(default=[], description="题目项列表")
|
||
correct_answers: List[CorrectAnswer] = Field(default=[], description="标准答案列表")
|
||
comment_max_length: int = Field(default=100, description="评语最大字数")
|
||
|
||
|
||
class CorrectionJudgeOutput(BaseModel):
|
||
"""批改判断节点输出"""
|
||
correction_results: List[CorrectionResult] = Field(default=[], description="批改结果列表")
|
||
|
||
|
||
# 4. 结果整合节点
|
||
class ResultMergeInput(BaseModel):
|
||
"""结果整合节点输入"""
|
||
question_items: List[QuestionItem] = Field(default=[], description="题目项列表")
|
||
correction_results: List[CorrectionResult] = Field(default=[], description="批改结果列表")
|
||
|
||
|
||
class ResultMergeOutput(BaseModel):
|
||
"""结果整合节点输出"""
|
||
annotations: List[Annotation] = Field(default=[], description="最终批注列表")
|
||
|
||
|
||
# === 循环节点 ===
|
||
class ProcessImagesInput(BaseModel):
|
||
"""多学生作业处理循环节点输入"""
|
||
student_homework: List[StudentHomework] = Field(default=[], description="学生作业列表")
|
||
correct_answers: List[CorrectAnswer] = Field(default=[], description="标准答案列表")
|
||
comment_max_length: int = Field(default=100, description="评语最大字数")
|
||
max_concurrent: int = Field(default=10, description="并行批改的最大数量")
|
||
grade_standards: dict = Field(default={}, description="评价等级标准")
|
||
|
||
|
||
class ProcessImagesOutput(BaseModel):
|
||
"""多学生作业处理循环节点输出"""
|
||
student_results: List[StudentResult] = Field(..., description="各学生的批改结果列表")
|