[
  {
    "instruction": "将以下中文翻译成英文",
    "input": "今天天气很好",
    "output": "The weather is very nice today."
  },
  {
    "instruction": "用一句话总结以下文章的主要观点",
    "input": "人工智能正在改变医疗诊断...(文章内容)",
    "output": "AI技术正在通过提升诊断准确率和效率来革新医疗行业。"
  }
]

[
  {
    "messages": [
      {"role": "system", "content": "你是一个Python专家助手，擅长用简单的语言解释编程概念。"},
      {"role": "user", "content": "什么是装饰器？"},
      {"role": "assistant", "content": "装饰器是Python中一种设计模式，允许你在不修改原函数代码的情况下，给函数添加额外功能。\n\n最简单的例子：\n```python\ndef log_decorator(func):\n    def wrapper(*args, **kwargs):\n        print(f'调用函数: {func.__name__}')\n        return func(*args, **kwargs)\n    return wrapper\n\n@log_decorator\ndef say_hello():\n    print('Hello!')\n```\n\n调用 `say_hello()` 时会自动打印日志。"}
    ]
  }
]

坏的 instruction：
"回答用户问题"

好的 instruction：
"你是一名电商客服，用户反映商品未按时到达。请以专业、同理心的态度回复，
说明可能的原因，并提供具体的解决方案（如查询物流、申请补偿）。
回复字数控制在100-200字。"

坏的 instruction：
"写Python代码"

好的 instruction：
"用Python编写一个函数，实现以下功能：
1. 接受一个字符串列表作为输入
2. 移除列表中的重复元素，同时保持原有顺序
3. 函数需要有类型注解和docstring
4. 提供使用示例"

坏的 instruction：
"解释这个医学概念"

好的 instruction：
"用通俗易懂的语言向没有医学背景的普通人解释以下医学概念。
避免使用专业术语，如需使用请加括号解释。回答结构清晰，
先给出定义，再举一个生活中的类比。"

# 安装依赖
# pip install openai

from openai import OpenAI
import json
import time

client = OpenAI()  # 需要设置 OPENAI_API_KEY 环境变量

