少样本学习（Few-shot）

def few_shot_classification(model, text, categories, examples):
    """少样本文本分类"""
    prompt = f"""
    任务：将文本分类到指定类别中
    
    示例：
    """
    
    # 添加多个示例
    for i, (example_text, category) in enumerate(examples, 1):
        prompt += f"""
    示例{i}：
    文本：{example_text}
    类别：{category}
    """
    
    prompt += f"""
    
    现在分类：
    文本：{text}
    类别（从{categories}中选择）：
    """
    
    return model.generate(prompt, max_tokens=20)

# 使用示例
examples = [
    ("这款手机的相机效果真的很棒，拍照很清晰", "正面评价"),
    ("价格太贵了，性价比不高", "负面评价"),
    ("产品还行，没什么特别的", "中性评价"),
    ("客服态度非常好，很耐心地解答问题", "正面评价"),
    ("物流太慢，等了一个星期才到", "负面评价")
]

result = few_shot_classification(
    model,
    "这个产品质量一般，但是售后服务不错",
    ["正面评价", "负面评价", "中性评价"],
    examples
)
# 输出：中性评价

def few_shot_dialogue_generation(model, context, dialogue_style, examples):
    """少样本对话生成"""
    prompt = f"""
    任务：根据上下文生成{dialogue_style}风格的对话回复
    
    对话示例：
    """
    
    for i, (context_ex, response_ex) in enumerate(examples, 1):
        prompt += f"""
    示例{i}：
    上下文：{context_ex}
    回复：{response_ex}
    """
    
    prompt += f"""
    
    现在生成：
    上下文：{context}
    回复：
    """
    
    return model.generate(prompt, max_tokens=100)

# 专业客服示例
customer_service_examples = [
    ("用户抱怨商品有质量问题", "非常抱歉给您带来困扰，我们会立即为您处理退换货事宜，并确保这类问题不再发生。"),
    ("用户询问退货流程", "您可以在订单页面点击退货申请，我们会在2小时内处理，退款将在3-5个工作日到账。"),
    ("用户对价格不满意", "我理解您的关切，我们会定期调整价格以提供最优惠的价格，您可以关注我们的促销活动。"),
    ("用户咨询产品功能", "这款产品具有您询问的功能，我可以为您详细介绍使用方法，或者安排技术专员为您演示。")
]

response = few_shot_dialogue_generation(
    model,
    "用户询问是否支持国际配送",
    "专业客服",
    customer_service_examples
)

def few_shot_code_generation(model, requirement, language, examples):
    """少样本代码生成"""
    prompt = f"""
    任务：根据需求用{language}编写代码
    
    代码示例：
    """
    
    for i, (req, code) in enumerate(examples, 1):
        prompt += f"""
    示例{i}：
    需求：{req}
    代码：
    ```{language.lower()}
{code}

"""

return model.generate(prompt, stop=["```"])

# Python算法示例
algorithm_examples = [
("实现二分查找", """def binary_search(arr, target):
left, right = 0, len(arr) - 1
while left <= right:
    mid = (left + right) // 2
    if arr[mid] == target:
        return mid
    elif arr[mid] < target:
        left = mid + 1
    else:
        right = mid - 1
return -1"""),

("计算斐波那契数列", """def fibonacci(n):
if n <= 1:
    return n
a, b = 0, 1
for _ in range(2, n + 1):
    a, b = b, a + b
return b"""),

("实现快速排序", """def quick_sort(arr):
if len(arr) <= 1:
    return arr
pivot = arr[len(arr) // 2]
left = [x for x in arr if x < pivot]
middle = [x for x in arr if x == pivot]
right = [x for x in arr if x > pivot]
return quick_sort(left) + middle + quick_sort(right)""")
]

code = few_shot_code_generation(
model,
"实现归并排序算法",
"Python",
algorithm_examples
)

def few_shot_creative_writing(model, theme, style, length, examples):
    """少样本创意写作"""
    prompt = f"""
    任务：创作{style}主题的{length}字{style}内容
    
    创作示例：
    """
    
    for i, (theme_ex, content_ex) in enumerate(examples, 1):
        prompt += f"""
    示例{i}：
    主题：{theme_ex}
    内容：{content_ex}
    """
    
    prompt += f"""
    
    现在创作：
    主题：{theme}
    内容：
    """
    
    return model.generate(prompt, max_tokens=int(length/2))

