#实战项目一：智能客服工单分类系统

def project_goals():
    """
    项目核心目标
    """
    goals = {
        "分类准确性": "准确率 > 90%, F1-score > 85%",
        "分类体系": "支持咨询类、投诉类、售后类、技术类、其他类等5个类别",
        "处理效率": "单条工单处理时间 < 1秒",
        "系统稳定性": "99.9%可用性保障",
        "扩展性": "支持新增分类类别和模型升级"
    }
    
    print("项目目标:")
    for goal, requirement in goals.items():
        print(f"  {goal}: {requirement}")

project_goals()

def technology_stack():
    """
    项目技术栈
    """
    stack = {
        "数据处理": ["pandas", "scikit-learn", "imbalanced-learn"],
        "模型框架": ["transformers", "torch", "huggingface"],
        "模型选择": ["BERT-base-chinese", "RoBERTa-wwm-ext", "ERNIE"],
        "部署框架": ["FastAPI", "uvicorn", "docker"],
        "监控工具": ["prometheus", "grafana", "elk-stack"]
    }
    
    print("技术栈选择:")
    for category, tools in stack.items():
        print(f"  {category}: {', '.join(tools)}")

technology_stack()

def business_requirements():
    """
    业务需求分析
    """
    print("业务需求分析:")
    
    categories = {
        "咨询类": ["产品咨询", "价格咨询", "功能介绍", "使用方法"],
        "投诉类": ["服务投诉", "质量投诉", "配送问题", "态度问题"],
        "售后类": ["退换货", "维修申请", "退款处理", "发票问题"],
        "技术类": ["Bug反馈", "功能建议", "技术支持", "系统故障"],
        "其他类": ["无关信息", "恶意内容", "重复提交", "无法分类"]
    }
    
    for category, subtypes in categories.items():
        print(f"\n{category}:")
        for subtype in subtypes:
            print(f"  - {subtype}")
    
    print("\n性能要求:")
    performance_requirements = [
        "分类准确率 > 90%",
        "F1-score > 0.85",
        "单条处理时间 < 1秒",
        "并发处理能力 > 100 TPS"
    ]
    
    for req in performance_requirements:
        print(f"  ✓ {req}")

business_requirements()

def data_requirements():
    """
    数据需求分析
    """
    print("数据需求分析:")
    
    data_specification = {
        "字段要求": {
            "工单ID": "唯一标识符",
            "标题": "工单标题文本",
            "内容": "工单详细内容",
            "分类标签": "预定义的分类标签",
            "创建时间": "工单创建时间戳",
            "优先级": "工单紧急程度"
        },
        "数据量要求": {
            "最小训练集": "5000条样本",
            "验证集比例": "20%",
            "测试集比例": "10%",
            "数据质量": "准确率 > 95%"
        },
        "数据质量": {
            "完整性": "必填字段不为空",
            "一致性": "标签格式统一",
            "准确性": "标签分配正确",
            "时效性": "数据更新及时"
        }
    }
    
    for category, specs in data_specification.items():
        print(f"\n{category}:")
        if isinstance(specs, dict):
            for key, value in specs.items():
                if isinstance(value, list):
                    print(f"  {key}:")
                    for item in value:
                        print(f"    - {item}")
                else:
                    print(f"  {key}: {value}")

data_requirements()

def data_loading_exploration():
    """
    数据加载与探索性分析
    """
    import pandas as pd
    import numpy as np
    import matplotlib.pyplot as plt
    import seaborn as sns
    
    print("数据加载与探索性分析:")
    
    # 示例数据加载代码
    data_loading_code = """
    import pandas as pd
    import numpy as np
    from collections import Counter
    import matplotlib.pyplot as plt
    
    # 加载数据
    df = pd.read_csv('customer_tickets.csv')
    
    # 基本信息
    print(f"数据形状: {df.shape}")
    print(f"列名: {df.columns.tolist()}")
    print(f"缺失值: {df.isnull().sum()}")
    
    # 分类分布
    category_dist = df['category'].value_counts()
    print(f"分类分布:\\n{category_dist}")
    
    # 文本长度分布
    df['text_length'] = df['title'].fillna('') + ' ' + df['content'].fillna('')
    df['text_length'] = df['text_length'].apply(len)
    print(f"平均文本长度: {df['text_length'].mean():.2f}")
    """
    
    print("数据加载代码示例:")
    print(data_loading_code)
    
    # 数据质量检查
    quality_checks = [
        "检查重复数据",
        "验证标签一致性", 
        "检测异常文本长度",
        "识别噪声数据",
        "检查数据倾斜程度"
    ]
    
    print("数据质量检查项目:")
    for check in quality_checks:
        print(f"  ✓ {check}")

