158_企业应用：客服自动化优化 - 2025年LLM驱动的智能客服A/B测试KPI指标体系设计与实践

用户输入 → 多模态处理层 → 意图识别模块 → LLM处理引擎 → 知识检索增强 → 回复生成模块 → 用户反馈收集

确定测试目标 → 设计测试方案 → 样本量计算 → 流量分配 → 数据收集 → 统计分析 → 结果解读 → 决策实施

def calculate_sample_size(baseline, expected_change, significance=0.05, power=0.8):
    # 简化版样本量计算公式
    z_score = 1.96  # 95%置信水平
    power_score = 0.84  # 80%统计功效
    
    p1 = baseline
    p2 = baseline * (1 + expected_change)
    
    # 计算合并比例
    p = (p1 + p2) / 2
    
    # 计算样本量
    sample_size = (z_score * math.sqrt(2 * p * (1 - p)) + power_score * math.sqrt(p1 * (1 - p1) + p2 * (1 - p2))) ** 2 / (p2 - p1) ** 2
    
    return math.ceil(sample_size)

def calculate_ahp_weights(comparison_matrix):
    # 计算特征向量（权重）
    import numpy as np
    
    # 正规化比较矩阵
    normalized_matrix = comparison_matrix / comparison_matrix.sum(axis=0)
    
    # 计算权重向量（行平均）
    weights = normalized_matrix.mean(axis=1)
    
    # 一致性检验
    eigenvalues, _ = np.linalg.eig(comparison_matrix)
    max_eigenvalue = np.max(eigenvalues)
    n = len(comparison_matrix)
    
    # 计算一致性指标CI
    ci = (max_eigenvalue - n) / (n - 1)
    
    # 随机一致性指标RI (简化版)
    ri_values = {
        1: 0.0, 2: 0.0, 3: 0.58, 4: 0.90, 5: 1.12,
        6: 1.24, 7: 1.32, 8: 1.41, 9: 1.45, 10: 1.49
    }
    ri = ri_values.get(n, 1.5)
    
    # 计算一致性比率CR
    cr = ci / ri
    
    return weights, cr

客服人力成本节约率 = (实施前人力成本 - 实施后人力成本) / 实施前人力成本 × 100%

客户获取成本降低率 = (实施前CAC - 实施后CAC) / 实施前CAC × 100%

平均处理成本 = 客服系统总运营成本 / 处理请求总数

ROI = (系统收益 - 系统成本) / 系统成本 × 100%

转化率提升率 = (AI辅助转化率 - 传统转化率) / 传统转化率 × 100%

客户终身价值增长 = (实施后LTV - 实施前LTV) / 实施前LTV × 100%

自助服务转化价值 = 自助服务完成的交易数量 × 平均交易价值

问题升级转化价值 = 升级人工处理的交易数量 × 平均交易价值

问题一次性解决率 = 一次交互解决的问题数量 / 总问题数量 × 100%

平均处理时间 = 总处理时间 / 处理请求总数

高峰期处理能力 = 高峰期每小时最大处理请求数

知识利用率 = 系统引用的知识条目数 / 知识库总条目数 × 100%

客户满意度评分 = 满意和非常满意的用户数 / 参与评分的用户总数 × 100%

净推荐值 = (推荐者百分比 - 贬损者百分比) × 100

情感分析满意度 = 正面情感表达数 / 总反馈数 × 100%

多渠道一致性满意度 = 各渠道满意度标准差的倒数

平均会话时长 = 总会话时间 / 会话总数

平均交互轮次 = 总交互轮数 / 会话总数

主动参与率 = 用户主动发起的会话数 / 总会话数 × 100%

重复使用率 = 30天内使用多次的用户数 / 总用户数 × 100%

第一响应时间 = 从用户发送请求到系统首次回复的平均时间

内容相关性评分 = 用户认为回复内容相关的会话数 / 总会话数 × 100%

个性化程度评分 = 用户认为回复个性化的会话数 / 总会话数 × 100%

情感共鸣指数 = 系统正确识别并适当回应用户情绪的案例数 / 含情感表达的案例总数 × 100%

客户流失率变化 = (实施前流失率 - 实施后流失率) / 实施前流失率 × 100%

自助服务放弃率 = 开始但未完成自助服务的会话数 / 开始自助服务的会话总数 × 100%

