小学标准化试卷AI批改Agent最佳工程实践

试卷图像 → 预处理层 → 识别层 → 理解层 → 评分层 → 结果输出
           ↓         ↓        ↓        ↓
         图像增强   OCR识别  语义理解  规则引擎
         定位切分   手写识别  答案匹配  置信度评估
         质量检测   字符识别  纠错容错  异常处理

import cv2
import numpy as np
from typing import Tuple, List, Dict
from dataclasses import dataclass

@dataclass
class ExamPage:
    """试卷页面数据结构"""
    image: np.ndarray
    page_number: int
    quality_score: float
    answer_regions: List[Dict]

class ExamPreprocessor:
    """试卷预处理器"""

    def __init__(self, config: Dict):
        self.config = config
        self.target_dpi = config.get('target_dpi', 300)
        self.quality_threshold = config.get('quality_threshold', 0.7)

    def process(self, image_path: str) -> ExamPage:
        """
        预处理试卷图像

        Args:
            image_path: 试卷图像路径

        Returns:
            ExamPage: 处理后的试卷页面对象
        """
        # 1. 读取图像
        image = cv2.imread(image_path)

        # 2. 图像增强
        enhanced = self._enhance_image(image)

        # 3. 倾斜矫正
        corrected = self._correct_skew(enhanced)

        # 4. 去噪处理
        denoised = self._denoise(corrected)

        # 5. 二值化
        binary = self._binarize(denoised)

        # 6. 质量评估
        quality = self._assess_quality(binary)

        # 7. 答题区域定位
        regions = self._locate_answer_regions(binary)

        return ExamPage(
            image=binary,
            page_number=1,
            quality_score=quality,
            answer_regions=regions
        )

    def _enhance_image(self, image: np.ndarray) -> np.ndarray:
        """图像增强:对比度、亮度调整"""
        # 转换到LAB色彩空间
        lab = cv2.cvtColor(image, cv2.COLOR_BGR2LAB)
        l, a, b = cv2.split(lab)

        # 应用CLAHE(对比度受限的自适应直方图均衡)
        clahe = cv2.createCLAHE(clipLimit=3.0, tileGridSize=(8, 8))
        enhanced_l = clahe.apply(l)

        # 合并通道
        enhanced_lab = cv2.merge([enhanced_l, a, b])
        enhanced = cv2.cvtColor(enhanced_lab, cv2.COLOR_LAB2BGR)

        return enhanced

    def _correct_skew(self, image: np.ndarray) -> np.ndarray:
        """倾斜矫正:确保试卷水平"""
        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)

        # 边缘检测
        edges = cv2.Canny(gray, 50, 150, apertureSize=3)

        # 霍夫变换检测直线
        lines = cv2.HoughLines(edges, 1, np.pi/180, 200)

        if lines is None:
            return image

        # 计算倾斜角度
        angles = []
        for rho, theta in lines[:, 0]:
            angle = (theta * 180 / np.pi) - 90
            if abs(angle) < 45:
                angles.append(angle)

        if not angles:
            return image

        # 使用中位数角度进行旋转
        median_angle = np.median(angles)

        # 旋转图像
        (h, w) = image.shape[:2]
        center = (w // 2, h // 2)
        M = cv2.getRotationMatrix2D(center, median_angle, 1.0)
        corrected = cv2.warpAffine(image, M, (w, h),
                                    flags=cv2.INTER_CUBIC,
                                    borderMode=cv2.BORDER_REPLICATE)

        return corrected

    def _denoise(self, image: np.ndarray) -> np.ndarray:
        """去噪:移除扫描噪点"""
        # 使用双边滤波保留边缘的同时去噪
        denoised = cv2.bilateralFilter(image, 9, 75, 75)
        return denoised

    def _binarize(self, image: np.ndarray) -> np.ndarray:
        """自适应二值化"""
        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)

        # 自适应阈值二值化
        binary = cv2.adaptiveThreshold(
            gray, 255,
            cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
            cv2.THRESH_BINARY,
            11, 2
        )

        return binary

    def _assess_quality(self, image: np.ndarray) -> float:
        """
        评估图像质量

        Returns:
            质量分数 0-1,越高越好
        """
        # 计算清晰度(拉普拉斯方差)
        laplacian = cv2.Laplacian(image, cv2.CV_64F)
        sharpness = laplacian.var()

        # 归一化到0-1
        quality = min(sharpness / 500, 1.0)

        return quality

    def _locate_answer_regions(self, image: np.ndarray) -> List[Dict]:
        """
        定位答题区域

        使用模板匹配或深度学习检测答题框
        """
        regions = []

        # 查找轮廓
        contours, _ = cv2.findContours(
            image, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE
        )

        for i, contour in enumerate(contours):
            # 计算边界框
            x, y, w, h = cv2.boundingRect(contour)

            # 过滤太小的区域
            if w < 50 or h < 20:
                continue

            # 提取答题区域
            region = {
                'id': i,
                'bbox': (x, y, w, h),
                'area': w * h,
                'aspect_ratio': w / h,
                'image': image[y:y+h, x:x+w]
            }

            regions.append(region)

        # 按照从上到下、从左到右排序
        regions.sort(key=lambda r: (r['bbox'][1], r['bbox'][0]))

        return regions

