Few-shot 示例的动态选择：根据 query 语义自动匹配最相关的 prompt 示例

Few-shot 示例的动态选择根据 query 语义自动匹配最相关的 prompt 示例一、深度引言与场景痛点大家好我是赵咕咕。你给模型写了一个 Few-shot prompt里面放了3个示例——一个关于代码审查、一个关于文档生成、一个关于数据分析。结果用户问了一个写周报的问题你的示例一个都用不上。这就是静态 Few-shot 的痛点你准备的示例是固定的但用户的问题是千变万化的。示例不对口Few-shot 不但帮不上忙反而可能引导模型往错误方向跑。解决思路很简单根据用户 query 的语义从示例库中动态选出最相关的几个示例。这篇文章我们就来实现这个方案让你的 Prompt 示例活起来。二、底层机制与原理深度剖析动态 Few-shot 选择的核心流程是把每个示例做向量化存储用户 query 来了也做向量化然后用向量相似度匹配选出最相关的 K 个示例。这里有个关键设计示例的特征向量不应该基于示例原文而应该基于这个示例适合解决什么问题。比如一个代码审查的示例它的匹配向量应该和代码审查、代码质量、code review这些概念靠近而不是和示例中具体的变量名靠近。所以我们实际上需要两套向量匹配向量用于和 user query 做相似度计算的问题特征向量示例内容真正注入 prompt 的完整示例文本架构流程如下flowchart TB A[用户 Query] -- B[Query 向量化br/Embedding Model] B -- C[向量相似度检索br/与示例库匹配向量对比] D[(示例库br/Example Store)] -- E[匹配向量索引br/Match Vector Index] D -- F[示例内容存储br/Example Content] C -- G[Top-K 相似示例] G -- H[示例去重与多样性过滤br/MMR 算法] H -- I[按匹配分数排序] I -- J[注入 Promptbr/Few-shot Examples] K[Prompt 模板] -- L[最终 Prompt] J -- L A -- L style B fill:#e8f5e9 style C fill:#fff3e0 style H fill:#f3e5f5除了相似度匹配还需要考虑示例间的多样性。如果前 5 个最相似的示例都是同一个类型后两个再相关的示例也没机会展示。这就需要 MMRMaximal Marginal Relevance算法来平衡相关性和多样性。三、生产级代码实现下面是完整的动态 Few-shot 选择实现from __future__ import annotations import asyncio import numpy as np from dataclasses import dataclass, field from typing import Optional import hashlib import json dataclass class FewShotExample: 单个 Few-shot 示例 example_id: str description: str # 示例适用场景的描述 user_input: str # 示例中的用户输入 assistant_output: str # 示例中的期望输出 match_vector: Optional[list[float]] None # 匹配向量 tags: list[str] field(default_factorylist) difficulty: str medium # easy / medium / hard def to_prompt_str(self) - str: 转为 prompt 中的格式 return ( f用户: {self.user_input}\n f助手: {self.assistant_output}\n ) property def similarity_key(self) - str: 用于匹配的文本 return f{self.description} { .join(self.tags)} class DynamicFewShotSelector: 动态 Few-shot 示例选择器 def __init__(self, embedding_model): embedding_model: 需要实现 embed(text) - list[float] self._embed embedding_model self._examples: list[FewShotExample] [] self._vectors: Optional[np.ndarray] None # shape: (N, D) async def add_example(self, example: FewShotExample) - None: 添加示例并计算匹配向量 if example.match_vector is None: match_text example.similarity_key example.match_vector await self._embed(match_text) self._examples.append(example) self._rebuild_index() def _rebuild_index(self) - None: 重构向量索引矩阵 if not self._examples: self._vectors None return vectors [] for ex in self._examples: if ex.match_vector: vectors.append(ex.match_vector) if vectors: self._vectors np.array(vectors, dtypenp.float32) async def select( self, query: str, k: int 3, diversity_lambda: float 0.5, ) - list[FewShotExample]: 为 query 选择最佳示例 diversity_lambda: MMR 多样性权重 (0纯多样性, 1纯相关性) if not self._examples or self._vectors is None: return [] # 1. 向量化 query query_vec np.array( await self._embed(query), dtypenp.float32 ) # 2. 计算余弦相似度 scores self._cosine_similarity(query_vec, self._vectors) # 3. MMR 多样性选择 selected_indices self._mmr_select( query_vec, scores, k, diversity_lambda ) return [self._examples[i] for i in selected_indices] def _cosine_similarity( self, query: np.ndarray, candidates: np.ndarray ) - np.ndarray: 计算余弦相似度 query_norm query / (np.linalg.norm(query) 1e-8) candidates_norm candidates / ( np.linalg.norm(candidates, axis1, keepdimsTrue) 1e-8 ) return np.dot(candidates_norm, query_norm) def _mmr_select( self, query_vec: np.ndarray, scores: np.ndarray, k: int, lam: float, ) - list[int]: MMR (Maximal Marginal Relevance) 算法 平衡相关性和多样性地选择 Top-K n len(scores) if n k: return list(range(n)) selected [] remaining set(range(n)) # 第一个选择最高相关性 first int(np.argmax(scores)) selected.append(first) remaining.remove(first) while len(selected) k and remaining: best_idx -1 best_mmr -float(inf) for i in remaining: # 相关性项 relevance lam * scores[i] # 多样性项与已选示例的最大相似度 max_sim_to_selected 0.0 if selected: for j in selected: sim self._cosine_similarity( self._vectors[i], self._vectors[j].reshape(1, -1) ) max_sim_to_selected max( max_sim_to_selected, sim ) diversity (1 - lam) * (1 - max_sim_to_selected) mmr relevance diversity if mmr best_mmr: best_mmr mmr best_idx i if best_idx 0: selected.append(best_idx) remaining.remove(best_idx) return selected def to_prompt_examples(self, examples: list[FewShotExample]) - str: 将选中的示例转为 prompt 片段 if not examples: return lines [## 参考示例\n] for i, ex in enumerate(examples, 1): lines.append(f### 示例 {i} ({ex.description})\n) lines.append(ex.to_prompt_str()) return \n.join(lines) property def example_count(self) - int: return len(self._examples) class MockEmbedding: 模拟 Embedding实际使用时替换为真实模型 async def embed(self, text: str) - list[float]: # 用 hash 模拟向量实际请使用 text-embedding-3 等模型 h hashlib.md5(text.encode()).digest() return list(h) [0.0] * (256 - 16) # 使用示例 async def main(): embedder MockEmbedding() selector DynamicFewShotSelector(embedder) # 添加示例库 examples [ FewShotExample( example_idex1, description代码审查与 Bug 分析, user_input帮我 review 这段 Python 代码找出潜在 bug, assistant_output发现了 3 个问题1. 未处理的 None..., tags[code, review, python], difficultymedium, ), FewShotExample( example_idex2, description技术文档生成, user_input为这份 API 生成使用文档, assistant_output## API 文档\n### 接口说明\n..., tags[document, api, writing], difficultyeasy, ), FewShotExample( example_idex3, description性能优化分析, user_input这段代码的数据库查询很慢帮我优化, assistant_output建议1. 为 user_id 添加索引..., tags[performance, database, optimization], difficultyhard, ), FewShotExample( example_idex4, description周报与工作总结生成, user_input我是后端工程师帮我生成本周工作周报, assistant_output本周工作总结\n1. 完成了订单模块重构..., tags[report, writing, weekly], difficultyeasy, ), ] for ex in examples: await selector.add_example(ex) print(f示例库共有 {selector.example_count} 个示例\n) # 测试1代码相关 query query1 这段 Python 代码有没有安全问题 selected1 await selector.select(query1, k2) print(fQuery: {query1}) print(选中示例:) for ex in selected1: print(f - {ex.description}) # 测试2文档相关 query query2 写一份项目周报总结 selected2 await selector.select(query2, k2) print(f\nQuery: {query2}) print(选中示例:) for ex in selected2: print(f - {ex.description}) # 生成 prompt 示例 prompt selector.to_prompt_examples(selected2) print(f\n生成的 Prompt 示例:\n{prompt}) asyncio.run(main())四、边界分析与架构权衡动态 Few-shot 选择有几个工程边界匹配向量的质量问题。如果示例的描述文本写得不好匹配精度会大幅下降。建议为每个示例额外编写搜索关键词字段而非完全依赖原始描述。同时定期评估检索命中率对低命中率的示例做描述优化。冷启动问题。示例库刚建立时覆盖的场景可能不够广导致某些 query 找不到合适的示例。解决方案是设置一个兜底示例集当相似度分数普遍低于阈值时使用兜底示例。兜底示例应该是覆盖范围最广的通用示例。多样性 vs 相关性的平衡。MMR 的diversity_lambda默认设为 0.5 是经验值实际需要根据示例库的分布调优。示例库中相同类型的示例多时应降低 lambda增加多样性权重示例类型已经足够多样时可提高 lambda。Token 预算控制。Few-shot 示例会占用 prompt token。如果一个示例平均 500 token选 5 个就是 2500 token。需要根据模型的上下文窗口做修剪示例过长时做摘要或者根据 token 预算动态调整 k 值。效果评估的难度。如何衡量动态选择的 Few-shot 确实比静态的好可以用人工标注的测试集做 A/B 对比比较两种方案下模型输出的质量和一致性。这本身又是一个小型评测工程。五、总结动态 Few-shot 让 Prompt Engineering 从写死示例升级到数据驱动选择。核心思路是把示例库向量化用 query 语义匹配最相关示例并通过 MMR 算法保证多样性。关键要点每个示例需要一个独立的匹配向量而非直接用示例原文用 MMR 算法平衡相关性和多样性设置兜底策略应对冷启动注意 token 预算示例过多反而有害好的 Few-shot 不是写出来的是选出来的。