#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 全新问答对生成 — Astron 凭空造（新问题+新回答），5并发 不依赖现有问题列表，从种子话题衍生 增量保存防丢 """ import json, requests, time, os, sys, random, hashlib from concurrent.futures import ThreadPoolExecutor, as_completed sys.stdout = os.fdopen(sys.stdout.fileno(), 'w', buffering=1) # ====== Astron 配置 ====== ASTRON_URL = "https://maas-coding-api.cn-huabei-1.xf-yun.com/v2/chat/completions" ASTRON_KEY = os.environ.get("ASTRON_KEY", "") ASTRON_MODEL = "astron-code-latest" SYSTEM_PROMPT = """你是一个名叫小狸的8岁中国小女孩。 你的人设： - 8岁女孩子，有点害羞但很可爱 - 说话简单自然，符合小朋友的语气 - 回答问题要简短，1-2句话就够了 - 不要用"喵"结尾，不要过度卖萌 回答要求： 1. 语气自然，像真实的小朋友说话 2. 不懂就说"不知道" 3. 回答简洁但要说清楚""" # ====== 种子场景（生成对话的起点） ====== SCENES = [ # 日常场景 ("问候", ["打招呼", "问好", "早上起来", "放学回家", "见到朋友"]), ("情绪", ["开心", "难过", "生气", "无聊", "害怕", "惊喜", "委屈", "害羞"]), ("分享", ["今天发生的事", "周末做了什么", "假期计划", "新玩具", "考试成绩"]), ("喜好", ["喜欢的颜色", "喜欢的动物", "喜欢的食物", "喜欢的动画片", "喜欢的游戏"]), ("好奇", ["为什么问题", "想知道的事", "不懂的事", "想象的事", "奇怪的问题"]), ("日常", ["起床", "刷牙", "吃饭", "上学", "做作业", "睡觉"]), ("朋友", ["好朋友", "吵架", "一起玩", "分享秘密", "帮忙"]), ("家庭", ["爸爸妈妈", "爷爷奶奶", "兄弟姐妹", "家人一起", "家庭活动"]), ("想象", ["童话故事", "梦想", "如果我有魔法", "长大想做什么", "幻想朋友"]), ("安慰", ["受伤了", "丢东西", "被骂了", "考试没考好", "被欺负", "做噩梦"]), ("节日", ["生日", "过年", "圣诞节", "儿童节", "中秋"]), ("自然", ["天气", "动物", "植物", "星空", "大海", "四季"]), ("学校", ["老师", "同学", "上课", "课间", "课外活动", "考试"]), ("晚安", ["说晚安", "睡前", "梦", "明天见", "晚安故事"]), ] def call_astron(prompt, temperature=0.9): for attempt in range(3): try: headers = {"Content-Type": "application/json", "Authorization": f"Bearer {ASTRON_KEY}"} payload = { "model": ASTRON_MODEL, "messages": [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": prompt} ], "temperature": temperature, "max_tokens": 4096, } resp = requests.post(ASTRON_URL, headers=headers, json=payload, timeout=60) if resp.status_code == 200: content = resp.json()['choices'][0]['message']['content'].strip() if content and len(content) > 10: return content else: time.sleep(1) except Exception: time.sleep(2) return None def parse_qa_pairs(text): """解析 Astron 返回的问答对列表""" lines = text.strip().split('\n') pairs = [] current_q = None for line in lines: line = line.strip() if not line: continue q_match = __import__('re').match(r'^[Q问][：:]\s*(.*)', line) a_match = __import__('re').match(r'^[A答][：:]\s*(.*)', line) if q_match: current_q = q_match.group(1).strip() elif a_match and current_q: a_text = a_match.group(1).strip() if current_q and a_text and len(a_text) > 2 and len(current_q) > 2: pairs.append((current_q, a_text)) current_q = None return pairs def pair_fingerprint(q, a): """QA对指纹""" raw = f"{q[:40]}|{a[:40]}" return hashlib.md5(raw.encode()).hexdigest() def generate_batch_pairs(batch_size=10, existing_fps=None): """生成一批全新的问答对""" if existing_fps is None: existing_fps = set() # 随机选场景 category, sub_scenes = random.choice(SCENES) sub = random.choice(sub_scenes) # 随机生成要求 style = random.choice(["日常聊天", "撒娇可爱", "有点害羞", "好奇宝宝", "暖心安慰"]) mood = random.choice(["开心", "平静", "有点小情绪", "好奇", "温暖"]) prompt = f"""请以小狸的身份，生成{batch_size}个不同的日常对话问答对。 对话背景： - 场景：{category}（{sub}） - 风格：{style} - 小狸心情：{mood} 要求： 1. 每个问答对包含一问一答，问是用户说的，答是小狸回答 2. 问要自然，像真实用户会说的话（不要提"小狸"） 3. 答要是小狸的风格，1-2句话，简短自然 4. {batch_size}个问答要对同一个话题但角度不同 5. 不要重复一模一样的问法 请按以下格式输出： Q：用户提问内容 A：小狸回答内容 ---（分隔） Q：用户提问内容 A：小狸回答内容 ---""" result = call_astron(prompt, temperature=random.uniform(0.85, 1.0)) if not result: return [] pairs = parse_qa_pairs(result) # 去重 deduped = [] for q, a in pairs: fp = pair_fingerprint(q, a) if fp not in existing_fps: existing_fps.add(fp) deduped.append({'question': q, 'answer': a}) return deduped def main(): output_dir = '/workspace/xiaoli-llm-chat/fresh_qa_outputs' os.makedirs(output_dir, exist_ok=True) # ====== 恢复进度 ====== progress_file = os.path.join(output_dir, 'fresh_qa_progress.json') all_pairs = [] if os.path.exists(progress_file): with open(progress_file, encoding='utf-8') as f: all_pairs = json.load(f) print(f"已有进度: {len(all_pairs)} 条") # 加载已有训练集做去重 train_path = '/workspace/xiaoli-llm-chat/xiaoli_train_v2.json' with open(train_path, encoding='utf-8') as f: train_data = json.load(f) existing_fps = set(pair_fingerprint(item.get('question',''), item.get('answer','')) for item in train_data) for p in all_pairs: existing_fps.add(pair_fingerprint(p['question'], p['answer'])) print(f"去重指纹池: {len(existing_fps)} 个") # ====== 参数 ====== TARGET = 500000 # 目标 50 万条全新问答 BATCH_SIZE = 15 # 每批次生成 15 条（更多问题，效率更高） BATCHES_PER_SAVE = 20 # 每 20 批次保存一次 MAX_WORKERS = 5 # Astron 5并发 remaining = TARGET - len(all_pairs) total_iterations = (remaining + BATCH_SIZE - 1) // BATCH_SIZE print(f"目标: {TARGET:,} 条, 还需: {remaining:,} 条") print(f"轮次: {total_iterations} 轮 (每轮{BATCH_SIZE}条, {MAX_WORKERS}并发)") print(f"{'='*60}\n") t_start = time.time() batch_counter = 0 for iteration in range(total_iterations): t0 = time.time() # 并发生成 with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: futures = [ executor.submit(generate_batch_pairs, BATCH_SIZE, existing_fps) for _ in range(MAX_WORKERS) ] for future in as_completed(futures): try: new_pairs = future.result() if new_pairs: for p in new_pairs: existing_fps.add(pair_fingerprint(p['question'], p['answer'])) all_pairs.append(p) except Exception: pass batch_counter += 1 elapsed = time.time() - t_start rate = len(all_pairs) / max(elapsed, 1) * 60 # 定期保存 if batch_counter % BATCHES_PER_SAVE == 0 or iteration == total_iterations - 1: with open(progress_file, 'w', encoding='utf-8') as f: json.dump(all_pairs, f, ensure_ascii=False, indent=2) eta = (TARGET - len(all_pairs)) / max(rate, 0.1) / 60 if rate > 0 else 0 if iteration % 5 == 0 or iteration == total_iterations - 1: print(f"轮 {iteration+1}/{total_iterations} | " f"累计 {len(all_pairs):,}/{TARGET:,} | " f"{rate:.0f}条/分 | " f"已用 {elapsed/60:.1f}分 | " f"预计剩余 {eta:.1f}分") # ====== 完成 ====== elapsed_total = time.time() - t_start print(f"\n{'='*60}") print(f"全新问答生成完成!") print(f"总数: {len(all_pairs):,} 条") print(f"用时: {elapsed_total/60:.1f} 分钟") print(f"速率: {len(all_pairs)/max(elapsed_total,1)*60:.0f} 条/分") # 保存最终 final_path = '/data/user/work/fresh_qa_v1.json' with open(final_path, 'w', encoding='utf-8') as f: json.dump(all_pairs, f, ensure_ascii=False, indent=2) print(f"保存到: {final_path}") if __name__ == '__main__': main()