# -*- 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)