升级人工率 = 转人工处理的会话数 / 总会话数 × 100%

长期留存率 = 3个月后仍使用系统的用户数 / 初始用户数 × 100%

系统可用性 = (计划运行时间 - 故障时间) / 计划运行时间 × 100%

并发处理能力 = 系统能够同时处理的最大会话数

意图识别准确率 = 正确识别用户意图的案例数 / 测试案例总数 × 100%

回复相关性分数 = 专家评分的加权平均值

知识检索准确率 = 检索到相关知识的查询数 / 总查询数 × 100%

模型幻觉率 = 生成包含错误或虚构信息的回复数 / 总回复数 × 100%

敏感信息泄露率 = 包含未授权敏感信息的回复数 / 总回复数 × 100%

合规性评分 = 各合规检查项得分的加权平均值

平均故障间隔时间 = 总运行时间 / 故障次数

平均修复时间 = 总故障修复时间 / 故障次数

备份恢复成功率 = 成功从备份恢复的测试次数 / 总测试次数 × 100%

综合评分 = Σ(指标权重 × 指标得分)

def create_kpi_dashboard(metrics_data):
    import matplotlib.pyplot as plt
    import seaborn as sns
    import pandas as pd
    
    # 设置中文字体
    plt.rcParams['font.sans-serif'] = ['SimHei']
    plt.rcParams['axes.unicode_minus'] = False
    
    # 创建仪表盘布局
    fig = plt.figure(figsize=(20, 15))
    gs = fig.add_gridspec(3, 2, height_ratios=[1, 1, 1])
    
    # 1. 核心KPI概览
    ax1 = fig.add_subplot(gs[0, 0])
    kpi_df = pd.DataFrame({
        '指标': ['CSAT', 'FCR', '平均处理时间', '人力成本节约率'],
        '当前值': [metrics_data['csat'], metrics_data['fcr'], 
                  metrics_data['avg_handle_time'], metrics_data['cost_saving_rate']],
        '目标值': [0.9, 0.85, 30, 0.3]
    })
    
    # 创建KPI雷达图
    ax2 = fig.add_subplot(gs[0, 1], polar=True)
    categories = list(kpi_df['指标'])
    N = len(categories)
    
    # 计算角度
    angles = [n / float(N) * 2 * 3.14159 for n in range(N)]
    angles += angles[:1]  # 闭合雷达图
    
    # 准备数据
    values = list(kpi_df['当前值']) + list(kpi_df['当前值'])[:1]
    targets = list(kpi_df['目标值']) + list(kpi_df['目标值'])[:1]
    
    # 绘制雷达图
    ax2.plot(angles, values, 'o-', linewidth=2, label='当前值')
    ax2.plot(angles, targets, 'o-', linewidth=2, label='目标值')
    ax2.fill(angles, values, alpha=0.1)
    ax2.set_thetagrids([a * 180 / 3.14159 for a in angles[:-1]], categories)
    ax2.set_ylim(0, 1.0)  # 根据需要调整
    ax2.legend(loc='upper right')
    ax2.set_title('核心KPI雷达图')
    
    # 2. A/B测试对比
    ax3 = fig.add_subplot(gs[1, 0])
    ab_df = pd.DataFrame({
        '指标': ['CSAT', 'FCR', '平均处理时间', '转化率'],
        'A版本': [0.85, 0.78, 35, 0.03],
        'B版本': [0.92, 0.87, 28, 0.05]
    })
    
    # 创建分组柱状图
    x = range(len(ab_df))
    width = 0.35
    ax3.bar([i - width/2 for i in x], ab_df['A版本'], width, label='A版本')
    ax3.bar([i + width/2 for i in x], ab_df['B版本'], width, label='B版本')
    ax3.set_xticks(x)
    ax3.set_xticklabels(ab_df['指标'])
    ax3.legend()
    ax3.set_title('A/B测试效果对比')
    
    # 3. 趋势分析
    ax4 = fig.add_subplot(gs[1, 1])
    dates = pd.date_range(start='2025-01-01', periods=6, freq='M')
    trend_df = pd.DataFrame({
        '日期': dates,
        'CSAT': [0.82, 0.84, 0.87, 0.89, 0.90, 0.92],
        'FCR': [0.75, 0.77, 0.80, 0.82, 0.84, 0.87]
    })
    
    ax4.plot(trend_df['日期'], trend_df['CSAT'], marker='o', label='CSAT')
    ax4.plot(trend_df['日期'], trend_df['FCR'], marker='s', label='FCR')
    ax4.set_title('关键指标趋势分析')
    ax4.legend()
    ax4.grid(True, linestyle='--', alpha=0.7)
    
    # 4. 细分分析
    ax5 = fig.add_subplot(gs[2, :])
    segment_df = pd.DataFrame({
        '问题类型': ['账户问题', '产品咨询', '订单查询', '退款申请', '技术支持'],
        '解决率': [0.92, 0.88, 0.95, 0.78, 0.82],
        '平均处理时间': [25, 42, 28, 55, 60],
        '客户满意度': [0.94, 0.89, 0.96, 0.82, 0.85]
    })
    
    # 创建热力图数据
    heat_data = segment_df.pivot_table(values=['解决率', '客户满意度'], 
                                      index='问题类型', aggfunc='mean')
    sns.heatmap(heat_data, annot=True, cmap='YlGnBu', ax=ax5)
    ax5.set_title('问题类型细分分析')
    
    plt.tight_layout()
    return fig