# generated by AI

from paddleocr import PaddleOCR
import torch
from transformers import TrOCRProcessor, VisionEncoderDecoderModel
from typing import List, Tuple

class HandwritingRecognizer:
    """手写文字识别器"""

    def __init__(self, model_type: str = 'paddle'):
        """
        初始化识别器

        Args:
            model_type: 'paddle' 或 'trocr'
        """
        self.model_type = model_type

        if model_type == 'paddle':
            # PaddleOCR对中文手写识别效果好
            self.ocr = PaddleOCR(
                use_angle_cls=True,
                lang='ch',
                use_gpu=torch.cuda.is_available()
            )
        elif model_type == 'trocr':
            # TrOCR对英文和数字识别效果好
            self.processor = TrOCRProcessor.from_pretrained(
                'microsoft/trocr-base-handwritten'
            )
            self.model = VisionEncoderDecoderModel.from_pretrained(
                'microsoft/trocr-base-handwritten'
            )

    def recognize(self, image: np.ndarray) -> List[Dict]:
        """
        识别图像中的手写文字

        Args:
            image: 输入图像

        Returns:
            识别结果列表,每个元素包含文字和置信度
        """
        if self.model_type == 'paddle':
            return self._recognize_with_paddle(image)
        else:
            return self._recognize_with_trocr(image)

    def _recognize_with_paddle(self, image: np.ndarray) -> List[Dict]:
        """使用PaddleOCR识别"""
        results = self.ocr.ocr(image, cls=True)

        recognized = []
        for line in results[0]:
            box, (text, confidence) = line
            recognized.append({
                'text': text,
                'confidence': confidence,
                'bbox': box
            })

        return recognized

    def _recognize_with_trocr(self, image: np.ndarray) -> List[Dict]:
        """使用TrOCR识别"""
        from PIL import Image

        # 转换为PIL Image
        pil_image = Image.fromarray(cv2.cvtColor(image, cv2.COLOR_BGR2RGB))

        # 预处理
        pixel_values = self.processor(pil_image, return_tensors='pt').pixel_values

        # 生成文本
        generated_ids = self.model.generate(pixel_values)
        text = self.processor.batch_decode(generated_ids, skip_special_tokens=True)[0]

        return [{
            'text': text,
            'confidence': 0.9,  # TrOCR不直接提供置信度
            'bbox': None
        }]

class AnswerExtractor:
    """答案提取器:结合识别器和区域定位"""

    def __init__(self):
        self.recognizer = HandwritingRecognizer(model_type='paddle')

    def extract(self, exam_page: ExamPage, question_config: Dict) -> Dict[int, str]:
        """
        从试卷页面提取所有答案

        Args:
            exam_page: 预处理后的试卷页面
            question_config: 题目配置(题号、答题区域映射)

        Returns:
            题号到答案的映射
        """
        answers = {}

        for region in exam_page.answer_regions:
            # 识别该区域的文字
            recognized = self.recognizer.recognize(region['image'])

            # 合并识别结果
            text = ' '.join([r['text'] for r in recognized])

            # 根据配置匹配题号
            question_id = self._match_question_id(region, question_config)

            if question_id:
                answers[question_id] = {
                    'raw_text': text,
                    'confidence': np.mean([r['confidence'] for r in recognized]),
                    'region_id': region['id']
                }

        return answers

    def _match_question_id(self, region: Dict, config: Dict) -> int:
        """根据区域位置匹配题号"""
        # 简化版:根据区域顺序映射题号
        # 实际应该用更复杂的模板匹配或坐标映射
        return region['id'] + 1