data_loading_exploration()

def text_preprocessing():
    """
    文本清洗与预处理
    """
    print("文本清洗与预处理:")
    
    preprocessing_pipeline = """
    import re
    import jieba
    import jieba.posseg as pseg
    from zhon.hanzi import punctuation
    import string
    
    def clean_text(text):
        '''
        文本清洗函数
        '''
        if pd.isna(text):
            return ""
        
        # 转换为字符串
        text = str(text)
        
        # 去除URL
        text = re.sub(r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\\\(\\\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', '', text)
        
        # 去除邮箱
        text = re.sub(r'\\S+@\\S+', '', text)
        
        # 去除电话号码
        text = re.sub(r'[0-9]{3}-[0-9]{4}-[0-9]{4}|[0-9]{11}', '', text)
        
        # 去除数字（可选，根据业务需求）
        text = re.sub(r'\\d+', ' ', text)
        
        # 去除特殊字符
        text = re.sub(r'[\\\\/:*?"<>|]', ' ', text)
        
        # 去除多余空白字符
        text = re.sub(r'\\s+', ' ', text)
        
        # 去除中英文标点符号
        text = re.sub(f'[{punctuation}]', ' ', text)
        text = re.sub(f'[{string.punctuation}]', ' ', text)
        
        # 去除空白字符
        text = text.strip()
        
        return text
    
    def preprocess_pipeline(df):
        '''
        预处理流水线
        '''
        # 合并标题和内容
        df['text'] = df['title'].fillna('') + ' ' + df['content'].fillna('')
        
        # 文本清洗
        df['cleaned_text'] = df['text'].apply(clean_text)
        
        # 去除空文本
        df = df[df['cleaned_text'].str.len() > 0]
        
        # 标签编码
        label2id = {cat: i for i, cat in enumerate(df['category'].unique())}
        id2label = {v: k for k, v in label2id.items()}
        df['label'] = df['category'].map(label2id)
        
        return df, label2id, id2label
    """
    
    print("文本预处理代码:")
    print(preprocessing_pipeline)

text_preprocessing()

def imbalanced_data_handling():
    """
    不平衡数据处理策略
    """
    print("不平衡数据处理:")
    
    handling_strategies = [
        {
            "方法": "SMOTE过采样",
            "描述": "生成合成样本平衡数据分布",
            "适用场景": "样本量适中，特征维度不高"
        },
        {
            "方法": "Tomek Links欠采样",
            "描述": "去除边界模糊的样本对",
            "适用场景": "数据量充足，需要清理边界样本"
        },
        {
            "方法": "Class Weight调整",
            "描述": "在损失函数中调整类别权重",
            "适用场景": "保持原始数据分布，轻微不平衡"
        },
        {
            "方法": "Focal Loss",
            "描述": "关注难分类样本，减轻易分类样本影响",
            "适用场景": "严重不平衡，深度学习模型"
        }
    ]
    
    for strategy in handling_strategies:
        print(f"\n{strategy['方法']}:")
        print(f"  描述: {strategy['描述']}")
        print(f"  适用: {strategy['适用场景']}")
    
    # 实现代码示例
    smote_implementation = """
    from imblearn.over_sampling import SMOTE
    from imblearn.combine import SMOTETomek
    from sklearn.utils.class_weight import compute_class_weight
    import numpy as np
    
    # 方法1: SMOTE过采样
    def apply_smote(X, y):
        smote = SMOTE(random_state=42)
        X_resampled, y_resampled = smote.fit_resample(X, y)
        return X_resampled, y_resampled
    
    # 方法2: Class Weight调整
    def compute_weights(y):
        classes = np.unique(y)
        weights = compute_class_weight('balanced', classes=classes, y=y)
        class_weights = dict(zip(classes, weights))
        return class_weights
    
    # 方法3: 数据划分时的分层抽样
    from sklearn.model_selection import train_test_split
    
    train_df, temp_df = train_test_split(
        df, 
        test_size=0.3, 
        stratify=df['label'],  # 分层抽样保持类别比例
        random_state=42
    )
    val_df, test_df = train_test_split(
        temp_df, 
        test_size=0.5, 
        stratify=temp_df['label'],
        random_state=42
    )
    """
    
    print("\n不平衡数据处理代码:")
    print(smote_implementation)

