jhzhang/rag
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(exam_pkg): 优化出题系统稳定性与推理模型适配

Browse Source 
- core/llm_utils: MiMo模型自动注入thinking=disabled参数，全局生效
- config: 新增LLM_DISABLE_THINKING配置项（默认true）
- generator: 推理模型自适应max_tokens（1.5x）、429限流重试+指数退避、v2管线补题机制
- generator: analyze_document_for_exam新增max_total参数控制AI出题上限
- generator: validate_questions_schema兼容type和question_type字段
- grader: 主观题max_tokens从1000提升至2000、fuzzy_match增加编辑距离容错
- manager: results变量初始化防NameError、透传max_total参数
- api: /exam/generate-smart支持max_total请求参数
This commit is contained in:
lacerate551
2026-06-22 18:51:11 +08:00
parent d589c27bce
commit 5ccfdaf9d1
6 changed files with 462 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

200
exam_pkg/generator.py
View File

@@ -26,19 +26,38 @@ import logging
logger = logging.getLogger(__name__)
# 导入 LLM 工具函数
from core.llm_utils import call_llm, parse_json_list_from_response
from core.llm_utils import call_llm, parse_json_list_from_response, extract_json_list, extract_json_object
# 导入 LLM 配置
try:
from config import API_KEY, BASE_URL, MODEL
from config import API_KEY, BASE_URL, MODEL, LLM_TEMPERATURE, LLM_MAX_TOKENS
LLM_AVAILABLE = True
except ImportError:
API_KEY = None
BASE_URL = None
MODEL = None
LLM_TEMPERATURE = 0.7
LLM_MAX_TOKENS = 4000
LLM_AVAILABLE = False
# 推理模型识别：这些模型会消耗额外 token 用于思考链，需要更大的 max_tokens 预算
_REASONING_MODEL_KEYWORDS = ('mimo', 'qwq', 'deepseek-r1', 'deepseek-reasoner', 'o1', 'o3')
def _is_reasoning_model(model_name: str) -> bool:
"""判断是否为推理模型（需要额外思考链 token 预算）"""
if not model_name:
return False
name_lower = model_name.lower()
return any(kw in name_lower for kw in _REASONING_MODEL_KEYWORDS)
def _get_effective_max_tokens(base_max: int, model_name: str) -> int:
"""根据模型类型计算实际 max_tokens。推理模型需要 1.5x 预算给思考链"""
if _is_reasoning_model(model_name):
return max(base_max, int(base_max * 1.5))
return base_max
# ==================== 辅助函数 ====================
def group_chunks_by_section(chunks: List[Dict]) -> Dict[str, List[Dict]]:
@@ -131,8 +150,9 @@ def validate_questions_schema(questions: List[Dict]) -> List[Dict]:
validated = []
for q in questions:
# 必须有 type
if q.get('type') not in VALID_TYPES:
# 必须有 type 或 question_type
q_type = q.get('question_type') or q.get('type')
if q_type not in VALID_TYPES:
continue
# 必须有 content
@@ -145,7 +165,7 @@ def validate_questions_schema(questions: List[Dict]) -> List[Dict]:
continue
# 选项题必须有 options
if q['type'] in ['single_choice', 'multiple_choice']:
if q_type in ['single_choice', 'multiple_choice']:
if not content.get('data', {}).get('options'):
continue
@@ -359,8 +379,8 @@ class QuestionGenerator:
client=self.client,
prompt=prompt,
model=self.model,
temperature=0.7,
max_tokens=4000,
temperature=LLM_TEMPERATURE,
max_tokens=_get_effective_max_tokens(4000, self.model),
messages=messages
)
if not content:
@@ -493,25 +513,37 @@ class QuestionGenerator:
return type_names.get(q_type, q_type)
def _call_llm(self, prompt: str) -> str:
"""调用本地 LLM（OpenAI 兼容接口）"""
"""调用本地 LLM（OpenAI 兼容接口），支持 429 限流重试"""
if not self.client:
raise ValueError("LLM 客户端未初始化，请检查 config.py 中的 API_KEY 配置")
import time
messages = [
{"role": "system", "content": "你是一个专业的出题专家，擅长根据文档内容生成各类考试题目。你必须严格按照JSON格式输出，不要有任何其他内容。"},
{"role": "user", "content": prompt}
]
result = call_llm(
client=self.client,
prompt=prompt,
model=self.model,
temperature=0.7,
max_tokens=4000,
messages=messages
)
if result is None:
raise Exception("LLM 调用失败")
return result
# 重试机制：429 限流时指数退避（最多重试 3 次）
for attempt in range(4):
result = call_llm(
client=self.client,
prompt=prompt,
model=self.model,
temperature=LLM_TEMPERATURE,
max_tokens=_get_effective_max_tokens(4000, self.model),
messages=messages
)
if result is not None:
return result
# 可能是 429 限流，等待后重试
if attempt < 3:
wait_time = 2 ** attempt # 1s, 2s, 4s
logger.warning(f" LLM 调用失败，{wait_time}s 后重试 ({attempt+1}/3)...")
time.sleep(wait_time)
raise Exception("LLM 调用失败（已重试 3 次）")
def _get_format_examples(self) -> str:
"""返回各题型格式示例（覆盖全部 5 种题型）"""
@@ -610,7 +642,6 @@ class QuestionGenerator:
)
# 清理纯标点符号
import re
all_content = re.sub(r'^[\s\*\-\d\.。、，：:；;]+$', '', all_content, flags=re.MULTILINE)
all_content = all_content.strip()
@@ -645,27 +676,24 @@ class QuestionGenerator:
请直接输出 JSON 数组："""
try:
response = self._call_llm(prompt)
for _attempt in range(2):
try:
response = self._call_llm(prompt)
if not response or not response.strip():
continue
# 清理响应（移除可能的 markdown 标记）
response = response.strip()
if response.startswith('```'):
lines = response.split('\n')
response = '\n'.join(lines[1:-1] if lines[-1] == '```' else lines[1:])
# 解析 JSON
result = json.loads(response)
if isinstance(result, list):
return [
{"name": kp, "section": section}
for kp in result[:max_points]
if isinstance(kp, str) and 3 <= len(kp) <= 30
]
except json.JSONDecodeError as e:
logger.error(f" 知识点 JSON 解析失败: {e}")
except Exception as e:
logger.error(f" 知识点提取失败: {e}")
# 使用多策略 JSON 提取（支持 markdown 代码块、正则回退）
result = extract_json_list(response)
if isinstance(result, list) and result:
return [
{"name": kp, "section": section}
for kp in result[:max_points]
if isinstance(kp, str) and 3 <= len(kp) <= 30
]
# 解析结果无效，重试一次
logger.warning(f" 知识点提取返回无效结果(尝试 {_attempt+1}/2)，重试中...")
except Exception as e:
logger.error(f" 知识点提取失败(尝试 {_attempt+1}/2): {e}")
return []
@@ -918,7 +946,7 @@ def generate_questions_from_content(
return generator.generate_questions_structured(chunks, document_name, question_types, difficulty)
def analyze_document_for_exam(chunks: List[Dict]) -> Dict[str, Any]:
def analyze_document_for_exam(chunks: List[Dict], max_total: int = None) -> Dict[str, Any]:
"""
AI 智能分析文档内容，决定适合的题型和数量
@@ -1026,14 +1054,14 @@ def analyze_document_for_exam(chunks: List[Dict]) -> Dict[str, Any]:
try:
response = generator._call_llm(prompt)
response = response.strip()
if not response or not response.strip():
return _generate_default_question_types(total_knowledge_points)
# 清理 markdown 代码块
if response.startswith('```'):
lines = response.split('\n')
response = '\n'.join(lines[1:-1] if lines[-1] == '```' else lines[1:])
result = json.loads(response)
# 使用多策略 JSON 提取（支持 markdown 代码块、正则回退）
result = extract_json_object(response)
if not isinstance(result, dict):
logger.error(f"AI 分析返回非对象类型: {type(result)}")
return _generate_default_question_types(total_knowledge_points)
# 验证和清理结果
valid_types = ['single_choice', 'multiple_choice', 'true_false', 'fill_blank', 'subjective']
@@ -1046,6 +1074,24 @@ def analyze_document_for_exam(chunks: List[Dict]) -> Dict[str, Any]:
# 过滤不适合的题型
suitable_types = [t for t, c in question_types.items() if c > 0]
# 总数上限校验：如果指定了 max_total，超过时按比例缩减
if max_total and max_total > 0:
total = sum(question_types.values())
if total > max_total:
ratio = max_total / total
question_types = {k: max(0, round(v * ratio)) for k, v in question_types.items()}
# 修正四舍五入误差
diff = max_total - sum(question_types.values())
if diff > 0:
# 把差额分配给最大的题型
for k in sorted(question_types, key=question_types.get, reverse=True):
question_types[k] += 1
diff -= 1
if diff <= 0:
break
logger.info(f" AI 推荐 {total} 题，按上限 {max_total} 缩减为 {sum(question_types.values())} 题")
suitable_types = [t for t, c in question_types.items() if c > 0]
return {
"total_knowledge_points": result.get('total_knowledge_points', total_knowledge_points),
"suitable_types": suitable_types,
@@ -1053,10 +1099,6 @@ def analyze_document_for_exam(chunks: List[Dict]) -> Dict[str, Any]:
"reason": result.get('reason', 'AI 分析完成')
}
except json.JSONDecodeError as e:
logger.error(f"AI 分析结果 JSON 解析失败: {e}")
# 降级：根据知识点数量生成默认配置
return _generate_default_question_types(total_knowledge_points)
except Exception as e:
logger.error(f"AI 分析失败: {e}")
return _generate_default_question_types(total_knowledge_points)
@@ -1156,6 +1198,10 @@ def generate_questions_structured_v2(
# 提取知识点
kps = generator._extract_knowledge_points(section, section_chunks, max_points=3)
# API 限流保护：连续 LLM 调用间隔 1 秒
import time
time.sleep(1)
# 全局去重
for kp in kps:
kp_name = kp['name']
@@ -1219,6 +1265,10 @@ def generate_questions_structured_v2(
prompt, {q_type: 1}, kp_chunks, document_name, max_retries=2
)
# API 限流保护：连续 LLM 调用间隔 1 秒
import time
time.sleep(1)
if success and questions:
all_questions.extend(questions)
else:
@@ -1241,6 +1291,28 @@ def generate_questions_structured_v2(
type_counts[q.get('question_type')] += 1
logger.info(f" 题型分布: {dict(type_counts)}")
# 4.4 补题：如果某题型数量不足，使用 chunks 补充
shortage_types = {}
for q_type, target_count in question_types.items():
actual_count = type_counts.get(q_type, 0)
if actual_count < target_count:
shortage_types[q_type] = target_count - actual_count
if shortage_types:
logger.info(f" [v2] 补题: 缺少题型 {dict(shortage_types)}")
for q_type, shortage in shortage_types.items():
logger.info(f" 补充 {q_type} {shortage} 道...")
extra = generator._makeup_questions(chunks, q_type, shortage, difficulty, document_name)
# 补题也需去重
extra_deduped = _deduplicate_questions(extra, exclude_stems=exclude_stems)
final.extend(extra_deduped[:shortage])
# 补题后重新统计
type_counts = defaultdict(int)
for q in final:
type_counts[q.get('question_type')] += 1
logger.info(f" 补题后题型分布: {dict(type_counts)}")
return final
@@ -1404,9 +1476,27 @@ def _deduplicate_questions(questions: List[Dict], exclude_stems: List[str] = Non
# 预填已有题目的题干前缀，使新生成的题目与已有题目冲突时被过滤
if exclude_stems:
_valid_types = ('single_choice', 'multiple_choice', 'true_false', 'fill_blank', 'subjective')
for stem in exclude_stems:
seen_stems.add(stem[:80])
seen_kp_type.add(f"{stem[:30]}_") # 通配题型匹配
# 用 exclude_stem 自身长度做前缀键（排除题干通常短于30字）
# 新题目的 stem[:30] 如果以此前缀开头，[:len(prefix)] 后就能匹配
_prefix = stem[:30]
for _qt in _valid_types:
seen_kp_type.add(f"{_prefix}_{_qt}")
# 辅助函数：检查新题干是否匹配任何 exclude 前缀
_exclude_prefixes = []
if exclude_stems:
_exclude_prefixes = [s[:30] for s in exclude_stems]
def _matches_exclude(stem_text: str) -> bool:
"""检查题干前30字是否以某个 exclude 前缀开头"""
_stem30 = stem_text[:30]
for _ep in _exclude_prefixes:
if _stem30.startswith(_ep):
return True
return False
deduped = []
@@ -1419,6 +1509,10 @@ def _deduplicate_questions(questions: List[Dict], exclude_stems: List[str] = Non
if stem_key in seen_stems:
continue
# 跨调用排除：题干前缀匹配到 exclude_stems 则跳过
if _matches_exclude(stem):
continue
# 知识点 + 题型去重
kp_type_key = f"{stem[:30]}_{q.get('question_type')}"
if kp_type_key in seen_kp_type: