Few-shot Prompting 2026：示例选择与排列优化

import random

def random_selection(examples: list[dict], k: int = 4) -> list[dict]:
    """随机选择k个示例"""
    return random.sample(examples, k)

import numpy as np
from sklearn.neighbors import NearestNeighbors

class KNNExampleSelector:
    """
    基于KNN的示例选择
    选择与当前查询语义最相似的示例
    """
    
    def __init__(self, examples: list[dict], embed_model):
        self.examples = examples
        self.embed_model = embed_model
        
        # 预计算示例的embedding
        self.example_embeddings = np.array([
            embed_model.encode(ex["input"]) for ex in examples
        ])
        
        # 构建KNN索引
        self.knn = NearestNeighbors(
            n_neighbors=min(20, len(examples)),
            metric='cosine'
        )
        self.knn.fit(self.example_embeddings)
    
    def select(self, query: str, k: int = 4) -> list[dict]:
        """选择与query最相似的k个示例"""
        query_embed = self.embed_model.encode(query).reshape(1, -1)
        
        # KNN搜索
        distances, indices = self.knn.kneighbors(query_embed)
        
        # 取top-k
        selected = [self.examples[i] for i in indices[0][:k]]
        
        return selected

class DiversityAwareSelector:
    """
    多样性感知的示例选择
    平衡相似性和多样性
    """
    
    def __init__(self, examples, embed_model):
        self.examples = examples
        self.embed_model = embed_model
        self.embeddings = np.array([
            embed_model.encode(ex["input"]) for ex in examples
        ])
    
    def select(self, query: str, k: int = 4, 
               alpha: float = 0.5) -> list[dict]:
        """
        alpha: 相似性权重 (0-1)
        1-alpha: 多样性权重
        """
        query_embed = self.embed_model.encode(query)
        
        # 计算与查询的相似度
        similarities = cosine_similarity(
            query_embed.reshape(1, -1), self.embeddings
        )[0]
        
        selected = []
        selected_indices = []
        
        for _ in range(k):
            scores = []
            
            for i in range(len(self.examples)):
                if i in selected_indices:
                    scores.append(-float('inf'))
                    continue
                
                # 相似性分数
                sim_score = similarities[i]
                
                # 多样性分数（与已选示例的最大距离）
                if selected_indices:
                    max_sim_to_selected = max(
                        cosine_similarity(
                            self.embeddings[i].reshape(1, -1),
                            self.embeddings[j].reshape(1, -1)
                        )[0][0]
                        for j in selected_indices
                    )
                    div_score = 1 - max_sim_to_selected
                else:
                    div_score = 1.0
                
                # 综合分数
                combined = alpha * sim_score + (1 - alpha) * div_score
                scores.append(combined)
            
            best_idx = np.argmax(scores)
            selected.append(self.examples[best_idx])
            selected_indices.append(best_idx)
        
        return selected

class RLExampleSelector:
    """
    基于强化学习的示例选择
    通过历史反馈学习最优选择策略
    """
    
    def __init__(self, examples, embed_model):
        self.examples = examples
        self.embed_model = embed_model
        self.q_table = {}  # state -> action values
        self.learning_rate = 0.1
        self.epsilon = 0.1
    
    def select(self, query: str, k: int = 4) -> list[dict]:
        query_embed = self.embed_model.encode(query)
        state = self._discretize_state(query_embed)
        
        if random.random() < self.epsilon:
            # 探索：随机选择
            return random.sample(self.examples, k)
        else:
            # 利用：选择Q值最高的示例
            selected = []
            remaining = list(range(len(self.examples)))
            
            for _ in range(k):
                # 选择Q值最高的
                q_values = [
                    self.q_table.get((state, i), 0.0)
                    for i in remaining
                ]
                best = remaining[np.argmax(q_values)]
                selected.append(self.examples[best])
                remaining.remove(best)
            
            return selected
    
    def update(self, query: str, selected_indices: list[int], 
               reward: float):
        """根据反馈更新Q值"""
        query_embed = self.embed_model.encode(query)
        state = self._discretize_state(query_embed)
        
        for idx in selected_indices:
            key = (state, idx)
            old_q = self.q_table.get(key, 0.0)
            self.q_table[key] = old_q + self.learning_rate * (
                reward - old_q
            )