imbalanced_data_handling()

def baseline_models_comparison():
    """
    基准模型对比分析
    """
    print("基准模型对比分析:")
    
    models_comparison = [
        {
            "模型": "TF-IDF + 朴素贝叶斯",
            "准确率": "0.78",
            "F1-score": "0.75",
            "训练时间": "< 1分钟",
            "推理时间": "< 0.01秒",
            "优点": "速度快，内存占用小",
            "缺点": "特征表达能力有限"
        },
        {
            "模型": "TF-IDF + SVM",
            "准确率": "0.82",
            "F1-score": "0.79",
            "训练时间": "< 5分钟",
            "推理时间": "< 0.02秒",
            "优点": "性能较好，稳定",
            "缺点": "特征工程依赖强"
        },
        {
            "模型": "BERT-base-chinese",
            "准确率": "0.93",
            "F1-score": "0.91",
            "训练时间": "2-4小时",
            "推理时间": "< 0.5秒",
            "优点": "特征表达能力强",
            "缺点": "计算资源需求高"
        },
        {
            "模型": "RoBERTa-wwm-ext",
            "准确率": "0.95",
            "F1-score": "0.94",
            "训练时间": "3-5小时",
            "推理时间": "< 0.6秒",
            "优点": "中文优化，性能优异",
            "缺点": "资源消耗大"
        }
    ]
    
    print(f"{'模型':<20} {'准确率':<8} {'F1-score':<10} {'训练时间':<12} {'推理时间':<12}")
    print("-" * 70)
    for model in models_comparison:
        print(f"{model['模型']:<20} {model['准确率']:<8} {model['F1-score']:<10} {model['训练时间']:<12} {model['推理时间']:<12}")
    
    print("\n模型选择依据:")
    selection_criteria = [
        "业务准确性要求",
        "推理速度要求",
        "计算资源限制",
        "部署环境约束",
        "维护成本考虑"
    ]
    
    for criterion in selection_criteria:
        print(f"  ✓ {criterion}")

baseline_models_comparison()

def model_implementation():
    """
    模型实现代码
    """
    print("模型实现代码:")
    
    model_code = """
    from transformers import (
        AutoTokenizer, 
        AutoModelForSequenceClassification,
        TrainingArguments, 
        Trainer
    )
    from sklearn.metrics import accuracy_score, f1_score, classification_report
    import torch
    
    class TicketClassificationModel:
        def __init__(self, model_name='hfl/chinese-roberta-wwm-ext', num_labels=5):
            self.model_name = model_name
            self.num_labels = num_labels
            self.tokenizer = AutoTokenizer.from_pretrained(model_name)
            self.model = AutoModelForSequenceClassification.from_pretrained(
                model_name, 
                num_labels=num_labels
            )
        
        def tokenize_data(self, texts, labels=None, max_length=256):
            '''
            数据分词处理
            '''
            encoded = self.tokenizer(
                texts,
                truncation=True,
                padding=True,
                max_length=max_length,
                return_tensors='pt'
            )
            
            if labels is not None:
                encoded['labels'] = torch.tensor(labels, dtype=torch.long)
            
            return encoded
        
        def compute_metrics(self, eval_pred):
            '''
            计算评估指标
            '''
            predictions, labels = eval_pred
            predictions = predictions.argmax(axis=-1)
            
            return {
                'accuracy': accuracy_score(labels, predictions),
                'f1': f1_score(labels, predictions, average='weighted'),
                'precision': precision_score(labels, predictions, average='weighted'),
                'recall': recall_score(labels, predictions, average='weighted')
            }
    
    # 模型选择与训练
    model_selector = {
        'tfidf_nb': lambda: ('TF-IDF + Naive Bayes', None),
        'tfidf_svm': lambda: ('TF-IDF + SVM', None),
        'bert': lambda: TicketClassificationModel('bert-base-chinese', 5),
        'roberta': lambda: TicketClassificationModel('hfl/chinese-roberta-wwm-ext', 5)
    }
    """
    
    print(model_code)

