# 1. 安装Anaconda（推荐）
wget https://repo.anaconda.com/archive/Anaconda3-2023.09-0-Linux-x86_64.sh
bash Anaconda3-2023.09-0-Linux-x86_64.sh

# 2. 创建虚拟环境
conda create -n llm_finetune python=3.10
conda activate llm_finetune

# 3. 安装CUDA（如果系统没有）
conda install cudatoolkit=11.8 -c pytorch -c conda-forge

# 4. 验证CUDA安装
nvidia-smi
nvcc --version

#!/usr/bin/env python3
"""
环境检查脚本 - 验证微调环境是否正确配置
"""

import sys
import subprocess
import importlib
import torch
import platform

def check_python_version():
    """检查Python版本"""
    version = sys.version_info
    print(f"Python版本: {version.major}.{version.minor}.{version.micro}")
    
    if version.major == 3 and version.minor >= 8:
        print("✅ Python版本符合要求")
        return True
    else:
        print("❌ Python版本过低，建议使用3.8+")
        return False

def check_gpu_availability():
    """检查GPU可用性"""
    print(f"\n{'='*50}")
    print("GPU环境检查")
    print(f"{'='*50}")
    
    # 检查CUDA是否可用
    cuda_available = torch.cuda.is_available()
    print(f"CUDA可用: {cuda_available}")
    
    if cuda_available:
        # GPU数量
        gpu_count = torch.cuda.device_count()
        print(f"GPU数量: {gpu_count}")
        
        # 每个GPU的信息
        for i in range(gpu_count):
            gpu_name = torch.cuda.get_device_name(i)
            gpu_memory = torch.cuda.get_device_properties(i).total_memory / 1024**3
            print(f"GPU {i}: {gpu_name} ({gpu_memory:.1f}GB)")
            
        # 当前GPU使用情况
        print(f"当前使用GPU: {torch.cuda.current_device()}")
        
        # 显存使用情况
        memory_allocated = torch.cuda.memory_allocated() / 1024**3
        memory_reserved = torch.cuda.memory_reserved() / 1024**3
        print(f"已分配显存: {memory_allocated:.2f}GB")
        print(f"已预留显存: {memory_reserved:.2f}GB")
        
        return True
    else:
        print("❌ CUDA不可用，将使用CPU训练（速度很慢）")
        return False

def check_required_packages():
    """检查必需的Python包"""
    print(f"\n{'='*50}")
    print("Python包检查")
    print(f"{'='*50}")
    
    required_packages = {
        'torch': '2.0.0',
        'transformers': '4.30.0',
        'datasets': '2.12.0',
        'accelerate': '0.20.0',
        'peft': '0.4.0',
        'bitsandbytes': '0.39.0',
        'scipy': '1.10.0',
        'numpy': '1.24.0',
        'pandas': '2.0.0',
        'tqdm': '4.65.0',
        'wandb': '0.15.0'
    }
    
    all_installed = True
    
    for package, min_version in required_packages.items():
        try:
            module = importlib.import_module(package)
            version = getattr(module, '__version__', 'unknown')
            print(f"✅ {package}: {version}")
        except ImportError:
            print(f"❌ {package}: 未安装")
            all_installed = False
    
    return all_installed

def check_system_resources():
    """检查系统资源"""
    print(f"\n{'='*50}")
    print("系统资源检查")
    print(f"{'='*50}")
    
    # 操作系统
    print(f"操作系统: {platform.system()} {platform.release()}")
    
    # CPU信息
    try:
        import psutil
        cpu_count = psutil.cpu_count()
        memory = psutil.virtual_memory()
        disk = psutil.disk_usage('/')
        
        print(f"CPU核心数: {cpu_count}")
        print(f"内存总量: {memory.total / 1024**3:.1f}GB")
        print(f"可用内存: {memory.available / 1024**3:.1f}GB")
        print(f"磁盘总量: {disk.total / 1024**3:.1f}GB")
        print(f"磁盘可用: {disk.free / 1024**3:.1f}GB")
        
        # 资源建议
        if memory.total / 1024**3 < 16:
            print("⚠️  内存不足16GB，建议增加内存")
        if disk.free / 1024**3 < 100:
            print("⚠️  磁盘空间不足100GB，建议清理空间")
            
    except ImportError:
        print("❌ psutil未安装，无法检查系统资源")

def test_basic_operations():
    """测试基本操作"""
    print(f"\n{'='*50}")
    print("基本功能测试")
    print(f"{'='*50}")
    
    try:
        # 测试张量操作
        x = torch.randn(1000, 1000)
        if torch.cuda.is_available():
            x = x.cuda()
            print("✅ GPU张量操作正常")
        else:
            print("✅ CPU张量操作正常")
        
        # 测试transformers
        from transformers import AutoTokenizer
        tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
        tokens = tokenizer("Hello, world!")
        print("✅ Transformers库正常")
        
        # 测试数据集
        from datasets import Dataset
        data = Dataset.from_dict({"text": ["Hello", "World"]})
        print("✅ Datasets库正常")
        
        return True
        
    except Exception as e:
        print(f"❌ 基本功能测试失败: {e}")
        return False

def main():
    """主函数"""
    print("🚀 大模型微调环境检查")
    print(f"{'='*60}")
    
    # 执行所有检查
    checks = [
        check_python_version(),
        check_gpu_availability(),
        check_required_packages(),
        test_basic_operations()
    ]
    
    # 系统资源检查（不影响总体结果）
    check_system_resources()
    
    # 总结
    print(f"\n{'='*60}")
    print("检查结果总结")
    print(f"{'='*60}")
    
    if all(checks):
        print("🎉 环境配置完美！可以开始微调了")
        print("\n建议的下一步操作：")
        print("1. 准备训练数据")
        print("2. 选择合适的基础模型")
        print("3. 配置微调参数")
        print("4. 开始训练")
    else:
        print("⚠️  环境配置存在问题，请根据上述提示进行修复")
        print("\n常见解决方案：")
        print("1. 更新Python版本")
        print("2. 安装缺失的包: pip install -r requirements.txt")
        print("3. 检查CUDA驱动安装")
        print("4. 重启系统后重试")

if __name__ == "__main__":
    main()

# 创建requirements.txt文件
cat > requirements.txt << EOF
# 核心深度学习框架
torch>=2.0.0
torchvision>=0.15.0
torchaudio>=2.0.0

# Transformers生态
transformers>=4.30.0
tokenizers>=0.13.0
datasets>=2.12.0
evaluate>=0.4.0

# 高效微调库
peft>=0.4.0
accelerate>=0.20.0
bitsandbytes>=0.39.0

# 分布式训练
deepspeed>=0.9.0
fairscale>=0.4.13

# 数据处理
numpy>=1.24.0
pandas>=2.0.0
scipy>=1.10.0
scikit-learn>=1.2.0

# 可视化和监控
matplotlib>=3.7.0
seaborn>=0.12.0
wandb>=0.15.0
tensorboard>=2.13.0

# 工具库
tqdm>=4.65.0
rich>=13.0.0
click>=8.1.0
pyyaml>=6.0
jsonlines>=3.1.0

# Jupyter支持
jupyter>=1.0.0
ipywidgets>=8.0.0
EOF

# 安装所有依赖
pip install -r requirements.txt

# 验证安装
python -c "import torch; print(f'PyTorch版本: {torch.__version__}')"
python -c "import transformers; print(f'Transformers版本: {transformers.__version__}')"

#!/bin/bash
# 大模型微调环境安装脚本

set -e  # 遇到错误立即退出

echo "🚀 开始安装大模型微调环境..."

# 1. 检查Python版本
python_version=$(python3 --version 2>&1 | awk '{print $2}' | cut -d. -f1,2)
echo "当前Python版本: $python_version"

if [[ $(echo "$python_version >= 3.8" | bc -l) -eq 0 ]]; then
    echo "❌ Python版本过低，需要3.8+版本"
    exit 1
fi

# 2. 升级pip
echo "📦 升级pip..."
pip install --upgrade pip setuptools wheel

