1573 lines
51 KiB
Python
1573 lines
51 KiB
Python
"""
|
||
出题生成器 - 结构化 RAG 出题架构
|
||
|
||
核心功能：
|
||
1. 按章节分组检索切片
|
||
2. 每章节提取知识点
|
||
3. 按知识点精准检索并出题
|
||
4. 覆盖控制 + 去重
|
||
|
||
架构：
|
||
文档 → 章节分组 → 知识点提取 → 定向检索 → 出题 → 合并去重
|
||
|
||
使用方式：
|
||
from exam_pkg.generator import QuestionGenerator
|
||
|
||
generator = QuestionGenerator()
|
||
questions = generator.generate_questions(source_content, question_types, difficulty)
|
||
"""
|
||
|
||
import json
|
||
import re
|
||
from collections import defaultdict
|
||
from typing import List, Dict, Any, Optional
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 导入 LLM 工具函数
|
||
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, 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]]:
|
||
"""
|
||
🔥 结构化出题 Step 1：按章节分组
|
||
|
||
将切片按 section 字段分组，便于后续按章节提取知识点
|
||
"""
|
||
section_map = defaultdict(list)
|
||
for chunk in chunks:
|
||
section = chunk.get('section', '') or '未分类'
|
||
# 清理章节名称（去除多余的标记）
|
||
section_clean = section.replace('**', '').strip()
|
||
section_map[section_clean].append(chunk)
|
||
return dict(section_map)
|
||
|
||
|
||
def build_semantic_query(question_types: Dict[str, int]) -> str:
|
||
"""
|
||
根据题型构建语义化检索 query（保留用于知识点检索）
|
||
"""
|
||
query_parts = []
|
||
query_parts.append("重点内容 关键概念 定义")
|
||
|
||
if question_types.get('fill_blank', 0) > 0:
|
||
query_parts.append("术语 公式 数值 标准")
|
||
|
||
if question_types.get('subjective', 0) > 0:
|
||
query_parts.append("流程 步骤 原则 方法")
|
||
|
||
if question_types.get('multiple_choice', 0) > 0:
|
||
query_parts.append("区别 对比 分类")
|
||
|
||
return " ".join(query_parts)
|
||
|
||
|
||
def build_knowledge_point_query(knowledge_point: str, question_type: str = None) -> str:
|
||
"""
|
||
🔥 结构化出题 Step 3：根据知识点构建精准检索 query
|
||
|
||
Args:
|
||
knowledge_point: 知识点名称（如 "请假申请流程"）
|
||
question_type: 题型，用于补充检索词
|
||
|
||
Returns:
|
||
精准检索 query
|
||
"""
|
||
# 基础：知识点本身
|
||
query_parts = [knowledge_point]
|
||
|
||
# 根据题型补充
|
||
if question_type in ['single_choice', 'multiple_choice']:
|
||
query_parts.append("定义 规则 标准")
|
||
elif question_type == 'fill_blank':
|
||
query_parts.append("具体数值 公式 术语")
|
||
elif question_type == 'subjective':
|
||
query_parts.append("详细说明 步骤 流程")
|
||
|
||
return " ".join(query_parts)
|
||
|
||
|
||
def safe_parse_questions(result: str) -> List[Dict]:
|
||
"""
|
||
🔥 P0 改进：安全解析 JSON，支持自动修复
|
||
|
||
尝试多种方式解析 LLM 返回的 JSON：
|
||
1. 直接解析
|
||
2. 提取 ```json 代码块
|
||
3. 提取数组 [...] 模式
|
||
"""
|
||
questions = parse_json_list_from_response(result)
|
||
if questions:
|
||
# 兼容 {"questions": [...]} 格式
|
||
if isinstance(questions, list):
|
||
return questions
|
||
return []
|
||
|
||
|
||
def validate_questions_schema(questions: List[Dict]) -> List[Dict]:
|
||
"""
|
||
🔥 P0 改进：JSON Schema 校验，过滤/修复无效题目
|
||
|
||
校验规则：
|
||
- 必须有 type 且在有效类型中
|
||
- 必须有 content.stem
|
||
- 必须有 content.answer
|
||
- 选项题必须有 options
|
||
"""
|
||
VALID_TYPES = {'single_choice', 'multiple_choice', 'true_false', 'fill_blank', 'subjective'}
|
||
validated = []
|
||
|
||
for q in questions:
|
||
# 必须有 type 或 question_type
|
||
q_type = q.get('question_type') or q.get('type')
|
||
if q_type not in VALID_TYPES:
|
||
continue
|
||
|
||
# 必须有 content
|
||
content = q.get('content', {})
|
||
if not content.get('stem'):
|
||
continue
|
||
|
||
# 必须有 answer
|
||
if 'answer' not in content:
|
||
continue
|
||
|
||
# 选项题必须有 options
|
||
if q_type in ['single_choice', 'multiple_choice']:
|
||
if not content.get('data', {}).get('options'):
|
||
continue
|
||
|
||
# 填空题答案格式归一化：扁平数组 → 二维数组
|
||
if q_type == 'fill_blank':
|
||
ans = content.get('answer')
|
||
if isinstance(ans, list) and ans and all(isinstance(item, str) for item in ans):
|
||
# 扁平数组 ["答案1", "答案2"] → [["答案1"], ["答案2"]]
|
||
content['answer'] = [[item] for item in ans]
|
||
logger.warning(f"填空题答案格式修正: 扁平数组 → 二维数组 ({len(ans)} 空)")
|
||
|
||
validated.append(q)
|
||
|
||
return validated
|
||
|
||
|
||
def build_source_context(chunks: List[Dict]) -> str:
|
||
"""
|
||
构建带溯源标记的上下文
|
||
|
||
🔥 改进：添加 chunk_id 标记，便于 LLM 精确引用
|
||
"""
|
||
context_parts = []
|
||
for chunk in chunks:
|
||
chunk_id = chunk.get('chunk_id', '')
|
||
page_info = f"第{chunk['page']}页" if chunk.get('page') else ""
|
||
section_info = chunk.get('section', '')
|
||
# 添加 chunk_id 标记，格式：[chunk_xxx | 第N页 章节]
|
||
context_parts.append(f"[chunk_id:{chunk_id} | {page_info} {section_info}]\n{chunk['content']}")
|
||
|
||
return "\n\n---\n\n".join(context_parts)
|
||
|
||
|
||
def find_referenced_chunks(question: Dict, chunks: List[Dict]) -> List[Dict]:
|
||
"""
|
||
找到题目引用的切片
|
||
|
||
🔥 改进：优先用 referenced_chunk_ids 精确匹配，其次用 referenced_pages
|
||
"""
|
||
# 1. 优先使用 chunk_id 精确匹配
|
||
referenced_chunk_ids = question.get('referenced_chunk_ids', [])
|
||
if referenced_chunk_ids:
|
||
referenced = []
|
||
for chunk in chunks:
|
||
chunk_id = chunk.get('chunk_id', '')
|
||
for ref_id in referenced_chunk_ids:
|
||
# 支持两种格式：直接匹配或去掉 chunk_id: 前缀后匹配
|
||
if chunk_id == ref_id or chunk_id == ref_id.replace('chunk_id:', ''):
|
||
referenced.append(chunk)
|
||
break
|
||
if referenced:
|
||
return referenced
|
||
|
||
# 2. 退而求其次，使用页码匹配
|
||
referenced_pages = question.get('referenced_pages', [])
|
||
if referenced_pages:
|
||
referenced = []
|
||
for chunk in chunks:
|
||
if chunk.get('page') in referenced_pages:
|
||
referenced.append(chunk)
|
||
if referenced:
|
||
return referenced
|
||
|
||
# 3. 都没有，返回第一个切片（作为 fallback）
|
||
return chunks[:1] if chunks else []
|
||
|
||
|
||
# ==================== QuestionGenerator 类 ====================
|
||
|
||
class QuestionGenerator:
|
||
"""本地出题生成器 - 使用本地 OpenAI 客户端"""
|
||
|
||
def __init__(self):
|
||
self.client = None
|
||
if LLM_AVAILABLE and API_KEY:
|
||
try:
|
||
from openai import OpenAI
|
||
self.client = OpenAI(api_key=API_KEY, base_url=BASE_URL)
|
||
except ImportError:
|
||
pass
|
||
self.model = MODEL
|
||
|
||
def generate_questions_structured(
|
||
self,
|
||
chunks: List[Dict],
|
||
document_name: str,
|
||
question_types: Dict[str, int],
|
||
difficulty: int = 3
|
||
) -> List[Dict]:
|
||
"""
|
||
🔥 结构化出题主流程（唯一入口）
|
||
|
||
流程：
|
||
1. 按章节分组
|
||
2. 每章节提取知识点
|
||
3. 按知识点分配题目数量
|
||
4. 按知识点出题
|
||
5. 合并去重
|
||
"""
|
||
total_questions = sum(question_types.values())
|
||
if not chunks or total_questions == 0:
|
||
return []
|
||
|
||
# Step 1: 按章节分组
|
||
section_map = group_chunks_by_section(chunks)
|
||
logger.info(f"[结构化出题] 章节数: {len(section_map)}")
|
||
|
||
# Step 2: 每章节提取知识点
|
||
all_knowledge_points = []
|
||
for section, section_chunks in section_map.items():
|
||
kps = self._extract_knowledge_points(section, section_chunks)
|
||
all_knowledge_points.extend(kps)
|
||
logger.info(f" 章节 [{section[:20]}...]: 提取 {len(kps)} 个知识点")
|
||
|
||
if not all_knowledge_points:
|
||
# 降级：直接使用 chunks 出题（不提取知识点）
|
||
logger.warning("[结构化出题] 知识点提取失败，直接使用 chunks 出题")
|
||
return self._generate_questions_fallback(chunks, document_name, question_types, difficulty)
|
||
|
||
logger.info(f"[结构化出题] 总知识点: {len(all_knowledge_points)}")
|
||
|
||
# Step 3: 分配题目数量
|
||
kp_assignments = self._assign_questions_to_kps(
|
||
all_knowledge_points, question_types, total_questions
|
||
)
|
||
|
||
# Step 4: 按知识点出题（带重试机制）
|
||
all_questions = []
|
||
failed_assignments = [] # 记录失败的分配，用于补题
|
||
|
||
for kp, q_types, q_count in kp_assignments:
|
||
# 找到该知识点相关的 chunks
|
||
kp_chunks = self._retrieve_kp_chunks(kp, chunks)
|
||
if not kp_chunks:
|
||
logger.warning(f" [警告] 知识点 [{kp[:20]}...] 无相关 chunks，跳过")
|
||
failed_assignments.append((kp, q_types, q_count, "无相关chunks"))
|
||
continue
|
||
|
||
# 构造上下文并出题
|
||
source_content = build_source_context(kp_chunks)
|
||
prompt = self._build_prompt_for_kp(source_content, kp, q_types, difficulty)
|
||
|
||
# 🔥 改进：带重试的出题
|
||
success, questions = self._generate_with_retry(prompt, q_types, kp_chunks, document_name, max_retries=2)
|
||
|
||
if success and questions:
|
||
# 🔥 新增：校验题型一致性
|
||
validated_questions = self._validate_question_types(questions, q_types)
|
||
all_questions.extend(validated_questions)
|
||
else:
|
||
failed_assignments.append((kp, q_types, q_count, "出题失败"))
|
||
|
||
# 🔥 Step 4.5: 补题机制 - 对不足的题型进行补充
|
||
current_counts = defaultdict(int)
|
||
for q in all_questions:
|
||
current_counts[q.get('question_type')] += 1
|
||
|
||
for q_type, target_count in question_types.items():
|
||
shortage = target_count - current_counts.get(q_type, 0)
|
||
if shortage > 0:
|
||
logger.info(f" [补题] {q_type} 缺少 {shortage} 道，尝试补充...")
|
||
extra_questions = self._makeup_questions(chunks, q_type, shortage, difficulty, document_name)
|
||
all_questions.extend(extra_questions)
|
||
|
||
# Step 5: 去重 + 数量校正
|
||
final_questions = self._deduplicate_and_balance(all_questions, question_types)
|
||
|
||
# 🔥 最终日志：输出题型分布
|
||
final_counts = defaultdict(int)
|
||
for q in final_questions:
|
||
final_counts[q.get('question_type')] += 1
|
||
logger.info(f"[出题完成] 题型分布: {dict(final_counts)}")
|
||
|
||
return final_questions
|
||
|
||
def _generate_questions_fallback(
|
||
self,
|
||
chunks: List[Dict],
|
||
document_name: str,
|
||
question_types: Dict[str, int],
|
||
difficulty: int = 3
|
||
) -> List[Dict]:
|
||
"""
|
||
降级出题方法：当知识点提取失败时，直接使用 chunks 出题
|
||
|
||
输出格式与正常路径一致（嵌套结构），确保 API 响应格式统一
|
||
"""
|
||
source_content = build_source_context(chunks)
|
||
total_questions = sum(question_types.values())
|
||
if not source_content or total_questions == 0:
|
||
return []
|
||
|
||
q_type_str = ", ".join(f"{t}:{n}道" for t, n in question_types.items())
|
||
|
||
# 构造出题 prompt —— 使用与正常路径相同的嵌套格式
|
||
prompt = f"""请根据以下参考资料生成题目。
|
||
|
||
## 参考资料
|
||
|
||
{source_content}
|
||
|
||
## 要求
|
||
|
||
- 题型及数量：{q_type_str}
|
||
- 难度等级：{difficulty}/5
|
||
- 答案必须基于参考资料
|
||
|
||
## 输出格式（JSON 数组）
|
||
{self._get_format_examples()}
|
||
|
||
请直接输出 JSON 数组："""
|
||
|
||
try:
|
||
messages = [
|
||
{"role": "system", "content": "你是一个专业的出题专家，必须严格按照JSON格式输出。"},
|
||
{"role": "user", "content": prompt}
|
||
]
|
||
content = 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 not content:
|
||
return []
|
||
# 解析 JSON
|
||
raw_questions = parse_json_list_from_response(content)
|
||
if not raw_questions:
|
||
return []
|
||
|
||
# Schema 校验
|
||
validated = validate_questions_schema(raw_questions)
|
||
|
||
# 补充溯源信息（与正常路径格式统一）
|
||
enriched = self._enrich_with_source_trace(validated, chunks, document_name)
|
||
return enriched
|
||
except Exception as e:
|
||
logger.error(f"[降级出题失败] {e}")
|
||
|
||
return []
|
||
|
||
def _generate_with_retry(
|
||
self,
|
||
prompt: str,
|
||
q_types: Dict[str, int],
|
||
chunks: List[Dict],
|
||
document_name: str,
|
||
max_retries: int = 2
|
||
) -> tuple:
|
||
"""
|
||
🔥 带重试的出题方法
|
||
|
||
Returns:
|
||
(success: bool, questions: List[Dict])
|
||
"""
|
||
for attempt in range(max_retries + 1):
|
||
try:
|
||
response = self._call_llm(prompt)
|
||
raw_questions = safe_parse_questions(response)
|
||
validated = validate_questions_schema(raw_questions)
|
||
|
||
if validated:
|
||
enriched = self._enrich_with_source_trace(validated, chunks, document_name)
|
||
return (True, enriched)
|
||
else:
|
||
logger.warning(f" [重试 {attempt+1}] JSON 解析成功但题目无效")
|
||
|
||
except Exception as e:
|
||
logger.error(f" [重试 {attempt+1}] 出题异常: {e}")
|
||
|
||
return (False, [])
|
||
|
||
def _validate_question_types(
|
||
self,
|
||
questions: List[Dict],
|
||
expected_types: Dict[str, int]
|
||
) -> List[Dict]:
|
||
"""
|
||
🔥 校验题型一致性，过滤不符合预期的题目
|
||
|
||
只保留 expected_types 中指定的题型
|
||
"""
|
||
valid_types = set(expected_types.keys())
|
||
validated = []
|
||
|
||
for q in questions:
|
||
q_type = q.get('question_type') or q.get('type')
|
||
if q_type in valid_types:
|
||
validated.append(q)
|
||
else:
|
||
logger.debug(f" [过滤] 题型 {q_type} 不在预期范围内")
|
||
|
||
return validated
|
||
|
||
def _makeup_questions(
|
||
self,
|
||
chunks: List[Dict],
|
||
q_type: str,
|
||
count: int,
|
||
difficulty: int,
|
||
document_name: str
|
||
) -> List[Dict]:
|
||
"""
|
||
补题机制：为缺少的题型补充题目
|
||
|
||
策略：从 chunks 中选择未使用的切片进行补题
|
||
"""
|
||
if not chunks or count <= 0:
|
||
return []
|
||
|
||
# 简单策略：使用前几个 chunks 进行补题
|
||
makeup_chunks = chunks[:5]
|
||
source_content = build_source_context(makeup_chunks)
|
||
|
||
prompt = f"""基于以下文档内容，生成 {count} 道 {self._get_q_type_name(q_type)}。
|
||
|
||
## 文档内容
|
||
{source_content}
|
||
|
||
## 出题要求
|
||
- 题型：{self._get_q_type_name(q_type)}
|
||
- 数量：{count} 道
|
||
- 难度：{difficulty}/5
|
||
|
||
## 输出格式（JSON 数组）
|
||
{self._get_format_examples()}
|
||
|
||
请直接输出 JSON 数组："""
|
||
|
||
try:
|
||
response = self._call_llm(prompt)
|
||
raw_questions = safe_parse_questions(response)
|
||
validated = validate_questions_schema(raw_questions)
|
||
enriched = self._enrich_with_source_trace(validated, makeup_chunks, document_name)
|
||
|
||
# 只取需要的数量
|
||
return enriched[:count]
|
||
except Exception as e:
|
||
logger.error(f" [补题失败] {e}")
|
||
return []
|
||
|
||
def _get_q_type_name(self, q_type: str) -> str:
|
||
"""获取题型的中文名称"""
|
||
type_names = {
|
||
'single_choice': '单选题',
|
||
'multiple_choice': '多选题',
|
||
'true_false': '判断题',
|
||
'fill_blank': '填空题',
|
||
'subjective': '简答题'
|
||
}
|
||
return type_names.get(q_type, q_type)
|
||
|
||
def _call_llm(self, prompt: str) -> str:
|
||
"""调用本地 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}
|
||
]
|
||
|
||
# 重试机制：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 种题型）"""
|
||
return '''
|
||
### 单选题示例
|
||
{
|
||
"type": "single_choice",
|
||
"content": {
|
||
"stem": "题干内容",
|
||
"data": {"options": [{"key": "A", "content": "选项A"}, {"key": "B", "content": "选项B"}, {"key": "C", "content": "选项C"}, {"key": "D", "content": "选项D"}]},
|
||
"answer": "B",
|
||
"explanation": "解析..."
|
||
},
|
||
"referenced_chunk_ids": ["chunk_001"]
|
||
}
|
||
|
||
### 多选题示例
|
||
{
|
||
"type": "multiple_choice",
|
||
"content": {
|
||
"stem": "题干内容",
|
||
"data": {"options": [{"key": "A", "content": "选项A"}, {"key": "B", "content": "选项B"}, {"key": "C", "content": "选项C"}, {"key": "D", "content": "选项D"}]},
|
||
"answer": ["A", "C"],
|
||
"explanation": "解析..."
|
||
},
|
||
"referenced_chunk_ids": ["chunk_001"]
|
||
}
|
||
|
||
### 判断题示例
|
||
{
|
||
"type": "true_false",
|
||
"content": {
|
||
"stem": "判断：某个陈述内容",
|
||
"data": {},
|
||
"answer": "对",
|
||
"explanation": "解析..."
|
||
},
|
||
"referenced_chunk_ids": ["chunk_002"]
|
||
}
|
||
|
||
### 填空题示例
|
||
{
|
||
"type": "fill_blank",
|
||
"content": {
|
||
"stem": "RAG的全称是___，其核心在于___。",
|
||
"data": {"blank_count": 2},
|
||
"answer": [["检索增强生成"], ["外部知识库", "检索"]],
|
||
"explanation": "解析..."
|
||
},
|
||
"referenced_chunk_ids": ["chunk_003"]
|
||
}
|
||
|
||
### 主观题示例
|
||
{
|
||
"type": "subjective",
|
||
"content": {
|
||
"stem": "请简述某个概念的含义及其应用场景。",
|
||
"data": {
|
||
"scoring_points": [
|
||
{"point": "要点1：概念定义", "weight": 0.4},
|
||
{"point": "要点2：应用场景", "weight": 0.3},
|
||
{"point": "要点3：举例说明", "weight": 0.3}
|
||
]
|
||
},
|
||
"answer": "参考答案范文...",
|
||
"explanation": "解析..."
|
||
},
|
||
"referenced_chunk_ids": ["chunk_004"]
|
||
}
|
||
'''
|
||
|
||
def _extract_knowledge_points(
|
||
self,
|
||
section: str,
|
||
chunks: List[Dict],
|
||
max_points: int = 5
|
||
) -> List[Dict]:
|
||
"""
|
||
🔥 Step 2: 从章节内容提取知识点
|
||
|
||
策略：
|
||
1. 长内容（>= 100字）：调用 LLM 提取知识点
|
||
2. 短内容（< 100字）：直接用内容本身作为知识点
|
||
|
||
Returns:
|
||
[{"name": "知识点名称", "section": "所属章节"}, ...]
|
||
"""
|
||
if not chunks:
|
||
return []
|
||
|
||
# 合并同一章节的所有切片内容
|
||
all_content = "\n\n".join(
|
||
c.get('content', '').strip()
|
||
for c in chunks
|
||
if c.get('content', '').strip()
|
||
)
|
||
|
||
# 清理纯标点符号
|
||
all_content = re.sub(r'^[\s\*\-\d\.。、，：:；;]+$', '', all_content, flags=re.MULTILINE)
|
||
all_content = all_content.strip()
|
||
|
||
if not all_content:
|
||
return []
|
||
|
||
# 🔥 策略分叉：根据内容长度选择不同处理方式
|
||
if len(all_content) < 100:
|
||
# 短内容：直接用内容作为知识点（不调用 LLM）
|
||
# 提取关键短语（去除标点和编号）
|
||
clean_content = re.sub(r'^[\d\.\-\*]+\s*', '', all_content)
|
||
clean_content = re.sub(r'[。、，：:；;\s]+$', '', clean_content)
|
||
|
||
if 5 <= len(clean_content) <= 30:
|
||
return [{"name": clean_content, "section": section}]
|
||
return []
|
||
|
||
# 长内容：调用 LLM 提取知识点
|
||
prompt = f"""从以下文档内容中提取 {max_points} 个关键知识点。
|
||
|
||
## 内容
|
||
{all_content[:2000]}
|
||
|
||
## 要求
|
||
1. 每个知识点用简短短语描述（5-15字）
|
||
2. 知识点应该适合用于出考试题
|
||
3. 知识点之间不要重复或重叠
|
||
4. 必须返回 JSON 数组格式，不要有其他内容
|
||
|
||
## 输出格式
|
||
["知识点1", "知识点2", "知识点3"]
|
||
|
||
请直接输出 JSON 数组："""
|
||
|
||
for _attempt in range(2):
|
||
try:
|
||
response = self._call_llm(prompt)
|
||
if not response or not response.strip():
|
||
continue
|
||
|
||
# 使用多策略 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 []
|
||
|
||
def _assign_questions_to_kps(
|
||
self,
|
||
knowledge_points: List[Dict],
|
||
question_types: Dict[str, int],
|
||
total_questions: int
|
||
) -> List[tuple]:
|
||
"""
|
||
🔥 Step 3: 将题目分配到各知识点
|
||
|
||
🔥 改进策略：
|
||
1. 为每种题型单独分配知识点（确保题型覆盖）
|
||
2. 轮流分配知识点，避免重复
|
||
3. 每个知识点最多负责 1 种题型的 1-2 道题
|
||
|
||
Returns:
|
||
[(知识点名称, 题型dict, 题目数量), ...]
|
||
"""
|
||
kp_count = len(knowledge_points)
|
||
if kp_count == 0:
|
||
return []
|
||
|
||
# 🔥 新策略：为每种题型分配知识点
|
||
# 使用 dict 来累积每个知识点的题型分配
|
||
kp_assignments = {} # {kp_name: {q_type: count, ...}}
|
||
|
||
# 按题型依次分配
|
||
for q_type, type_count in question_types.items():
|
||
if type_count <= 0:
|
||
continue
|
||
|
||
assigned_for_type = 0
|
||
kp_idx = 0
|
||
|
||
while assigned_for_type < type_count:
|
||
# 循环使用知识点
|
||
kp_idx = kp_idx % kp_count
|
||
kp = knowledge_points[kp_idx]
|
||
kp_name = kp['name']
|
||
|
||
# 初始化该知识点的分配记录
|
||
if kp_name not in kp_assignments:
|
||
kp_assignments[kp_name] = {}
|
||
|
||
# 每个知识点对每种题型最多出 1 道
|
||
current_count = kp_assignments[kp_name].get(q_type, 0)
|
||
if current_count < 1:
|
||
kp_assignments[kp_name][q_type] = current_count + 1
|
||
assigned_for_type += 1
|
||
|
||
kp_idx += 1
|
||
|
||
# 防止无限循环（知识点不够用时）
|
||
if kp_idx > kp_count * type_count:
|
||
break
|
||
|
||
# 转换为 List[tuple] 格式
|
||
assignments = []
|
||
for kp_name, q_types in kp_assignments.items():
|
||
total = sum(q_types.values())
|
||
assignments.append((kp_name, q_types, total))
|
||
|
||
return assignments
|
||
|
||
def _retrieve_kp_chunks(
|
||
self,
|
||
knowledge_point: str,
|
||
all_chunks: List[Dict],
|
||
top_k: int = 5
|
||
) -> List[Dict]:
|
||
"""
|
||
🔥 Step 3.5: 根据知识点检索相关 chunks
|
||
|
||
策略：
|
||
1. 先尝试语义检索
|
||
2. 如果结果不足，用关键词匹配补充
|
||
"""
|
||
# 构建精准 query
|
||
query = f"{knowledge_point} 定义 规则 说明"
|
||
|
||
# 简单的关键词匹配（不调用向量检索，因为 chunks 已经是过滤后的）
|
||
scored_chunks = []
|
||
kp_lower = knowledge_point.lower()
|
||
|
||
for chunk in all_chunks:
|
||
content = chunk.get('content', '').lower()
|
||
score = 0
|
||
|
||
# 知识点直接匹配
|
||
if kp_lower in content:
|
||
score += 10
|
||
|
||
# 关键词匹配
|
||
for keyword in ['定义', '规则', '说明', '标准', '流程']:
|
||
if keyword in content:
|
||
score += 1
|
||
|
||
if score > 0:
|
||
scored_chunks.append((score, chunk))
|
||
|
||
# 按分数排序
|
||
scored_chunks.sort(key=lambda x: x[0], reverse=True)
|
||
|
||
# 如果匹配不足，补充前面的 chunks
|
||
result = [c for _, c in scored_chunks[:top_k]]
|
||
if len(result) < top_k:
|
||
for chunk in all_chunks:
|
||
if chunk not in result:
|
||
result.append(chunk)
|
||
if len(result) >= top_k:
|
||
break
|
||
|
||
return result
|
||
|
||
def _build_prompt_for_kp(
|
||
self,
|
||
source_content: str,
|
||
knowledge_point: str,
|
||
question_types: Dict[str, int],
|
||
difficulty: int
|
||
) -> str:
|
||
"""
|
||
🔥 Step 4: 构造针对知识点的出题 Prompt
|
||
"""
|
||
q_type_str = ", ".join(f"{t}:{n}道" for t, n in question_types.items())
|
||
|
||
return f"""基于以下文档内容，围绕知识点【{knowledge_point}】生成考试题目。
|
||
|
||
## 文档内容
|
||
{source_content}
|
||
|
||
## 出题要求
|
||
- 核心知识点：{knowledge_point}
|
||
- 难度等级：{difficulty}/5
|
||
- 题型及数量：{q_type_str}
|
||
|
||
## 🔥 关键约束
|
||
1. **必须围绕【{knowledge_point}】出题**，题目内容要与该知识点直接相关
|
||
2. **每道题必须基于不同的角度或细节**，避免重复考察同一内容
|
||
3. **严禁输出非 JSON 内容**
|
||
4. **必须从文档内容中找到依据**
|
||
|
||
## 输出格式（JSON 数组）
|
||
{self._get_format_examples()}
|
||
|
||
请直接输出 JSON 数组："""
|
||
|
||
def _deduplicate_and_balance(
|
||
self,
|
||
questions: List[Dict],
|
||
target_types: Dict[str, int]
|
||
) -> List[Dict]:
|
||
"""
|
||
🔥 Step 5: 去重 + 题型平衡
|
||
"""
|
||
if not questions:
|
||
return []
|
||
|
||
# 按题型分组
|
||
by_type = defaultdict(list)
|
||
for q in questions:
|
||
q_type = q.get('question_type', 'unknown')
|
||
by_type[q_type].append(q)
|
||
|
||
# 去重：同题型的题目，题干相似度高的去重
|
||
deduped = []
|
||
for q_type, q_list in by_type.items():
|
||
seen_stems = set()
|
||
type_questions = []
|
||
|
||
for q in q_list:
|
||
stem = q.get('content', {}).get('stem', '')
|
||
stem_key = stem[:50] # 用前50字作为去重key
|
||
|
||
if stem_key not in seen_stems:
|
||
seen_stems.add(stem_key)
|
||
type_questions.append(q)
|
||
|
||
# 按目标数量截取
|
||
target_count = target_types.get(q_type, 0)
|
||
deduped.extend(type_questions[:target_count])
|
||
|
||
return deduped
|
||
|
||
def _enrich_with_source_trace(
|
||
self,
|
||
questions: List[Dict],
|
||
chunks: List[Dict],
|
||
document_name: str
|
||
) -> List[Dict]:
|
||
"""
|
||
补充溯源信息
|
||
|
||
🔥 只返回 RAG 负责的字段，后端负责的字段（question_id, score, tags）不生成
|
||
"""
|
||
enriched = []
|
||
|
||
for q in questions:
|
||
# 找到引用的切片
|
||
referenced_chunks = find_referenced_chunks(q, chunks)
|
||
|
||
question = {
|
||
# 题型（RAG 负责）
|
||
"question_type": q.get('type'),
|
||
|
||
# 难度（RAG 负责）
|
||
"difficulty": q.get('difficulty', 3),
|
||
|
||
# 题目内容（RAG 负责）
|
||
"content": q.get('content', {}),
|
||
|
||
# 溯源信息（RAG 负责）
|
||
"source_trace": {
|
||
"document_name": document_name,
|
||
"chunk_ids": [c.get('chunk_id', '') for c in referenced_chunks],
|
||
"page_numbers": sorted(set([c.get('page') for c in referenced_chunks if c.get('page')])),
|
||
"sources": [
|
||
{
|
||
"chunk_id": c.get('chunk_id', ''),
|
||
"page": c.get('page'),
|
||
"section": c.get('section', ''),
|
||
"snippet": c['content'][:200] if c.get('content') else ''
|
||
}
|
||
for c in referenced_chunks
|
||
]
|
||
}
|
||
}
|
||
enriched.append(question)
|
||
|
||
return enriched
|
||
|
||
|
||
# ==================== 便捷函数 ====================
|
||
|
||
def generate_questions_from_content(
|
||
source_content: str,
|
||
document_name: str,
|
||
question_types: Dict[str, int],
|
||
difficulty: int = 3
|
||
) -> List[Dict]:
|
||
"""便捷函数：从文本内容生成题目（将纯文本包装为单 chunk 后调用结构化出题）"""
|
||
if not source_content or not source_content.strip():
|
||
return []
|
||
|
||
# 将纯文本包装为一个 chunk
|
||
chunks = [{"content": source_content, "section": "", "page": None, "chunk_id": "content_0"}]
|
||
generator = QuestionGenerator()
|
||
return generator.generate_questions_structured(chunks, document_name, question_types, difficulty)
|
||
|
||
|
||
def analyze_document_for_exam(chunks: List[Dict], max_total: int = None) -> Dict[str, Any]:
|
||
"""
|
||
AI 智能分析文档内容，决定适合的题型和数量
|
||
|
||
Args:
|
||
chunks: 文档切片列表
|
||
|
||
Returns:
|
||
{
|
||
"total_knowledge_points": int,
|
||
"suitable_types": List[str],
|
||
"question_types": Dict[str, int],
|
||
"reason": str
|
||
}
|
||
"""
|
||
if not chunks:
|
||
return {
|
||
"total_knowledge_points": 0,
|
||
"suitable_types": [],
|
||
"question_types": {},
|
||
"reason": "文档内容为空，无法出题"
|
||
}
|
||
|
||
generator = QuestionGenerator()
|
||
|
||
# Step 1: 按章节分组
|
||
section_map = group_chunks_by_section(chunks)
|
||
logger.info(f"文档共 {len(section_map)} 个章节，{len(chunks)} 个切片")
|
||
|
||
# Step 2: 统计每章节知识点数量
|
||
total_knowledge_points = 0
|
||
section_stats = []
|
||
|
||
for section_name, section_chunks in section_map.items():
|
||
# 合并章节内容
|
||
all_content = "\n".join(
|
||
c.get('content', '').strip()
|
||
for c in section_chunks
|
||
if c.get('content', '').strip()
|
||
)
|
||
|
||
# 统计知识点（简化版：根据内容长度估算）
|
||
if len(all_content) < 50:
|
||
kp_count = 0
|
||
elif len(all_content) < 200:
|
||
kp_count = 1
|
||
elif len(all_content) < 500:
|
||
kp_count = 2
|
||
else:
|
||
kp_count = min(5, len(all_content) // 200)
|
||
|
||
total_knowledge_points += kp_count
|
||
section_stats.append({
|
||
"name": section_name[:50],
|
||
"content_length": len(all_content),
|
||
"knowledge_points": kp_count
|
||
})
|
||
|
||
# Step 3: 调用 LLM 分析适合的题型和数量
|
||
prompt = f"""你是一个出题专家。根据以下文档分析结果，决定适合的题型和数量。
|
||
|
||
## 文档章节统计
|
||
共 {len(section_map)} 个章节，约 {total_knowledge_points} 个知识点
|
||
|
||
## 章节详情
|
||
{json.dumps(section_stats[:10], ensure_ascii=False, indent=2)}
|
||
|
||
## 内容示例（前3个章节）
|
||
"""
|
||
# 添加前3个章节的内容示例
|
||
for i, (section_name, section_chunks) in enumerate(list(section_map.items())[:3]):
|
||
content_preview = "\n".join(
|
||
c.get('content', '')[:100]
|
||
for c in section_chunks[:2]
|
||
if c.get('content')
|
||
)
|
||
prompt += f"\n### {section_name[:30]}\n{content_preview[:300]}\n"
|
||
|
||
question_limit = min(total_knowledge_points * 2, max_total if max_total else 20)
|
||
prompt += """
|
||
## 要求
|
||
根据文档内容特点，决定：
|
||
1. 适合哪些题型（single_choice/multiple_choice/true_false/fill_blank/subjective）
|
||
2. 每种题型出多少道题
|
||
3. 总题目数应与知识点数量匹配（知识点少则少出，多则多出）
|
||
|
||
## 输出格式（严格 JSON）
|
||
{
|
||
"total_knowledge_points": 知识点总数,
|
||
"suitable_types": ["题型1", "题型2"],
|
||
"question_types": {
|
||
"single_choice": 数量,
|
||
"multiple_choice": 数量,
|
||
"true_false": 数量,
|
||
"fill_blank": 数量,
|
||
"subjective": 数量
|
||
},
|
||
"reason": "分析理由（50字以内）"
|
||
}
|
||
|
||
注意：
|
||
- 不适合的题型数量设为 0
|
||
- 所有数量之和不要超过 %d
|
||
- 必须返回合法 JSON，不要有其他内容
|
||
|
||
请直接输出 JSON：""" % question_limit
|
||
|
||
try:
|
||
response = generator._call_llm(prompt)
|
||
if not response or not response.strip():
|
||
return _generate_default_question_types(total_knowledge_points)
|
||
|
||
# 使用多策略 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']
|
||
question_types = {}
|
||
for q_type in valid_types:
|
||
count = result.get('question_types', {}).get(q_type, 0)
|
||
if isinstance(count, int) and count >= 0:
|
||
question_types[q_type] = count
|
||
|
||
# 过滤不适合的题型
|
||
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,
|
||
"question_types": question_types,
|
||
"reason": result.get('reason', 'AI 分析完成')
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"AI 分析失败: {e}")
|
||
return _generate_default_question_types(total_knowledge_points)
|
||
|
||
|
||
def _generate_default_question_types(knowledge_points: int) -> Dict[str, Any]:
|
||
"""
|
||
根据知识点数量生成默认题型配置（AI 分析失败时的降级方案）
|
||
|
||
Args:
|
||
knowledge_points: 知识点数量
|
||
|
||
Returns:
|
||
与 analyze_document_for_exam 相同格式
|
||
"""
|
||
if knowledge_points <= 3:
|
||
question_types = {
|
||
"single_choice": 2,
|
||
"true_false": 2,
|
||
"fill_blank": 1,
|
||
"subjective": 0,
|
||
"multiple_choice": 0
|
||
}
|
||
reason = "知识点较少，以客观题为主"
|
||
elif knowledge_points <= 8:
|
||
question_types = {
|
||
"single_choice": 3,
|
||
"multiple_choice": 1,
|
||
"true_false": 2,
|
||
"fill_blank": 2,
|
||
"subjective": 1
|
||
}
|
||
reason = "知识点适中，题型均衡分布"
|
||
else:
|
||
question_types = {
|
||
"single_choice": 5,
|
||
"multiple_choice": 2,
|
||
"true_false": 3,
|
||
"fill_blank": 3,
|
||
"subjective": 2
|
||
}
|
||
reason = "知识点丰富，全面覆盖各类题型"
|
||
|
||
return {
|
||
"total_knowledge_points": knowledge_points,
|
||
"suitable_types": [t for t, c in question_types.items() if c > 0],
|
||
"question_types": question_types,
|
||
"reason": reason
|
||
}
|
||
|
||
|
||
# ==================== 重构版出题逻辑 ====================
|
||
|
||
def generate_questions_structured_v2(
|
||
chunks: List[Dict],
|
||
document_name: str,
|
||
question_types: Dict[str, int],
|
||
difficulty: int = 3,
|
||
exclude_stems: List[str] = None
|
||
) -> List[Dict]:
|
||
"""
|
||
🔥 重构版：结构化出题 v2
|
||
|
||
最佳设计：
|
||
1. Phase 1: 文档结构分析
|
||
2. Phase 2: 知识点规划（全局唯一）
|
||
3. Phase 3: 精准出题（每个知识点单独检索）
|
||
4. Phase 4: 质量校验
|
||
|
||
Args:
|
||
chunks: 文档切片列表
|
||
document_name: 文档名称
|
||
question_types: 题型及数量 {"single_choice": 5, ...}
|
||
difficulty: 难度 1-5
|
||
exclude_stems: 需要排除的已有题目题干列表（跨调用去重）
|
||
|
||
Returns:
|
||
题目列表
|
||
"""
|
||
total_questions = sum(question_types.values())
|
||
if not chunks or total_questions == 0:
|
||
return []
|
||
|
||
generator = QuestionGenerator()
|
||
|
||
# ========== Phase 1: 文档结构分析 ==========
|
||
logger.info("[v2] Phase 1: 文档结构分析")
|
||
section_map = group_chunks_by_section(chunks)
|
||
logger.info(f" 章节数: {len(section_map)}")
|
||
|
||
# ========== Phase 2: 知识点规划（全局唯一） ==========
|
||
logger.info("[v2] Phase 2: 知识点规划")
|
||
all_knowledge_points = []
|
||
seen_kp_names = set() # 全局去重
|
||
|
||
for section, section_chunks in section_map.items():
|
||
# 提取知识点
|
||
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']
|
||
if kp_name not in seen_kp_names:
|
||
seen_kp_names.add(kp_name)
|
||
kp['section'] = section # 标记来源章节
|
||
all_knowledge_points.append(kp)
|
||
|
||
logger.info(f" 全局唯一知识点: {len(all_knowledge_points)}")
|
||
|
||
if not all_knowledge_points:
|
||
# 降级：直接使用 chunks 出题（格式已由 fallback 内部统一）
|
||
logger.warning("[v2] 知识点提取失败，降级使用 chunks 出题")
|
||
fallback_questions = generator._generate_questions_fallback(chunks, document_name, question_types, difficulty)
|
||
# 降级路径也执行去重和数量平衡
|
||
deduped = _deduplicate_questions(fallback_questions, exclude_stems=exclude_stems)
|
||
final = _balance_question_types(deduped, question_types)
|
||
logger.info(f" [降级] 最终题目数: {len(final)}")
|
||
return final
|
||
|
||
# AI 分配：哪些知识点出什么题型
|
||
assignments = _ai_assign_question_types(all_knowledge_points, question_types)
|
||
logger.info(f" 分配任务: {len(assignments)} 个")
|
||
|
||
# ========== Phase 3: 精准出题 ==========
|
||
logger.info("[v2] Phase 3: 精准出题")
|
||
all_questions = []
|
||
|
||
for assignment in assignments:
|
||
kp_name = assignment['knowledge_point']
|
||
q_type = assignment['question_type']
|
||
kp_section = assignment.get('section', '')
|
||
|
||
# 找到该知识点的原始对象
|
||
kp_obj = next((kp for kp in all_knowledge_points if kp['name'] == kp_name), None)
|
||
if not kp_obj:
|
||
continue
|
||
|
||
# 根据章节找到相关 chunks
|
||
section_chunks = section_map.get(kp_section, [])
|
||
if not section_chunks:
|
||
section_chunks = chunks # 降级使用全部 chunks
|
||
|
||
# 精准检索该知识点的相关 chunks
|
||
kp_chunks = _retrieve_kp_chunks_v2(kp_name, section_chunks, top_k=5)
|
||
|
||
if not kp_chunks:
|
||
logger.warning(f" [v2] 知识点 [{kp_name[:20]}...] 无相关 chunks，跳过")
|
||
continue
|
||
|
||
# 构造上下文
|
||
source_content = build_source_context(kp_chunks)
|
||
|
||
# 出题 prompt
|
||
prompt = generator._build_prompt_for_kp(
|
||
source_content, kp_name, {q_type: 1}, difficulty
|
||
)
|
||
|
||
# 出题（带重试）
|
||
success, questions = generator._generate_with_retry(
|
||
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:
|
||
logger.warning(f" [v2] 知识点 [{kp_name[:20]}...] 出题失败")
|
||
|
||
# ========== Phase 4: 质量校验 ==========
|
||
logger.info("[v2] Phase 4: 质量校验")
|
||
|
||
# 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)
|
||
logger.info(f" 最终题目数: {len(final)}")
|
||
|
||
# 4.3 题型分布
|
||
type_counts = defaultdict(int)
|
||
for q in final:
|
||
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)}")
|
||
# 收集 final 中已有题干，补题时一并排除
|
||
existing_stems = [q.get('content', {}).get('stem', '') for q in final if q.get('content', {}).get('stem')]
|
||
combined_exclude = list(exclude_stems or []) + existing_stems
|
||
|
||
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=combined_exclude)
|
||
final.extend(extra_deduped[:shortage])
|
||
# 将新补的题干也加入排除列表，防止后续题型补出重复
|
||
for q in extra_deduped:
|
||
stem = q.get('content', {}).get('stem', '')
|
||
if stem:
|
||
combined_exclude.append(stem)
|
||
|
||
# 补题后重新统计
|
||
type_counts = defaultdict(int)
|
||
for q in final:
|
||
type_counts[q.get('question_type')] += 1
|
||
logger.info(f" 补题后题型分布: {dict(type_counts)}")
|
||
|
||
return final
|
||
|
||
|
||
def _ai_assign_question_types(
|
||
knowledge_points: List[Dict],
|
||
question_types: Dict[str, int]
|
||
) -> List[Dict]:
|
||
"""
|
||
AI 分配：哪些知识点出什么题型
|
||
|
||
策略：
|
||
1. 按章节轮询知识点（确保章节覆盖）
|
||
2. 每个知识点最多出 1 道题
|
||
3. 确保不同知识点都有覆盖
|
||
|
||
Args:
|
||
knowledge_points: 知识点列表
|
||
question_types: 题型及数量
|
||
|
||
Returns:
|
||
[{"knowledge_point": "...", "question_type": "...", "section": "..."}, ...]
|
||
"""
|
||
assignments = []
|
||
|
||
# 按章节分组知识点
|
||
kp_by_section = defaultdict(list)
|
||
for kp in knowledge_points:
|
||
section = kp.get('section', '未分类')
|
||
kp_by_section[section].append(kp['name'])
|
||
|
||
# 为每种题型分配知识点
|
||
for q_type, type_count in question_types.items():
|
||
if type_count <= 0:
|
||
continue
|
||
|
||
# 轮询章节
|
||
section_names = list(kp_by_section.keys())
|
||
assigned = 0
|
||
section_idx = 0
|
||
|
||
while assigned < type_count:
|
||
if not section_names:
|
||
break
|
||
|
||
# 当前章节
|
||
section = section_names[section_idx % len(section_names)]
|
||
kp_list = kp_by_section[section]
|
||
|
||
# 找一个还没分配过该题型的知识点
|
||
for kp_name in kp_list:
|
||
# 检查是否已经分配过
|
||
already_assigned = any(
|
||
a['knowledge_point'] == kp_name and a['question_type'] == q_type
|
||
for a in assignments
|
||
)
|
||
|
||
if not already_assigned:
|
||
assignments.append({
|
||
'knowledge_point': kp_name,
|
||
'question_type': q_type,
|
||
'section': section
|
||
})
|
||
assigned += 1
|
||
break
|
||
|
||
section_idx += 1
|
||
|
||
# 防止无限循环
|
||
if section_idx > len(section_names) * 2:
|
||
break
|
||
|
||
return assignments
|
||
|
||
|
||
def _retrieve_kp_chunks_v2(
|
||
knowledge_point: str,
|
||
section_chunks: List[Dict],
|
||
top_k: int = 5
|
||
) -> List[Dict]:
|
||
"""
|
||
v2 版本：根据知识点检索相关 chunks
|
||
|
||
改进：
|
||
1. 优先使用章节内的 chunks
|
||
2. 关键词匹配更精准
|
||
3. 支持多个关键词组合
|
||
|
||
Args:
|
||
knowledge_point: 知识点名称
|
||
section_chunks: 该章节的 chunks
|
||
top_k: 返回数量
|
||
|
||
Returns:
|
||
相关 chunks 列表
|
||
"""
|
||
if not section_chunks:
|
||
return []
|
||
|
||
# 提取关键词
|
||
keywords = knowledge_point.replace(' ', '').replace('，', ',').split(',')
|
||
keywords = [kw.strip() for kw in keywords if len(kw.strip()) >= 2]
|
||
|
||
# 评分
|
||
scored_chunks = []
|
||
for chunk in section_chunks:
|
||
content = chunk.get('content', '').lower()
|
||
score = 0
|
||
|
||
# 知识点直接匹配（最高优先级）
|
||
if knowledge_point.lower() in content:
|
||
score += 100
|
||
|
||
# 关键词匹配
|
||
for kw in keywords:
|
||
if kw.lower() in content:
|
||
score += 10
|
||
|
||
# 内容长度适中加分
|
||
content_len = len(content)
|
||
if 100 <= content_len <= 500:
|
||
score += 2
|
||
|
||
if score > 0:
|
||
scored_chunks.append((score, chunk))
|
||
|
||
# 按分数排序
|
||
scored_chunks.sort(key=lambda x: x[0], reverse=True)
|
||
|
||
# 返回 top_k
|
||
result = [c for _, c in scored_chunks[:top_k]]
|
||
|
||
# 如果不足，补充前面的 chunks
|
||
if len(result) < top_k:
|
||
for chunk in section_chunks:
|
||
if chunk not in result:
|
||
result.append(chunk)
|
||
if len(result) >= top_k:
|
||
break
|
||
|
||
return result
|
||
|
||
|
||
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:
|
||
_valid_types = ('single_choice', 'multiple_choice', 'true_false', 'fill_blank', 'subjective')
|
||
for stem in exclude_stems:
|
||
seen_stems.add(stem[:80])
|
||
# 用 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 = []
|
||
|
||
for q in questions:
|
||
content = q.get('content', {})
|
||
stem = content.get('stem', '')
|
||
|
||
# 题干去重（前 80 字）
|
||
stem_key = stem[:80]
|
||
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:
|
||
continue
|
||
|
||
seen_stems.add(stem_key)
|
||
seen_kp_type.add(kp_type_key)
|
||
deduped.append(q)
|
||
|
||
return deduped
|
||
|
||
|
||
def _balance_question_types(
|
||
questions: List[Dict],
|
||
target_types: Dict[str, int]
|
||
) -> List[Dict]:
|
||
"""
|
||
题型平衡：按目标数量截取
|
||
|
||
Args:
|
||
questions: 题目列表
|
||
target_types: 目标题型数量
|
||
|
||
Returns:
|
||
平衡后的题目列表
|
||
"""
|
||
# 按题型分组
|
||
by_type = defaultdict(list)
|
||
for q in questions:
|
||
q_type = q.get('question_type', 'unknown')
|
||
by_type[q_type].append(q)
|
||
|
||
# 按目标数量截取
|
||
result = []
|
||
for q_type, target_count in target_types.items():
|
||
type_questions = by_type.get(q_type, [])
|
||
result.extend(type_questions[:target_count])
|
||
|
||
return result
|