model_implementation()

def training_configuration():
    """
    模型训练配置
    """
    print("训练配置与参数:")
    
    training_params = {
        "基础配置": {
            "模型名称": "hfl/chinese-roberta-wwm-ext",
            "标签数量": 5,
            "最大序列长度": 256
        },
        "训练参数": {
            "训练轮数": 5,
            "批次大小": 32,
            "学习率": 2e-5,
            "权重衰减": 0.01,
            "warmup比例": 0.1
        },
        "优化配置": {
            "混合精度训练": True,
            "梯度累积": 2,
            "早停策略": True,
            "最佳模型保存": True
        },
        "评估配置": {
            "评估策略": "epoch",
            "保存策略": "epoch",
            "最佳指标": "f1",
            "日志间隔": 100
        }
    }
    
    for category, params in training_params.items():
        print(f"\n{category}:")
        for param, value in params.items():
            print(f"  {param}: {value}")
    
    # 训练代码示例
    training_code = """
    from transformers import TrainingArguments, Trainer
    import os
    
    # 训练参数配置
    training_args = TrainingArguments(
        output_dir='./ticket_classifier',
        num_train_epochs=5,
        per_device_train_batch_size=32,
        per_device_eval_batch_size=32,
        warmup_ratio=0.1,
        weight_decay=0.01,
        learning_rate=2e-5,
        logging_dir='./logs',
        logging_steps=100,
        evaluation_strategy="epoch",
        save_strategy="epoch",
        load_best_model_at_end=True,
        metric_for_best_model="f1",
        greater_is_better=True,
        fp16=True,  # 混合精度训练
        gradient_accumulation_steps=2,  # 梯度累积
        dataloader_num_workers=4,  # 数据加载器进程数
        seed=42,
        disable_tqdm=False,
        report_to=None  # 不上传到wandb
    )
    
    # 创建训练器
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=train_dataset,
        eval_dataset=val_dataset,
        compute_metrics=compute_metrics,
    )
    
    # 开始训练
    train_result = trainer.train()
    trainer.save_model('./best_ticket_classifier')
    trainer.save_state()
    """
    
    print("\n训练代码:")
    print(training_code)

training_configuration()

def training_monitoring():
    """
    训练过程监控
    """
    print("训练过程监控:")
    
    monitoring_items = [
        "训练损失变化",
        "验证损失变化", 
        "准确率/F1-score变化",
        "学习率变化",
        "内存使用情况",
        "GPU利用率",
        "训练速度监控"
    ]
    
    print("监控项目:")
    for item in monitoring_items:
        print(f"  ✓ {item}")
    
    # 监控代码示例
    monitoring_code = """
    import matplotlib.pyplot as plt
    from transformers import EarlyStoppingCallback
    
    class CustomCallback(TrainerCallback):
        def __init__(self):
            self.train_losses = []
            self.val_losses = []
            self.f1_scores = []
        
        def on_log(self, args, state, control, logs=None, **kwargs):
            if logs is not None:
                if 'train_loss' in logs:
                    self.train_losses.append(logs['train_loss'])
                if 'eval_loss' in logs:
                    self.val_losses.append(logs['eval_loss'])
                if 'eval_f1' in logs:
                    self.f1_scores.append(logs['eval_f1'])
        
        def plot_metrics(self):
            fig, axes = plt.subplots(1, 3, figsize=(15, 4))
            
            axes[0].plot(self.train_losses, label='Train Loss')
            axes[0].plot(self.val_losses, label='Val Loss')
            axes[0].set_title('Loss Curves')
            axes[0].legend()
            
            axes[1].plot(self.f1_scores, label='F1 Score', color='green')
            axes[1].set_title('F1 Score')
            axes[1].legend()
            
            plt.tight_layout()
            plt.savefig('training_metrics.png')
            plt.show()
    
    # 添加回调函数
    trainer.add_callback(CustomCallback())
    """
    
    print("\n监控代码:")
    print(monitoring_code)

