import requests
import time
import random
import json
import os
categories = [
{"name": "结婚登记与条件", "desc": "涉及结婚年龄、自愿原则、禁止近亲结婚、疾病限制、登记程序及法律效力等。"},
{"name": "夫妻权利义务", "desc": "涵盖共同财产管理、相互扶养义务、姓名权、生育权及日常家事代理权等。"},
{"name": "离婚程序与条件", "desc": "包括协议离婚冷静期、诉讼离婚标准、分居认定、感情破裂证明及调解程序等。"},
{"name": "财产分割", "desc": "涉及共同财产界定、个人财产保护、债务承担、房产分割规则及隐匿财产追责等。"},
{"name": "子女抚养与监护权", "desc": "包含抚养费计算标准、探视权实施、非婚生子女权益、抚养关系变更及教育责任等。"},
{"name": "家庭暴力与保护令", "desc": "涵盖暴力行为认定、人身保护令申请、证据收集、紧急庇护措施及刑事责任关联等。"},
{"name": "继承权与婚姻关系", "desc": "涉及配偶继承顺位、遗嘱效力、遗产分割冲突、再婚继承权及代位继承问题等。"},
{"name": "无效婚姻与可撤销婚姻", "desc": "包括重婚无效、胁迫婚姻撤销、隐瞒疾病撤销的程序及法律后果等。"},
{"name": "涉外婚姻", "desc": "涉及跨国婚姻登记、域外结婚效力认定、财产跨境分割及国际子女抚养公约适用等。"},
{"name": "法律责任与救济措施", "desc": "包含虚假登记处罚、拒不执行判决后果、损害赔偿计算及强制执行程序等。"}
]
SYSTEM_PROMPT_TEMPLATE = (
"你是一位婚姻法方面的专家，根据【{category}】的定义，"
"针对其中夫妻权利义务方面的规定，向学生提供20道练习题，"
"你能提出哪些问题？每个问题单独一行，"
"只允许输出问题本身，不允许输出任何其他字符。"
)
# 每个类别目标生成数量
TARGET_COUNT_PER_CATEGORY = 5
# 每次生成数量
BATCH_SIZE = 2
def callServer(systemPrompt, userPrompt, model="deepseek-ai/DeepSeek-V3.2"):
url = "https://api.siliconflow.cn/v1/chat/completions"
headers = {
"Authorization": f"Bearer sk-cgteigkshpgccdbsfzefmvdjpaicjmydidliorsnjorbqclg",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [
{"role": "system", "content": systemPrompt},
{"role": "user", "content": userPrompt}
],
"temperature": 0.6,
"max_tokens": 4000
}
success = True
msg = "OK"
think = ""
result = ""
try:
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
resp_json = response.json()
msg_dict = resp_json['choices'][0]['message']
# V3.2 没有 reasoning_content 字段，安全获取
think = msg_dict.get('reasoning_content', '')
result = msg_dict['content']
except Exception as e:
success = False
msg = "API调用异常:" + str(e)
return success, msg, think, result
def callServerStimu(systemPrompt, userPrompt, model="deepseek-ai/DeepSeek-V3.2"):
time.sleep(random.randint(1, 3))
success = True
msg = "OK"
think = ''
result = '\n'.join([f'关于{userPrompt[:15]}的练习题第{i+1}题？' for i in range(20)])
return success, msg, think, result
def loadExistingQuestions(targetFile):
"""读取已有结果文件，用于断点续传和去重"""
existing = {} # {category_name: set(question_str)}
existing_ids = {} # {category_name: count}
if os.path.exists(targetFile):
with open(targetFile, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if not line:
continue
try:
data = json.loads(line)
cate = data.get("category", "")
instr = data.get("instruction", "")
if cate and instr:
if cate not in existing:
existing[cate] = set()
existing_ids[cate] = 0
existing[cate].add(instr)
existing_ids[cate] = max(existing_ids[cate], data.get("id", 0))
except json.JSONDecodeError:
continue
return existing, existing_ids
def generateQuestions(targetFile, useStimu=False):
# 加载已有数据（断点续传 + 去重）
existingQuestions, existingIds = loadExistingQuestions(targetFile)
# 初始化各类别的去重集合和当前最大ID
questionSets = {}
currentId = 1
for cat in categories:
name = cat["name"]
if name in existingQuestions:
questionSets[name] = existingQuestions[name]
if name in existingIds:
currentId = max(currentId, existingIds[name] + 1)
else:
questionSets[name] = set()
print(f"断点续传：已有数据，当前起始ID = {currentId}")
# 以追加模式打开文件，增量写入
fileExisted = os.path.exists(targetFile)
with open(targetFile, 'a', encoding='utf-8') as f:
if not fileExisted:
f.write("")
for cat in categories:
name = cat["name"]
desc = cat["desc"]
systemPrompt = SYSTEM_PROMPT_TEMPLATE.format(category=name)
currentCount = len(questionSets[name])
batchNum = 0
maxBatches = TARGET_COUNT_PER_CATEGORY // BATCH_SIZE # 25轮
print(f"\n{'='*60}")
print(f"类别: {name} | 已有: {currentCount} | 目标: {TARGET_COUNT_PER_CATEGORY}")
print(f"{'='*60}")
# 如果已有数量已达标，跳过
if currentCount >= TARGET_COUNT_PER_CATEGORY:
print(f" {name} 已达到目标数量，跳过")
continue
while currentCount < TARGET_COUNT_PER_CATEGORY and batchNum < maxBatches:
batchNum += 1
# 构造user prompt，告知已有数量，引导生成不同问题
userPrompt = (
f"类别：{name}\n"
f"描述：{desc}\n\n"
f"请根据以上类别描述，针对{name}方面的规定，生成20道练习题。"
f"要求每道题角度不同，尽量覆盖该类别的各个方面。"
)
if useStimu:
success, msg, think, result = callServerStimu(systemPrompt, userPrompt)
else:
success, msg, think, result = callServer(systemPrompt, userPrompt)
print(f" [{name}] 第 {batchNum}/{maxBatches} 轮: success={success}, msg={msg}")
if success:
lines = result.strip().split('\n')
newCount = 0
for line in lines:
line = line.strip()
# 去除可能的序号前缀（如 "1." "2、" 等）
if len(line) > 0:
# 去掉开头数字序号
cleaned = line
for prefix in ['1.', '2.', '3.', '4.', '5.', '6.', '7.', '8.', '9.', '10.',
'11.', '12.', '13.', '14.', '15.', '16.', '17.', '18.', '19.', '20.',
'1、', '2、', '3、', '4、', '5、', '6、', '7、', '8、', '9、', '10、',
'11、', '12、', '13、', '14、', '15、', '16、', '17、', '18、', '19、', '20、',
'1）', '2）', '3）', '4）', '5）', '6）', '7）', '8）', '9）', '10）',
'11）', '12）', '13）', '14）', '15）', '16）', '17）', '18）', '19）', '20）']:
if cleaned.startswith(prefix):
cleaned = cleaned[len(prefix):].strip()
break
if len(cleaned) > 0 and cleaned not in questionSets[name]:
questionSets[name].add(cleaned)
record = {
"id": currentId,
"category": name,
"instruction": cleaned
}
f.write(json.dumps(record, ensure_ascii=False) + '\n')
f.flush()
currentId += 1
newCount += 1
currentCount = len(questionSets[name])
print(f" 新增有效问题: {newCount} | 累计去重后: {currentCount}/{TARGET_COUNT_PER_CATEGORY}")
else:
print(f" think: {think}")
print(f" result: {result}")
# 避免请求过于频繁
time.sleep(1)
print(f" {name} 完成，共 {currentCount} 道题目")
total = sum(len(questionSets[cat["name"]]) for cat in categories)
print(f"\n{'='*60}")
print(f"全部完成！共生成 {total} 道题目，已保存到 {targetFile}")
print(f"{'='*60}")
return total
if __name__ == "__main__":
targetFile = "../data/testDataTemp/queryData.jsonl"
# useStimu=True 可用模拟模式测试流程，不消耗API额度
generateQuestions(targetFile, useStimu=False)