""" 语料采集引擎 · 核心大脑 ======================== 判断标准 → 脱敏规则 → 格式化输出 """ import re import json import hashlib from typing import List, Dict, Optional # ============================================================ # 第1层：脱敏引擎 # ============================================================ SENSITIVE_PATTERNS = [ # IP地址（用 lookahead/lookbehind 替代 \b，避免中文干扰） (r'(? 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, }