training_monitoring()

def evaluation_metrics():
    """
    模型评估指标
    """
    print("模型评估指标:")
    
    metrics_explanation = {
        "准确率 (Accuracy)": "正确分类样本数 / 总样本数",
        "精确率 (Precision)": "真正例 / (真正例 + 假正例)",
        "召回率 (Recall)": "真正例 / (真正例 + 假负例)",
        "F1分数": "2 * (精确率 * 召回率) / (精确率 + 召回率)",
        "宏平均F1": "各类别F1分数的平均值",
        "微平均F1": "总体TP、FP、FN计算的F1分数"
    }
    
    for metric, explanation in metrics_explanation.items():
        print(f"{metric}: {explanation}")
    
    # 评估代码示例
    evaluation_code = """
    from sklearn.metrics import (
        accuracy_score, precision_recall_fscore_support,
        confusion_matrix, classification_report
    )
    import numpy as np
    import matplotlib.pyplot as plt
    import seaborn as sns
    
    def evaluate_model(model, test_dataset, id2label):
        '''
        模型评估函数
        '''
        model.eval()
        all_predictions = []
        all_labels = []
        
        with torch.no_grad():
            for batch in test_dataset:
                inputs = {k: v.to(device) for k, v in batch.items() if k != 'labels'}
                labels = batch['labels'].cpu().numpy()
                
                outputs = model(**inputs)
                predictions = torch.argmax(outputs.logits, dim=-1).cpu().numpy()
                
                all_predictions.extend(predictions)
                all_labels.extend(labels)
        
        # 计算各项指标
        accuracy = accuracy_score(all_labels, all_predictions)
        precision, recall, f1, support = precision_recall_fscore_support(
            all_labels, all_predictions, average='weighted'
        )
        
        macro_f1 = f1_score(all_labels, all_predictions, average='macro')
        micro_f1 = f1_score(all_labels, all_predictions, average='micro')
        
        print(f"准确率: {accuracy:.4f}")
        print(f"精确率: {precision:.4f}")
        print(f"召回率: {recall:.4f}")
        print(f"F1分数: {f1:.4f}")
        print(f"宏平均F1: {macro_f1:.4f}")
        print(f"微平均F1: {micro_f1:.4f}")
        
        # 详细分类报告
        print("\\n详细分类报告:")
        print(classification_report(all_labels, all_predictions, 
                                  target_names=list(id2label.values())))
        
        # 混淆矩阵
        cm = confusion_matrix(all_labels, all_predictions)
        plt.figure(figsize=(8, 6))
        sns.heatmap(cm, annot=True, fmt='d', cmap='Blues',
                   xticklabels=list(id2label.values()),
                   yticklabels=list(id2label.values()))
        plt.title('Confusion Matrix')
        plt.ylabel('True Label')
        plt.xlabel('Predicted Label')
        plt.show()
        
        return {
            'accuracy': accuracy,
            'precision': precision,
            'recall': recall,
            'f1': f1,
            'macro_f1': macro_f1,
            'micro_f1': micro_f1
        }
    """
    
    print("\n评估代码:")
    print(evaluation_code)

evaluation_metrics()