# 3. 安装PyTorch（根据CUDA版本选择）
echo "🔥 安装PyTorch..."
if command -v nvidia-smi &> /dev/null; then
    cuda_version=$(nvidia-smi | grep "CUDA Version" | awk '{print $9}' | cut -d. -f1,2)
    echo "检测到CUDA版本: $cuda_version"
    
    if [[ "$cuda_version" == "11.8" ]]; then
        pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118
    elif [[ "$cuda_version" == "12.1" ]]; then
        pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121
    else
        echo "⚠️  未识别的CUDA版本，安装CPU版本"
        pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cpu
    fi
else
    echo "⚠️  未检测到NVIDIA GPU，安装CPU版本"
    pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cpu
fi

# 4. 安装Transformers生态
echo "🤗 安装Transformers生态..."
pip install transformers[torch] datasets evaluate tokenizers

# 5. 安装高效微调库
echo "⚡ 安装高效微调库..."
pip install peft accelerate bitsandbytes

# 6. 安装分布式训练库
echo "🌐 安装分布式训练库..."
pip install deepspeed fairscale

# 7. 安装数据处理库
echo "📊 安装数据处理库..."
pip install numpy pandas scipy scikit-learn

# 8. 安装可视化库
echo "📈 安装可视化库..."
pip install matplotlib seaborn wandb tensorboard

# 9. 安装工具库
echo "🛠️  安装工具库..."
pip install tqdm rich click pyyaml jsonlines

# 10. 验证安装
echo "✅ 验证安装..."
python3 << EOF
import torch
import transformers
import datasets
import peft
import accelerate

print(f"✅ PyTorch: {torch.__version__}")
print(f"✅ Transformers: {transformers.__version__}")
print(f"✅ Datasets: {datasets.__version__}")
print(f"✅ PEFT: {peft.__version__}")
print(f"✅ Accelerate: {accelerate.__version__}")

if torch.cuda.is_available():
    print(f"✅ CUDA可用，GPU数量: {torch.cuda.device_count()}")
    for i in range(torch.cuda.device_count()):
        print(f"   GPU {i}: {torch.cuda.get_device_name(i)}")
else:
    print("⚠️  CUDA不可用，将使用CPU")
EOF

echo "🎉 环境安装完成！"
echo ""
echo "下一步操作："
echo "1. 运行环境检查脚本: python check_environment.py"
echo "2. 准备训练数据"
echo "3. 开始微调实验"

"""
常见安装问题及解决方案
"""

def solve_common_issues():
    """解决常见安装问题"""
    
    issues_solutions = {
        "CUDA版本不匹配": {
            "症状": "RuntimeError: CUDA runtime error",
            "原因": "PyTorch CUDA版本与系统CUDA版本不匹配",
            "解决方案": [
                "1. 检查系统CUDA版本: nvidia-smi",
                "2. 卸载现有PyTorch: pip uninstall torch torchvision torchaudio",
                "3. 安装匹配版本: pip install torch --index-url https://download.pytorch.org/whl/cu118"
            ]
        },
        
        "内存不足": {
            "症状": "CUDA out of memory",
            "原因": "GPU显存不足",
            "解决方案": [
                "1. 减小batch_size",
                "2. 使用梯度累积: gradient_accumulation_steps",
                "3. 启用梯度检查点: gradient_checkpointing=True",
                "4. 使用QLoRA量化微调"
            ]
        },
        
        "包版本冲突": {
            "症状": "ImportError或AttributeError",
            "原因": "不同包版本之间不兼容",
            "解决方案": [
                "1. 创建新的虚拟环境",
                "2. 按照指定版本安装: pip install package==version",
                "3. 使用conda管理依赖: conda install package"
            ]
        },
        
        "网络连接问题": {
            "症状": "下载模型或数据集失败",
            "原因": "网络连接不稳定或被墙",
            "解决方案": [
                "1. 使用镜像源: export HF_ENDPOINT=https://hf-mirror.com",
                "2. 手动下载模型文件",
                "3. 使用代理: export https_proxy=http://proxy:port"
            ]
        }
    }
    
    print("🔧 常见问题解决方案")
    print("="*60)
    
    for issue, info in issues_solutions.items():
        print(f"\n❌ {issue}")
        print(f"症状: {info['症状']}")
        print(f"原因: {info['原因']}")
        print("解决方案:")
        for solution in info['解决方案']:
            print(f"   {solution}")

if __name__ == "__main__":
    solve_common_issues()

"""
微调数据格式标准化示例
"""

# 1. 指令微调格式（Alpaca风格）
instruction_format = {
    "instruction": "请将以下英文翻译成中文",
    "input": "Hello, how are you today?",
    "output": "你好，你今天怎么样？"
}

# 2. 对话格式（ChatML风格）
chat_format = {
    "messages": [
        {"role": "system", "content": "你是一个有用的AI助手"},
        {"role": "user", "content": "请介绍一下Python编程语言"},
        {"role": "assistant", "content": "Python是一种高级编程语言..."}
    ]
}

# 3. 问答格式
qa_format = {
    "question": "什么是深度学习？",
    "answer": "深度学习是机器学习的一个分支，使用多层神经网络...",
    "context": "在人工智能领域中..."  # 可选的上下文信息
}

# 4. 文本分类格式
classification_format = {
    "text": "这家餐厅的服务态度很好，食物也很美味",
    "label": "positive",
    "confidence": 0.95  # 可选的置信度
}

# 5. 代码生成格式
code_format = {
    "instruction": "编写一个Python函数来计算斐波那契数列",
    "input": "def fibonacci(n):",
    "output": """def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)"""
}

# 6. 多轮对话格式
multi_turn_format = {
    "conversations": [
        {"from": "human", "value": "你好"},
        {"from": "gpt", "value": "你好！有什么可以帮助你的吗？"},
        {"from": "human", "value": "请介绍一下机器学习"},
        {"from": "gpt", "value": "机器学习是人工智能的一个重要分支..."}
    ]
}

def create_training_data_template():
    """创建训练数据模板"""
    
    templates = {
        "alpaca": {
            "description": "Alpaca指令微调格式，适用于指令跟随任务",
            "required_fields": ["instruction", "output"],
            "optional_fields": ["input"],
            "example": instruction_format
        },
        
        "chatml": {
            "description": "ChatML对话格式，适用于对话系统",
            "required_fields": ["messages"],
            "optional_fields": [],
            "example": chat_format
        },
        
        "qa": {
            "description": "问答格式，适用于知识问答任务",
            "required_fields": ["question", "answer"],
            "optional_fields": ["context"],
            "example": qa_format
        },
        
        "classification": {
            "description": "分类格式，适用于文本分类任务",
            "required_fields": ["text", "label"],
            "optional_fields": ["confidence"],
            "example": classification_format
        }
    }
    
    return templates

def validate_data_format(data, format_type):
    """验证数据格式是否正确"""
    
    templates = create_training_data_template()
    
    if format_type not in templates:
        return False, f"不支持的格式类型: {format_type}"
    
    template = templates[format_type]
    required_fields = template["required_fields"]
    
    # 检查必需字段
    for field in required_fields:
        if field not in data:
            return False, f"缺少必需字段: {field}"
    
    # 特殊格式验证
    if format_type == "chatml":
        messages = data.get("messages", [])
        if not isinstance(messages, list) or len(messages) == 0:
            return False, "messages字段必须是非空列表"
        
        for msg in messages:
            if "role" not in msg or "content" not in msg:
                return False, "每条消息必须包含role和content字段"
    
    return True, "格式验证通过"

# 使用示例
if __name__ == "__main__":
    # 验证数据格式
    is_valid, message = validate_data_format(instruction_format, "alpaca")
    print(f"Alpaca格式验证: {message}")
    
    is_valid, message = validate_data_format(chat_format, "chatml")
    print(f"ChatML格式验证: {message}")

"""
大模型微调数据预处理工具包
"""

import json
import re
import pandas as pd
import numpy as np
from typing import List, Dict, Any, Tuple
from collections import Counter
import hashlib
from tqdm import tqdm
import jieba
from transformers import AutoTokenizer