# generated by AI

from anthropic import Anthropic
from typing import List, Dict, Tuple
import json

class AnswerUnderstandingAgent:
    """答案理解Agent:使用LLM理解答案含义"""

    def __init__(self, api_key: str):
        self.client = Anthropic(api_key=api_key)
        self.cache = {}  # 缓存相似答案的理解结果

    async def understand(
        self,
        student_answer: str,
        standard_answer: str,
        question_text: str,
        question_type: str
    ) -> Dict:
        """
        理解学生答案是否正确

        Args:
            student_answer: 学生答案
            standard_answer: 标准答案
            question_text: 题目内容
            question_type: 题型(choice/fill_blank/true_false/subjective)

        Returns:
            理解结果:{is_correct, score, reason, confidence}
        """
        # 检查缓存
        cache_key = f"{question_text}:{student_answer}"
        if cache_key in self.cache:
            return self.cache[cache_key]

        # 根据题型选择理解策略
        if question_type in ['choice', 'true_false']:
            result = self._exact_match(student_answer, standard_answer)
        elif question_type == 'fill_blank':
            result = await self._fuzzy_match_with_llm(
                student_answer, standard_answer, question_text
            )
        elif question_type == 'subjective':
            result = await self._semantic_match_with_llm(
                student_answer, standard_answer, question_text
            )
        else:
            result = {'is_correct': False, 'score': 0, 'reason': 'Unknown type'}

        # 缓存结果
        self.cache[cache_key] = result
        return result

    def _exact_match(self, student: str, standard: str) -> Dict:
        """精确匹配:选择题、判断题"""
        # 归一化处理
        student_norm = self._normalize_answer(student)
        standard_norm = self._normalize_answer(standard)

        is_correct = student_norm == standard_norm

        return {
            'is_correct': is_correct,
            'score': 1.0 if is_correct else 0.0,
            'reason': 'Exact match' if is_correct else 'Does not match',
            'confidence': 1.0
        }

    def _normalize_answer(self, answer: str) -> str:
        """答案归一化"""
        # 转小写、去空格、去标点
        import re
        normalized = answer.lower().strip()
        normalized = re.sub(r'[^\w\s]', '', normalized)
        normalized = re.sub(r'\s+', '', normalized)
        return normalized

    async def _fuzzy_match_with_llm(
        self,
        student: str,
        standard: str,
        question: str
    ) -> Dict:
        """
        模糊匹配:填空题
        允许同义词、不同表达方式
        """
        prompt = f"""你是一个专业的小学试卷批改助手。请判断学生的答案是否正确。

题目: {question}
标准答案: {standard}
学生答案: {student}

判断标准:
1. 如果含义完全一致,即使表达略有不同也算正确
2. 同义词算正确(如"很大"和"非常大")
3. 允许合理的单位、标点符号差异
4. 拼写错误但不影响理解的算部分正确

请返回JSON格式:
{{
    "is_correct": true/false,
    "score": 0.0-1.0,
    "reason": "判断理由",
    "confidence": 0.0-1.0
}}"""

        message = self.client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}]
        )

        response_text = message.content[0].text
        result = json.loads(response_text)

        return result

    async def _semantic_match_with_llm(
        self,
        student: str,
        standard: str,
        question: str
    ) -> Dict:
        """
        语义匹配:主观题
        需要理解答案的深层含义
        """
        prompt = f"""你是一个专业的小学试卷批改助手。请对学生的主观题答案进行评分。

题目: {question}
标准答案: {standard}
学生答案: {student}

评分标准:
1. 完全答对所有要点: 1.0分
2. 答对主要要点,细节略有不足: 0.7-0.9分
3. 部分答对,有明显遗漏: 0.4-0.6分
4. 基本偏离主题: 0.1-0.3分
5. 完全错误或空白: 0分

请返回JSON格式:
{{
    "is_correct": true/false,
    "score": 0.0-1.0,
    "reason": "详细的评分理由,指出答对和答错的点",
    "confidence": 0.0-1.0,
    "missing_points": ["遗漏的要点1", "遗漏的要点2"]
}}"""

        message = self.client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=2048,
            messages=[{"role": "user", "content": prompt}]
        )

        response_text = message.content[0].text
        result = json.loads(response_text)

        return result