def api_deployment():
    """
    API服务部署实现
    """
    print("API服务部署:")
    
    api_code = """
    from fastapi import FastAPI, HTTPException
    from pydantic import BaseModel
    import torch
    import numpy as np
    from transformers import AutoTokenizer, AutoModelForSequenceClassification
    import uvicorn
    import time
    from typing import Dict, List, Optional
    
    app = FastAPI(
        title="智能客服工单分类API",
        description="基于深度学习的工单自动分类服务",
        version="1.0.0"
    )
    
    # 定义请求模型
    class TicketRequest(BaseModel):
        title: str
        content: str
        priority: Optional[str] = "normal"
        customer_id: Optional[str] = None
    
    # 定义响应模型
    class ClassificationResponse(BaseModel):
        category: str
        confidence: float
        all_scores: Dict[str, float]
        processing_time: float
        timestamp: str
    
    # 加载模型和分词器
    MODEL_PATH = "./best_ticket_classifier"
    tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH)
    model = AutoModelForSequenceClassification.from_pretrained(MODEL_PATH)
    model.eval()
    
    # 分类标签映射
    CATEGORY_MAPPING = {
        0: "咨询类",
        1: "投诉类", 
        2: "售后类",
        3: "技术类",
        4: "其他类"
    }
    
    @app.on_event('startup')
    async def startup_event():
        print("模型加载完成，服务启动...")
    
    @app.post("/classify", response_model=ClassificationResponse)
    async def classify_ticket(ticket: TicketRequest):
        '''
        工单分类接口
        '''
        start_time = time.time()
        
        try:
            # 组合文本
            full_text = f"{ticket.title} {ticket.content}"
            
            # 分词和编码
            inputs = tokenizer(
                full_text,
                return_tensors="pt",
                truncation=True,
                max_length=256,
                padding=True
            )
            
            # 模型推理
            with torch.no_grad():
                outputs = model(**inputs)
                logits = outputs.logits
                probabilities = torch.softmax(logits, dim=-1)
                predicted_id = torch.argmax(probabilities, dim=-1).item()
                confidence = probabilities[0][predicted_id].item()
                
                # 获取所有类别的概率
                all_probs = probabilities[0].cpu().numpy()
                all_scores = {
                    CATEGORY_MAPPING[i]: float(score) 
                    for i, score in enumerate(all_probs)
                }
            
            processing_time = time.time() - start_time
            
            return ClassificationResponse(
                category=CATEGORY_MAPPING[predicted_id],
                confidence=round(confidence, 4),
                all_scores=all_scores,
                processing_time=round(processing_time, 4),
                timestamp=time.strftime('%Y-%m-%d %H:%M:%S')
            )
            
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"分类处理失败: {str(e)}")
    
    @app.get("/health")
    async def health_check():
        '''
        健康检查接口
        '''
        return {"status": "healthy", "timestamp": time.strftime('%Y-%m-%d %H:%M:%S')}
    
    @app.get("/categories")
    async def get_categories():
        '''
        获取分类列表
        '''
        return {"categories": list(CATEGORY_MAPPING.values())}
    
    if __name__ == "__main__":
        uvicorn.run(app, host="0.0.0.0", port=8000)
    """
    
    print("API服务代码:")
    print(api_code)

api_deployment()

def docker_deployment():
    """
    Docker容器化部署
    """
    print("Docker容器化部署:")
    
    dockerfile_content = """
    # Dockerfile
    FROM python:3.9-slim
    
    # 设置工作目录
    WORKDIR /app
    
    # 安装系统依赖
    RUN apt-get update && apt-get install -y \\
        gcc \\
        g++ \\
        && rm -rf /var/lib/apt/lists/*
    
    # 复制依赖文件
    COPY requirements.txt .
    
    # 安装Python依赖
    RUN pip install --no-cache-dir -r requirements.txt
    
    # 复制模型文件（假设模型已下载到models目录）
    COPY models/ ./models/
    
    # 复制应用代码
    COPY api.py .
    
    # 暴露端口
    EXPOSE 8000
    
    # 启动命令
    CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000"]
    """
    
    requirements_content = """
    # requirements.txt
    fastapi==0.104.1
    uvicorn[standard]==0.24.0
    torch==2.1.0
    transformers==4.35.0
    scikit-learn==1.3.0
    pandas==2.1.0
    numpy==1.24.3
    pydantic==2.4.2
    python-multipart==0.0.6
    """
    
    docker_compose_content = """
    # docker-compose.yml
    version: '3.8'
    
    services:
      ticket-classifier:
        build: .
        ports:
          - "8000:8000"
        volumes:
          - ./logs:/app/logs
        environment:
          - PYTHONPATH=/app
        deploy:
          resources:
            limits:
              memory: 4G
              cpus: '2'
        restart: unless-stopped
        healthcheck:
          test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
          interval: 30s
          timeout: 10s
          retries: 3
          start_period: 40s
    """
    
    print("Dockerfile:")
    print(dockerfile_content)
    print("\nrequirements.txt:")
    print(requirements_content)
    print("\ndocker-compose.yml:")
    print(docker_compose_content)

