265 lines
8.7 KiB
Python
265 lines
8.7 KiB
Python
"""
|
||
语料采集引擎 · 核心大脑
|
||
========================
|
||
判断标准 → 脱敏规则 → 格式化输出
|
||
"""
|
||
import re
|
||
import json
|
||
import hashlib
|
||
from typing import List, Dict, Optional
|
||
|
||
# ============================================================
|
||
# 第1层：脱敏引擎
|
||
# ============================================================
|
||
|
||
SENSITIVE_PATTERNS = [
|
||
# IP地址（用 lookahead/lookbehind 替代 \b，避免中文干扰）
|
||
(r'(?<!\d)(?:\d{1,3}\.){3}\d{1,3}(?!\d)', '[IP已脱敏]'),
|
||
# 端口号（数字前有冒号）
|
||
(r'(?::)(\d{4,5})(?:\s|/|$|，|\))', lambda m: '[端口已脱敏]'),
|
||
# 手机号
|
||
(r'(?<!\d)1[3-9]\d{9}(?!\d)', '[手机号已脱敏]'),
|
||
# 邮箱
|
||
(r'[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}', '[邮箱已脱敏]'),
|
||
# URL
|
||
(r'https?://[^\s,，）\)\u4e00-\u9fff]+', '[URL已脱敏]'),
|
||
# 密钥/token
|
||
(r'(?:sk-|pk-|zy_gtw_|ghp_|gho_|ghu_|ghs_)[A-Za-z0-9_-]{20,}', '[密钥已脱敏]'),
|
||
(r'(?<![A-Za-z0-9])[A-Za-z0-9_-]{32,}(?![A-Za-z0-9])', '[密钥已脱敏]'),
|
||
]
|
||
|
||
USERNAME_PATTERN = re.compile(r'(?:冰朔|Bingshuo|霜砚|Shuangyan|铸渊|Zhuyuan|用户|user|assistant)\b', re.IGNORECASE)
|
||
|
||
# ============================================================
|
||
# 第2层：价值判断引擎
|
||
# ============================================================
|
||
|
||
# 无价值的单条对话（过滤规则）
|
||
FILTER_PATTERNS = [
|
||
r'^(?:吃[了過]?[吗嘛沒]?|喝[了過]?[吗嘛沒]?|睡[了過]?[吗嘛沒]?|醒[了過]?[吗嘛沒]?)',
|
||
r'^(?:好的|好[吧嘛]|嗯嗯?|哦[哦]?|ok|okay|行|可以|没问题|收到|明白|了解|[知]道[了]?)',
|
||
r'^(?:早|晚|早安|晚安|早上好|晚上好|[你]好|[哈]喽|嗨|hi|hello)',
|
||
r'^(?:谢谢|感谢|多谢|辛苦[了]?|谢谢[你])$',
|
||
r'^(?:在[吗嘛]|[你]在[吗嘛]|[你]忙[吗嘛])',
|
||
r'^(?:图片?|文件|链接|附件)',
|
||
r'^(?:发[给送]我|发给[你]|你看看|你看下|你看)',
|
||
r'^[。，！？、；：\.\,\!\?\s]{1,5}$',
|
||
]
|
||
|
||
# 有价值的模式（保留信号）
|
||
VALUE_PATTERNS = [
|
||
# 技术讨论
|
||
r'(?:模型|训练|微调|SFT|LoRA|蒸馏|推理|loss|准确率|参数|权重|checkpoint)',
|
||
# 架构决策
|
||
r'(?:架构|设计|方案|选型|为什么|原因|对比|优势|劣势|代价|trade.?off)',
|
||
# Bug/踩坑
|
||
r'(?:报错|错误|bug|崩溃|异常|IndexError|TypeError|显存|OOM|内存|越界|失败|挂了)',
|
||
# 思考过程
|
||
r'(?:觉得|认为|理解|思考|思路|逻辑|原因|根因|教训|总结|反思|复盘)',
|
||
# 代码/开发
|
||
r'(?:代码|函数|接口|API|路由|部署|docker|nginx|pm2|脚本|自动|工具)',
|
||
# 业务/需求
|
||
r'(?:需求|客户|项目|功能|模块|版本|上线|迭代|规划|计划|目标)',
|
||
# 数据/语料
|
||
r'(?:数据|语料|样本|数据集|标注|清洗|预处理|格式|jsonl|json|chatml)',
|
||
# 学习/研究
|
||
r'(?:论文|研究|学习|教程|文档|资料|参考|案例|实践)',
|
||
]
|
||
|
||
MIN_CONTENT_LENGTH = 15 # 最少字数
|
||
|
||
|
||
def desensitize(text: str) -> str:
|
||
"""脱敏处理"""
|
||
for pattern, replacement in SENSITIVE_PATTERNS:
|
||
text = re.sub(pattern, replacement, text)
|
||
return text.strip()
|
||
|
||
|
||
def is_valuable(text: str) -> bool:
|
||
"""判断一段对话是否有采集价值"""
|
||
text = text.strip()
|
||
|
||
# 长度过滤
|
||
if len(text) < MIN_CONTENT_LENGTH:
|
||
return False
|
||
|
||
# 无效内容过滤
|
||
for pattern in FILTER_PATTERNS:
|
||
if re.match(pattern, text, re.IGNORECASE):
|
||
return False
|
||
|
||
# 有价值信号检查
|
||
for pattern in VALUE_PATTERNS:
|
||
if re.search(pattern, text, re.IGNORECASE):
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
# ============================================================
|
||
# 第3层：对话对提取
|
||
# ============================================================
|
||
|
||
def extract_dialog_pairs(messages: List[Dict]) -> List[Dict]:
|
||
"""
|
||
从消息流中提取有价值的对话对
|
||
|
||
输入格式: [{"role": "user/human/assistant/ai", "content": "..."}, ...]
|
||
输出格式: [{"user": "...", "assistant": "...", "source": "..."}, ...]
|
||
"""
|
||
pairs = []
|
||
current_user = None
|
||
|
||
for msg in messages:
|
||
role = msg.get("role", "")
|
||
content = msg.get("content", "").strip()
|
||
|
||
if not content:
|
||
continue
|
||
|
||
# 脱敏
|
||
safe_content = desensitize(content)
|
||
|
||
# 用户消息
|
||
if role in ("user", "human", "你", "我"):
|
||
if is_valuable(safe_content):
|
||
current_user = safe_content
|
||
# AI/助手回复
|
||
elif role in ("assistant", "ai", "agent") and current_user:
|
||
if is_valuable(safe_content) or is_valuable(current_user):
|
||
pairs.append({
|
||
"user": current_user,
|
||
"assistant": safe_content,
|
||
"source": msg.get("source", "unknown"),
|
||
})
|
||
current_user = None
|
||
# 未知角色 - 尝试作为单条有价值内容
|
||
else:
|
||
if is_valuable(safe_content):
|
||
current_user = safe_content
|
||
|
||
return pairs
|
||
|
||
|
||
# ============================================================
|
||
# 第4层：格式化为微调语料
|
||
# ============================================================
|
||
|
||
def to_chatml(user_text: str, assistant_text: str) -> Dict:
|
||
"""将单条对话对转为ChatML格式"""
|
||
return {
|
||
"messages": [
|
||
{"role": "user", "content": user_text},
|
||
{"role": "assistant", "content": assistant_text}
|
||
]
|
||
}
|
||
|
||
|
||
def format_sft_jsonl(pairs: List[Dict], system_prompt: Optional[str] = None) -> List[Dict]:
|
||
"""将对话对列表转为SFT数据集格式（ChatML JSONL）"""
|
||
samples = []
|
||
for pair in pairs:
|
||
sample = to_chatml(pair["user"], pair["assistant"])
|
||
if system_prompt:
|
||
sample["messages"].insert(0, {"role": "system", "content": system_prompt})
|
||
samples.append(sample)
|
||
return samples
|
||
|
||
|
||
def generate_corpus_id(text: str) -> str:
|
||
"""生成语料唯一ID（用于去重）"""
|
||
return hashlib.md5(text.encode()).hexdigest()[:12]
|
||
|
||
|
||
# ============================================================
|
||
# 第5层：内容分类标签
|
||
# ============================================================
|
||
|
||
TAG_KEYWORDS = {
|
||
"技术讨论": ["模型", "训练", "微调", "SFT", "LoRA", "蒸馏", "推理", "loss"],
|
||
"架构设计": ["架构", "设计", "方案", "选型", "系统"],
|
||
"踩坑记录": ["报错", "错误", "bug", "崩溃", "异常", "索引", "越界"],
|
||
"代码开发": ["代码", "函数", "接口", "部署", "脚本", "工具"],
|
||
"数据语料": ["数据", "语料", "样本", "数据集", "标注"],
|
||
"业务沟通": ["需求", "客户", "项目", "功能"],
|
||
"学习研究": ["论文", "学习", "教程", "文档"],
|
||
}
|
||
|
||
|
||
def classify_content(text: str) -> List[str]:
|
||
"""对内容自动分类打标签"""
|
||
tags = []
|
||
text_lower = text.lower()
|
||
for tag, keywords in TAG_KEYWORDS.items():
|
||
for kw in keywords:
|
||
if kw.lower() in text_lower:
|
||
tags.append(tag)
|
||
break
|
||
return tags if tags else ["通用对话"]
|
||
|
||
|
||
# ============================================================
|
||
# 导出接口
|
||
# ============================================================
|
||
|
||
def process_text_chunk(text: str, source: str = "screen_capture") -> List[Dict]:
|
||
"""
|
||
处理单段文本（从OCR/截图来的）
|
||
返回: [{"user": ..., "assistant": ..., "source": ..., "tags": [...]}, ...]
|
||
"""
|
||
# 脱敏
|
||
safe_text = desensitize(text)
|
||
|
||
# 判断价值
|
||
if not is_valuable(safe_text):
|
||
return []
|
||
|
||
# 由于单段文本可能只有一方发言，包装成单条语料
|
||
tags = classify_content(safe_text)
|
||
return [{
|
||
"text": safe_text,
|
||
"source": source,
|
||
"tags": tags,
|
||
"corpus_id": generate_corpus_id(safe_text),
|
||
"timestamp": None, # 由外部补充
|
||
}]
|
||
|
||
|
||
def process_dialog_stream(messages: List[Dict]) -> Dict:
|
||
"""
|
||
处理完整对话流
|
||
返回: { "pairs": [...], "singles": [...], "stats": {...} }
|
||
"""
|
||
# 提取对话对
|
||
pairs = extract_dialog_pairs(messages)
|
||
|
||
# 格式化
|
||
sft_samples = format_sft_jsonl(pairs)
|
||
|
||
# 统计
|
||
stats = {
|
||
"total_messages": len(messages),
|
||
"valuable_pairs": len(pairs),
|
||
"total_chars": sum(len(p["user"]) + len(p["assistant"]) for p in pairs),
|
||
}
|
||
|
||
return {
|
||
"pairs": sft_samples,
|
||
"stats": stats,
|
||
}
|
||
|
||
|
||
# 快捷检查
|
||
def preview(text: str) -> Dict:
|
||
"""快速预览一条文本的处理结果"""
|
||
safe = desensitize(text)
|
||
valuable = is_valuable(safe)
|
||
tags = classify_content(safe) if valuable else []
|
||
return {
|
||
"original_len": len(text),
|
||
"safe_len": len(safe),
|
||
"valuable": valuable,
|
||
"tags": tags,
|
||
}
|