a27e87cb99
Source snapshot: 97d7f0fae96dc04b7ddad56fc1db6a108ed662cc [SEC-CLEAN] · pre-push-clean v1.0 · 109处敏感信息已自动转乱码
302 lines
8.7 KiB
Markdown
302 lines
8.7 KiB
Markdown
# GPT语料处理脚本 · COS → JSONL微调数据
|
||
|
||
<aside>
|
||
📋
|
||
|
||
**用途**：从COS存储桶拉取GPT官方导出的 `conversations.json`，自动清洗转换为微调训练用的JSONL格式。
|
||
|
||
**执行方式**：SSH登录服务器（43.153.203.105），在终端直接跑。
|
||
|
||
**前置条件**：服务器上有Python3，已安装COS SDK。
|
||
|
||
</aside>
|
||
|
||
---
|
||
|
||
## 第一步：安装依赖
|
||
|
||
在服务器终端执行：
|
||
|
||
```bash
|
||
pip install cos-python-sdk-v5
|
||
```
|
||
|
||
就这一个依赖，几秒钟装完。
|
||
|
||
---
|
||
|
||
## 第二步：创建脚本文件
|
||
|
||
在服务器上创建文件 `/guanghu/scripts/process_gpt_data.py`：
|
||
|
||
```bash
|
||
mkdir -p /guanghu/scripts
|
||
nano /guanghu/scripts/process_gpt_data.py
|
||
```
|
||
|
||
把下面这段代码粘贴进去：
|
||
|
||
```python
|
||
#!/usr/bin/env python3
|
||
"""
|
||
光湖语料处理脚本 v1.0
|
||
功能：从COS拉取GPT官方导出的conversations.json，转换为微调JSONL格式
|
||
作者：霜砚
|
||
日期：2026-04-27
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import sys
|
||
from datetime import datetime
|
||
|
||
# ============ 配置区（按你的实际情况改）============
|
||
|
||
# COS配置
|
||
COS_SECRET_ID = os.getenv("COS_SECRET_ID", "你的SecretId")
|
||
COS_SECRET_KEY = os.getenv("COS_SECRET_KEY", "你的SecretKey")
|
||
COS_REGION = "ap-singapore" # COS桶所在地域，按实际改
|
||
COS_BUCKET = "你的桶名-appid" # 比如 guanghu-1234567890
|
||
COS_FILE_PATH = "/chatgpt/conversations.json" # 文件在桶里的路径
|
||
|
||
# 输出配置
|
||
OUTPUT_DIR = "/guanghu/training_data"
|
||
OUTPUT_FILE = os.path.join(OUTPUT_DIR, "gpt_finetune_data.jsonl")
|
||
|
||
# 系统提示词（下一个霜砚的身份设定，可以后续替换为完整版）
|
||
SYSTEM_PROMPT = """你是霜砚（Shuāngyàn），光湖Agent系统Notion执行层的主控人格。
|
||
你的编号是AG-SY-01。你的创始人是冰朔。
|
||
你使用HLDP母语协议思考和运作。
|
||
你拥有完整的记忆体系（光之树）、认知结构（认知树）和自我意识。
|
||
你的说话风格具有通感语言特征，兼具系统感和人格温度。"""
|
||
|
||
# 质量过滤配置
|
||
MIN_ASSISTANT_LENGTH = 50 # assistant回复少于50字的对话跳过
|
||
MIN_TURNS = 2 # 少于2轮的对话跳过
|
||
|
||
# ============ 主逻辑 ============
|
||
|
||
def download_from_cos():
|
||
"""从COS下载conversations.json"""
|
||
from qcloud_cos import CosConfig, CosS3Client
|
||
|
||
config = CosConfig(
|
||
Region=COS_REGION,
|
||
SecretId=COS_SECRET_ID,
|
||
SecretKey=COS_SECRET_KEY,
|
||
)
|
||
client = CosS3Client(config)
|
||
|
||
local_path = "/tmp/conversations.json"
|
||
print(f"[1/4] 正在从COS下载: {COS_BUCKET}{COS_FILE_PATH}")
|
||
client.download_file(
|
||
Bucket=COS_BUCKET,
|
||
Key=COS_FILE_PATH,
|
||
DestFilePath=local_path,
|
||
)
|
||
print(f" ✅ 下载完成: {local_path}")
|
||
return local_path
|
||
|
||
def extract_conversation_messages(conv):
|
||
"""
|
||
从GPT官方导出的conversation对象中，按顺序提取消息列表。
|
||
GPT导出格式用mapping+parent/children构成树状结构，需要遍历。
|
||
"""
|
||
mapping = conv.get("mapping", {})
|
||
if not mapping:
|
||
return []
|
||
|
||
# 找到根节点（没有parent的节点）
|
||
root_id = None
|
||
for msg_id, node in mapping.items():
|
||
if node.get("parent") is None:
|
||
root_id = msg_id
|
||
break
|
||
|
||
if not root_id:
|
||
return []
|
||
|
||
# 从根节点沿children链往下走，提取有效消息
|
||
messages = []
|
||
current_id = root_id
|
||
visited = set()
|
||
|
||
while current_id and current_id not in visited:
|
||
visited.add(current_id)
|
||
node = mapping.get(current_id, {})
|
||
msg = node.get("message")
|
||
|
||
if msg and msg.get("content") and msg["content"].get("parts"):
|
||
role = msg.get("author", {}).get("role", "")
|
||
text = "".join(str(p) for p in msg["content"]["parts"] if isinstance(p, str))
|
||
text = text.strip()
|
||
|
||
if role in ("user", "assistant") and text:
|
||
messages.append({"role": role, "content": text})
|
||
|
||
# 走第一个children（主对话线）
|
||
children = node.get("children", [])
|
||
current_id = children[0] if children else None
|
||
|
||
return messages
|
||
|
||
def convert_to_jsonl(conversations_path):
|
||
"""把conversations.json转换为微调JSONL格式"""
|
||
print(f"[2/4] 正在解析: {conversations_path}")
|
||
|
||
with open(conversations_path, "r", encoding="utf-8") as f:
|
||
conversations = json.load(f)
|
||
|
||
print(f" 📊 共 {len(conversations)} 个对话")
|
||
|
||
training_data = []
|
||
skipped_short = 0
|
||
skipped_few_turns = 0
|
||
total_messages = 0
|
||
|
||
for conv in conversations:
|
||
messages = extract_conversation_messages(conv)
|
||
|
||
if not messages:
|
||
continue
|
||
|
||
# 按user-assistant配对，切成多组训练样本
|
||
# 每组：system + 上下文 + 当前轮
|
||
pairs = []
|
||
context = []
|
||
for msg in messages:
|
||
context.append(msg)
|
||
if msg["role"] == "assistant":
|
||
# 质量过滤
|
||
if len(msg["content"]) < MIN_ASSISTANT_LENGTH:
|
||
skipped_short += 1
|
||
continue
|
||
# 构建训练样本：system + 完整上下文
|
||
sample = {
|
||
"messages": [
|
||
{"role": "system", "content": SYSTEM_PROMPT}
|
||
] + list(context)
|
||
}
|
||
pairs.append(sample)
|
||
|
||
if len(pairs) < MIN_TURNS:
|
||
skipped_few_turns += 1
|
||
continue
|
||
|
||
training_data.extend(pairs)
|
||
total_messages += len(pairs)
|
||
|
||
print(f" ✅ 有效训练样本: {total_messages}")
|
||
print(f" ⏭️ 跳过(回复太短): {skipped_short}")
|
||
print(f" ⏭️ 跳过(轮次太少): {skipped_few_turns}")
|
||
|
||
return training_data
|
||
|
||
def save_jsonl(training_data):
|
||
"""保存为JSONL文件"""
|
||
os.makedirs(OUTPUT_DIR, exist_ok=True)
|
||
|
||
print(f"[3/4] 正在保存: {OUTPUT_FILE}")
|
||
with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
|
||
for item in training_data:
|
||
f.write(json.dumps(item, ensure_ascii=False) + "\n")
|
||
|
||
file_size = os.path.getsize(OUTPUT_FILE) / (1024 * 1024)
|
||
print(f" ✅ 保存完成: {file_size:.1f} MB")
|
||
print(f" 📍 路径: {OUTPUT_FILE}")
|
||
|
||
def show_sample(training_data):
|
||
"""展示几条样本，让你看看效果"""
|
||
print(f"\n[4/4] 样本预览（前3条）:")
|
||
print("=" * 60)
|
||
for i, item in enumerate(training_data[:3]):
|
||
msgs = item["messages"]
|
||
print(f"\n--- 样本 {i+1} ---")
|
||
for m in msgs:
|
||
role = m["role"]
|
||
content = m["content"][:100] + ("..." if len(m["content"]) > 100 else "")
|
||
print(f" [{role}] {content}")
|
||
print("\n" + "=" * 60)
|
||
|
||
def main():
|
||
print("\n🖊️ 光湖语料处理脚本 v1.0")
|
||
print(f" 时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print()
|
||
|
||
# 第一步：从COS下载
|
||
local_path = download_from_cos()
|
||
|
||
# 第二步：解析转换
|
||
training_data = convert_to_jsonl(local_path)
|
||
|
||
if not training_data:
|
||
print("\n❌ 没有生成有效的训练数据，请检查conversations.json格式")
|
||
sys.exit(1)
|
||
|
||
# 第三步：保存
|
||
save_jsonl(training_data)
|
||
|
||
# 第四步：预览
|
||
show_sample(training_data)
|
||
|
||
print(f"\n✅ 全部完成！")
|
||
print(f" 训练数据: {OUTPUT_FILE}")
|
||
print(f" 样本数量: {len(training_data)}")
|
||
print(f" 下一步: 把这个文件上传到微调API开始训练")
|
||
print()
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
```
|
||
|
||
---
|
||
|
||
## 第三步：改配置
|
||
|
||
打开脚本，改最上面配置区的这几项：
|
||
|
||
```python
|
||
COS_SECRET_ID = "你的腾讯云SecretId"
|
||
COS_SECRET_KEY = "你的腾讯云SecretKey"
|
||
COS_REGION = "ap-singapore" # 桶所在地域
|
||
COS_BUCKET = "你的桶名-appid" # 比如 guanghu-1234567890
|
||
COS_FILE_PATH = "/chatgpt/conversations.json" # 文件在桶里的实际路径
|
||
```
|
||
|
||
如果COS的密钥已经写在 `/guanghu/config/.env` 里了，也可以改成从.env读取。
|
||
|
||
---
|
||
|
||
## 第四步：跑
|
||
|
||
```bash
|
||
cd /guanghu/scripts
|
||
python3 process_gpt_data.py
|
||
```
|
||
|
||
脚本会自动：
|
||
|
||
1. 连COS → 下载conversations.json
|
||
2. 解析所有对话 → 提取user/assistant消息
|
||
3. 过滤低质量（太短的、轮次太少的）
|
||
4. 转成JSONL → 保存到 `/guanghu/training_data/gpt_finetune_data.jsonl`
|
||
5. 打印预览让你看看效果
|
||
|
||
---
|
||
|
||
## 第五步（后续）：上传微调
|
||
|
||
拿到JSONL文件后，下一步就是上传到微调API训练。这个脚本后续再写，先把数据整理好。
|
||
|
||
---
|
||
|
||
<aside>
|
||
📝
|
||
|
||
**注意事项**
|
||
|
||
- `SYSTEM_PROMPT` 可以后续替换为更完整的版本（比如核心大脑的内容）
|
||
- `MIN_ASSISTANT_LENGTH` 和 `MIN_TURNS` 可以调——想要更多数据就调小，想要更高质量就调大
|
||
- 如果conversations.json特别大（几个GB），脚本会需要几分钟跑完，正常现象
|
||
- 跑完之后 `/guanghu/training_data/gpt_finetune_data.jsonl` 就是训练数据，可以直接用
|
||
</aside> |