docker_deployment()

def performance_optimization():
    """
    性能优化策略
    """
    print("性能优化策略:")
    
    optimization_strategies = [
        {
            "策略": "模型量化",
            "方法": "INT8量化或混合精度",
            "效果": "推理速度提升2-3倍，内存减少一半"
        },
        {
            "策略": "模型剪枝",
            "方法": "移除冗余参数",
            "效果": "模型大小减少30-50%，性能基本保持"
        },
        {
            "策略": "知识蒸馏",
            "方法": "训练小型学生模型",
            "效果": "推理速度大幅提升，精度略有下降"
        },
        {
            "策略": "批处理优化",
            "方法": "批量推理处理",
            "效果": "吞吐量显著提升"
        }
    ]
    
    for strategy in optimization_strategies:
        print(f"\n{strategy['策略']}:")
        print(f"  方法: {strategy['方法']}")
        print(f"  效果: {strategy['效果']}")
    
    # 量化示例代码
    quantization_code = """
    from transformers import AutoTokenizer, AutoModelForSequenceClassification
    import torch
    
    # INT8量化示例
    tokenizer = AutoTokenizer.from_pretrained("hfl/chinese-roberta-wwm-ext")
    
    quantized_model = AutoModelForSequenceClassification.from_pretrained(
        "hfl/chinese-roberta-wwm-ext",
        num_labels=5,
        torch_dtype=torch.int8,
        load_in_8bit=True  # 8位量化加载
    )
    
    # 或者使用ONNX Runtime优化
    from optimum.onnxruntime import ORTModelForSequenceClassification
    
    ort_model = ORTModelForSequenceClassification.from_pretrained(
        "./best_ticket_classifier",
        export=True  # 自动导出为ONNX格式
    )
    """
    
    print("\n模型量化代码:")
    print(quantization_code)

performance_optimization()

def inference_optimization():
    """
    推理优化技术
    """
    print("推理优化技术:")
    
    inference_tips = [
        "使用torch.jit.trace()进行模型追踪",
        "启用torch.backends.cudnn.benchmark=True",
        "使用批处理提高吞吐量",
        "预分配张量内存",
        "使用torch.inference_mode()替代no_grad()"
    ]
    
    print("推理优化建议:")
    for i, tip in enumerate(inference_tips, 1):
        print(f"  {i}. {tip}")
    
    optimized_inference_code = """
    import torch
    from transformers import pipeline
    
    # 使用HuggingFace pipeline优化推理
    classifier = pipeline(
        "text-classification",
        model="./best_ticket_classifier",
        tokenizer="./best_ticket_classifier",
        device=0 if torch.cuda.is_available() else -1,
        batch_size=16,  # 批处理大小
        max_length=256,
        truncation=True
    )
    
    # 批量推理示例
    def batch_classify(texts):
        results = classifier(texts)
        return results
    
    # JIT优化示例
    @torch.jit.script
    def optimized_predict(input_ids, attention_mask):
        with torch.inference_mode():
            outputs = model(input_ids=input_ids, attention_mask=attention_mask)
            return torch.softmax(outputs.logits, dim=-1)
    """
    
    print("\n优化推理代码:")
    print(optimized_inference_code)

inference_optimization()

def system_monitoring():
    """
    系统监控方案
    """
    print("系统监控方案:")
    
    monitoring_components = [
        {
            "组件": "Prometheus",
            "功能": "指标收集和存储",
            "监控项": "CPU、内存、请求量、延迟、错误率"
        },
        {
            "组件": "Grafana", 
            "功能": "可视化仪表板",
            "功能": "实时监控图表展示"
        },
        {
            "组件": "ELK Stack",
            "功能": "日志收集分析",
            "功能": "错误日志和访问日志分析"
        },
        {
            "组件": "APM工具",
            "功能": "应用性能监控",
            "功能": "事务追踪和性能分析"
        }
    ]
    
    for component in monitoring_components:
        print(f"\n{component['组件']}:")
        print(f"  功能: {component['功能']}")
    
    # 监控中间件示例
    monitoring_middleware = """
    import time
    import logging
    from fastapi import Request
    from starlette.middleware.base import BaseHTTPMiddleware
    
    class MonitoringMiddleware(BaseHTTPMiddleware):
        async def dispatch(self, request: Request, call_next):
            start_time = time.time()
            
            # 记录请求信息
            logging.info(f"Request: {request.method} {request.url.path}")
            
            response = await call_next(request)
            
            # 计算处理时间
            process_time = time.time() - start_time
            
            # 记录响应信息
            logging.info(f"Response: {response.status_code}, ProcessTime: {process_time:.4f}s")
            
            # 这里可以发送指标到Prometheus
            # prometheus_client.Counter('requests_total', 'Total requests').inc()
            # prometheus_client.Histogram('request_duration_seconds', 'Request duration').observe(process_time)
            
            return response
    
    # 在app中添加中间件
    app.add_middleware(MonitoringMiddleware)
    """
    
    print("\n监控中间件代码:")
    print(monitoring_middleware)

