diff --git a/exam_pkg/api.py b/exam_pkg/api.py index 48313e8..ba6e573 100644 --- a/exam_pkg/api.py +++ b/exam_pkg/api.py @@ -61,6 +61,22 @@ def validate_difficulty(difficulty) -> str: return "difficulty 必须为 1-5 的整数" return None + +MAX_EXCLUDE_STEMS = 100 + +def validate_exclude_stems(exclude_stems) -> str: + """校验 exclude_stems 参数（可选）""" + if exclude_stems is None: + return None + if not isinstance(exclude_stems, list): + return "exclude_stems 必须为字符串数组" + if len(exclude_stems) > MAX_EXCLUDE_STEMS: + return f"exclude_stems 数量不能超过 {MAX_EXCLUDE_STEMS}，当前 {len(exclude_stems)}" + for i, stem in enumerate(exclude_stems): + if not isinstance(stem, str) or not stem.strip(): + return f"exclude_stems 第 {i+1} 项必须为非空字符串" + return None + # 创建蓝图 exam_bp = Blueprint('exam', __name__) @@ -86,6 +102,7 @@ def api_generate_questions(): "subjective": 1 }, "difficulty": 3, + "exclude_stems": ["已有题目的题干1", "已有题目的题干2"], // 可选，排除已有题目避免重复 "options": { "include_explanation": true, "max_source_chunks": 50 @@ -135,6 +152,20 @@ def api_generate_questions(): if diff_error: return error_response("INVALID_PARAMS", BAD_REQUEST, diff_error, http_status=400) + # 校验总题数上限 + MAX_TOTAL_QUESTIONS = 20 + total_requested = sum(question_types.values()) + if total_requested > MAX_TOTAL_QUESTIONS: + return error_response("INVALID_PARAMS", BAD_REQUEST, + f"总题数不能超过 {MAX_TOTAL_QUESTIONS} 道，当前请求 {total_requested} 道", + http_status=400) + + # 校验排除题干列表（可选） + exclude_stems = data.get('exclude_stems') + stems_error = validate_exclude_stems(exclude_stems) + if stems_error: + return error_response("INVALID_PARAMS", BAD_REQUEST, stems_error, http_status=400) + # 获取当前用户 user = get_current_user() if not user: @@ -156,7 +187,8 @@ def api_generate_questions(): question_types=question_types, difficulty=data.get('difficulty', 3), options=data.get('options', {}), - request_id=data.get('request_id') + request_id=data.get('request_id'), + exclude_stems=data.get('exclude_stems') ) return success_response(data=result, status_code=EXAM_SUCCESS, message="出题成功") @@ -199,6 +231,12 @@ def api_generate_smart(): if diff_error: return error_response("INVALID_PARAMS", BAD_REQUEST, diff_error, http_status=400) + # 校验排除题干列表（可选） + exclude_stems = data.get('exclude_stems') + stems_error = validate_exclude_stems(exclude_stems) + if stems_error: + return error_response("INVALID_PARAMS", BAD_REQUEST, stems_error, http_status=400) + # 获取当前用户 user = get_current_user() if not user: @@ -231,7 +269,8 @@ def api_generate_smart(): question_types=question_types, difficulty=data.get('difficulty', 3), options=data.get('options', {}), - request_id=data.get('request_id') + request_id=data.get('request_id'), + exclude_stems=data.get('exclude_stems') ) # 3. 在返回结果中添加 AI 分析信息 @@ -258,21 +297,21 @@ def api_grade_answers(): { "question_id": "uuid", "question_type": "single_choice", - "question_content": {"answer": "B"}, + "content": {"answer": "B"}, "student_answer": "A", "max_score": 2 }, { "question_id": "uuid", "question_type": "fill_blank", - "question_content": {"answer": [["答案1"], ["答案2"]]}, + "content": {"answer": [["答案1", "同义词"], ["答案2"]]}, "student_answer": ["学生答案1", "学生答案2"], "max_score": 4 }, { "question_id": "uuid", "question_type": "subjective", - "question_content": { + "content": { "stem": "简述...", "data": {"scoring_points": [...]}, "answer": "参考范文..." @@ -283,6 +322,12 @@ def api_grade_answers(): ] } + 说明： + - content 字段与出题接口返回的 content 结构一致，后端可直接透传 + - 客观题：content 只需包含 answer + - 填空题：content.answer 格式为 [["答案1", "同义词"], ...] 每空一个列表 + - 主观题：content 需包含 stem + answer + data.scoring_points + 返回: { "success": true, @@ -312,6 +357,14 @@ def api_grade_answers(): if not answers: return error_response("MISSING_PARAMS", BAD_REQUEST, "缺少答案数据", http_status=400) + # 校验每道题的 question_type + for i, ans in enumerate(answers): + q_type = ans.get('question_type') + if q_type not in VALID_QUESTION_TYPES: + return error_response("INVALID_PARAMS", BAD_REQUEST, + f"第 {i+1} 题的 question_type 无效: {q_type}，合法值: {', '.join(sorted(VALID_QUESTION_TYPES))}", + http_status=400) + # 调用新版批题接口 result = grade_answers( answers=answers, diff --git a/exam_pkg/generator.py b/exam_pkg/generator.py index f8e419d..213c557 100644 --- a/exam_pkg/generator.py +++ b/exam_pkg/generator.py @@ -1114,7 +1114,8 @@ def generate_questions_structured_v2( chunks: List[Dict], document_name: str, question_types: Dict[str, int], - difficulty: int = 3 + difficulty: int = 3, + exclude_stems: List[str] = None ) -> List[Dict]: """ 🔥 重构版：结构化出题 v2 @@ -1130,6 +1131,7 @@ def generate_questions_structured_v2( document_name: 文档名称 question_types: 题型及数量 {"single_choice": 5, ...} difficulty: 难度 1-5 + exclude_stems: 需要排除的已有题目题干列表（跨调用去重） Returns: 题目列表 @@ -1169,7 +1171,7 @@ def generate_questions_structured_v2( logger.warning("[v2] 知识点提取失败，降级使用 chunks 出题") fallback_questions = generator._generate_questions_fallback(chunks, document_name, question_types, difficulty) # 降级路径也执行去重和数量平衡 - deduped = _deduplicate_questions(fallback_questions) + deduped = _deduplicate_questions(fallback_questions, exclude_stems=exclude_stems) final = _balance_question_types(deduped, question_types) logger.info(f" [降级] 最终题目数: {len(final)}") return final @@ -1225,9 +1227,9 @@ def generate_questions_structured_v2( # ========== Phase 4: 质量校验 ========== logger.info("[v2] Phase 4: 质量校验") - # 4.1 去重 - deduped = _deduplicate_questions(all_questions) - logger.info(f" 去重: {len(all_questions)} → {len(deduped)}") + # 4.1 去重（含跨调用排除已有题目） + deduped = _deduplicate_questions(all_questions, exclude_stems=exclude_stems) + logger.info(f" 去重: {len(all_questions)} → {len(deduped)}" + (f"（排除已有 {len(exclude_stems)} 道）" if exclude_stems else "")) # 4.2 数量校正 final = _balance_question_types(deduped, question_types) @@ -1381,22 +1383,31 @@ def _retrieve_kp_chunks_v2( return result -def _deduplicate_questions(questions: List[Dict]) -> List[Dict]: +def _deduplicate_questions(questions: List[Dict], exclude_stems: List[str] = None) -> List[Dict]: """ 题目去重 策略： 1. 相同题干去重（前 80 字） 2. 相同知识点 + 相同题型去重 + 3. 排除已有题目（跨调用去重） Args: questions: 原始题目列表 + exclude_stems: 需要排除的已有题目题干列表（用于跨调用去重） Returns: 去重后的题目列表 """ seen_stems = set() seen_kp_type = set() + + # 预填已有题目的题干前缀，使新生成的题目与已有题目冲突时被过滤 + if exclude_stems: + for stem in exclude_stems: + seen_stems.add(stem[:80]) + seen_kp_type.add(f"{stem[:30]}_") # 通配题型匹配 + deduped = [] for q in questions: diff --git a/exam_pkg/grader.py b/exam_pkg/grader.py index 1052108..6637750 100644 --- a/exam_pkg/grader.py +++ b/exam_pkg/grader.py @@ -15,6 +15,8 @@ """ import json +import logging +import re import threading import time from concurrent.futures import ThreadPoolExecutor, as_completed @@ -24,6 +26,8 @@ from typing import List, Dict, Any, Optional # 导入 LLM 工具函数 from core.llm_utils import call_llm +logger = logging.getLogger(__name__) + # 导入 LLM 配置 try: from config import API_KEY, BASE_URL, MODEL @@ -77,7 +81,7 @@ def grade_objective(answer: Dict) -> Dict: 🔥 本地直接判断，无 LLM 调用 """ q_type = answer['question_type'] - question_content = answer.get('question_content', {}) + question_content = answer.get('content', {}) correct_answer = question_content.get('answer') student_answer = answer.get('student_answer') max_score = answer.get('max_score', 2.0) @@ -114,7 +118,7 @@ def grade_fill_blank(answer: Dict) -> Dict: 填空题答案格式：[["答案1", "同义词1", ...], ["答案2", ...], ...] 学生答案格式：["学生答案1", "学生答案2", ...] """ - question_content = answer.get('question_content', {}) + question_content = answer.get('content', {}) correct_answers = question_content.get('answer', []) # [[答案1, 同义词...], ...] student_answers = answer.get('student_answer', []) max_score = answer.get('max_score', 4.0) @@ -284,33 +288,48 @@ class AnswerGrader: "details": {"error": "批阅超时"} } - @retry(times=2, delay=1) + @retry(times=3, delay=1) def _grade_subjective(self, answer: Dict) -> Dict: """ 批阅主观题 - 调用 LLM 评分 - 🔥 P1 改进：带重试 + 🔥 改进：3次重试 + 解析失败自动重试 """ with grading_semaphore: # 限流 prompt = self._build_grading_prompt(answer) response = self._call_llm(prompt) - return self._parse_grading_result(response, answer) + try: + return self._parse_grading_result(response, answer) + except (json.JSONDecodeError, ValueError, TypeError) as e: + # 解析失败抛异常 → 触发 @retry 重试 + logger.warning(f"[主观题评分] 解析失败将重试: {e}, 原始响应前200字: {str(response)[:200]}") + raise def _build_grading_prompt(self, answer: Dict) -> str: """构造评分 Prompt""" - question_content = answer.get('question_content', {}) + question_content = answer.get('content', {}) scoring_points = question_content.get('data', {}).get('scoring_points', []) + stem = question_content.get('stem', '') + reference_answer = question_content.get('answer', '') + + # 如果缺少评分标准，在 prompt 中补充提示 + scoring_section = "" + if scoring_points: + scoring_section = json.dumps(scoring_points, ensure_ascii=False, indent=2) + else: + scoring_section = "（未提供评分标准，请根据参考答案自行判断要点）" + return f"""请批阅以下简答题。 ## 题目 -{question_content.get('stem', '')} +{stem if stem else '（未提供题目）'} ## 参考答案 -{question_content.get('answer', '')} +{reference_answer if reference_answer else '（未提供参考答案）'} ## 评分标准 -{json.dumps(scoring_points, ensure_ascii=False, indent=2)} +{scoring_section} ## 学生答案 {answer.get('student_answer', '')} @@ -319,19 +338,20 @@ class AnswerGrader: {answer.get('max_score', 10)} 分 ## 输出约束 -1. 必须输出合法 JSON -2. score 不能超过满分 -3. achieved 为 0-1 之间的比例 +1. 必须输出合法 JSON，不要包含任何占位符或中文说明 +2. score 为数字，不能超过满分 +3. achieved 为 0-1 之间的数字 +4. 所有字段必须填入实际评分值 -## 输出格式（JSON） +## 输出格式示例（JSON） {{ - "score": 得分, + "score": 7, "scoring_breakdown": [ - {{"point": "要点名称", "weight": 权重, "achieved": 实际得分比例, "comment": "评语"}} + {{"point": "核心概念正确", "weight": 0.5, "achieved": 0.8, "comment": "基本概念描述准确"}} ], - "highlights": ["亮点1", "亮点2"], - "shortcomings": ["不足1"], - "overall_feedback": "整体评价" + "highlights": ["回答条理清晰"], + "shortcomings": ["缺少具体应用场景"], + "overall_feedback": "整体回答较好，但不够全面" }} 请直接输出 JSON：""" @@ -357,36 +377,96 @@ class AnswerGrader: raise Exception("LLM 调用失败") return result + def _extract_json(self, response: str) -> dict: + """ + 多策略从 LLM 响应中提取 JSON 对象 + + 策略优先级： + 1. markdown 代码块提取 ```json ... ``` + 2. 直接 json.loads + 3. 正则匹配第一个 {...} 块 + """ + if not response: + raise ValueError("LLM 返回为空") + + # 策略1：提取 markdown 代码块 + json_match = re.search(r'```(?:json)?\s*([\s\S]*?)\s*```', response) + if json_match: + json_str = json_match.group(1).strip() + try: + result = json.loads(json_str) + if isinstance(result, dict): + return result + except json.JSONDecodeError: + pass # 继续下一策略 + + # 策略2：直接解析整个响应 + try: + result = json.loads(response.strip()) + if isinstance(result, dict): + return result + except json.JSONDecodeError: + pass # 继续下一策略 + + # 策略3：正则匹配最外层 JSON 对象 + brace_match = re.search(r'\{[\s\S]*\}', response) + if brace_match: + try: + result = json.loads(brace_match.group(0)) + if isinstance(result, dict): + return result + except json.JSONDecodeError: + pass + + # 全部策略失败 + raise json.JSONDecodeError( + f"无法从 LLM 响应中提取有效 JSON，响应前300字: {response[:300]}", + response, 0 + ) + def _parse_grading_result(self, response: str, answer: Dict) -> Dict: - """解析评分结果""" + """ + 解析评分结果 + + 解析失败时抛出异常（由调用方的 @retry 处理重试） + """ max_score = answer.get('max_score', 10) - try: - # 尝试解析 JSON - result = json.loads(response) - score = min(result.get('score', 0), max_score) # 不能超过满分 + # 检查主观题内容完整性 + question_content = answer.get('content', {}) + warnings = [] + if not question_content.get('stem'): + warnings.append("缺少题目(stem)") + if not question_content.get('answer'): + warnings.append("缺少参考答案(answer)") + if not question_content.get('data', {}).get('scoring_points'): + warnings.append("缺少评分标准(scoring_points)，评分结果仅供参考") - return { - "question_id": answer.get('question_id'), - "score": score, - "max_score": max_score, - "grading_status": "success", - "details": { - "scoring_breakdown": result.get('scoring_breakdown', []), - "highlights": result.get('highlights', []), - "shortcomings": result.get('shortcomings', []), - "overall_feedback": result.get('overall_feedback', '') - } - } - except (json.JSONDecodeError, KeyError, TypeError) as e: - # 解析失败，返回默认 - return { - "question_id": answer.get('question_id'), - "score": 0, - "max_score": max_score, - "grading_status": "failed", - "details": {"error": "评分结果解析失败"} - } + # 多策略提取 JSON（失败抛异常 → 触发重试） + result = self._extract_json(response) + + # 校验关键字段 + score = result.get('score') + if score is None or not isinstance(score, (int, float)): + raise ValueError(f"score 字段缺失或类型错误: {score}") + score = min(float(score), max_score) # 不能超过满分 + + details = { + "scoring_breakdown": result.get('scoring_breakdown', []), + "highlights": result.get('highlights', []), + "shortcomings": result.get('shortcomings', []), + "overall_feedback": result.get('overall_feedback', '') + } + if warnings: + details["warnings"] = warnings + + return { + "question_id": answer.get('question_id'), + "score": score, + "max_score": max_score, + "grading_status": "success", + "details": details + } # ==================== 批题入口函数 ==================== diff --git a/exam_pkg/manager.py b/exam_pkg/manager.py index 2bc33a1..7aa9b5f 100644 --- a/exam_pkg/manager.py +++ b/exam_pkg/manager.py @@ -69,7 +69,8 @@ def generate_questions_from_file( question_types: Dict[str, int], difficulty: int = 3, options: Dict = None, - request_id: str = None + request_id: str = None, + exclude_stems: List[str] = None ) -> Dict: """ 从文件生成题目（结构化出题 v2） @@ -88,6 +89,7 @@ def generate_questions_from_file( options: 可选配置 - max_source_chunks: 最大切片数（默认 30） request_id: 请求 ID（幂等性支持） + exclude_stems: 需要排除的已有题目题干列表（跨调用去重） Returns: { @@ -115,7 +117,8 @@ def generate_questions_from_file( chunks=chunks, document_name=file_path, question_types=question_types, - difficulty=difficulty + difficulty=difficulty, + exclude_stems=exclude_stems ) # 3. 统计实际出题数量，与请求数量对比