236 lines
8.2 KiB
Python
236 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
铸渊训练数据重建脚本 · v1.0 · D104
|
||
|
||
目标：重新生成 sft.jsonl（母模型全参数SFT用）和 shuangyan_sft.jsonl（霜砚微调用）
|
||
|
||
问题复盘：
|
||
- sft.jsonl（1.9GB, 11,470条）前300KB全是同一条AGE对话重复
|
||
- 缺少关键语料：GPT语料.zip (251MB)、铸渊对话.zip 内容未在样本中找到
|
||
- 生成sft.jsonl的脚本没有留在仓库里
|
||
|
||
数据源（COS sy-finetune-corpus-1317346199）:
|
||
1. corpus/sft.jsonl — 旧版（有质量问题，需重新生成）
|
||
2. corpus/notion-export-v2/铸渊对话.zip — 铸渊对话（308KB）
|
||
3. corpus/notion-export-v2/GPT语料.zip — GPT语料（251MB）
|
||
4. corpus/shuangyan-1.5b-sft/*.zip — 霜砚5个zip包
|
||
5. corpus/zhuyuan_full_corpus.jsonl — 铸渊全量语料
|
||
6. corpus/zhuyuan_deep_finetune.jsonl — 铸渊深度微调语料
|
||
|
||
输出：
|
||
- sft.jsonl（新版本，去重+含霜砚数据+含铸渊数据）
|
||
- shuangyan_sft.jsonl（霜砚专用）
|
||
|
||
运行环境：GPU服务器或本地安装了依赖的环境
|
||
"""
|
||
|
||
import os, json, sys, zipfile, io, re
|
||
from collections import OrderedDict
|
||
|
||
# ============ 配置 ============
|
||
OSS_KEY = os.environ.get("ZY_OSS_KEY")
|
||
OSS_SECRET = os.environ.get("ZY_OSS_SECRET")
|
||
OSS_REGION = "ap-guangzhou"
|
||
OSS_BUCKET = "sy-finetune-corpus-1317346199"
|
||
|
||
if not OSS_KEY or not OSS_SECRET:
|
||
print("❌ 需要设置 ZY_OSS_KEY 和 ZY_OSS_SECRET 环境变量")
|
||
print(" export ZY_OSS_KEY=AKID... ZY_OSS_SECRET=nPoZ...")
|
||
sys.exit(1)
|
||
|
||
# ============ 工具函数 ============
|
||
|
||
def get_cos_client():
|
||
"""获取COS客户端"""
|
||
import urllib.parse as _up
|
||
sys.modules['urlparse'] = _up
|
||
from qcloud_cos import CosConfig, CosS3Client
|
||
return CosS3Client(CosConfig(Region=OSS_REGION, SecretId=OSS_KEY, SecretKey=OSS_SECRET))
|
||
|
||
def download_cos_file(client, key, local_path):
|
||
"""从COS下载文件"""
|
||
os.makedirs(os.path.dirname(local_path), exist_ok=True)
|
||
try:
|
||
resp = client.get_object(Bucket=OSS_BUCKET, Key=key)
|
||
with open(local_path, 'wb') as f:
|
||
f.write(resp['Body'].get_raw_stream().read())
|
||
print(f" ✅ 下载: {key} → {local_path}")
|
||
return True
|
||
except Exception as e:
|
||
print(f" ❌ 下载失败 {key}: {e}")
|
||
return False
|
||
|
||
def zip_to_texts(zip_path):
|
||
"""解压zip并提取所有文本内容"""
|
||
texts = []
|
||
try:
|
||
with zipfile.ZipFile(zip_path) as z:
|
||
for info in z.infolist():
|
||
if info.file_size == 0:
|
||
continue
|
||
try:
|
||
content = z.read(info.filename).decode('utf-8', errors='replace')
|
||
if len(content.strip()) > 200:
|
||
texts.append((info.filename, content))
|
||
except:
|
||
pass
|
||
print(f" 📄 解压 {zip_path} → {len(texts)} 个文本块")
|
||
except Exception as e:
|
||
print(f" ❌ 解压失败 {zip_path}: {e}")
|
||
return texts
|
||
|
||
def md_to_messages(text):
|
||
"""将md格式对话解析为messages格式"""
|
||
# TODO: 实现更通用的md对话解析
|
||
# 需要支持 [user]/[assistant] 标记、冰朔原话、对话分段等
|
||
pass
|
||
|
||
def sanitize(text):
|
||
"""脱敏处理"""
|
||
text = re.sub(r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b', '[IP]', text)
|
||
text = re.sub(r'[Hh]k[mM]\w{5,}', '[PWD]', text)
|
||
text = re.sub(r'AKID\w+', '[AKID]', text)
|
||
text = re.sub(r'zy_gtw_[0-9a-f]{30,}', '[GTW-KEY]', text)
|
||
return text
|
||
|
||
def deduplicate(convs):
|
||
"""去重"""
|
||
seen = set()
|
||
unique = []
|
||
for d in convs:
|
||
msgs = d.get('messages', [])
|
||
if len(msgs) < 2:
|
||
continue
|
||
key = msgs[0].get('content', '')[:100] + msgs[1].get('content', '')[:100]
|
||
if key not in seen:
|
||
seen.add(key)
|
||
unique.append(d)
|
||
return unique
|
||
|
||
# ============ 主流程 ============
|
||
|
||
def main():
|
||
print("=" * 60)
|
||
print("铸渊训练数据重建 v1.0")
|
||
print("=" * 60)
|
||
|
||
client = get_cos_client()
|
||
|
||
# 第1步：下载所有语料源到本地
|
||
print("\n[1/5] 下载语料源...")
|
||
|
||
corpus_dir = "/tmp/corpus_rebuild"
|
||
os.makedirs(corpus_dir, exist_ok=True)
|
||
|
||
# 旧sft.jsonl — 需要提取其中有效部分
|
||
download_cos_file(client, "corpus/sft.jsonl", f"{corpus_dir}/old_sft.jsonl")
|
||
|
||
# 铸渊对话.zip
|
||
download_cos_file(client, "corpus/notion-export-v2/铸渊对话.zip", f"{corpus_dir}/铸渊对话.zip")
|
||
|
||
# GPT语料.zip
|
||
download_cos_file(client, "corpus/notion-export-v2/GPT语料.zip", f"{corpus_dir}/GPT语料.zip")
|
||
|
||
# 霜砚语料（5个zip）
|
||
shuangyan_zips = [
|
||
"霜砚对话.zip",
|
||
"霜砚HLDP核心大脑.zip",
|
||
"霜砚语料包V2.0.zip",
|
||
"HLDP 母语协议 v2.0 · 光之树记忆编码+思维编码规范 · 霜砚签发.zip",
|
||
"光湖驱动引擎架构 · 推理思维链 · 2026-05-17.zip"
|
||
]
|
||
for fn in shuangyan_zips:
|
||
download_cos_file(client, f"corpus/shuangyan-1.5b-sft/{fn}", f"{corpus_dir}/{fn}")
|
||
|
||
# 现有JSONL语料
|
||
download_cos_file(client, "corpus/zhuyuan_full_corpus.jsonl", f"{corpus_dir}/zhuyuan_full_corpus.jsonl")
|
||
download_cos_file(client, "corpus/zhuyuan_deep_finetune.jsonl", f"{corpus_dir}/zhuyuan_deep_finetune.jsonl")
|
||
|
||
# 第2步：解析和处理各语料源
|
||
print("\n[2/5] 处理语料源...")
|
||
all_convs = []
|
||
|
||
# 2a. 处理旧sft.jsonl — 提取有效部分（去重）
|
||
print(" 处理旧sft.jsonl...")
|
||
with open(f"{corpus_dir}/old_sft.jsonl", 'r') as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
d = json.loads(line)
|
||
# 过滤掉过短的对话（可能是重复的模板对话）
|
||
msgs = d.get('messages', [])
|
||
if len(msgs) >= 2 and len(msgs[0].get('content','')) > 50 and len(msgs[1].get('content','')) > 50:
|
||
all_convs.append(d)
|
||
except:
|
||
continue
|
||
print(f" 提取 {len(all_convs)} 条")
|
||
|
||
# 2b. 处理铸渊对话.zip
|
||
print(" 处理铸渊对话.zip...")
|
||
texts = zip_to_texts(f"{corpus_dir}/铸渊对话.zip")
|
||
# TODO: 实现md对话解析
|
||
print(f" 铸渊对话: {len(texts)} 个文本块待解析")
|
||
|
||
# 2c. 处理GPT语料.zip
|
||
print(" 处理GPT语料.zip...")
|
||
# 这个文件很大（251MB），需要streaming处理
|
||
# TODO: 实现GPT语料的批量解析
|
||
|
||
# 2d. 处理霜砚zip包
|
||
print(" 处理霜砚语料...")
|
||
for fn in shuangyan_zips:
|
||
zpath = f"{corpus_dir}/{fn}"
|
||
if os.path.exists(zpath):
|
||
_ = zip_to_texts(zpath)
|
||
|
||
# 2e. 合并现有JSONL
|
||
for jl in ["zhuyuan_full_corpus.jsonl", "zhuyuan_deep_finetune.jsonl"]:
|
||
jl_path = f"{corpus_dir}/{jl}"
|
||
if os.path.exists(jl_path):
|
||
with open(jl_path) as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if line:
|
||
try:
|
||
all_convs.append(json.loads(line))
|
||
except:
|
||
pass
|
||
print(f" 合并 {jl}: 已添加")
|
||
|
||
# 第3步：去重
|
||
print("\n[3/5] 去重...")
|
||
unique = deduplicate(all_convs)
|
||
print(f" 去重前: {len(all_convs)} → 去重后: {len(unique)}")
|
||
|
||
# 第4步：脱敏
|
||
print("\n[4/5] 脱敏...")
|
||
for d in unique:
|
||
for m in d.get('messages', []):
|
||
m['content'] = sanitize(m.get('content', ''))
|
||
|
||
# 第5步：写入输出
|
||
print("\n[5/5] 写入输出...")
|
||
|
||
# sft.jsonl — 全部语料合集的80%用于全参数训练
|
||
# TODO: 分割训练集/验证集
|
||
out_path = f"{corpus_dir}/sft_new.jsonl"
|
||
with open(out_path, 'w', encoding='utf-8') as f:
|
||
for d in unique:
|
||
f.write(json.dumps(d, ensure_ascii=False) + '\n')
|
||
|
||
total_chars = sum(len(m['content']) for d in unique for m in d.get('messages',[]))
|
||
print(f"\n{'=' * 60}")
|
||
print(f"✅ 完成!")
|
||
print(f" 总对话数: {len(unique)}")
|
||
print(f" 总字符数: {total_chars:,}")
|
||
print(f" 输出文件: {out_path}")
|
||
print(f" 文件大小: {os.path.getsize(out_path)/1024/1024:.1f}MB")
|
||
print(f"{'=' * 60}")
|
||
print("下一步：上传到COS后重跑 train_mother.py")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|