class ExampleOrderOptimizer:
    """
    示例排列优化器
    研究：相同示例不同排列，准确率差异可达15-20%
    """
    
    def __init__(self, model):
        self.model = model
    
    def evaluate_ordering(self, examples: list[dict], 
                          eval_set: list[dict]) -> float:
        """评估特定排列的准确率"""
        correct = 0
        
        for eval_item in eval_set:
            prompt = self._build_prompt(examples, eval_item["input"])
            response = self.model.generate(prompt)
            
            if self._check_answer(response, eval_item["output"]):
                correct += 1
        
        return correct / len(eval_set)
    
    def find_optimal_order(self, examples: list[dict],
                          eval_set: list[dict]) -> list[dict]:
        """寻找最优排列（贪心搜索）"""
        from itertools import permutations
        
        best_acc = 0
        best_order = examples
        
        # 对于少量示例，可以穷举
        if len(examples) <= 5:
            for perm in permutations(examples):
                acc = self.evaluate_ordering(list(perm), eval_set)
                if acc > best_acc:
                    best_acc = acc
                    best_order = list(perm)
        else:
            # 贪心搜索
            best_order = self._greedy_search(examples, eval_set)
        
        return best_order

ORDERING_GUIDELINES = """
=== Few-shot 示例排列最佳实践 ===

1. 最近效应（Recency Effect）
   - 模型更容易受最后一个示例的影响
   - 将最相关的示例放在最后

2. 难度递进
   - 从简单到复杂排列
   - 帮助模型逐步理解任务

3. 正负交替
   - 正例-反例-正例-反例
   - 比连续正例更有效

4. 避免偏见
   - 不要将所有同一类别的示例放在一起
   - 打乱类别顺序减少偏见

5. 答案分布平衡
   - 如果是分类任务，确保各类别示例数量均衡
   - 避免模型偏向多数类
"""

class NegativeExamplePrompting:
    """
    负面示例Prompting
    展示错误案例及其修正，比纯正面示例更有效
    """
    
    @staticmethod
    def build_prompt(positive_examples: list[dict],
                     negative_examples: list[dict],
                     query: str) -> str:
        """
        构建包含正负示例的Prompt
        """
        prompt = "请根据以下示例完成任务。\n\n"
        
        # 正面示例
        prompt += "✅ 正确示例：\n"
        for ex in positive_examples:
            prompt += f"输入：{ex['input']}\n"
            prompt += f"输出：{ex['output']}\n"
            prompt += f"说明：{ex.get('explanation', '')}\n\n"
        
        # 负面示例
        if negative_examples:
            prompt += "❌ 错误示例（请避免以下错误）：\n"
            for ex in negative_examples:
                prompt += f"输入：{ex['input']}\n"
                prompt += f"❌ 错误输出：{ex['wrong_output']}\n"
                prompt += f"✅ 正确输出：{ex['correct_output']}\n"
                prompt += f"错误原因：{ex['error_reason']}\n\n"
        
        # 查询
        prompt += f"现在请处理：\n输入：{query}\n输出："
        
        return prompt

class CrossLingualFewShot:
    """
    跨语言Few-shot Prompting
    用英语示例指导中文任务（或反向）
    """
    
    def __init__(self, model):
        self.model = model
    
    def cross_lingual_prompt(self, 
                             source_examples: list[dict],
                             target_query: str,
                             source_lang: str = "en",
                             target_lang: str = "zh") -> str:
        """
        构建跨语言Few-shot Prompt
        """
        prompt = f"""以下是用{source_lang}语言展示的任务示例。
请理解示例中的任务模式，并用{target_lang}语言完成下面的查询。

示例：
"""
        for ex in source_examples:
            prompt += f"Input: {ex['input']}\n"
            prompt += f"Output: {ex['output']}\n\n"
        
        prompt += f"现在请用{target_lang}回答：\n"
        prompt += f"Input: {target_query}\n"
        prompt += f"Output: "
        
        return prompt
    
    def translated_examples_prompt(self, 
                                   examples: list[dict],
                                   query: str,
                                   target_lang: str = "zh") -> str:
        """
        翻译示例到目标语言后再使用
        """
        translated = []
        for ex in examples:
            translated_input = self.model.translate(
                ex['input'], target_lang=target_lang
            )
            translated_output = self.model.translate(
                ex['output'], target_lang=target_lang
            )
            translated.append({
                'input': translated_input,
                'output': translated_output
            })
        
        return self._build_standard_prompt(translated, query)