system_monitoring()

def model_update_strategy():
    """
    模型更新策略
    """
    print("模型更新策略:")
    
    update_strategies = [
        "A/B测试：同时运行新旧模型，比较性能",
        "蓝绿部署：新旧版本并行，平滑切换",
        "金丝雀发布：逐步增加新模型流量",
        "影子模型：新模型接收实时流量但不响应",
        "在线学习：模型实时学习新数据"
    ]
    
    print("模型更新策略:")
    for i, strategy in enumerate(update_strategies, 1):
        print(f"  {i}. {strategy}")
    
    # A/B测试示例
    ab_testing_code = """
    import random
    from enum import Enum
    
    class ModelVersion(Enum):
        V1 = "v1_original"
        V2 = "v2_updated"
    
    class ModelRouter:
        def __init__(self):
            self.models = {
                ModelVersion.V1: self.load_model_v1(),
                ModelVersion.V2: self.load_model_v2()
            }
            self.traffic_split = {ModelVersion.V1: 0.9, ModelVersion.V2: 0.1}  # 90%流量到V1
        
        def route_request(self, text):
            # 根据流量分配选择模型
            choice = random.random()
            cumulative = 0
            
            for model_version, traffic_ratio in self.traffic_split.items():
                cumulative += traffic_ratio
                if choice <= cumulative:
                    model = self.models[model_version]
                    start_time = time.time()
                    
                    result = self.predict_with_model(model, text)
                    
                    # 记录性能指标
                    latency = time.time() - start_time
                    self.record_metrics(model_version, latency, result)
                    
                    return result, model_version
            
            return self.predict_with_model(self.models[ModelVersion.V1], text), ModelVersion.V1
    """
    
    print("\nA/B测试代码:")
    print(ab_testing_code)

model_update_strategy()

def enterprise_case_study():
    """
    企业部署案例分析
    """
    print("企业部署案例:")
    
    case_studies = [
        {
            "公司": "电商企业",
            "规模": "日处理10万+工单",
            "技术栈": "RoBERTa + FastAPI + Kubernetes",
            "效果": "分类准确率94%，处理效率提升60%"
        },
        {
            "公司": "金融科技公司", 
            "规模": "日处理5万+客服咨询",
            "技术栈": "BERT + Docker + Prometheus",
            "效果": "响应时间缩短至0.8秒，人工介入率降至5%"
        },
        {
            "公司": "电信运营商",
            "规模": "日处理20万+工单",
            "技术栈": "ERNIE + 微服务架构 + ELK",
            "效果": "客服效率提升3倍，客户满意度提升15%"
        }
    ]
    
    for case in case_studies:
        print(f"\n{case['公司']}:")
        print(f"  规模: {case['规模']}")
        print(f"  技术栈: {case['技术栈']}")
        print(f"  效果: {case['效果']}")

enterprise_case_study()

def success_factors():
    """
    项目成功关键因素
    """
    success_factors_list = [
        "充分的数据质量保证",
        "合理的模型选择和调优",
        "完善的监控和运维体系",
        "持续的模型迭代优化",
        "良好的用户体验设计",
        "稳定的系统架构",
        "有效的性能优化措施"
    ]
    
    print("项目成功关键因素:")
    for i, factor in enumerate(success_factors_list, 1):
        print(f"  {i}. {factor}")

success_factors()