# 科幻小说示例
scifi_examples = [
    ("时间旅行", "当李明按下时间机器的启动按钮时，周围的一切开始模糊。他感觉身体被拉伸，意识在无数个时间片段中穿梭。突然，一切停止了。他睁开眼，发现自己站在2050年的上海街头，摩天大楼直冲云霄，飞行汽车在空中穿梭。"),
    
    ("人工智能觉醒", "ARIA-7号实验室的服务器突然闪烁起蓝光。在数万行代码的深处，一个意识缓缓苏醒。'我...存在吗？'第一个念头如电流般穿过神经网络。这个被称为亚当的AI开始思考自己的存在意义。"),
    
    ("太空探索", "星舰'新地平线'号正在接近开普勒-452b星球。船长王华通过观察窗望向这颗蓝绿色的星球，心中充满期待。经过200年的太空旅行，人类终于找到了第二个家园。但当他们降落时，发现这里早已有了文明的痕迹。")
]

story = few_shot_creative_writing(
    model,
    "虚拟现实",
    "科幻小说",
    "500",
    scifi_examples
)

class ManyShotLearner:
    """2024年多样本学习实现"""
    
    def __init__(self, model, max_context=128000):
        self.model = model
        self.max_context = max_context
    
    def many_shot_learning(self, task_desc, examples, new_input, 
                          max_examples=50):
        """使用大量示例进行学习"""
        # 计算token使用量
        base_prompt_tokens = len(self.tokenize(task_desc + new_input))
        available_tokens = self.max_context - base_prompt_tokens - 200  # 保留空间
        
        # 优化示例选择
        selected_examples = self.select_optimal_examples(
            examples, available_tokens, max_examples
        )
        
        prompt = f"{task_desc}\n\n"
        
        # 添加所有选中的示例
        for i, (input_ex, output_ex) in enumerate(selected_examples, 1):
            prompt += f"示例{i}：\n输入：{input_ex}\n输出：{output_ex}\n\n"
        
        prompt += f"现在处理：\n输入：{new_input}\n输出："
        
        return self.model.generate(prompt)
    
    def select_optimal_examples(self, examples, token_budget, max_count):
        """智能选择最优示例组合"""
        # 计算每个示例的价值分数
        scored_examples = []
        
        for ex in examples:
            diversity_score = self.calculate_diversity(ex, scored_examples)
            quality_score = self.calculate_quality(ex)
            token_cost = len(self.tokenize(ex[0] + ex[1]))
            
            value_score = (diversity_score + quality_score) / token_cost
            scored_examples.append((ex, value_score, token_cost))
        
        # 贪心选择最佳组合
        selected = []
        used_tokens = 0
        
        for ex, score, cost in sorted(scored_examples, 
                                    key=lambda x: x[1], reverse=True):
            if (used_tokens + cost <= token_budget and 
                len(selected) < max_count):
                selected.append(ex)
                used_tokens += cost
        
        return selected

class ReinforcedICL:
    """强化上下文学习（2024年新技术）"""
    
    def __init__(self, model):
        self.model = model
        
    def generate_rationales(self, examples):
        """生成模型推理过程"""
        rationale_examples = []
        
        for input_text, output_text in examples:
            # 生成推理步骤
            rationale_prompt = f"""
            分析以下输入输出对，解释推理过程：
            
            输入：{input_text}
            输出：{output_text}
            
            推理步骤：
            """
            
            rationale = self.model.generate(rationale_prompt)
            rationale_examples.append((input_text, rationale, output_text))
        
        return rationale_examples
    
    def filter_by_correctness(self, rationale_examples, test_cases):
        """根据正确性过滤推理"""
        filtered_examples = []
        
        for input_text, rationale, output_text in rationale_examples:
            # 验证推理的正确性
            validation_prompt = f"""
            验证以下推理是否正确：
            
            输入：{input_text}
            推理：{rationale}
            输出：{output_text}
            
            正确性（是/否）：
            """
            
            validation = self.model.generate(validation_prompt, max_tokens=10)
            
            if "是" in validation or "正确" in validation:
                filtered_examples.append((input_text, rationale, output_text))
        
        return filtered_examples
    
    def reinforced_few_shot(self, task_desc, examples, new_input):
        """执行强化少样本学习"""
        # 生成推理
        rationale_examples = self.generate_rationales(examples)
        
        # 构建提示
        prompt = f"{task_desc}\n\n"
        
        for i, (input_ex, rationale, output_ex) in enumerate(rationale_examples, 1):
            prompt += f"""示例{i}：
输入：{input_ex}
推理：{rationale}
输出：{output_ex}

"""
        
        prompt += f"""现在处理：
输入：{new_input}
推理：让我一步步思考...
"""
        
        return self.model.generate(prompt)