客户端 → API网关 → 流量分配服务 → 实验服务管理器 → 客服引擎(多版本) → 数据收集服务 → 分析仪表板

class TrafficAllocationService:
    def __init__(self):
        self.experiments = {}
        self.user_assignments = {}  # 缓存用户分配结果
    
    def create_experiment(self, experiment_id, variants, weights=None, start_time=None, end_time=None):
        """创建新的实验
        
        Args:
            experiment_id: 实验ID
            variants: 变体列表，如['variant_a', 'variant_b']
            weights: 权重列表，如[0.5, 0.5]，默认均匀分配
            start_time: 开始时间
            end_time: 结束时间
        """
        if weights is None:
            weights = [1/len(variants)] * len(variants)
        
        self.experiments[experiment_id] = {
            'variants': variants,
            'weights': weights,
            'start_time': start_time,
            'end_time': end_time,
            'created_at': datetime.now()
        }
        
        return experiment_id
    
    def allocate_user(self, experiment_id, user_id):
        """为用户分配实验变体
        
        Args:
            experiment_id: 实验ID
            user_id: 用户ID
            
        Returns:
            分配的变体名称
        """
        # 检查用户是否已有分配结果（缓存）
        cache_key = f"{experiment_id}:{user_id}"
        if cache_key in self.user_assignments:
            return self.user_assignments[cache_key]
        
        # 获取实验配置
        experiment = self.experiments.get(experiment_id)
        if not experiment:
            raise ValueError(f"Experiment {experiment_id} not found")
        
        # 检查实验时间有效性
        now = datetime.now()
        if experiment['start_time'] and now < experiment['start_time']:
            raise ValueError("Experiment not started yet")
        
        if experiment['end_time'] and now > experiment['end_time']:
            raise ValueError("Experiment already ended")
        
        # 基于用户ID的确定性分配（确保一致性）
        import hashlib
        hash_value = int(hashlib.md5(f"{user_id}:{experiment_id}".encode()).hexdigest(), 16) / (2**128)
        
        cumulative_weight = 0
        for variant, weight in zip(experiment['variants'], experiment['weights']):
            cumulative_weight += weight
            if hash_value <= cumulative_weight:
                # 缓存分配结果
                self.user_assignments[cache_key] = variant
                return variant
        
        # 兜底返回第一个变体
        return experiment['variants'][0]
    
    def get_experiment_stats(self, experiment_id):
        """获取实验统计信息
        
        Args:
            experiment_id: 实验ID
            
        Returns:
            各变体的用户分配数量
        """
        stats = {variant: 0 for variant in self.experiments[experiment_id]['variants']}
        
        for key, variant in self.user_assignments.items():
            if key.startswith(f"{experiment_id}:"):
                stats[variant] += 1
        
        return stats