class DataPreprocessor:
    """数据预处理器"""
    
    def __init__(self, tokenizer_name: str = "bert-base-chinese"):
        """初始化预处理器"""
        self.tokenizer = AutoTokenizer.from_pretrained(tokenizer_name)
        self.stats = {}
        
    def load_raw_data(self, file_path: str, format_type: str = "json") -> List[Dict]:
        """加载原始数据"""
        print(f"📂 加载数据文件: {file_path}")
        
        if format_type == "json":
            with open(file_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
        elif format_type == "jsonl":
            data = []
            with open(file_path, 'r', encoding='utf-8') as f:
                for line in f:
                    data.append(json.loads(line.strip()))
        elif format_type == "csv":
            df = pd.read_csv(file_path)
            data = df.to_dict('records')
        else:
            raise ValueError(f"不支持的文件格式: {format_type}")
        
        print(f"✅ 成功加载 {len(data)} 条数据")
        return data
    
    def remove_duplicates(self, data: List[Dict]) -> List[Dict]:
        """去除重复数据"""
        print("🔄 去除重复数据...")
        
        seen_hashes = set()
        unique_data = []
        
        for item in tqdm(data, desc="去重处理"):
            # 创建数据的哈希值
            content = str(item)
            hash_value = hashlib.md5(content.encode()).hexdigest()
            
            if hash_value not in seen_hashes:
                seen_hashes.add(hash_value)
                unique_data.append(item)
        
        removed_count = len(data) - len(unique_data)
        print(f"✅ 去除 {removed_count} 条重复数据，剩余 {len(unique_data)} 条")
        
        self.stats['duplicates_removed'] = removed_count
        return unique_data
    
    def clean_text(self, text: str) -> str:
        """清洗文本内容"""
        if not isinstance(text, str):
            return ""
        
        # 去除多余空白字符
        text = re.sub(r'\s+', ' ', text)
        
        # 去除特殊控制字符
        text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]', '', text)
        
        # 去除过多的标点符号
        text = re.sub(r'[。！？]{3,}', '。', text)
        text = re.sub(r'[,，]{2,}', '，', text)
        
        # 去除HTML标签
        text = re.sub(r'<[^>]+>', '', text)
        
        # 去除URL
        text = re.sub(r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', '', text)
        
        return text.strip()
    
    def filter_by_length(self, data: List[Dict], min_length: int = 10, max_length: int = 2048) -> List[Dict]:
        """根据长度过滤数据"""
        print(f"📏 过滤长度不合适的数据 (min={min_length}, max={max_length})...")
        
        filtered_data = []
        length_stats = []
        
        for item in tqdm(data, desc="长度过滤"):
            # 获取主要文本内容
            text_content = self._extract_main_text(item)
            text_length = len(text_content)
            length_stats.append(text_length)
            
            if min_length <= text_length <= max_length:
                filtered_data.append(item)
        
        # 统计信息
        removed_count = len(data) - len(filtered_data)
        avg_length = np.mean(length_stats)
        
        print(f"✅ 过滤 {removed_count} 条长度不合适的数据")
        print(f"📊 平均文本长度: {avg_length:.1f} 字符")
        
        self.stats['length_filtered'] = removed_count
        self.stats['avg_length'] = avg_length
        
        return filtered_data
    
    def _extract_main_text(self, item: Dict) -> str:
        """提取主要文本内容"""
        # 根据不同格式提取文本
        if 'instruction' in item and 'output' in item:
            return item['instruction'] + item.get('input', '') + item['output']
        elif 'messages' in item:
            return ' '.join([msg['content'] for msg in item['messages']])
        elif 'question' in item and 'answer' in item:
            return item['question'] + item['answer']
        elif 'text' in item:
            return item['text']
        else:
            return str(item)
    
    def standardize_format(self, data: List[Dict], target_format: str = "alpaca") -> List[Dict]:
        """标准化数据格式"""
        print(f"🔄 标准化为 {target_format} 格式...")
        
        standardized_data = []
        
        for item in tqdm(data, desc="格式标准化"):
            try:
                if target_format == "alpaca":
                    standardized_item = self._to_alpaca_format(item)
                elif target_format == "chatml":
                    standardized_item = self._to_chatml_format(item)
                else:
                    raise ValueError(f"不支持的目标格式: {target_format}")
                
                if standardized_item:
                    standardized_data.append(standardized_item)
                    
            except Exception as e:
                print(f"⚠️  格式转换失败: {e}")
                continue
        
        print(f"✅ 成功标准化 {len(standardized_data)} 条数据")
        return standardized_data
    
    def _to_alpaca_format(self, item: Dict) -> Dict:
        """转换为Alpaca格式"""
        # 如果已经是Alpaca格式
        if 'instruction' in item and 'output' in item:
            return {
                'instruction': self.clean_text(item['instruction']),
                'input': self.clean_text(item.get('input', '')),
                'output': self.clean_text(item['output'])
            }
        
        # 从对话格式转换
        elif 'messages' in item:
            messages = item['messages']
            if len(messages) >= 2:
                user_msg = next((msg for msg in messages if msg['role'] == 'user'), None)
                assistant_msg = next((msg for msg in messages if msg['role'] == 'assistant'), None)
                
                if user_msg and assistant_msg:
                    return {
                        'instruction': self.clean_text(user_msg['content']),
                        'input': '',
                        'output': self.clean_text(assistant_msg['content'])
                    }
        
        # 从问答格式转换
        elif 'question' in item and 'answer' in item:
            return {
                'instruction': self.clean_text(item['question']),
                'input': self.clean_text(item.get('context', '')),
                'output': self.clean_text(item['answer'])
            }
        
        return None
    
    def _to_chatml_format(self, item: Dict) -> Dict:
        """转换为ChatML格式"""
        # 如果已经是ChatML格式
        if 'messages' in item:
            cleaned_messages = []
            for msg in item['messages']:
                cleaned_messages.append({
                    'role': msg['role'],
                    'content': self.clean_text(msg['content'])
                })
            return {'messages': cleaned_messages}
        
        # 从Alpaca格式转换
        elif 'instruction' in item and 'output' in item:
            messages = []
            
            # 添加系统消息（如果需要）
            if item.get('input'):
                user_content = f"{item['instruction']}\n\n{item['input']}"
            else:
                user_content = item['instruction']
            
            messages.append({'role': 'user', 'content': self.clean_text(user_content)})
            messages.append({'role': 'assistant', 'content': self.clean_text(item['output'])})
            
            return {'messages': messages}
        
        return None
    
    def analyze_data_distribution(self, data: List[Dict]) -> Dict:
        """分析数据分布"""
        print("📊 分析数据分布...")
        
        analysis = {
            'total_samples': len(data),
            'text_lengths': [],
            'token_counts': [],
            'format_types': Counter()
        }
        
        for item in tqdm(data, desc="数据分析"):
            # 文本长度分析
            text_content = self._extract_main_text(item)
            analysis['text_lengths'].append(len(text_content))
            
            # Token数量分析
            tokens = self.tokenizer.encode(text_content, add_special_tokens=False)
            analysis['token_counts'].append(len(tokens))
            
            # 格式类型统计
            if 'instruction' in item:
                analysis['format_types']['alpaca'] += 1
            elif 'messages' in item:
                analysis['format_types']['chatml'] += 1
            elif 'question' in item:
                analysis['format_types']['qa'] += 1
            else:
                analysis['format_types']['other'] += 1
        
        # 计算统计信息
        analysis['length_stats'] = {
            'mean': np.mean(analysis['text_lengths']),
            'median': np.median(analysis['text_lengths']),
            'std': np.std(analysis['text_lengths']),
            'min': np.min(analysis['text_lengths']),
            'max': np.max(analysis['text_lengths'])
        }
        
        analysis['token_stats'] = {
            'mean': np.mean(analysis['token_counts']),
            'median': np.median(analysis['token_counts']),
            'std': np.std(analysis['token_counts']),
            'min': np.min(analysis['token_counts']),
            'max': np.max(analysis['token_counts'])
        }
        
        # 打印分析结果
        print(f"📈 数据分析结果:")
        print(f"   总样本数: {analysis['total_samples']}")
        print(f"   平均文本长度: {analysis['length_stats']['mean']:.1f} 字符")
        print(f"   平均Token数: {analysis['token_stats']['mean']:.1f}")
        print(f"   格式分布: {dict(analysis['format_types'])}")
        
        return analysis
    
    def split_dataset(self, data: List[Dict], train_ratio: float = 0.8, 
                     val_ratio: float = 0.1, test_ratio: float = 0.1) -> Tuple[List[Dict], List[Dict], List[Dict]]:
        """分割数据集"""
        print(f"✂️  分割数据集 (train:{train_ratio}, val:{val_ratio}, test:{test_ratio})...")
        
        # 验证比例
        if abs(train_ratio + val_ratio + test_ratio - 1.0) > 1e-6:
            raise ValueError("训练、验证、测试比例之和必须等于1")
        
        # 随机打乱数据
        np.random.shuffle(data)
        
        total_size = len(data)
        train_size = int(total_size * train_ratio)
        val_size = int(total_size * val_ratio)
        
        train_data = data[:train_size]
        val_data = data[train_size:train_size + val_size]
        test_data = data[train_size + val_size:]
        
        print(f"✅ 数据分割完成:")
        print(f"   训练集: {len(train_data)} 样本")
        print(f"   验证集: {len(val_data)} 样本")
        print(f"   测试集: {len(test_data)} 样本")
        
        return train_data, val_data, test_data
    
    def save_processed_data(self, data: List[Dict], output_path: str, format_type: str = "jsonl"):
        """保存处理后的数据"""
        print(f"💾 保存数据到: {output_path}")
        
        if format_type == "jsonl":
            with open(output_path, 'w', encoding='utf-8') as f:
                for item in data:
                    f.write(json.dumps(item, ensure_ascii=False) + '\n')
        elif format_type == "json":
            with open(output_path, 'w', encoding='utf-8') as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
        else:
            raise ValueError(f"不支持的保存格式: {format_type}")
        
        print(f"✅ 成功保存 {len(data)} 条数据")

def main():
    """主函数 - 数据预处理流程示例"""
    
    # 初始化预处理器
    preprocessor = DataPreprocessor()
    
    # 1. 加载原始数据
    raw_data = preprocessor.load_raw_data("raw_data.jsonl", "jsonl")
    
    # 2. 去除重复数据
    unique_data = preprocessor.remove_duplicates(raw_data)
    
    # 3. 长度过滤
    filtered_data = preprocessor.filter_by_length(unique_data, min_length=10, max_length=1024)
    
    # 4. 标准化格式
    standardized_data = preprocessor.standardize_format(filtered_data, target_format="alpaca")
    
    # 5. 数据分析
    analysis = preprocessor.analyze_data_distribution(standardized_data)
    
    # 6. 分割数据集
    train_data, val_data, test_data = preprocessor.split_dataset(standardized_data)
    
    # 7. 保存处理后的数据
    preprocessor.save_processed_data(train_data, "train.jsonl")
    preprocessor.save_processed_data(val_data, "val.jsonl")
    preprocessor.save_processed_data(test_data, "test.jsonl")
    
    print("🎉 数据预处理完成！")

if __name__ == "__main__":
    main()

"""
数据质量控制工具
"""

import re
import json
import numpy as np
from typing import List, Dict, Tuple, Set
from collections import Counter, defaultdict
import langdetect
from textstat import flesch_reading_ease
import jieba.analyse

class DataQualityController:
    """数据质量控制器"""
    
    def __init__(self):
        self.quality_rules = self._load_quality_rules()
        self.quality_scores = defaultdict(list)
        
    def _load_quality_rules(self) -> Dict:
        """加载质量检查规则"""
        return {
            'min_length': 10,
            'max_length': 2048,
            'min_words': 3,
            'max_repetition_ratio': 0.3,
            'forbidden_patterns': [
                r'测试',
                r'test',
                r'[a-zA-Z]{50,}',  # 过长的英文字符串
                r'(.)\1{10,}',     # 重复字符
            ],
            'required_punctuation_ratio': 0.01,
            'max_special_char_ratio': 0.1
        }
    
    def comprehensive_quality_check(self, data: List[Dict]) -> Dict:
        """综合质量检查"""
        print("🔍 开始综合质量检查...")
        
        results = {
            'total_samples': len(data),
            'passed_samples': 0,
            'failed_samples': 0,
            'quality_issues': defaultdict(int),
            'quality_distribution': [],
            'recommendations': []
        }
        
        high_quality_data = []
        
        for i, item in enumerate(data):
            quality_score, issues = self._check_single_sample(item, i)
            results['quality_distribution'].append(quality_score)
            
            # 记录问题
            for issue in issues:
                results['quality_issues'][issue] += 1
            
            # 质量阈值判断
            if quality_score >= 0.7:  # 70分以上认为是高质量
                results['passed_samples'] += 1
                high_quality_data.append(item)
            else:
                results['failed_samples'] += 1
        
        # 生成质量报告
        results['avg_quality_score'] = np.mean(results['quality_distribution'])
        results['quality_std'] = np.std(results['quality_distribution'])
        results['recommendations'] = self._generate_recommendations(results)
        
        self._print_quality_report(results)
        
        return results, high_quality_data
    
    def _check_single_sample(self, item: Dict, index: int) -> Tuple[float, List[str]]:
        """检查单个样本的质量"""
        issues = []
        scores = []
        
        # 提取主要文本
        text_content = self._extract_text_content(item)
        
        # 1. 长度检查
        length_score, length_issues = self._check_length(text_content)
        scores.append(length_score)
        issues.extend(length_issues)
        
        # 2. 内容质量检查
        content_score, content_issues = self._check_content_quality(text_content)
        scores.append(content_score)
        issues.extend(content_issues)
        
        # 3. 格式检查
        format_score, format_issues = self._check_format(item)
        scores.append(format_score)
        issues.extend(format_issues)
        
        # 4. 语言检查
        language_score, language_issues = self._check_language(text_content)
        scores.append(language_score)
        issues.extend(language_issues)
        
        # 5. 逻辑一致性检查
        logic_score, logic_issues = self._check_logic_consistency(item)
        scores.append(logic_score)
        issues.extend(logic_issues)
        
        # 计算总体质量分数
        overall_score = np.mean(scores)
        
        return overall_score, issues
    
    def _extract_text_content(self, item: Dict) -> str:
        """提取文本内容"""
        if 'instruction' in item and 'output' in item:
            return item['instruction'] + ' ' + item.get('input', '') + ' ' + item['output']
        elif 'messages' in item:
            return ' '.join([msg['content'] for msg in item['messages']])
        elif 'question' in item and 'answer' in item:
            return item['question'] + ' ' + item['answer']
        else:
            return str(item)
    
    def _check_length(self, text: str) -> Tuple[float, List[str]]:
        """检查文本长度"""
        issues = []
        
        length = len(text)
        word_count = len(text.split())
        
        # 长度检查
        if length < self.quality_rules['min_length']:
            issues.append(f"文本过短({length}字符)")
            return 0.2, issues
        elif length > self.quality_rules['max_length']:
            issues.append(f"文本过长({length}字符)")
            return 0.6, issues
        
        # 词数检查
        if word_count < self.quality_rules['min_words']:
            issues.append(f"词数过少({word_count}词)")
            return 0.3, issues
        
        # 理想长度范围
        if 50 <= length <= 500:
            return 1.0, issues
        elif 20 <= length <= 1000:
            return 0.8, issues
        else:
            return 0.6, issues
    
    def _check_content_quality(self, text: str) -> Tuple[float, List[str]]:
        """检查内容质量"""
        issues = []
        score = 1.0
        
        # 1. 重复内容检查
        repetition_ratio = self._calculate_repetition_ratio(text)
        if repetition_ratio > self.quality_rules['max_repetition_ratio']:
            issues.append(f"重复内容过多({repetition_ratio:.2f})")
            score -= 0.3
        
        # 2. 禁用模式检查
        for pattern in self.quality_rules['forbidden_patterns']:
            if re.search(pattern, text, re.IGNORECASE):
                issues.append(f"包含禁用模式: {pattern}")
                score -= 0.2
        
        # 3. 标点符号检查
        punctuation_ratio = len(re.findall(r'[。！？，、；：]', text)) / len(text)
        if punctuation_ratio < self.quality_rules['required_punctuation_ratio']:
            issues.append("标点符号过少")
            score -= 0.1
        
        # 4. 特殊字符检查
        special_char_ratio = len(re.findall(r'[^\w\s\u4e00-\u9fff。！？，、；：]', text)) / len(text)
        if special_char_ratio > self.quality_rules['max_special_char_ratio']:
            issues.append("特殊字符过多")
            score -= 0.2
        
        # 5. 可读性检查（仅对英文文本）
        try:
            if re.search(r'[a-zA-Z]', text):
                readability = flesch_reading_ease(text)
                if readability < 30:  # 很难阅读
                    issues.append("文本可读性差")
                    score -= 0.1
        except:
            pass
        
        return max(0.0, score), issues
    
    def _calculate_repetition_ratio(self, text: str) -> float:
        """计算重复内容比例"""
        words = text.split()
        if len(words) < 10:
            return 0.0
        
        word_counts = Counter(words)
        repeated_words = sum(count - 1 for count in word_counts.values() if count > 1)
        
        return repeated_words / len(words)
    
    def _check_format(self, item: Dict) -> Tuple[float, List[str]]:
        """检查格式规范"""
        issues = []
        score = 1.0
        
        # 检查必需字段
        if 'instruction' in item:
            if not item.get('instruction', '').strip():
                issues.append("instruction字段为空")
                score -= 0.5
            if not item.get('output', '').strip():
                issues.append("output字段为空")
                score -= 0.5
        elif 'messages' in item:
            messages = item.get('messages', [])
            if not messages:
                issues.append("messages字段为空")
                score -= 0.5
            else:
                for i, msg in enumerate(messages):
                    if 'role' not in msg or 'content' not in msg:
                        issues.append(f"消息{i}缺少必需字段")
                        score -= 0.2
        else:
            issues.append("未识别的数据格式")
            score -= 0.3
        
        return max(0.0, score), issues
    
    def _check_language(self, text: str) -> Tuple[float, List[str]]:
        """检查语言一致性"""
        issues = []
        
        try:
            # 检测语言
            detected_lang = langdetect.detect(text)
            
            # 中英文混合检查
            chinese_chars = len(re.findall(r'[\u4e00-\u9fff]', text))
            english_chars = len(re.findall(r'[a-zA-Z]', text))
            total_chars = chinese_chars + english_chars
            
            if total_chars > 0:
                chinese_ratio = chinese_chars / total_chars
                english_ratio = english_chars / total_chars
                
                # 判断主要语言
                if chinese_ratio > 0.7:
                    expected_lang = 'zh-cn'
                elif english_ratio > 0.7:
                    expected_lang = 'en'
                else:
                    expected_lang = 'mixed'
                
                # 语言一致性检查
                if detected_lang not in ['zh-cn', 'zh', 'en'] and expected_lang != 'mixed':
                    issues.append(f"语言检测异常: {detected_lang}")
                    return 0.5, issues
            
            return 1.0, issues
            
        except Exception as e:
            issues.append(f"语言检测失败: {str(e)}")
            return 0.7, issues
    
    def _check_logic_consistency(self, item: Dict) -> Tuple[float, List[str]]:
        """检查逻辑一致性"""
        issues = []
        score = 1.0
        
        if 'instruction' in item and 'output' in item:
            instruction = item['instruction'].lower()
            output = item['output'].lower()
            
            # 检查问答逻辑
            if '翻译' in instruction and len(re.findall(r'[a-zA-Z]', output)) == 0 and len(re.findall(r'[\u4e00-\u9fff]', instruction)) > 0:
                issues.append("翻译任务输出语言不匹配")
                score -= 0.3
            
            # 检查长度合理性
            if len(instruction) > len(output) * 3:
                issues.append("输出相对于指令过短")
                score -= 0.2
            
            # 检查相关性（简单关键词匹配）
            instruction_keywords = set(jieba.analyse.extract_tags(instruction, topK=5))
            output_keywords = set(jieba.analyse.extract_tags(output, topK=10))
            
            if instruction_keywords and output_keywords:
                overlap = len(instruction_keywords & output_keywords)
                if overlap == 0 and len(instruction) > 20:
                    issues.append("指令与输出相关性低")
                    score -= 0.2
        
        return max(0.0, score), issues
    
    def _generate_recommendations(self, results: Dict) -> List[str]:
        """生成改进建议"""
        recommendations = []
        
        # 基于质量问题生成建议
        issues = results['quality_issues']
        
        if issues.get('文本过短', 0) > results['total_samples'] * 0.1:
            recommendations.append("建议增加更多详细的训练样本，避免过短的文本")
        
        if issues.get('重复内容过多', 0) > results['total_samples'] * 0.05:
            recommendations.append("建议去除重复内容，提高数据多样性")
        
        if issues.get('标点符号过少', 0) > results['total_samples'] * 0.1:
            recommendations.append("建议规范标点符号使用，提高文本可读性")
        
        if results['avg_quality_score'] < 0.7:
            recommendations.append("整体数据质量偏低，建议进行人工审核和清洗")
        
        if results['failed_samples'] > results['total_samples'] * 0.3:
            recommendations.append("低质量样本过多，建议提高数据收集和标注标准")
        
        return recommendations
    
    def _print_quality_report(self, results: Dict):
        """打印质量报告"""
        print(f"\n{'='*60}")
        print("📊 数据质量检查报告")
        print(f"{'='*60}")
        
        print(f"总样本数: {results['total_samples']}")
        print(f"高质量样本: {results['passed_samples']} ({results['passed_samples']/results['total_samples']*100:.1f}%)")
        print(f"低质量样本: {results['failed_samples']} ({results['failed_samples']/results['total_samples']*100:.1f}%)")
        print(f"平均质量分数: {results['avg_quality_score']:.3f}")
        print(f"质量分数标准差: {results['quality_std']:.3f}")
        
        print(f"\n主要质量问题:")
        for issue, count in sorted(results['quality_issues'].items(), key=lambda x: x[1], reverse=True)[:10]:
            print(f"  • {issue}: {count} 次 ({count/results['total_samples']*100:.1f}%)")
        
        print(f"\n改进建议:")
        for i, rec in enumerate(results['recommendations'], 1):
            print(f"  {i}. {rec}")

# 使用示例
def main():
    """质量控制示例"""
    
    # 加载数据
    with open('processed_data.jsonl', 'r', encoding='utf-8') as f:
        data = [json.loads(line) for line in f]
    
    # 质量检查
    controller = DataQualityController()
    results, high_quality_data = controller.comprehensive_quality_check(data)
    
    # 保存高质量数据
    with open('high_quality_data.jsonl', 'w', encoding='utf-8') as f:
        for item in high_quality_data:
            f.write(json.dumps(item, ensure_ascii=False) + '\n')
    
    print(f"✅ 质量控制完成，保存了 {len(high_quality_data)} 条高质量数据")

if __name__ == "__main__":
    main()

"""
全参数微调完整实现
支持多种模型和任务类型
"""

import os
import json
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
from transformers import (
    AutoTokenizer, AutoModelForCausalLM, AutoModelForSeq2SeqLM,
    TrainingArguments, Trainer, DataCollatorForSeq2Seq,
    get_linear_schedule_with_warmup
)
from datasets import load_dataset
import numpy as np
from tqdm import tqdm
import wandb
from typing import Dict, List, Optional, Union
import logging

# 设置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class FullFineTuningConfig:
    """全参数微调配置类"""
    
    def __init__(self):
        # 模型配置
        self.model_name = "baichuan-inc/Baichuan2-7B-Chat"  # 基础模型
        self.model_type = "causal_lm"  # causal_lm 或 seq2seq_lm
        
        # 数据配置
        self.train_file = "train.jsonl"
        self.val_file = "val.jsonl"
        self.max_length = 1024
        self.data_format = "alpaca"  # alpaca, chatml, qa
        
        # 训练配置
        self.output_dir = "./full_finetune_output"
        self.num_epochs = 3
        self.batch_size = 4
        self.gradient_accumulation_steps = 4
        self.learning_rate = 2e-5
        self.weight_decay = 0.01
        self.warmup_ratio = 0.1
        self.max_grad_norm = 1.0
        
        # 保存和评估配置
        self.save_steps = 500
        self.eval_steps = 500
        self.logging_steps = 100
        self.save_total_limit = 3
        
        # 硬件配置
        self.fp16 = True  # 混合精度训练
        self.gradient_checkpointing = True  # 梯度检查点
        self.dataloader_num_workers = 4
        
        # 监控配置
        self.use_wandb = True
        self.wandb_project = "llm-full-finetune"
        self.wandb_run_name = "baichuan2-7b-full"

class AlpacaDataset(Dataset):
    """Alpaca格式数据集"""
    
    def __init__(self, data_file: str, tokenizer, max_length: int = 1024):
        self.tokenizer = tokenizer
        self.max_length = max_length
        
        # 加载数据
        self.data = []
        with open(data_file, 'r', encoding='utf-8') as f:
            for line in f:
                self.data.append(json.loads(line.strip()))
        
        logger.info(f"加载了 {len(self.data)} 条训练数据")
    
    def __len__(self):
        return len(self.data)
    
    def __getitem__(self, idx):
        item = self.data[idx]
        
        # 构建输入文本
        instruction = item['instruction']
        input_text = item.get('input', '')
        output_text = item['output']
        
        # 格式化prompt
        if input_text:
            prompt = f"### Instruction:\n{instruction}\n\n### Input:\n{input_text}\n\n### Response:\n"
        else:
            prompt = f"### Instruction:\n{instruction}\n\n### Response:\n"
        
        # 完整文本 = prompt + 回答
        full_text = prompt + output_text
        
        # Tokenize
        tokenized = self.tokenizer(
            full_text,
            truncation=True,
            max_length=self.max_length,
            padding=False,
            return_tensors="pt"
        )
        
        # 计算labels（只对回答部分计算损失）
        prompt_tokenized = self.tokenizer(
            prompt,
            truncation=True,
            max_length=self.max_length,
            padding=False,
            return_tensors="pt"
        )
        
        labels = tokenized["input_ids"].clone()
        # 将prompt部分的标签设为-100（不计算损失）
        labels[:, :prompt_tokenized["input_ids"].shape[1]] = -100
        
        return {
            "input_ids": tokenized["input_ids"].squeeze(),
            "attention_mask": tokenized["attention_mask"].squeeze(),
            "labels": labels.squeeze()
        }

class ChatMLDataset(Dataset):
    """ChatML格式数据集"""
    
    def __init__(self, data_file: str, tokenizer, max_length: int = 1024):
        self.tokenizer = tokenizer
        self.max_length = max_length
        
        # 加载数据
        self.data = []
        with open(data_file, 'r', encoding='utf-8') as f:
            for line in f:
                self.data.append(json.loads(line.strip()))
        
        logger.info(f"加载了 {len(self.data)} 条对话数据")
    
    def __len__(self):
        return len(self.data)
    
    def __getitem__(self, idx):
        item = self.data[idx]
        messages = item['messages']
        
        # 构建对话文本
        conversation = ""
        for msg in messages:
            role = msg['role']
            content = msg['content']
            
            if role == 'system':
                conversation += f"<|system|>\n{content}\n"
            elif role == 'user':
                conversation += f"<|user|>\n{content}\n"
            elif role == 'assistant':
                conversation += f"<|assistant|>\n{content}\n"
        
        # Tokenize
        tokenized = self.tokenizer(
            conversation,
            truncation=True,
            max_length=self.max_length,
            padding=False,
            return_tensors="pt"
        )
        
        # 对于对话数据，我们通常对整个对话计算损失
        labels = tokenized["input_ids"].clone()
        
        return {
            "input_ids": tokenized["input_ids"].squeeze(),
            "attention_mask": tokenized["attention_mask"].squeeze(),
            "labels": labels.squeeze()
        }

class DataCollatorForCausalLM:
    """因果语言模型数据整理器"""
    
    def __init__(self, tokenizer, max_length: int = 1024):
        self.tokenizer = tokenizer
        self.max_length = max_length
    
    def __call__(self, batch):
        # 获取batch中的最大长度
        max_len = min(
            max(len(item["input_ids"]) for item in batch),
            self.max_length
        )
        
        # 准备批次数据
        input_ids = []
        attention_masks = []
        labels = []
        
        for item in batch:
            # 截断或填充
            input_id = item["input_ids"][:max_len]
            attention_mask = item["attention_mask"][:max_len]
            label = item["labels"][:max_len]
            
            # 右填充
            pad_length = max_len - len(input_id)
            if pad_length > 0:
                input_id = torch.cat([
                    input_id,
                    torch.full((pad_length,), self.tokenizer.pad_token_id)
                ])
                attention_mask = torch.cat([
                    attention_mask,
                    torch.zeros(pad_length)
                ])
                label = torch.cat([
                    label,
                    torch.full((pad_length,), -100)
                ])
            
            input_ids.append(input_id)
            attention_masks.append(attention_mask)
            labels.append(label)
        
        return {
            "input_ids": torch.stack(input_ids),
            "attention_mask": torch.stack(attention_masks),
            "labels": torch.stack(labels)
        }

class FullFineTuner:
    """全参数微调训练器"""
    
    def __init__(self, config: FullFineTuningConfig):
        self.config = config
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        
        # 初始化wandb
        if config.use_wandb:
            wandb.init(
                project=config.wandb_project,
                name=config.wandb_run_name,
                config=vars(config)
            )
        
        # 加载tokenizer和模型
        self._load_model_and_tokenizer()
        
        # 准备数据
        self._prepare_datasets()
    
    def _load_model_and_tokenizer(self):
        """加载模型和分词器"""
        logger.info(f"加载模型: {self.config.model_name}")
        
        # 加载tokenizer
        self.tokenizer = AutoTokenizer.from_pretrained(
            self.config.model_name,
            trust_remote_code=True,
            padding_side="right"
        )
        
        # 设置pad_token
        if self.tokenizer.pad_token is None:
            self.tokenizer.pad_token = self.tokenizer.eos_token
        
        # 加载模型
        if self.config.model_type == "causal_lm":
            self.model = AutoModelForCausalLM.from_pretrained(
                self.config.model_name,
                trust_remote_code=True,
                torch_dtype=torch.float16 if self.config.fp16 else torch.float32,
                device_map="auto"
            )
        else:
            self.model = AutoModelForSeq2SeqLM.from_pretrained(
                self.config.model_name,
                trust_remote_code=True,
                torch_dtype=torch.float16 if self.config.fp16 else torch.float32,
                device_map="auto"
            )
        
        # 启用梯度检查点
        if self.config.gradient_checkpointing:
            self.model.gradient_checkpointing_enable()
        
        logger.info(f"模型参数量: {self.model.num_parameters():,}")
        logger.info(f"可训练参数量: {sum(p.numel() for p in self.model.parameters() if p.requires_grad):,}")
    
    def _prepare_datasets(self):
        """准备数据集"""
        logger.info("准备训练数据集...")
        
        # 选择数据集类
        if self.config.data_format == "alpaca":
            dataset_class = AlpacaDataset
        elif self.config.data_format == "chatml":
            dataset_class = ChatMLDataset
        else:
            raise ValueError(f"不支持的数据格式: {self.config.data_format}")
        
        # 创建数据集
        self.train_dataset = dataset_class(
            self.config.train_file,
            self.tokenizer,
            self.config.max_length
        )
        
        if os.path.exists(self.config.val_file):
            self.val_dataset = dataset_class(
                self.config.val_file,
                self.tokenizer,
                self.config.max_length
            )
        else:
            self.val_dataset = None
            logger.warning("未找到验证集文件")
        
        # 创建数据整理器
        self.data_collator = DataCollatorForCausalLM(
            self.tokenizer,
            self.config.max_length
        )
    
    def train(self):
        """开始训练"""
        logger.info("开始全参数微调训练...")
        
        # 训练参数
        training_args = TrainingArguments(
            output_dir=self.config.output_dir,
            num_train_epochs=self.config.num_epochs,
            per_device_train_batch_size=self.config.batch_size,
            per_device_eval_batch_size=self.config.batch_size,
            gradient_accumulation_steps=self.config.gradient_accumulation_steps,
            learning_rate=self.config.learning_rate,
            weight_decay=self.config.weight_decay,
            warmup_ratio=self.config.warmup_ratio,
            max_grad_norm=self.config.max_grad_norm,
            
            # 保存和评估
            save_steps=self.config.save_steps,
            eval_steps=self.config.eval_steps,
            logging_steps=self.config.logging_steps,
            save_total_limit=self.config.save_total_limit,
            evaluation_strategy="steps" if self.val_dataset else "no",
            
            # 硬件优化
            fp16=self.config.fp16,
            gradient_checkpointing=self.config.gradient_checkpointing,
            dataloader_num_workers=self.config.dataloader_num_workers,
            
            # 其他设置
            remove_unused_columns=False,
            report_to="wandb" if self.config.use_wandb else None,
            run_name=self.config.wandb_run_name,
            load_best_model_at_end=True if self.val_dataset else False,
            metric_for_best_model="eval_loss" if self.val_dataset else None,
        )
        
        # 创建训练器
        trainer = Trainer(
            model=self.model,
            args=training_args,
            train_dataset=self.train_dataset,
            eval_dataset=self.val_dataset,
            data_collator=self.data_collator,
            tokenizer=self.tokenizer,
        )
        
        # 开始训练
        trainer.train()
        
        # 保存最终模型
        trainer.save_model()
        self.tokenizer.save_pretrained(self.config.output_dir)
        
        logger.info(f"训练完成！模型保存在: {self.config.output_dir}")
    
    def evaluate_model(self, test_file: str = None):
        """评估模型性能"""
        if test_file is None:
            test_file = self.config.val_file
        
        if not os.path.exists(test_file):
            logger.warning(f"测试文件不存在: {test_file}")
            return
        
        logger.info("开始模型评估...")
        
        # 加载测试数据
        if self.config.data_format == "alpaca":
            test_dataset = AlpacaDataset(test_file, self.tokenizer, self.config.max_length)
        else:
            test_dataset = ChatMLDataset(test_file, self.tokenizer, self.config.max_length)
        
        # 创建数据加载器
        test_loader = DataLoader(
            test_dataset,
            batch_size=self.config.batch_size,
            collate_fn=self.data_collator,
            shuffle=False
        )
        
        # 评估模式
        self.model.eval()
        total_loss = 0
        total_samples = 0
        
        with torch.no_grad():
            for batch in tqdm(test_loader, desc="评估中"):
                # 移动到设备
                batch = {k: v.to(self.device) for k, v in batch.items()}
                
                # 前向传播
                outputs = self.model(**batch)
                loss = outputs.loss
                
                total_loss += loss.item() * batch['input_ids'].size(0)
                total_samples += batch['input_ids'].size(0)
        
        avg_loss = total_loss / total_samples
        perplexity = torch.exp(torch.tensor(avg_loss))
        
        logger.info(f"评估结果:")
        logger.info(f"  平均损失: {avg_loss:.4f}")
        logger.info(f"  困惑度: {perplexity:.4f}")
        
        if self.config.use_wandb:
            wandb.log({
                "eval/loss": avg_loss,
                "eval/perplexity": perplexity
            })
        
        return {"loss": avg_loss, "perplexity": perplexity.item()}
    
    def generate_sample(self, prompt: str, max_new_tokens: int = 256, temperature: float = 0.7):
        """生成样本文本"""
        self.model.eval()
        
        # 编码输入
        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.device)
        
        # 生成
        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_new_tokens=max_new_tokens,
                temperature=temperature,
                do_sample=True,
                pad_token_id=self.tokenizer.eos_token_id
            )
        
        # 解码输出
        generated_text = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
        
        # 移除输入部分
        response = generated_text[len(prompt):].strip()
        
        return response