# generated by AI

from typing import List, Dict
from enum import Enum

class GradingStatus(Enum):
    """批改状态"""
    AUTO_GRADED = "auto_graded"  # 自动批改成功
    LOW_CONFIDENCE = "low_confidence"  # 置信度不足
    NEED_MANUAL_REVIEW = "need_manual_review"  # 需要人工复核
    ERROR = "error"  # 批改失败

class GradingResult:
    """批改结果"""
    def __init__(
        self,
        question_id: int,
        status: GradingStatus,
        score: float,
        max_score: float,
        confidence: float,
        reason: str = "",
        metadata: Dict = None
    ):
        self.question_id = question_id
        self.status = status
        self.score = score
        self.max_score = max_score
        self.confidence = confidence
        self.reason = reason
        self.metadata = metadata or {}

class ExamGradingAgent:
    """试卷批改Agent:完整流程编排"""

    def __init__(self, config: Dict):
        self.preprocessor = ExamPreprocessor(config)
        self.extractor = AnswerExtractor()
        self.understanding_agent = AnswerUnderstandingAgent(
            api_key=config['anthropic_api_key']
        )
        self.confidence_threshold = config.get('confidence_threshold', 0.75)

    async def grade_exam(
        self,
        image_path: str,
        exam_template: Dict
    ) -> List[GradingResult]:
        """
        批改整份试卷

        Args:
            image_path: 试卷图像路径
            exam_template: 试卷模板(包含题目、标准答案、分值等)

        Returns:
            批改结果列表
        """
        results = []

        try:
            # 1. 预处理
            exam_page = self.preprocessor.process(image_path)

            # 质量检查
            if exam_page.quality_score < 0.5:
                return self._create_error_results(
                    exam_template,
                    "Image quality too low"
                )

            # 2. 提取答案
            student_answers = self.extractor.extract(
                exam_page,
                exam_template['question_config']
            )

            # 3. 逐题批改
            for question in exam_template['questions']:
                qid = question['id']

                # 检查是否提取到该题答案
                if qid not in student_answers:
                    results.append(GradingResult(
                        question_id=qid,
                        status=GradingStatus.NEED_MANUAL_REVIEW,
                        score=0,
                        max_score=question['score'],
                        confidence=0,
                        reason="Answer not detected"
                    ))
                    continue

                student_answer = student_answers[qid]

                # OCR置信度检查
                if student_answer['confidence'] < 0.6:
                    results.append(GradingResult(
                        question_id=qid,
                        status=GradingStatus.LOW_CONFIDENCE,
                        score=0,
                        max_score=question['score'],
                        confidence=student_answer['confidence'],
                        reason="OCR confidence too low",
                        metadata={'raw_text': student_answer['raw_text']}
                    ))
                    continue

                # 4. 答案理解和评分
                understanding = await self.understanding_agent.understand(
                    student_answer=student_answer['raw_text'],
                    standard_answer=question['answer'],
                    question_text=question['text'],
                    question_type=question['type']
                )

                # 综合置信度:OCR置信度 × 理解置信度
                combined_confidence = (
                    student_answer['confidence'] * understanding['confidence']
                )

                # 5. 确定批改状态
                if combined_confidence >= self.confidence_threshold:
                    status = GradingStatus.AUTO_GRADED
                else:
                    status = GradingStatus.LOW_CONFIDENCE

                final_score = understanding['score'] * question['score']

                results.append(GradingResult(
                    question_id=qid,
                    status=status,
                    score=final_score,
                    max_score=question['score'],
                    confidence=combined_confidence,
                    reason=understanding['reason'],
                    metadata={
                        'student_answer': student_answer['raw_text'],
                        'standard_answer': question['answer'],
                        'understanding': understanding
                    }
                ))

            return results

        except Exception as e:
            return self._create_error_results(
                exam_template,
                f"Grading failed: {str(e)}"
            )

    def _create_error_results(
        self,
        exam_template: Dict,
        error_message: str
    ) -> List[GradingResult]:
        """创建错误结果"""
        return [
            GradingResult(
                question_id=q['id'],
                status=GradingStatus.ERROR,
                score=0,
                max_score=q['score'],
                confidence=0,
                reason=error_message
            )
            for q in exam_template['questions']
        ]

    def generate_report(self, results: List[GradingResult]) -> Dict:
        """生成批改报告"""
        total_score = sum(r.score for r in results)
        max_total_score = sum(r.max_score for r in results)

        auto_graded = [r for r in results if r.status == GradingStatus.AUTO_GRADED]
        need_review = [r for r in results
                       if r.status in [GradingStatus.LOW_CONFIDENCE,
                                      GradingStatus.NEED_MANUAL_REVIEW]]

        return {
            'total_score': total_score,
            'max_score': max_total_score,
            'percentage': total_score / max_total_score * 100,
            'auto_graded_count': len(auto_graded),
            'need_review_count': len(need_review),
            'automation_rate': len(auto_graded) / len(results) * 100,
            'avg_confidence': np.mean([r.confidence for r in results]),
            'details': [
                {
                    'question_id': r.question_id,
                    'score': r.score,
                    'max_score': r.max_score,
                    'status': r.status.value,
                    'confidence': r.confidence,
                    'reason': r.reason
                }
                for r in results
            ]
        }