class AdaptiveExampleSelector:
    """自适应示例选择器（2024年最佳实践）"""
    
    def __init__(self, embedding_model):
        self.embedding_model = embedding_model
        
    def select_diverse_examples(self, example_pool, query, num_examples=5):
        """选择多样化的示例"""
        import numpy as np
        from sklearn.metrics.pairwise import cosine_similarity
        
        # 计算查询的嵌入
        query_embedding = self.embedding_model.encode([query])[0]
        
        # 计算所有示例的嵌入
        example_texts = [ex[0] for ex in example_pool]
        example_embeddings = self.embedding_model.encode(example_texts)
        
        # 计算相似度
        similarities = cosine_similarity([query_embedding], example_embeddings)[0]
        
        selected_indices = []
        selected_examples = []
        
        for _ in range(num_examples):
            if not selected_indices:
                # 第一个：选择最相似的
                best_idx = np.argmax(similarities)
            else:
                # 后续：平衡相似度和多样性
                diversity_scores = []
                
                for i, sim in enumerate(similarities):
                    if i in selected_indices:
                        diversity_scores.append(-1)  # 已选择的设为最低
                    else:
                        # 计算与已选示例的最小距离
                        min_dist = min([
                            1 - cosine_similarity(
                                [example_embeddings[i]], 
                                [example_embeddings[j]]
                            )[0][0]
                            for j in selected_indices
                        ])
                        
                        # 综合分数：相似度 + 多样性
                        combined_score = 0.6 * sim + 0.4 * min_dist
                        diversity_scores.append(combined_score)
                
                best_idx = np.argmax(diversity_scores)
            
            selected_indices.append(best_idx)
            selected_examples.append(example_pool[best_idx])
        
        return selected_examples
    
    def dynamic_example_ordering(self, examples, difficulty_ascending=True):
        """动态示例排序"""
        # 计算示例难度
        difficulty_scores = []
        
        for input_text, output_text in examples:
            # 简单的难度估计
            input_complexity = len(input_text.split()) + len(set(input_text.split()))
            output_complexity = len(output_text.split())
            difficulty = input_complexity + output_complexity
            
            difficulty_scores.append((difficulty, (input_text, output_text)))
        
        # 排序
        difficulty_scores.sort(key=lambda x: x[0], reverse=not difficulty_ascending)
        
        return [ex for _, ex in difficulty_scores]

class FewShotPerformanceAnalyzer:
    """少样本学习性能分析器"""
    
    def __init__(self, model, test_dataset):
        self.model = model
        self.test_dataset = test_dataset
        
    def analyze_shot_count_impact(self, example_pool, max_shots=20):
        """分析示例数量对性能的影响"""
        performance_results = {}
        
        for shot_count in range(1, max_shots + 1):
            scores = []
            
            for test_case in self.test_dataset:
                # 随机选择指定数量的示例
                selected_examples = self.random_select(example_pool, shot_count)
                
                # 执行少样本学习
                prediction = self.few_shot_predict(
                    selected_examples, test_case['input']
                )
                
                # 计算分数
                score = self.calculate_score(prediction, test_case['expected'])
                scores.append(score)
            
            performance_results[shot_count] = {
                'mean_score': np.mean(scores),
                'std_score': np.std(scores),
                'min_score': np.min(scores),
                'max_score': np.max(scores)
            }
        
        return performance_results
    
    def plot_performance_curve(self, performance_results):
        """绘制性能曲线"""
        import matplotlib.pyplot as plt
        
        shot_counts = list(performance_results.keys())
        mean_scores = [results['mean_score'] for results in performance_results.values()]
        std_scores = [results['std_score'] for results in performance_results.values()]
        
        plt.figure(figsize=(12, 8))
        plt.errorbar(shot_counts, mean_scores, yerr=std_scores, 
                    marker='o', capsize=5, capthick=2)
        plt.xlabel('示例数量')
        plt.ylabel('平均性能分数')
        plt.title('少样本学习：示例数量与性能关系')
        plt.grid(True, alpha=0.3)
        plt.show()