def main():
    """主函数"""
    # 创建配置
    config = FullFineTuningConfig()
    
    # 创建训练器
    trainer = FullFineTuner(config)
    
    # 开始训练
    trainer.train()
    
    # 评估模型
    trainer.evaluate_model()
    
    # 测试生成
    test_prompt = "### Instruction:\n请介绍一下人工智能的发展历程\n\n### Response:\n"
    response = trainer.generate_sample(test_prompt)
    print(f"测试生成:\n{response}")

if __name__ == "__main__":
    main()

#!/bin/bash
# 全参数微调训练脚本

# 设置环境变量
export CUDA_VISIBLE_DEVICES=0,1,2,3  # 使用4张GPU
export WANDB_PROJECT="llm-full-finetune"
export WANDB_RUN_NAME="baichuan2-7b-$(date +%Y%m%d_%H%M%S)"

# 创建输出目录
mkdir -p ./full_finetune_output

# 启动训练
python full_finetune.py \
    --model_name "baichuan-inc/Baichuan2-7B-Chat" \
    --train_file "train.jsonl" \
    --val_file "val.jsonl" \
    --output_dir "./full_finetune_output" \
    --num_epochs 3 \
    --batch_size 2 \
    --gradient_accumulation_steps 8 \
    --learning_rate 2e-5 \
    --max_length 1024 \
    --fp16 \
    --gradient_checkpointing \
    --use_wandb

