""" 出题与批题系统 API 蓝图 提供 REST API 接口： - 出题：生成题目（返回 JSON 给后端） - 批题：批阅答案（返回结果给后端） 职责边界： - RAG 服务负责：生成题目 + 批阅答案 - 后端服务负责：审核入库 + 状态管理 使用方式： from exam_pkg.api import exam_bp app.register_blueprint(exam_bp, url_prefix='/exam') """ from flask import Blueprint, request, jsonify import os # 导入考试管理模块 from exam_pkg.manager import ( generate_questions_from_file, grade_answers, ) # 导入网关认证模块 from auth.gateway import ( require_gateway_auth, require_role, check_collection_permission, get_current_user ) # 导入统一响应格式 from core.status_codes import EXAM_SUCCESS, GRADE_SUCCESS, EXAM_ERROR, GRADE_ERROR, BAD_REQUEST, NO_CONTENT, LLM_ERROR from api.response_utils import success_response, error_response # 合法题型 VALID_QUESTION_TYPES = {'single_choice', 'multiple_choice', 'true_false', 'fill_blank', 'subjective'} def validate_question_types(question_types: dict) -> str: """ 校验 question_types 参数 Returns: None 表示校验通过，字符串表示错误信息 """ if not isinstance(question_types, dict): return "question_types 必须为对象" for q_type, count in question_types.items(): if q_type not in VALID_QUESTION_TYPES: return f"不支持的题型: {q_type}，合法值: {', '.join(sorted(VALID_QUESTION_TYPES))}" if not isinstance(count, int) or count < 0: return f"题型 {q_type} 的数量必须为非负整数，当前值: {count}" return None def validate_difficulty(difficulty) -> str: """校验 difficulty 参数""" if not isinstance(difficulty, int) or difficulty < 1 or difficulty > 5: 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__) # ==================== 出题 API ==================== @exam_bp.route('/generate', methods=['POST']) @require_gateway_auth def api_generate_questions(): """ 🔥 新版出题接口 请求体: { "request_id": "uuid-optional", // 幂等性支持 "file_path": "public/产品手册.pdf", "collection": "public_kb", "question_types": { "single_choice": 3, "multiple_choice": 2, "true_false": 2, "fill_blank": 2, "subjective": 1 }, "difficulty": 3, "exclude_stems": ["已有题目的题干1", "已有题目的题干2"], // 可选，排除已有题目避免重复 "options": { "include_explanation": true, "max_source_chunks": 50 } } 返回: { "success": true, "request_id": "uuid", "questions": [ { "question_type": "single_choice", "difficulty": 3, "content": {"stem": "...", "data": {...}, "answer": "...", "explanation": "..."}, "source_trace": {"document_name": "...", "chunk_ids": [...], "sources": [...]} } ], "total": 10, "requested_types": {"single_choice": 3, "fill_blank": 2}, "actual_types": {"single_choice": 3, "fill_blank": 2}, "warnings": ["fill_blank: 请求 3 道，实际生成 2 道"], "source_chunks_used": 15 } """ try: data = request.json file_path = data.get('file_path') collection = data.get('collection') question_types = data.get('question_types', {}) if not file_path or not collection: return error_response("MISSING_PARAMS", BAD_REQUEST, "缺少 file_path 或 collection 参数", http_status=400) if not question_types: return error_response("MISSING_PARAMS", BAD_REQUEST, "缺少 question_types 参数", http_status=400) # 校验题型和数量 type_error = validate_question_types(question_types) if type_error: return error_response("INVALID_PARAMS", BAD_REQUEST, type_error, http_status=400) # 校验难度 difficulty = data.get('difficulty', 3) diff_error = validate_difficulty(difficulty) 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: return error_response("UNAUTHORIZED", BAD_REQUEST, "未认证", http_status=401) # 检查向量库访问权限 if not check_collection_permission( role=user['role'], department=user.get('department', ''), collection_name=collection, operation="read" ): return error_response("FORBIDDEN", BAD_REQUEST, "权限不足", http_status=403) # 调用新版出题接口 result = generate_questions_from_file( file_path=file_path, collection=collection, question_types=question_types, difficulty=data.get('difficulty', 3), options=data.get('options', {}), request_id=data.get('request_id'), exclude_stems=data.get('exclude_stems') ) return success_response(data=result, status_code=EXAM_SUCCESS, message="出题成功") except Exception as e: return error_response("EXAM_ERROR", EXAM_ERROR, str(e), http_status=500) @exam_bp.route('/generate-smart', methods=['POST']) @require_gateway_auth def api_generate_smart(): """ AI 智能出题 - 自动分析文件并决定题型和数量 与 /exam/generate 的区别：不需要传 question_types，AI 自动分析文档后决定 请求体: { "file_path": "public/产品手册.pdf", "collection": "public_kb", "difficulty": 3, // 可选，默认 3 "max_total": 20, // 可选，AI出题总数上限。不传则不限制 "options": {} // 可选 } 返回: 与 /exam/generate 相同格式，额外包含 ai_analysis 字段 """ try: data = request.json file_path = data.get('file_path') collection = data.get('collection') if not file_path or not collection: return error_response("MISSING_PARAMS", BAD_REQUEST, "缺少 file_path 或 collection 参数", http_status=400) # 校验难度 difficulty = data.get('difficulty', 3) diff_error = validate_difficulty(difficulty) if diff_error: return error_response("INVALID_PARAMS", BAD_REQUEST, diff_error, http_status=400) # 可选：AI出题总数上限（不传则不限制） max_total = data.get('max_total') if max_total is not None: try: max_total = int(max_total) if max_total <= 0: return error_response("INVALID_PARAMS", BAD_REQUEST, "max_total 必须为正整数", http_status=400) except (ValueError, TypeError): return error_response("INVALID_PARAMS", BAD_REQUEST, "max_total 必须为整数", 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: return error_response("UNAUTHORIZED", BAD_REQUEST, "未认证", http_status=401) # 检查向量库访问权限 if not check_collection_permission( role=user['role'], department=user.get('department', ''), collection_name=collection, operation="read" ): return error_response("FORBIDDEN", BAD_REQUEST, "权限不足", http_status=403) # 1. 调用 AI 分析文件，获取推荐的题型和数量 from exam_pkg.manager import analyze_file_for_exam ai_analysis = analyze_file_for_exam( file_path=file_path, collection=collection, max_total=max_total ) question_types = ai_analysis.get('question_types', {}) if not question_types or sum(question_types.values()) == 0: return error_response("EXAM_ERROR", EXAM_ERROR, "AI 分析后未生成有效题型配置", http_status=500) # 2. 使用 AI 推荐的题型调用出题接口 result = generate_questions_from_file( file_path=file_path, collection=collection, question_types=question_types, difficulty=data.get('difficulty', 3), options=data.get('options', {}), request_id=data.get('request_id'), exclude_stems=data.get('exclude_stems') ) # 3. 在返回结果中添加 AI 分析信息 result['ai_analysis'] = ai_analysis return success_response(data=result, status_code=EXAM_SUCCESS, message="AI 智能出题成功") except Exception as e: return error_response("EXAM_ERROR", EXAM_ERROR, str(e), http_status=500) # ==================== 批题 API ==================== @exam_bp.route('/grade', methods=['POST']) @require_gateway_auth def api_grade_answers(): """ 🔥 新版批题接口 请求体: { "request_id": "uuid-optional", "answers": [ { "question_id": "uuid", "question_type": "single_choice", "content": {"answer": "B"}, "student_answer": "A", "max_score": 2 }, { "question_id": "uuid", "question_type": "fill_blank", "content": {"answer": [["答案1", "同义词"], ["答案2"]]}, "student_answer": ["学生答案1", "学生答案2"], "max_score": 4 }, { "question_id": "uuid", "question_type": "subjective", "content": { "stem": "简述...", "data": {"scoring_points": [...]}, "answer": "参考范文..." }, "student_answer": "学生作答内容...", "max_score": 10 } ] } 说明： - content 字段与出题接口返回的 content 结构一致，后端可直接透传 - 客观题：content 只需包含 answer - 填空题：content.answer 格式为 [["答案1", "同义词"], ...] 每空一个列表 - 主观题：content 需包含 stem + answer + data.scoring_points 返回: { "success": true, "request_id": "uuid", "results": [ { "question_id": "uuid", "score": 2, "max_score": 2, "grading_status": "success", "details": {"correct": true, "student_answer": "A", "correct_answer": "B", "feedback": "..."} } ], "total_score": 10.5, "total_max_score": 16.0, "score_rate": 65.6 } grading_status 取值： - "success": 评分成功 - "failed": 评分失败（LLM 超时/解析失败等），score 为 0 """ try: data = request.json answers = data.get('answers', []) if not answers: return error_response("MISSING_PARAMS", BAD_REQUEST, "缺少答案数据", http_status=400) if not isinstance(answers, list): return error_response("INVALID_PARAMS", BAD_REQUEST, "answers 必须为数组", 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, request_id=data.get('request_id') ) return success_response(data=result, status_code=GRADE_SUCCESS, message="批阅完成") except Exception as e: return error_response("GRADE_ERROR", GRADE_ERROR, str(e), http_status=500) # ==================== 健康检查 ==================== @exam_bp.route('/health', methods=['GET']) def api_health(): """健康检查""" return jsonify({ "status": "ok", "service": "exam-api", "version": "2.0" })