class CustomerServiceEngineManager:
    def __init__(self):
        self.engines = {}
    
    def register_engine(self, engine_id, engine_instance):
        """注册新的客服引擎实例
        
        Args:
            engine_id: 引擎ID
            engine_instance: 引擎实例
        """
        self.engines[engine_id] = engine_instance
    
    def get_response(self, engine_id, user_input, context=None):
        """获取指定引擎的回复
        
        Args:
            engine_id: 引擎ID
            user_input: 用户输入
            context: 上下文信息
            
        Returns:
            引擎生成的回复
        """
        if engine_id not in self.engines:
            raise ValueError(f"Engine {engine_id} not found")
        
        engine = self.engines[engine_id]
        return engine.generate_response(user_input, context)
    
    def create_variant_engine(self, base_engine_id, variant_params):
        """创建引擎变体
        
        Args:
            base_engine_id: 基础引擎ID
            variant_params: 变体参数
            
        Returns:
            新的引擎ID
        """
        import copy
        
        if base_engine_id not in self.engines:
            raise ValueError(f"Base engine {base_engine_id} not found")
        
        base_engine = self.engines[base_engine_id]
        
        # 创建引擎副本并应用变体参数
        variant_engine = copy.deepcopy(base_engine)
        for param, value in variant_params.items():
            if hasattr(variant_engine, param):
                setattr(variant_engine, param, value)
        
        # 生成新的引擎ID
        variant_engine_id = f"{base_engine_id}_variant_{len(self.engines)}"
        self.register_engine(variant_engine_id, variant_engine)
        
        return variant_engine_id