echo "训练完成！"

"""
训练监控和调优工具
"""

import torch
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from typing import Dict, List, Optional
import wandb
import psutil
import GPUtil
from collections import defaultdict, deque
import time
import json

class TrainingMonitor:
    """训练监控器"""
    
    def __init__(self, model, tokenizer, config):
        self.model = model
        self.tokenizer = tokenizer
        self.config = config
        
        # 监控数据存储
        self.metrics_history = defaultdict(list)
        self.gradient_history = deque(maxlen=100)
        self.loss_history = deque(maxlen=1000)
        
        # 监控配置
        self.monitor_interval = 10  # 每10步监控一次
        self.sample_prompts = [
            "请介绍一下机器学习",
            "什么是深度学习？",
            "解释一下神经网络的工作原理"
        ]
        
        # 异常检测阈值
        self.loss_spike_threshold = 2.0  # 损失突增阈值
        self.gradient_explosion_threshold = 10.0  # 梯度爆炸阈值
        self.gpu_memory_threshold = 0.95  # GPU内存使用阈值
    
    def log_training_step(self, step: int, loss: float, learning_rate: float, 
                         grad_norm: Optional[float] = None):
        """记录训练步骤"""
        
        # 记录基本指标
        self.metrics_history['step'].append(step)
        self.metrics_history['loss'].append(loss)
        self.metrics_history['learning_rate'].append(learning_rate)
        self.loss_history.append(loss)
        
        if grad_norm is not None:
            self.metrics_history['grad_norm'].append(grad_norm)
            self.gradient_history.append(grad_norm)
        
        # 检测异常
        self._detect_anomalies(step, loss, grad_norm)
        
        # 定期监控
        if step % self.monitor_interval == 0:
            self._comprehensive_monitoring(step)
    
    def _detect_anomalies(self, step: int, loss: float, grad_norm: Optional[float]):
        """检测训练异常"""
        
        # 检测损失突增
        if len(self.loss_history) > 10:
            recent_avg = np.mean(list(self.loss_history)[-10:])
            if loss > recent_avg * self.loss_spike_threshold:
                self._log_warning(f"步骤 {step}: 检测到损失突增 {loss:.4f} (平均: {recent_avg:.4f})")
        
        # 检测梯度爆炸
        if grad_norm is not None and grad_norm > self.gradient_explosion_threshold:
            self._log_warning(f"步骤 {step}: 检测到梯度爆炸 {grad_norm:.4f}")
        
        # 检测梯度消失
        if grad_norm is not None and grad_norm < 1e-7:
            self._log_warning(f"步骤 {step}: 检测到梯度消失 {grad_norm:.7f}")
    
    def _comprehensive_monitoring(self, step: int):
        """综合监控"""
        
        # GPU监控
        gpu_stats = self._monitor_gpu()
        
        # 内存监控
        memory_stats = self._monitor_memory()
        
        # 模型状态监控
        model_stats = self._monitor_model_state()
        
        # 生成样本（每100步）
        if step % 100 == 0:
            sample_outputs = self._generate_samples()
            
            # 记录到wandb
            if self.config.use_wandb:
                wandb.log({
                    "monitor/gpu_utilization": gpu_stats.get('utilization', 0),
                    "monitor/gpu_memory_used": gpu_stats.get('memory_used', 0),
                    "monitor/cpu_percent": memory_stats.get('cpu_percent', 0),
                    "monitor/memory_percent": memory_stats.get('memory_percent', 0),
                    "monitor/weight_norm": model_stats.get('weight_norm', 0),
                    "monitor/sample_outputs": wandb.Table(
                        columns=["prompt", "output"],
                        data=[[p, o] for p, o in sample_outputs.items()]
                    )
                }, step=step)
    
    def _monitor_gpu(self) -> Dict:
        """监控GPU状态"""
        try:
            gpus = GPUtil.getGPUs()
            if gpus:
                gpu = gpus[0]  # 监控第一个GPU
                
                # 检查GPU内存使用
                if gpu.memoryUtil > self.gpu_memory_threshold:
                    self._log_warning(f"GPU内存使用率过高: {gpu.memoryUtil:.2%}")
                
                return {
                    'utilization': gpu.load * 100,
                    'memory_used': gpu.memoryUtil * 100,
                    'temperature': gpu.temperature
                }
        except Exception as e:
            print(f"GPU监控失败: {e}")
        
        return {}
    
    def _monitor_memory(self) -> Dict:
        """监控系统内存"""
        try:
            # CPU使用率
            cpu_percent = psutil.cpu_percent(interval=1)
            
            # 内存使用率
            memory = psutil.virtual_memory()
            
            return {
                'cpu_percent': cpu_percent,
                'memory_percent': memory.percent,
                'memory_available': memory.available / (1024**3)  # GB
            }
        except Exception as e:
            print(f"内存监控失败: {e}")
        
        return {}
    
    def _monitor_model_state(self) -> Dict:
        """监控模型状态"""
        try:
            # 计算权重范数
            total_norm = 0
            param_count = 0
            
            for param in self.model.parameters():
                if param.requires_grad:
                    param_norm = param.data.norm(2)
                    total_norm += param_norm.item() ** 2
                    param_count += 1
            
            weight_norm = total_norm ** (1. / 2)
            
            return {
                'weight_norm': weight_norm,
                'param_count': param_count
            }
        except Exception as e:
            print(f"模型状态监控失败: {e}")
        
        return {}
    
    def _generate_samples(self) -> Dict[str, str]:
        """生成样本文本"""
        samples = {}
        
        self.model.eval()
        with torch.no_grad():
            for prompt in self.sample_prompts:
                try:
                    # 编码输入
                    inputs = self.tokenizer(prompt, return_tensors="pt")
                    if torch.cuda.is_available():
                        inputs = {k: v.cuda() for k, v in inputs.items()}
                    
                    # 生成
                    outputs = self.model.generate(
                        **inputs,
                        max_new_tokens=100,
                        temperature=0.7,
                        do_sample=True,
                        pad_token_id=self.tokenizer.eos_token_id
                    )
                    
                    # 解码
                    generated_text = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
                    response = generated_text[len(prompt):].strip()
                    
                    samples[prompt] = response
                    
                except Exception as e:
                    samples[prompt] = f"生成失败: {str(e)}"
        
        self.model.train()
        return samples
    
    def _log_warning(self, message: str):
        """记录警告信息"""
        print(f"⚠️  {message}")
        
        if self.config.use_wandb:
            wandb.log({"warnings": message})
    
    def plot_training_curves(self, save_path: str = "training_curves.png"):
        """绘制训练曲线"""
        
        fig, axes = plt.subplots(2, 2, figsize=(15, 10))
        
        # 损失曲线
        if 'loss' in self.metrics_history:
            axes[0, 0].plot(self.metrics_history['step'], self.metrics_history['loss'])
            axes[0, 0].set_title('Training Loss')
            axes[0, 0].set_xlabel('Step')
            axes[0, 0].set_ylabel('Loss')
            axes[0, 0].grid(True)
        
        # 学习率曲线
        if 'learning_rate' in self.metrics_history:
            axes[0, 1].plot(self.metrics_history['step'], self.metrics_history['learning_rate'])
            axes[0, 1].set_title('Learning Rate')
            axes[0, 1].set_xlabel('Step')
            axes[0, 1].set_ylabel('Learning Rate')
            axes[0, 1].grid(True)
        
        # 梯度范数曲线
        if 'grad_norm' in self.metrics_history:
            axes[1, 0].plot(self.metrics_history['step'], self.metrics_history['grad_norm'])
            axes[1, 0].set_title('Gradient Norm')
            axes[1, 0].set_xlabel('Step')
            axes[1, 0].set_ylabel('Gradient Norm')
            axes[1, 0].grid(True)
        
        # 损失分布
        if 'loss' in self.metrics_history:
            axes[1, 1].hist(self.metrics_history['loss'], bins=50, alpha=0.7)
            axes[1, 1].set_title('Loss Distribution')
            axes[1, 1].set_xlabel('Loss')
            axes[1, 1].set_ylabel('Frequency')
            axes[1, 1].grid(True)
        
        plt.tight_layout()
        plt.savefig(save_path, dpi=300, bbox_inches='tight')
        plt.close()
        
        print(f"训练曲线已保存到: {save_path}")
    
    def generate_training_report(self, save_path: str = "training_report.json"):
        """生成训练报告"""
        
        report = {
            "training_summary": {
                "total_steps": len(self.metrics_history['step']),
                "final_loss": self.metrics_history['loss'][-1] if self.metrics_history['loss'] else None,
                "min_loss": min(self.metrics_history['loss']) if self.metrics_history['loss'] else None,
                "avg_loss": np.mean(self.metrics_history['loss']) if self.metrics_history['loss'] else None,
            },
            "gradient_analysis": {
                "avg_grad_norm": np.mean(self.gradient_history) if self.gradient_history else None,
                "max_grad_norm": max(self.gradient_history) if self.gradient_history else None,
                "min_grad_norm": min(self.gradient_history) if self.gradient_history else None,
            },
            "stability_analysis": {
                "loss_variance": np.var(self.metrics_history['loss']) if self.metrics_history['loss'] else None,
                "loss_trend": self._calculate_loss_trend(),
            }
        }
        
        # 保存报告
        with open(save_path, 'w', encoding='utf-8') as f:
            json.dump(report, f, indent=2, ensure_ascii=False)
        
        print(f"训练报告已保存到: {save_path}")
        return report
    
    def _calculate_loss_trend(self) -> str:
        """计算损失趋势"""
        if len(self.metrics_history['loss']) < 10:
            return "数据不足"
        
        recent_losses = self.metrics_history['loss'][-100:]  # 最近100步
        early_losses = self.metrics_history['loss'][:100]    # 前100步
        
        recent_avg = np.mean(recent_losses)
        early_avg = np.mean(early_losses)
        
        if recent_avg < early_avg * 0.9:
            return "下降"
        elif recent_avg > early_avg * 1.1:
            return "上升"
        else:
            return "稳定"

