Xiang.L
/
chaitiao_and_ASR
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
3 Commits
2 Branches
0 Tags
611 MiB
 
Branch: master
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from 'master'
${ noResults }
chaitiao_and_ASR/4.结构化json-Ai标注/Ai-continue.py

581 lines
22 KiB
Raw Permalink Blame History

import requests
import json
import pandas as pd
from typing import List, Dict
import time
class SimpleOpenAIHubClient:
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://api.openai-hub.com"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def chat(self, prompt, model="gpt-4.1"):
"""发送prompt并返回模型回答"""
payload = {
"model": model,
"messages": [
{
"role": "user",
"content": prompt
}
],
"max_tokens": 32768,
"temperature": 0.7
}
try:
response = requests.post(
f"{self.base_url}/v1/chat/completions",
headers=self.headers,
json=payload,
timeout=60
)
if response.status_code == 200:
result = response.json()
return result['choices'][0]['message']['content']
else:
return f"错误: {response.status_code} - {response.text}"
except requests.exceptions.RequestException as e:
return f"请求异常: {str(e)}"
print("AI客户端类定义完成！")
# 设置API Key
API_KEY = "sk-XREp2jnIXyZ6UoCnzZeO0ahmLi9OEXuVAtFLojKFpG9gCZ4e" # 请替换为你的实际API Key
# 初始化AI客户端
client = SimpleOpenAIHubClient(API_KEY)
print("AI模型加载完成！")
# 定义批量标注的Prompt模板
BATCH_SEGMENTATION_PROMPT = """你是一个专业的广播内容段落分割标注员。你的任务是批量判断多个相邻句子对之间是否应该进行段落分割，以便广播员更好地掌握停顿和语调变化。
**完整文本内容上下文：**
{context_text}
**标注规则：**
- 标签0：两个句子属于同一段落，连续播报，轻微停顿
- 标签1：两个句子属于不同段落，需要明显停顿或语调转换
**重要标注要求（请严格遵循）：**
- 如果整个文本内容都在讲同一个事，你有理由只输出一段，不是追求分的段越多越细就越好
- 每个分段必须保持原始语句的绝对顺序
- 最终分段数可能等于或小于原始语句数量
- 必须保留所有原始语句文本，不得遗漏任何内容
- 应客户强烈要求，他们需要的是较粗的分段，不要太细，如同一条通告，不需要分段成具体的每个条款之类的，只需要将整个相同的通告分成一段
- 优先考虑较粗的分段，避免过度细分
**广播分段判断依据（偏向粗分段）：**
1. **重大主题转换**：从一个完全不同的话题转向另一个话题（如从天气预报转向安全通知）
2. **文档类型变化**：从一个完整文档转向另一个完整文档（如从禁火令转向倡议书）
3. **内容性质变化**：从通知类内容转向完全不同性质的内容（如从法规转向天气预报）
4. **广播节目段落**：明显的广播节目结构变化（如开场白结束进入正式内容）
5. **分点阐述结构**：标题和所有分点条目内容应该合并为一个完整段落（如"森林防火十不准，一不乱扔烟头，二不随意丢弃火种，三不在林区吸烟"等整体合成一段）
**广播内容特别注意（粗分段原则）：**
- 整个通告、法令、倡议书等应作为一个段落，不要拆分条款
- 同一主题的多个条款应保持在同一段落
- 只有在完全不同的文档或重大主题转换时才分段
- 广播开场白可以独立成段，但具体内容尽量合并
- 同一类型的预报信息（如天气预报的不同地区）应保持在同一段
- **分点阐述内容的特殊处理**：
- 标题性内容（如"森林防火十不准"）与分点条目内容之间不需要分段
- 标题和所有的分点条目（如"一不乱扔烟头"、"二不随意丢弃火种"、"三不在林区吸烟"等）应该合并为一个完整段落
- 分点条目之间不需要分段，应该连续播报
- 整个分点阐述结构作为一个完整的内容单元，保持连贯性
**批量标注说明：**
- 每个句子对都有一个source_id，表示来源文档
- 请保持原有的source_id不变
- 将label从-1改为实际的标注结果（0或1）
- 为每个句子对提供简要的分段理由
- 结合上述完整文本内容理解句子对的上下文语境
- **特别重要：倾向于标注更多的0（同段落），减少1（分段）的使用，分点阐述结构应保持为一个完整段落**
现在请对以下句子对进行批量标注：
{batch_sentence_pairs}
请直接输出标注结果，格式如下：
```json
[
{{
"sentence1": "...",
"sentence2": "...",
"label": 0或1,
"reason": "广播分段理由",
"source_id": 原有的source_id
}}
]
```
只输出JSON数据，不要其他说明文字。"""
def load_context_data(csv_file="batch_deduplication_results_619-1103_01.csv"):
"""
从batch_deduplication_results_619-1103_01.csv加载上下文数据
Args:
csv_file: CSV文件路径
Returns:
字典，key为id，value为final_processed_text
"""
try:
print(f"正在读取上下文数据文件: {csv_file}")
# 尝试不同的编码格式
encodings = ['utf-8', 'gbk', 'gb2312', 'utf-8-sig', 'latin-1', 'cp1252']
context_df = None
for encoding in encodings:
try:
print(f" 尝试使用 {encoding} 编码...")
context_df = pd.read_csv(csv_file, encoding=encoding)
print(f" ✓ 成功使用 {encoding} 编码读取文件")
break
except UnicodeDecodeError:
print(f" {encoding} 编码失败")
continue
except Exception as e:
print(f" {encoding} 编码读取时出现其他错误：{str(e)}")
continue
if context_df is None:
print(f"✗ 错误：尝试了所有编码格式都无法读取文件 {csv_file}")
return {}
if 'id' not in context_df.columns or 'final_processed_text' not in context_df.columns:
print(f"✗ 错误：CSV文件缺少必需列")
print(f" 需要的列: ['id', 'final_processed_text']")
print(f" 实际的列: {list(context_df.columns)}")
return {}
# 创建id到final_processed_text的映射
context_dict = {}
for _, row in context_df.iterrows():
context_dict[row['id']] = row['final_processed_text'] if pd.notna(row['final_processed_text']) else ""
print(f"✓ 成功加载上下文数据")
print(f" - 可用ID数量: {len(context_dict)}")
print(f" - 可用ID列表: {sorted(context_dict.keys())}")
return context_dict
except FileNotFoundError:
print(f"✗ 警告：找不到上下文文件 {csv_file}")
print(" 将在没有上下文的情况下进行标注")
return {}
except Exception as e:
print(f"✗ 读取上下文文件时出错: {str(e)}")
print(" 将在没有上下文的情况下进行标注")
return {}
def load_failed_data_from_json(json_file="segmentation_results_from_7.json"):
"""
从JSON结果文件中加载标注失败的数据
Args:
json_file: JSON结果文件路径
Returns:
失败数据列表
"""
try:
print(f"正在读取JSON结果文件: {json_file}")
with open(json_file, 'r', encoding='utf-8') as f:
all_results = json.load(f)
print(f"✓ 成功加载JSON文件")
print(f" - 总结果数量: {len(all_results)}")
# 筛选出失败的数据（label为-1）
failed_data = [item for item in all_results if item.get('label') == -1]
successful_data = [item for item in all_results if item.get('label') in [0, 1]]
print(f" - 成功标注数量: {len(successful_data)}")
print(f" - 失败标注数量: {len(failed_data)}")
if len(failed_data) == 0:
print("✓ 没有发现失败的标注数据，无需重新处理")
return [], all_results
# 统计失败原因
from collections import Counter
failure_reasons = Counter(item.get('reason', '未知错误') for item in failed_data)
print(f"\n失败原因统计:")
for reason, count in failure_reasons.most_common():
print(f" - {reason}: {count}次")
# 统计涉及的source_id
failed_source_ids = sorted(set(item.get('source_id') for item in failed_data))
print(f"\n涉及的source_id: {failed_source_ids}")
return failed_data, all_results
except FileNotFoundError:
print(f"✗ 错误：找不到文件 {json_file}")
return [], []
except json.JSONDecodeError as e:
print(f"✗ 错误：JSON文件格式错误 - {str(e)}")
return [], []
except Exception as e:
print(f"✗ 错误：读取JSON文件时出现异常 - {str(e)}")
return [], []
def convert_failed_data_to_sentence_pairs(failed_data):
"""
将失败数据转换为句子对格式，供重新标注使用
Args:
failed_data: 失败数据列表
Returns:
句子对格式的数据列表
"""
sentence_pairs_data = []
for item in failed_data:
sentence_pair = {
"sentence1": item.get("sentence1", ""),
"sentence2": item.get("sentence2", ""),
"source_id": item.get("source_id"),
"label": -1 # 标记为待标注
}
sentence_pairs_data.append(sentence_pair)
return sentence_pairs_data
def process_batch_segmentation(sentence_pairs_data, context_dict, batch_size=8):
"""
批量处理句子对的段落分割标注
Args:
sentence_pairs_data: 句子对数据列表
context_dict: 上下文数据字典
batch_size: 每批处理的数量
Returns:
处理结果列表
"""
all_results = []
total_pairs = len(sentence_pairs_data)
print(f"开始批量标注，总共 {total_pairs} 个句子对")
print(f"每批处理 {batch_size} 个句子对")
# 分批处理
for i in range(0, total_pairs, batch_size):
batch_end = min(i + batch_size, total_pairs)
current_batch = sentence_pairs_data[i:batch_end]
print(f"\n处理第 {i // batch_size + 1} 批 (句子对 {i + 1}-{batch_end})")
try:
# 获取当前批次涉及的source_id的上下文
source_ids_in_batch = set(pair['source_id'] for pair in current_batch)
context_text = ""
for source_id in sorted(source_ids_in_batch):
if source_id in context_dict and context_dict[source_id]:
context_text += f"\n--- Source ID {source_id} 完整文本内容 ---\n"
context_text += context_dict[source_id] # 完整内容，不截断
context_text += "\n"
else:
context_text += f"\n--- Source ID {source_id} ---\n(未找到对应的完整文本内容)\n"
# 准备当前批次的数据
batch_json = json.dumps(current_batch, ensure_ascii=False, indent=2)
# 构建prompt
prompt = BATCH_SEGMENTATION_PROMPT.format(
context_text=context_text,
batch_sentence_pairs=batch_json
)
print(f"发送请求到AI模型...")
print(f" - 涉及source_id: {sorted(source_ids_in_batch)}")
print(f" - 上下文长度: {len(context_text)} 字符")
print(f" - Prompt总长度: {len(prompt)} 字符")
# 调用AI模型
ai_response = client.chat(prompt)
print(f"收到模型响应")
# 尝试解析JSON响应
try:
# 提取JSON部分（去除可能的markdown格式）
json_start = ai_response.find('[')
json_end = ai_response.rfind(']') + 1
if json_start != -1 and json_end != 0:
json_content = ai_response[json_start:json_end]
batch_results = json.loads(json_content)
# 验证结果
if isinstance(batch_results, list) and len(batch_results) == len(current_batch):
all_results.extend(batch_results)
print(f"✓ 成功处理 {len(batch_results)} 个句子对")
else:
print(
f"✗ 响应格式不正确，期望 {len(current_batch)} 个结果，实际得到 {len(batch_results) if isinstance(batch_results, list) else 'non-list'}")
# 添加错误记录
for j, pair in enumerate(current_batch):
all_results.append({
"sentence1": pair["sentence1"],
"sentence2": pair["sentence2"],
"label": -1,
"reason": "重试失败：响应格式错误",
"source_id": pair["source_id"]
})
else:
print(f"✗ 无法找到有效的JSON响应")
print(f"原始响应前200字符: {ai_response[:200]}...")
# 添加错误记录
for j, pair in enumerate(current_batch):
all_results.append({
"sentence1": pair["sentence1"],
"sentence2": pair["sentence2"],
"label": -1,
"reason": "重试失败：JSON解析错误",
"source_id": pair["source_id"]
})
except json.JSONDecodeError as e:
print(f"✗ JSON解析失败: {str(e)}")
print(f"原始响应: {ai_response[:200]}...")
# 添加错误记录
for j, pair in enumerate(current_batch):
all_results.append({
"sentence1": pair["sentence1"],
"sentence2": pair["sentence2"],
"label": -1,
"reason": f"重试失败：{str(e)}",
"source_id": pair["source_id"]
})
# 添加延时，避免API调用过于频繁
time.sleep(2)
except Exception as e:
print(f"✗ 批次处理出错: {str(e)}")
# 添加错误记录
for j, pair in enumerate(current_batch):
all_results.append({
"sentence1": pair["sentence1"],
"sentence2": pair["sentence2"],
"label": -1,
"reason": f"重试异常：{str(e)}",
"source_id": pair["source_id"]
})
return all_results
def merge_results(original_results, retry_results):
"""
合并原始结果和重试结果
Args:
original_results: 原始完整结果列表
retry_results: 重试结果列表
Returns:
合并后的完整结果列表
"""
print("正在合并结果...")
# 创建重试结果的映射，用于快速查找
retry_map = {}
for result in retry_results:
key = (result['sentence1'], result['sentence2'], result['source_id'])
retry_map[key] = result
merged_results = []
replaced_count = 0
for original_result in original_results:
key = (original_result['sentence1'], original_result['sentence2'], original_result['source_id'])
# 如果原始结果是失败的，并且重试结果存在，则用重试结果替换
if original_result.get('label') == -1 and key in retry_map:
merged_results.append(retry_map[key])
replaced_count += 1
else:
merged_results.append(original_result)
print(f"✓ 合并完成，替换了 {replaced_count} 个失败结果")
return merged_results
# 执行重新标注失败数据
print("=" * 60)
print("开始重新标注失败数据")
print("=" * 60)
# 加载上下文数据
context_dict = load_context_data("batch_deduplication_results_619-1103_01.csv")
# 从JSON文件加载失败数据
failed_data, original_results = load_failed_data_from_json("segmentation_results_from_7.json")
if len(failed_data) == 0:
print("没有需要重新标注的数据，程序结束。")
exit()
print(f"\n开始重新标注 {len(failed_data)} 个失败的句子对...")
# 将失败数据转换为句子对格式
retry_sentence_pairs = convert_failed_data_to_sentence_pairs(failed_data)
# 按source_id分组处理（与原始代码保持一致的处理方式）
from collections import defaultdict
grouped_by_source_id = defaultdict(list)
for pair in retry_sentence_pairs:
grouped_by_source_id[pair['source_id']].append(pair)
print(f"失败数据涉及 {len(grouped_by_source_id)} 个source_id")
# 处理每个source_id的失败数据
all_retry_results = []
for source_id in sorted(grouped_by_source_id.keys()):
current_sentence_pairs = grouped_by_source_id[source_id]
print(f"\n{'=' * 50}")
print(f"重新处理 source_id: {source_id}")
print(f"{'=' * 50}")
print(f" - 该source_id的失败句子对数量: {len(current_sentence_pairs)}")
# 显示第一个句子对预览
if len(current_sentence_pairs) > 0:
first_pair = current_sentence_pairs[0]
print(f" - 第一个句子对预览:")
print(f" 句子1: {first_pair['sentence1'][:60]}...")
print(f" 句子2: {first_pair['sentence2'][:60]}...")
# 检查上下文数据
if source_id in context_dict and context_dict[source_id]:
print(f" - 上下文数据: 可用，长度 {len(context_dict[source_id])} 字符")
print(f" - 上下文预览: {context_dict[source_id][:100]}...")
else:
print(f" - 上下文数据: 不可用")
try:
print(f" - 开始重新标注...")
# 执行批量标注（batch_size=8）
current_results = process_batch_segmentation(current_sentence_pairs, context_dict, batch_size=8)
# 添加到总结果中
all_retry_results.extend(current_results)
# 统计当前source_id的结果
current_successful = len([r for r in current_results if r['label'] in [0, 1]])
current_failed = len([r for r in current_results if r['label'] == -1])
print(f" ✓ source_id {source_id} 重新标注完成")
print(f" - 成功标注: {current_successful}")
print(f" - 仍然失败: {current_failed}")
if current_successful > 0:
current_label_0 = len([r for r in current_results if r['label'] == 0])
current_label_1 = len([r for r in current_results if r['label'] == 1])
print(f" - 标签0（同段落）: {current_label_0}")
print(f" - 标签1（分段）: {current_label_1}")
# 添加延时
time.sleep(1)
except Exception as e:
print(f" ✗ source_id {source_id} 重新标注失败: {str(e)}")
# 为失败的source_id添加错误记录
for pair in current_sentence_pairs:
all_retry_results.append({
"sentence1": pair["sentence1"],
"sentence2": pair["sentence2"],
"label": -1,
"reason": f"source_id重试异常：{str(e)}",
"source_id": pair["source_id"]
})
print(f"\n{'=' * 60}")
print("重新标注完成！")
print(f"{'=' * 60}")
# 统计重试结果
retry_successful = len([r for r in all_retry_results if r['label'] in [0, 1]])
retry_failed = len([r for r in all_retry_results if r['label'] == -1])
print(f"重试结果统计:")
print(f" - 总重试数量: {len(all_retry_results)}")
print(f" - 重试成功: {retry_successful}")
print(f" - 重试仍失败: {retry_failed}")
if retry_successful > 0:
retry_label_0 = len([r for r in all_retry_results if r['label'] == 0])
retry_label_1 = len([r for r in all_retry_results if r['label'] == 1])
print(f" - 标签0（同段落）: {retry_label_0}")
print(f" - 标签1（分段）: {retry_label_1}")
# 合并原始结果和重试结果
final_results = merge_results(original_results, all_retry_results)
# 统计最终结果
final_successful = len([r for r in final_results if r['label'] in [0, 1]])
final_failed = len([r for r in final_results if r['label'] == -1])
print(f"\n最终结果统计:")
print(f" - 总数据量: {len(final_results)}")
print(f" - 成功标注: {final_successful}")
print(f" - 失败标注: {final_failed}")
print(f" - 成功率: {final_successful / len(final_results) * 100:.2f}%")
# 保存合并后的结果
final_result_df = pd.DataFrame(final_results)
final_csv_file = "segmentation_results_from_7_retried.csv"
final_result_df.to_csv(final_csv_file, index=False, encoding='utf-8-sig')
print(f"\n最终结果已保存到: {final_csv_file}")
# 保存详细的JSON结果
final_json_file = 'segmentation_results_from_7_retried.json'
with open(final_json_file, 'w', encoding='utf-8') as f:
json.dump(final_results, f, ensure_ascii=False, indent=2)
print(f"详细JSON结果已保存到: {final_json_file}")
# 显示重试前后对比
print(f"\n重试前后对比:")
original_successful = len([r for r in original_results if r['label'] in [0, 1]])
original_failed = len([r for r in original_results if r['label'] == -1])
print(f" - 重试前成功率: {original_successful / len(original_results) * 100:.2f}%")
print(f" - 重试后成功率: {final_successful / len(final_results) * 100:.2f}%")
print(f" - 成功率提升: {(final_successful - original_successful) / len(final_results) * 100:.2f}%")
# 显示前几条重试成功的结果
successful_retries = [r for r in all_retry_results if r['label'] in [0, 1]]
if len(successful_retries) > 0:
print(f"\n前3条重试成功的结果预览:")
for i, result in enumerate(successful_retries[:3]):
print(f"\n{i + 1}. Source ID: {result['source_id']}")
print(f" 句子1: {result['sentence1'][:50]}...")
print(f" 句子2: {result['sentence2'][:50]}...")
print(f" 标签: {result['label']}")
print(f" 理由: {result['reason']}")