class DataCollectionService:
    def __init__(self, database_url):
        self.database_url = database_url
        self.setup_database()
    
    def setup_database(self):
        """初始化数据库连接"""
        import sqlite3
        self.db_conn = sqlite3.connect(self.database_url)
        self.create_tables()
    
    def create_tables(self):
        """创建数据表"""
        cursor = self.db_conn.cursor()
        
        # 创建会话表
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS conversations (
            conversation_id TEXT PRIMARY KEY,
            user_id TEXT,
            experiment_id TEXT,
            variant_id TEXT,
            start_time TIMESTAMP,
            end_time TIMESTAMP,
            total_duration INTEGER,
            interaction_count INTEGER,
            csat_score INTEGER,
            resolved BOOLEAN,
            escalated_to_human BOOLEAN
        )
        ''')
        
        # 创建交互表
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS interactions (
            interaction_id TEXT PRIMARY KEY,
            conversation_id TEXT,
            user_input TEXT,
            system_response TEXT,
            timestamp TIMESTAMP,
            response_time INTEGER,
            FOREIGN KEY (conversation_id) REFERENCES conversations (conversation_id)
        )
        ''')
        
        # 创建指标表
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS metrics (
            metric_id TEXT PRIMARY KEY,
            conversation_id TEXT,
            metric_name TEXT,
            metric_value REAL,
            timestamp TIMESTAMP,
            FOREIGN KEY (conversation_id) REFERENCES conversations (conversation_id)
        )
        ''')
        
        self.db_conn.commit()
    
    def record_conversation_start(self, conversation_id, user_id, experiment_id, variant_id):
        """记录会话开始"""
        cursor = self.db_conn.cursor()
        cursor.execute(
            "INSERT INTO conversations (conversation_id, user_id, experiment_id, variant_id, start_time) VALUES (?, ?, ?, ?, ?)",
            (conversation_id, user_id, experiment_id, variant_id, datetime.now())
        )
        self.db_conn.commit()
    
    def record_interaction(self, interaction_id, conversation_id, user_input, system_response, response_time):
        """记录交互"""
        cursor = self.db_conn.cursor()
        cursor.execute(
            "INSERT INTO interactions (interaction_id, conversation_id, user_input, system_response, timestamp, response_time) VALUES (?, ?, ?, ?, ?, ?)",
            (interaction_id, conversation_id, user_input, system_response, datetime.now(), response_time)
        )
        self.db_conn.commit()
    
    def record_conversation_end(self, conversation_id, total_duration, interaction_count, csat_score=None, resolved=None, escalated_to_human=None):
        """记录会话结束"""
        cursor = self.db_conn.cursor()
        
        # 更新会话信息
        update_fields = []
        update_values = []
        
        update_fields.append("end_time = ?")
        update_values.append(datetime.now())
        
        if total_duration is not None:
            update_fields.append("total_duration = ?")
            update_values.append(total_duration)
        
        if interaction_count is not None:
            update_fields.append("interaction_count = ?")
            update_values.append(interaction_count)
        
        if csat_score is not None:
            update_fields.append("csat_score = ?")
            update_values.append(csat_score)
        
        if resolved is not None:
            update_fields.append("resolved = ?")
            update_values.append(resolved)
        
        if escalated_to_human is not None:
            update_fields.append("escalated_to_human = ?")
            update_values.append(escalated_to_human)
        
        update_values.append(conversation_id)
        
        query = f"UPDATE conversations SET {', '.join(update_fields)} WHERE conversation_id = ?"
        cursor.execute(query, update_values)
        self.db_conn.commit()
    
    def record_metric(self, metric_id, conversation_id, metric_name, metric_value):
        """记录指标"""
        cursor = self.db_conn.cursor()
        cursor.execute(
            "INSERT INTO metrics (metric_id, conversation_id, metric_name, metric_value, timestamp) VALUES (?, ?, ?, ?, ?)",
            (metric_id, conversation_id, metric_name, metric_value, datetime.now())
        )
        self.db_conn.commit()
    
    def get_experiment_data(self, experiment_id, start_time=None, end_time=None):
        """获取实验数据"""
        cursor = self.db_conn.cursor()
        
        query = "SELECT * FROM conversations WHERE experiment_id = ?"
        params = [experiment_id]
        
        if start_time:
            query += " AND start_time >= ?"
            params.append(start_time)
        
        if end_time:
            query += " AND start_time <= ?"
            params.append(end_time)
        
        cursor.execute(query, params)
        return cursor.fetchall()

class AnalyticsService:
    def __init__(self, data_collection_service):
        self.data_service = data_collection_service
    
    def calculate_kpi_metrics(self, experiment_id, variant_id=None, start_time=None, end_time=None):
        """计算KPI指标
        
        Args:
            experiment_id: 实验ID
            variant_id: 变体ID（可选）
            start_time: 开始时间
            end_time: 结束时间
            
        Returns:
            计算的KPI指标字典
        """
        # 获取会话数据
        conversations = self.data_service.get_experiment_data(experiment_id, start_time, end_time)
        
        # 如果指定了变体ID，过滤数据
        if variant_id:
            conversations = [c for c in conversations if c[3] == variant_id]  # 假设variant_id在第4列
        
        if not conversations:
            return {}
        
        # 计算基础指标
        total_conversations = len(conversations)
        resolved_conversations = sum(1 for c in conversations if c[10])  # 假设resolved在第11列
        escalated_conversations = sum(1 for c in conversations if c[11])  # 假设escalated_to_human在第12列
        
        # 计算FCR
        fcr = resolved_conversations / total_conversations if total_conversations > 0 else 0
        
        # 计算升级人工率
        escalation_rate = escalated_conversations / total_conversations if total_conversations > 0 else 0
        
        # 计算平均处理时间
        total_durations = [c[7] for c in conversations if c[7] is not None]  # 假设total_duration在第8列
        avg_handle_time = sum(total_durations) / len(total_durations) if total_durations else 0
        
        # 计算平均交互轮次
        interaction_counts = [c[8] for c in conversations if c[8] is not None]  # 假设interaction_count在第9列
        avg_interactions = sum(interaction_counts) / len(interaction_counts) if interaction_counts else 0
        
        # 计算CSAT（如果有数据）
        csat_scores = [c[9] for c in conversations if c[9] is not None]  # 假设csat_score在第10列
        avg_csat = sum(csat_scores) / len(csat_scores) if csat_scores else None
        
        return {
            'total_conversations': total_conversations,
            'fcr': fcr,
            'escalation_rate': escalation_rate,
            'avg_handle_time': avg_handle_time,
            'avg_interactions': avg_interactions,
            'avg_csat': avg_csat
        }
    
    def compare_variants(self, experiment_id, start_time=None, end_time=None):
        """比较不同变体的表现
        
        Args:
            experiment_id: 实验ID
            start_time: 开始时间
            end_time: 结束时间
            
        Returns:
            各变体的指标对比字典
        """
        # 获取所有会话
        conversations = self.data_service.get_experiment_data(experiment_id, start_time, end_time)
        
        # 获取所有唯一的变体ID
        variant_ids = set(c[3] for c in conversations)  # 假设variant_id在第4列
        
        # 计算每个变体的指标
        comparison = {}
        for variant_id in variant_ids:
            comparison[variant_id] = self.calculate_kpi_metrics(
                experiment_id, variant_id, start_time, end_time
            )
        
        return comparison
    
    def perform_statistical_test(self, metric_name, variant_a_data, variant_b_data, test_type='ttest'):
        """执行统计显著性测试
        
        Args:
            metric_name: 指标名称
            variant_a_data: 变体A的数据列表
            variant_b_data: 变体B的数据列表
            test_type: 测试类型，支持'ttest'和'chisquare'
            
        Returns:
            测试结果字典
        """
        import scipy.stats as stats
        
        result = {
            'metric_name': metric_name,
            'variant_a_mean': sum(variant_a_data) / len(variant_a_data) if variant_a_data else 0,
            'variant_b_mean': sum(variant_b_data) / len(variant_b_data) if variant_b_data else 0,
            'difference': 0,
            'percent_change': 0,
            'p_value': None,
            'significant': False
        }
        
        # 计算差异和百分比变化
        if result['variant_a_mean'] != 0:
            result['difference'] = result['variant_b_mean'] - result['variant_a_mean']
            result['percent_change'] = (result['difference'] / result['variant_a_mean']) * 100
        
        # 执行统计测试
        if test_type == 'ttest' and len(variant_a_data) > 1 and len(variant_b_data) > 1:
            # 执行独立样本t检验
            _, p_value = stats.ttest_ind(variant_a_data, variant_b_data)
            result['p_value'] = p_value
            result['significant'] = p_value < 0.05
        
        elif test_type == 'chisquare' and len(variant_a_data) > 0 and len(variant_b_data) > 0:
            # 构建列联表
            # 假设数据是二元的（成功/失败）
            a_success = sum(variant_a_data)
            a_failure = len(variant_a_data) - a_success
            b_success = sum(variant_b_data)
            b_failure = len(variant_b_data) - b_success
            
            contingency_table = [[a_success, a_failure], [b_success, b_failure]]
            
            # 执行卡方检验
            _, p_value, _, _ = stats.chi2_contingency(contingency_table)
            result['p_value'] = p_value
            result['significant'] = p_value < 0.05
        
        return result

def setup_mvp_system():
    # 1. 创建流量分配服务
    traffic_service = TrafficAllocationService()
    
    # 2. 创建客服引擎管理器
    engine_manager = CustomerServiceEngineManager()
    
    # 3. 创建并注册基础客服引擎
    class BasicCustomerServiceEngine:
        def __init__(self, model_name="gpt-4"):
            self.model_name = model_name
            self.prompt_template = "你是一个智能客服助手，请以友好、专业的方式回答用户问题。用户问题: {user_input}"
        
        def generate_response(self, user_input, context=None):
            import time
            start_time = time.time()
            
            # 构建提示
            prompt = self.prompt_template.format(user_input=user_input)
            
            # 这里应该调用实际的LLM API
            # 为了示例，我们返回模拟的响应
            response = f"[{self.model_name}] 我是智能客服助手，我将帮助您解决问题。您的问题是: {user_input}"
            
            response_time = int((time.time() - start_time) * 1000)  # 毫秒
            
            return {
                'response': response,
                'response_time': response_time,
                'model_used': self.model_name
            }
    
    # 创建并注册基础引擎
    base_engine = BasicCustomerServiceEngine()
    engine_manager.register_engine("base_engine", base_engine)
    
    # 创建变体引擎
    # 变体1: 使用GPT-4.5
    variant_1_params = {"model_name": "gpt-4.5"}
    variant_1_id = engine_manager.create_variant_engine("base_engine", variant_1_params)
    
    # 变体2: 使用优化的提示模板
    variant_2_params = {
        "prompt_template": "你是一个专业的智能客服助手。请仔细分析用户问题，提供准确、有用的信息。\n用户问题: {user_input}\n\n请确保你的回答:\n1. 直接解决用户的核心问题\n2. 提供相关的额外信息\n3. 使用友好、专业的语言\n4. 如有必要，提供后续步骤建议"
    }
    variant_2_id = engine_manager.create_variant_engine("base_engine", variant_2_params)
    
    # 4. 创建数据收集服务
    data_service = DataCollectionService("customer_service_ab_test.db")
    
    # 5. 创建分析服务
    analytics_service = AnalyticsService(data_service)
    
    # 6. 创建实验
    experiment_id = traffic_service.create_experiment(
        "llm_model_optimization",
        ["control", "variant_1", "variant_2"],
        [0.3, 0.35, 0.35],
        start_time=datetime.now()
    )
    
    print(f"MVP系统设置完成。\n实验ID: {experiment_id}\n变体ID: {variant_1_id}, {variant_2_id}")
    
    return {
        'traffic_service': traffic_service,
        'engine_manager': engine_manager,
        'data_service': data_service,
        'analytics_service': analytics_service,
        'experiment_id': experiment_id,
        'variant_ids': [variant_1_id, variant_2_id]
    }

# 运行系统示例
def run_example_conversation(system, user_id, user_input):
    # 获取流量分配服务
    traffic_service = system['traffic_service']
    engine_manager = system['engine_manager']
    data_service = system['data_service']
    experiment_id = system['experiment_id']
    
    # 为用户分配变体
    variant = traffic_service.allocate_user(experiment_id, user_id)
    
    # 根据变体选择引擎
    engine_id = "base_engine"  # 默认使用基础引擎
    if variant == "variant_1":
        engine_id = system['variant_ids'][0]
    elif variant == "variant_2":
        engine_id = system['variant_ids'][1]
    
    # 生成会话ID
    import uuid
    conversation_id = str(uuid.uuid4())
    
    # 记录会话开始
    data_service.record_conversation_start(conversation_id, user_id, experiment_id, variant)
    
    # 获取引擎响应
    response_data = engine_manager.get_response(engine_id, user_input)
    
    # 生成交互ID
    interaction_id = str(uuid.uuid4())
    
    # 记录交互
    data_service.record_interaction(
        interaction_id,
        conversation_id,
        user_input,
        response_data['response'],
        response_data['response_time']
    )
    
    # 模拟用户反馈
    csat_score = 5  # 假设用户给出5分（满分）
    resolved = True  # 假设问题已解决
    
    # 记录会话结束
    data_service.record_conversation_end(
        conversation_id,
        total_duration=response_data['response_time'],
        interaction_count=1,
        csat_score=csat_score,
        resolved=resolved,
        escalated_to_human=False
    )
    
    # 记录额外指标
    data_service.record_metric(str(uuid.uuid4()), conversation_id, "response_quality", 4.8)
    
    print(f"用户 {user_id} 被分配到变体: {variant}")
    print(f"系统响应: {response_data['response']}")
    print(f"响应时间: {response_data['response_time']}ms")
    
    return conversation_id

# 运行示例
def main():
    # 设置MVP系统
    system = setup_mvp_system()
    
    # 模拟多个用户会话
    user_ids = [f"user_{i}" for i in range(10)]
    user_inputs = [
        "我的账户无法登录，该怎么办？",
        "我想了解你们的退款政策",
        "如何更改我的配送地址？",
        "我的订单什么时候能发货？",
        "产品出现故障，如何申请保修？"
    ]
    
    for user_id in user_ids:
        # 随机选择一个用户输入
        import random
        user_input = random.choice(user_inputs)
        run_example_conversation(system, user_id, user_input)
    
    # 分析实验结果
    analytics_service = system['analytics_service']
    experiment_id = system['experiment_id']
    
    # 获取各变体的指标对比
    comparison = analytics_service.compare_variants(experiment_id)
    
    print("\n实验结果分析:")
    for variant, metrics in comparison.items():
        print(f"\n变体 {variant}:")
        print(f"  会话总数: {metrics['total_conversations']}")
        print(f"  FCR: {metrics['fcr']:.2%}")
        print(f"  升级人工率: {metrics['escalation_rate']:.2%}")
        print(f"  平均处理时间: {metrics['avg_handle_time']:.2f}ms")
        print(f"  平均交互轮次: {metrics['avg_interactions']:.2f}")
        print(f"  平均CSAT: {metrics['avg_csat']:.2f}")

if __name__ == "__main__":
    main()

python -m venv venv
source venv/bin/activate  # Linux/Mac
# 或
venv\Scripts\activate  # Windows

pip install scipy matplotlib pandas seaborn

python mvp_system.py