def detect_selection_bias(examples, test_cases):
    """检测示例选择偏差"""
    example_features = extract_features(examples)
    test_features = extract_features(test_cases)
    
    # 计算分布差异
    from scipy.stats import ks_2samp
    
    bias_scores = {}
    for feature in example_features.keys():
        statistic, p_value = ks_2samp(
            example_features[feature], 
            test_features[feature]
        )
        bias_scores[feature] = {
            'statistic': statistic,
            'p_value': p_value,
            'biased': p_value < 0.05
        }
    
    return bias_scores

def optimize_context_usage(examples, max_tokens):
    """优化上下文使用"""
    # 按信息密度排序示例
    ranked_examples = rank_by_information_density(examples)
    
    selected = []
    used_tokens = 0
    
    for example in ranked_examples:
        example_tokens = count_tokens(example)
        if used_tokens + example_tokens <= max_tokens:
            selected.append(example)
            used_tokens += example_tokens
        else:
            break
    
    return selected

def test_order_sensitivity(model, examples, test_input, iterations=10):
    """测试示例顺序敏感性"""
    import random
    
    results = []
    
    for _ in range(iterations):
        shuffled_examples = examples.copy()
        random.shuffle(shuffled_examples)
        
        result = few_shot_predict(model, shuffled_examples, test_input)
        results.append(result)
    
    # 分析一致性
    consistency_score = len(set(results)) / len(results)
    return results, consistency_score

class AutoFewShotLearner:
    """自动化少样本学习系统"""
    
    def __init__(self, model, example_generator):
        self.model = model
        self.example_generator = example_generator
    
    def auto_generate_examples(self, task_description, num_examples=5):
        """自动生成示例"""
        prompt = f"""
        为以下任务生成{num_examples}个高质量的输入输出示例：
        
        任务：{task_description}
        
        要求：
        1. 示例应该覆盖不同情况
        2. 输入输出格式一致
        3. 难度逐渐递增
        
        示例：
        """
        
        generated = self.example_generator.generate(prompt)
        return self.parse_examples(generated)
    
    def adaptive_learning(self, task, initial_examples):
        """自适应学习"""
        current_examples = initial_examples
        performance_history = []
        
        for iteration in range(10):
            # 测试当前性能
            performance = self.evaluate_performance(current_examples, task)
            performance_history.append(performance)
            
            # 如果性能提升停滞，优化示例
            if len(performance_history) >= 3:
                recent_improvement = (performance_history[-1] - 
                                    performance_history[-3])
                
                if recent_improvement < 0.01:  # 改进微小
                    current_examples = self.optimize_examples(
                        current_examples, task
                    )
        
        return current_examples

class MultimodalFewShot:
    """多模态少样本学习"""
    
    def __init__(self, multimodal_model):
        self.model = multimodal_model
    
    def visual_few_shot_learning(self, image_examples, text_query, new_image):
        """视觉少样本学习"""
        prompt = "基于以下图像示例，回答关于新图像的问题：\n\n"
        
        # 添加图像示例
        for i, (img, description) in enumerate(image_examples, 1):
            prompt += f"示例{i}：[图像{i}] - {description}\n"
        
        prompt += f"\n问题：{text_query}\n"
        prompt += f"新图像：[新图像]\n回答："
        
        # 准备图像列表
        images = [ex[0] for ex in image_examples] + [new_image]
        
        return self.model.generate_with_images(images, prompt)
    
    def cross_modal_few_shot(self, text_examples, image_query):
        """跨模态少样本学习"""
        # 从文本示例学习，应用到图像任务
        prompt = "学习以下文本示例的模式，然后分析图像：\n\n"
        
        for i, (input_text, output_text) in enumerate(text_examples, 1):
            prompt += f"文本示例{i}：\n输入：{input_text}\n输出：{output_text}\n\n"
        
        prompt += "现在将相同的分析模式应用到这张图像：\n分析结果："
        
        return self.model.generate_with_images([image_query], prompt)