class DynamicFewShotSystem:
    """
    动态Few-shot系统
    每次查询动态选择最相关的示例
    """
    
    def __init__(self, example_pool: list[dict], embed_model, llm):
        self.example_pool = example_pool
        self.embed_model = embed_model
        self.llm = llm
        self.selector = DiversityAwareSelector(example_pool, embed_model)
        self.feedback_store = []
    
    async def answer(self, query: str, k: int = 4) -> dict:
        """动态选择示例并回答"""
        
        # 1. 选择示例
        examples = self.selector.select(query, k=k)
        
        # 2. 构建Prompt
        prompt = self._build_prompt(examples, query)
        
        # 3. 生成回答
        response = await self.llm.generate(prompt)
        
        # 4. 记录用于后续优化
        self.feedback_store.append({
            "query": query,
            "selected_examples": examples,
            "response": response,
            "timestamp": datetime.now()
        })
        
        return {
            "answer": response,
            "examples_used": examples,
            "prompt": prompt
        }
    
    def optimize_pool(self):
        """基于历史反馈优化示例池"""
        # 分析哪些示例被高频选中且效果好
        example_stats = {}
        
        for record in self.feedback_store:
            for ex in record["selected_examples"]:
                ex_id = ex["id"]
                if ex_id not in example_stats:
                    example_stats[ex_id] = {
                        "count": 0,
                        "success": 0
                    }
                example_stats[ex_id]["count"] += 1
        
        # 保留高频且高效的示例，淘汰低效的
        # ...

class FewShotEvaluator:
    """Few-shot Prompting评估工具"""
    
    def __init__(self, model, eval_dataset):
        self.model = model
        self.eval_set = eval_dataset
    
    async def evaluate_configuration(self,
                                     selector_class,
                                     k: int,
                                     ordering: str = "similarity_desc",
                                     use_negative: bool = False) -> dict:
        """评估特定Few-shot配置"""
        
        selector = selector_class(self.eval_set, self.model)
        
        results = []
        for item in self.eval_set:
            # 选择示例
            examples = selector.select(item["input"], k=k)
            
            # 排列
            if ordering == "similarity_desc":
                examples = sorted(examples, key=lambda x: x["similarity"])
            elif ordering == "difficulty_asc":
                examples = sorted(examples, key=lambda x: x["difficulty"])
            
            # 构建Prompt
            prompt = self._build_prompt(examples, item["input"])
            
            # 生成
            response = await self.model.generate(prompt)
            
            # 评估
            correct = self._check(response, item["output"])
            results.append({
                "correct": correct,
                "response": response,
                "expected": item["output"]
            })
        
        accuracy = sum(r["correct"] for r in results) / len(results)
        
        return {
            "accuracy": accuracy,
            "k": k,
            "selector": selector_class.__name__,
            "ordering": ordering,
            "use_negative": use_negative,
            "detailed_results": results
        }

FEW_SHOT_GOLDEN_RULES_2026 = """
=== Few-shot Prompting 黄金法则 ===

1. 质量 > 数量
   - 3-5个精选示例优于20个随机示例
   - 每个示例都应展示不同的模式

2. 相似性选择
   - 使用KNN或语义搜索选择与查询最相关的示例
   - 但保持一定多样性

3. 排列有讲究
   - 最相关的示例放最后（近因效应）
   - 简单到复杂排列帮助理解

4. 包含负面示例
   - 展示"不该怎么做"比只展示"该怎么做"更有效
   - 附带错误原因说明

5. 答案平衡
   - 分类任务中各类别示例数量均等
   - 避免模型产生频率偏见

6. 动态选择
   - 不同查询用不同示例
   - 建立示例池，按需选择

7. 持续优化
   - 记录每次查询的示例选择和效果
   - 定期评估和更新示例池
"""