# -*- coding: utf-8 -*-
import requests
import time
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": "包含虚假登记处罚、拒不执行判决后果、损害赔偿计算及强制执行程序等。"}
]
# 构建类别说明文本
CATEGORY_LIST_TEXT = "\n".join([f'{c["name"]}：{c["desc"]}' for c in categories])
# 提示词模板（{category} 会被替换为实际问题所属的类别名）
PROMPT_TEMPLATE = (
"你是一位婚姻法方面的专家，如果要把【输入文本】分类到【特定类别】中，"
"那么【输入文本】属于{category}是否准确？给出一个准确性评分，"
"该评分是0到9之间的正整数，其中0表示完全错误，9表示非常准确。"
"【输入文本】中的每一行是一条独立数据，需要单独判定。"
"输出时每条数据的结果单独输出一行，"
"只允许输出准确性评分，不允许输出其他任何字符。\n\n"
"【特定类别】\n"
'["结婚登记与条件","夫妻权利义务","离婚程序与条件","财产分割",'
'"子女抚养与监护权","家庭暴力与保护令","继承权与婚姻关系",'
'"无效婚姻与可撤销婚姻","涉外婚姻","法律责任与救济措施"]\n'
"各类别详细说明如下：\n"
"{category_list}\n\n"
"【输入文本】\n"
"{input_text}"
)
# 每批处理数量
BATCH_SIZE = 20
# ==================== API 调用 ====================
def callServer(prompt, model="Qwen/Qwen2.5-72B-Instruct"):
"""批量调用，返回多行评分文本"""
url = "https://api.siliconflow.cn/v1/chat/completions"
headers = {
"Authorization": f"Bearer sk-cgteigkshpgccdbsfzefmvdjpaicjmydidliorsnjorbqclg",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.2,
"max_tokens": 200
}
success = True
msg = "OK"
result = ""
try:
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
result = response.json()['choices'][0]['message']['content']
except Exception as e:
success = False
msg = "API调用异常:" + str(e)
return success, msg, result
def callServerSingle(prompt, model="Qwen/Qwen2.5-72B-Instruct"):
"""单条调用，返回单个整数评分"""
url = "https://api.siliconflow.cn/v1/chat/completions"
headers = {
"Authorization": f"Bearer sk-cgteigkshpgccdbsfzefmvdjpaicjmydidliorsnjorbqclg",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.2,
"max_tokens": 10
}
success = True
msg = "OK"
score = -1
try:
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
content = response.json()['choices'][0]['message']['content'].strip()
score = int(content)
except Exception as e:
success = False
msg = "API调用异常:" + str(e)
return success, msg, score
# ==================== 评分解析 ====================
def parseScores(result, expectedCount):
"""解析模型返回的多行评分，每行一个0-9的整数"""
scores = []
lines = result.strip().split('\n')
for line in lines:
line = line.strip()
if len(line) == 0:
continue
try:
score = int(line)
if 0 <= score <= 9:
scores.append(score)
else:
scores.append(-1)
except ValueError:
scores.append(-1)
return scores
# ==================== 单条打分提示词构造 ====================
def buildSinglePrompt(instruction, category):
"""构造单条打分的提示词（批量解析失败时的回退方案）"""
return (
"你是一位婚姻法方面的专家，如果要把【输入文本】分类到【特定类别】中，"
"那么【输入文本】属于" + category + "是否准确？给出一个准确性评分，"
"该评分是0到9之间的正整数，其中0表示完全错误，9表示非常准确。"
"只允许输出准确性评分，不允许输出其他任何字符。\n\n"
"【特定类别】\n"
'["结婚登记与条件","夫妻权利义务","离婚程序与条件","财产分割",'
'"子女抚养与监护权","家庭暴力与保护令","继承权与婚姻关系",'
'"无效婚姻与可撤销婚姻","涉外婚姻","法律责任与救济措施"]\n'
"各类别详细说明如下：\n"
+ CATEGORY_LIST_TEXT + "\n\n"
"【输入文本】\n"
+ instruction
)
# ==================== 主流程 ====================
def scoreQuestions(targetFile):
# ---------- 读取全部记录 ----------
records = []
with open(targetFile, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if line:
records.append(json.loads(line))
print(f"共读取 {len(records)} 条记录")
# ---------- 按类别分组 ----------
byCategory = {}
for i, rec in enumerate(records):
cate = rec.get("category", "")
if cate not in byCategory:
byCategory[cate] = []
byCategory[cate].append(i)
# ---------- 逐类别处理 ----------
for cate, indices in byCategory.items():
# 过滤已打分的记录（断点续传）
pending = [i for i in indices if "questionScore" not in records[i]]
if not pending:
print(f"类别[{cate}] 全部已打分，跳过")
continue
print(f"\n{'='*60}")
print(f"类别[{cate}]: 待打分 {len(pending)} 条")
print(f"{'='*60}")
# ---------- 分批处理 ----------
for batchStart in range(0, len(pending), BATCH_SIZE):
batchIndices = pending[batchStart:batchStart + BATCH_SIZE]
batchQuestions = [records[i]["instruction"] for i in batchIndices]
inputText = "\n".join(batchQuestions)
prompt = PROMPT_TEMPLATE.format(
category=cate,
category_list=CATEGORY_LIST_TEXT,
input_text=inputText
)
batchNum = batchStart // BATCH_SIZE + 1
totalBatches = (len(pending) + BATCH_SIZE - 1) // BATCH_SIZE
print(f"\n 批次 {batchNum}/{totalBatches} ({len(batchQuestions)} 条)")
# ---------- 重试机制 ----------
scores = None
for attempt in range(3):
success, msg, result = callServer(prompt)
if success:
scores = parseScores(result, len(batchQuestions))
if len(scores) == len(batchQuestions):
print(f" 批量打分成功: {scores}")
break
else:
print(f" 评分数量不匹配(期望{len(batchQuestions)}, 实际{len(scores)}), "
f"第{attempt+1}次重试...")
scores = None
else:
print(f" API调用失败(第{attempt+1}次): {msg}")
time.sleep(2)
# ---------- 批量失败则回退逐条打分 ----------
if scores is None:
print(f" 批量打分失败，回退为逐条打分...")
scores = []
for j, idx in enumerate(batchIndices):
instruction = records[idx]["instruction"]
singlePrompt = buildSinglePrompt(instruction, cate)
singleScore = -1
for attempt in range(3):
success, msg, singleScore = callServerSingle(singlePrompt)
if success and 0 <= singleScore <= 9:
break
print(f" 单条打分失败(第{attempt+1}次): {msg}")
time.sleep(2)
scores.append(singleScore)
print(f" [{j + 1}/{len(batchIndices)}] score={singleScore}")
time.sleep(0.5)
# ---------- 写入评分并增量保存 ----------
for j, idx in enumerate(batchIndices):
records[idx]["questionScore"] = scores[j]
with open(targetFile, 'w', encoding='utf-8') as f:
for rec in records:
f.write(json.dumps(rec, ensure_ascii=False) + '\n')
print(f" 已保存到文件")
time.sleep(1)
# ---------- 最终统计 ----------
scored = sum(1 for r in records if "questionScore" in r)
print(f"\n{'='*60}")
print(f"打分完成: {scored}/{len(records)} 条已打分")
# 按类别统计平均分
print(f"\n各类别评分统计:")
for cate, indices in byCategory.items():
cateScores = [records[i].get("questionScore", -1) for i in indices]
validScores = [s for s in cateScores if s >= 0]
if validScores:
avg = sum(validScores) / len(validScores)
print(f" {cate}: 平均分={avg:.1f}, 数量={len(validScores)}/{len(indices)}")
else:
print(f" {cate}: 无有效评分")
if __name__ == "__main__":
targetFile = "../data/testDataTemp/queryData.jsonl"
scoreQuestions(targetFile)