# 使用示例
async def main():
    config = {
        'target_dpi': 300,
        'quality_threshold': 0.7,
        'confidence_threshold': 0.75,
        'anthropic_api_key': 'your-api-key'
    }

    exam_template = {
        'questions': [
            {
                'id': 1,
                'type': 'choice',
                'text': '1+1等于几? A.1 B.2 C.3 D.4',
                'answer': 'B',
                'score': 5
            },
            {
                'id': 2,
                'type': 'fill_blank',
                'text': '中国的首都是____',
                'answer': '北京',
                'score': 5
            }
        ],
        'question_config': {}
    }

    agent = ExamGradingAgent(config)
    results = await agent.grade_exam('exam.jpg', exam_template)
    report = agent.generate_report(results)

    print(f"总分: {report['total_score']}/{report['max_score']}")
    print(f"自动化率: {report['automation_rate']:.1f}%")

# generated by AI

原始数据 → 质量评估 → 增强处理 → 验证 → AI处理
           ↓
        不合格 → 拒绝/预警

物理层 → 感知层 → 理解层 → 决策层
(原始) → (识别) → (语义) → (行动)

def hybrid_processing(data, confidence_threshold=0.75):
    """混合处理策略"""
    # 第一步:规则处理(快速、0成本)
    rule_result = rule_based_process(data)
    if rule_result.confidence >= 0.95:
        return rule_result

    # 第二步:传统ML(中速、低成本)
    ml_result = ml_based_process(data)
    if ml_result.confidence >= confidence_threshold:
        return ml_result

    # 第三步:LLM处理(慢速、高成本但准确)
    llm_result = llm_based_process(data)
    if llm_result.confidence >= confidence_threshold:
        return llm_result

    # 第四步:人工介入
    return human_review_required(data)

# generated by AI

class DataPipeline:
    """带监控的数据处理Pipeline"""

    def __init__(self):
        self.metrics = {
            'quality_distribution': [],
            'confidence_distribution': [],
            'automation_rate': [],
            'error_patterns': {}
        }

    def process(self, data):
        """处理数据并记录指标"""
        # 记录输入数据质量
        quality = assess_quality(data)
        self.metrics['quality_distribution'].append(quality)

        # 处理
        result = self.pipeline.run(data)

        # 记录置信度
        self.metrics['confidence_distribution'].append(result.confidence)

        # 记录自动化率
        is_automated = result.confidence >= self.threshold
        self.metrics['automation_rate'].append(is_automated)

        # 记录错误模式
        if result.error:
            error_type = classify_error(result.error)
            self.metrics['error_patterns'][error_type] = \
                self.metrics['error_patterns'].get(error_type, 0) + 1

        return result

    def analyze_bottlenecks(self):
        """分析瓶颈"""
        # 找出低质量数据的来源
        low_quality = [q for q in self.metrics['quality_distribution'] if q < 0.5]

        # 找出频繁的错误模式
        top_errors = sorted(
            self.metrics['error_patterns'].items(),
            key=lambda x: x[1],
            reverse=True
        )[:5]

        # 分析自动化率趋势
        automation_rate = np.mean(self.metrics['automation_rate'])

        return {
            'low_quality_rate': len(low_quality) / len(self.metrics['quality_distribution']),
            'top_error_patterns': top_errors,
            'automation_rate': automation_rate
        }

# generated by AI