def generate_training_data(topic: str, n_examples: int = 20) -> list:
    """
    用GPT-4生成微调训练数据。

    Args:
        topic: 数据主题，如"Python异常处理"
        n_examples: 每次生成的条数（建议不超过30，避免质量下降）

    Returns:
        包含 instruction/input/output 的字典列表
    """
    prompt = f"""生成{n_examples}条关于"{topic}"的高质量训练数据。

每条数据必须包含以下三个字段：
- instruction: 具体、清晰的任务描述（不少于15个字）
- input: 任务处理的原始材料（如果任务不需要额外输入，填空字符串""）
- output: 高质量的期望回答（不少于50个字）

要求：
1. 覆盖不同难度（初级/中级/高级各占约三分之一）
2. 覆盖不同角度（概念解释、代码示例、常见错误、最佳实践等）
3. 回答要专业、准确、有实用价值
4. instruction字段要具体，不能是"解释X"这种过于宽泛的描述

必须以如下JSON格式返回，外层键名必须是"data"：
{{"data": [
  {{"instruction": "...", "input": "...", "output": "..."}},
  ...
]}}"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},  # 强制JSON格式输出
        temperature=0.8  # 稍高的温度增加多样性
    )

    raw = json.loads(response.choices[0].message.content)

    # 容错处理：GPT-4有时会用不同的键名
    # 按优先级尝试常见键名
    examples = (
        raw.get("data") or          # 我们要求的格式
        raw.get("examples") or      # 另一种常见键名
        raw.get("training_data") or # 另一种可能的键名
        raw.get("items") or
        []
    )

    # 如果上面都找不到，尝试找第一个列表类型的值
    if not examples:
        for value in raw.values():
            if isinstance(value, list) and len(value) > 0:
                examples = value
                break

    print(f"成功生成 {len(examples)} 条训练数据")
    return examples


def batch_generate(topic: str, total: int = 200, batch_size: int = 20) -> list:
    """
    分批生成大量训练数据，避免单次请求质量下降。

    Args:
        topic: 主题
        total: 总条数
        batch_size: 每批条数
    """
    all_data = []
    batches = (total + batch_size - 1) // batch_size  # 向上取整

    for i in range(batches):
        print(f"生成第 {i+1}/{batches} 批...")
        batch = generate_training_data(topic, batch_size)
        all_data.extend(batch)
        time.sleep(1)  # 避免触发API限流

    print(f"总计生成: {len(all_data)} 条")
    return all_data


# 使用示例
if __name__ == "__main__":
    # 生成关于"Python异常处理"的训练数据
    data = batch_generate(topic="Python异常处理", total=100, batch_size=20)

    # 保存到文件
    with open("training_data_raw.json", "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

    print(f"数据已保存到 training_data_raw.json")

# pip install datasets

from datasets import load_dataset

# 加载经典的Alpaca数据集（52000条英文指令数据）
dataset = load_dataset("tatsu-lab/alpaca", split="train")
print(f"Alpaca数据集: {len(dataset)} 条")

# 过滤出特定类型的数据（比如代码相关）
code_data = dataset.filter(
    lambda x: any(kw in x['instruction'].lower()
                  for kw in ['python', 'code', 'function', 'program'])
)
print(f"代码相关数据: {len(code_data)} 条")

# 中文数据集推荐
# load_dataset("BelleGroup/train_1M_CN")  # 中文指令数据集，100万条
# load_dataset("shibing624/alpaca-zh")    # 中文Alpaca数据集

import pandas as pd

# 假设你有一份客服对话记录的CSV文件
# 列名：question, answer, satisfaction_score
df = pd.read_csv("customer_service_logs.csv")

# 只保留用户满意度高的对话（高质量数据）
high_quality = df[df['satisfaction_score'] >= 4]

# 转换为训练格式
training_data = []
for _, row in high_quality.iterrows():
    training_data.append({
        "instruction": "你是一名专业的客服代表，请回答用户问题。",
        "input": row['question'],
        "output": row['answer']
    })

print(f"从业务数据提取了 {len(training_data)} 条训练样本")

import json
import statistics
from collections import Counter

def analyze_dataset(data_path: str) -> dict:
    """
    全面分析数据集质量，输出关键统计指标。

    Args:
        data_path: JSON数据文件路径

    Returns:
        包含各项统计指标的字典
    """
    with open(data_path, 'r', encoding='utf-8') as f:
        data = json.load(f)

    print(f"{'='*50}")
    print(f"数据集分析报告")
    print(f"{'='*50}")
    print(f"总数据量: {len(data)} 条")

    # 1. 长度分布分析
    output_lengths = [len(d.get('output', '')) for d in data]
    instruction_lengths = [len(d.get('instruction', '')) for d in data]

    print(f"\n[输出长度分布]")
    print(f"  最短: {min(output_lengths)} 字符")
    print(f"  最长: {max(output_lengths)} 字符")
    print(f"  平均: {statistics.mean(output_lengths):.0f} 字符")
    print(f"  中位数: {statistics.median(output_lengths):.0f} 字符")

    # 长度分段统计（直观了解分布）
    length_buckets = Counter()
    for l in output_lengths:
        if l < 50:
            length_buckets["<50字符（极短）"] += 1
        elif l < 200:
            length_buckets["50-200字符（短）"] += 1
        elif l < 500:
            length_buckets["200-500字符（中）"] += 1
        elif l < 1000:
            length_buckets["500-1000字符（长）"] += 1
        else:
            length_buckets[">1000字符（极长）"] += 1

    print(f"\n[输出长度分段]")
    for bucket, count in sorted(length_buckets.items()):
        pct = count / len(data) * 100
        print(f"  {bucket}: {count}条 ({pct:.1f}%)")

    # 2. 重复数据检测
    outputs = [d.get('output', '') for d in data]
    instructions = [d.get('instruction', '') + d.get('input', '') for d in data]
    output_duplicates = len(outputs) - len(set(outputs))
    instruction_duplicates = len(instructions) - len(set(instructions))

    print(f"\n[重复数据]")
    print(f"  重复输出: {output_duplicates} 条")
    print(f"  重复指令+输入组合: {instruction_duplicates} 条")

    # 3. 多样性分析（instruction 关键词分布）
    # 统计 instruction 中出现频率最高的词，判断是否过于单一
    all_words = []
    for d in data:
        words = d.get('instruction', '').split()[:5]  # 取前5个词
        all_words.extend(words)

    top_words = Counter(all_words).most_common(10)
    print(f"\n[Instruction 关键词 Top10]（多样性参考）")
    for word, count in top_words:
        print(f"  '{word}': {count}次")

    # 4. 格式一致性检查
    missing_instruction = sum(1 for d in data if not d.get('instruction', '').strip())
    missing_output = sum(1 for d in data if not d.get('output', '').strip())

    print(f"\n[格式检查]")
    print(f"  缺少instruction字段: {missing_instruction} 条")
    print(f"  缺少output字段: {missing_output} 条")

    # 5. 质量评分（综合得分）
    problems = output_duplicates + instruction_duplicates + missing_instruction + missing_output
    short_count = length_buckets.get("<50字符（极短）", 0)
    problems += short_count

    quality_score = max(0, 100 - (problems / len(data)) * 100)
    print(f"\n[综合质量评分]: {quality_score:.1f}/100")
    if quality_score >= 90:
        print("  评级：优秀，可以直接用于训练")
    elif quality_score >= 70:
        print("  评级：良好，建议清洗后再训练")
    else:
        print("  评级：需要大量清洗工作")

    return {
        "total": len(data),
        "output_lengths": output_lengths,
        "duplicates": output_duplicates,
        "quality_score": quality_score
    }


# 使用示例
stats = analyze_dataset("training_data_raw.json")

import re

def clean_dataset(data: list,
                  min_output_length: int = 50,
                  max_output_length: int = 2000) -> list:
    """
    清洗数据集，移除低质量和重复数据。

    参数说明：
    - min_output_length=50：过滤掉50字符以下的输出。
      依据：经验表明，少于50字符的回答通常是"好的"、"知道了"等无实质内容的
      回复，或者是生成失败的截断输出。50字符大约等于2-3句完整的中文短句，
      是一个有意义回答的最低标准。如果你的任务天然就需要短回答（如情感分类
      返回"正面"/"负面"），应把这个值调小到5-10。

    - max_output_length=2000：过滤掉2000字符以上的极长输出。
      依据：过长的输出在训练时会截断，被截断的训练样本会让模型学到"回答可以
      被中途打断"的错误行为。2000字符约等于800-1000汉字，对于大多数问答场景
      已经足够。如果你的任务需要长文生成，应该增大这个值。

    Args:
        data: 原始数据列表
        min_output_length: output字段的最小字符数
        max_output_length: output字段的最大字符数

    Returns:
        清洗后的数据列表
    """
    cleaned = []
    seen_keys = set()  # 用于去重（基于 instruction + input 的组合）

    removed_stats = {
        "empty": 0,
        "too_short": 0,
        "too_long": 0,
        "duplicate": 0,
        "format_error": 0
    }

    for item in data:
        # 检查必填字段是否存在
        if not isinstance(item, dict):
            removed_stats["format_error"] += 1
            continue

        instruction = item.get('instruction', '').strip()
        output = item.get('output', '').strip()
        input_text = item.get('input', '').strip()

        # 过滤空输出
        if not output:
            removed_stats["empty"] += 1
            continue

        # 过滤空 instruction
        if not instruction:
            removed_stats["format_error"] += 1
            continue

        # 过滤过短输出（去掉截断失败的生成结果）
        if len(output) < min_output_length:
            removed_stats["too_short"] += 1
            continue

        # 过滤过长输出（避免训练时截断）
        if len(output) > max_output_length:
            removed_stats["too_long"] += 1
            continue

        # 去除重复（相同的 instruction + input 组合）
        dedup_key = instruction + "|||" + input_text
        if dedup_key in seen_keys:
            removed_stats["duplicate"] += 1
            continue
        seen_keys.add(dedup_key)

        # 文本规范化：去除多余空白
        item['instruction'] = instruction
        item['input'] = input_text
        item['output'] = output

        cleaned.append(item)

    # 输出清洗报告
    print(f"清洗前: {len(data)} 条")
    print(f"清洗后: {len(cleaned)} 条")
    print(f"\n移除原因：")
    for reason, count in removed_stats.items():
        if count > 0:
            labels = {
                "empty": "空输出",
                "too_short": f"输出过短（<{min_output_length}字符）",
                "too_long": f"输出过长（>{max_output_length}字符）",
                "duplicate": "重复数据",
                "format_error": "格式错误"
            }
            print(f"  {labels[reason]}: {count} 条")

    return cleaned


# 使用示例
with open("training_data_raw.json", "r", encoding="utf-8") as f:
    raw_data = json.load(f)

# 根据你的任务调整参数
# 如果是分类任务（输出是"是"/"否"/"正面"等），min_output_length 设为 1
cleaned_data = clean_dataset(
    raw_data,
    min_output_length=50,   # 通用问答任务的推荐值
    max_output_length=2000  # 根据你的模型最大序列长度调整
)

# 保存清洗后的数据
with open("training_data_clean.json", "w", encoding="utf-8") as f:
    json.dump(cleaned_data, f, ensure_ascii=False, indent=2)

print(f"\n清洗后数据已保存到 training_data_clean.json")

from collections import Counter
import math

def measure_diversity(data: list) -> float:
    """
    用词汇多样性粗略衡量数据集的多样性。
    返回值越接近1，多样性越高。
    """
    # 统计所有 instruction 中出现的词
    all_words = []
    for item in data:
        words = item.get('instruction', '').replace('，', ' ').replace('。', ' ').split()
        all_words.extend(words)

    total_words = len(all_words)
    unique_words = len(set(all_words))

    diversity_ratio = unique_words / total_words if total_words > 0 else 0
    print(f"词汇多样性: {diversity_ratio:.3f} (总词数{total_words}, 不重复词数{unique_words})")

    if diversity_ratio > 0.5:
        print("  -> 多样性良好")
    elif diversity_ratio > 0.3:
        print("  -> 多样性一般，建议增加更多覆盖角度")
    else:
        print("  -> 多样性较差，数据集过于单一")

    return diversity_ratio


### 9.2. 指标2：长度分布（Length Distribution）

理想的数据集长度分布应接近正态分布，而不是两极分化：

def check_length_distribution(data: list):
    """检查输出长度是否分布均匀。"""
    lengths = [len(d.get('output', '')) for d in data]

    mean = statistics.mean(lengths)
    std = statistics.stdev(lengths) if len(lengths) > 1 else 0

    # 变异系数（Coefficient of Variation）衡量分布均匀程度
    # CV < 0.5 表示分布较均匀，CV > 1 表示分布非常不均匀
    cv = std / mean if mean > 0 else 0

    print(f"长度分布: 均值={mean:.0f}, 标准差={std:.0f}, 变异系数={cv:.2f}")
    if cv < 0.5:
        print("  -> 长度分布均匀，良好")
    elif cv < 1.0:
        print("  -> 长度分布有些不均匀，可接受")
    else:
        print("  -> 长度分布非常不均匀，建议检查是否有数据生成异常")

def check_format_consistency(data: list, expected_pattern: str = None) -> float:
    """
    检查数据集中有多少条符合预期格式。

    Args:
        expected_pattern: 正则表达式，如 r'^\d+\.' 检查是否以数字开头

    Returns:
        符合格式的比例（0-1）
    """
    if not expected_pattern:
        print("未指定格式要求，跳过格式一致性检查")
        return 1.0

    pattern = re.compile(expected_pattern)
    consistent = sum(1 for d in data if pattern.search(d.get('output', '')))
    ratio = consistent / len(data)

    print(f"格式一致性: {ratio:.1%} ({consistent}/{len(data)} 条符合格式要求)")
    return ratio


# 综合评估示例
print("\n=== 综合质量评估 ===")
with open("training_data_clean.json", "r", encoding="utf-8") as f:
    clean_data = json.load(f)

diversity = measure_diversity(clean_data)
check_length_distribution(clean_data)
# 如果你的输出要求以序号开头，可以这样检查：
# check_format_consistency(clean_data, expected_pattern=r'^\d+[\.\、]')

import json
import time
from openai import OpenAI
import statistics

client = OpenAI()

# 第1步：生成原始数据
print("步骤1：生成原始训练数据...")
raw_data = []
topics = [
    "Python装饰器的使用场景和最佳实践",
    "Python异常处理与错误调试",
    "Python列表推导式和生成器表达式",
]

for topic in topics:
    batch = generate_training_data(topic, n_examples=20)
    raw_data.extend(batch)
    time.sleep(2)

print(f"原始数据: {len(raw_data)} 条")

# 第2步：数据清洗
print("\n步骤2：数据清洗...")
cleaned = clean_dataset(raw_data, min_output_length=50, max_output_length=2000)

# 第3步：质量评估
print("\n步骤3：质量评估...")
measure_diversity(cleaned)
check_length_distribution(cleaned)

# 第4步：保存最终数据集
output_path = "training_data_final.json"
with open(output_path, "w", encoding="utf-8") as f:
    json.dump(cleaned, f, ensure_ascii=False, indent=2)

print(f"\n最终数据集: {len(cleaned)} 条，已保存到 {output_path}")
print(f"下一步：用这份数据集运行 QLoRA 微调（见下一章）")