class HyperparameterTuner:
    """超参数调优器"""
    
    def __init__(self):
        self.tuning_history = []
    
    def suggest_learning_rate(self, current_loss: float, loss_history: List[float]) -> float:
        """建议学习率调整"""
        
        if len(loss_history) < 10:
            return None  # 数据不足
        
        # 计算损失变化趋势
        recent_trend = np.mean(loss_history[-5:]) - np.mean(loss_history[-10:-5])
        
        if recent_trend > 0:  # 损失上升
            return "减小学习率到当前的0.5倍"
        elif recent_trend < -0.01:  # 损失下降很快
            return "可以适当增加学习率到当前的1.2倍"
        else:
            return "学习率合适，保持当前设置"
    
    def suggest_batch_size(self, gpu_memory_usage: float) -> str:
        """建议批次大小调整"""
        
        if gpu_memory_usage > 0.9:
            return "GPU内存使用率过高，建议减小batch_size"
        elif gpu_memory_usage < 0.6:
            return "GPU内存使用率较低，可以增加batch_size提高训练效率"
        else:
            return "batch_size设置合理"
    
    def suggest_gradient_accumulation(self, current_batch_size: int, target_batch_size: int) -> int:
        """建议梯度累积步数"""
        
        if target_batch_size <= current_batch_size:
            return 1
        
        return target_batch_size // current_batch_size

# 使用示例
def create_monitor_example():
    """创建监控器示例"""
    
    # 假设已有模型、分词器和配置
    # monitor = TrainingMonitor(model, tokenizer, config)
    
    # 在训练循环中使用
    """
    for step, batch in enumerate(train_loader):
        # 训练步骤
        loss = train_step(batch)
        
        # 计算梯度范数
        grad_norm = torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
        
        # 记录监控数据
        monitor.log_training_step(
            step=step,
            loss=loss.item(),
            learning_rate=optimizer.param_groups[0]['lr'],
            grad_norm=grad_norm.item()
        )
    
    # 训练结束后生成报告
    monitor.plot_training_curves()
    monitor.generate_training_report()
    """
    
    print("监控器使用示例已创建")

if __name__ == "__main__":
    create_monitor_example()