当前位置：首页 > news >正文

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

news 2026/7/8 21:50:18

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[注入 Prompt<br/>Few-shot Examples] K[Prompt 模板] --> L[最终 Prompt] J --> L A --> L style B fill:#e8f5e9 style C fill:#fff3e0 style H fill:#f3e5f5

除了相似度匹配，还需要考虑示例间的多样性。如果前 5 个最相似的示例都是同一个类型，后两个再相关的示例也没机会展示。这就需要 MMR（Maximal 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_factory=list) 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, dtype=np.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), dtype=np.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, axis=1, keepdims=True) + 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_id="ex1", description="代码审查与 Bug 分析", user_input="帮我 review 这段 Python 代码，找出潜在 bug", assistant_output="发现了 3 个问题：1. 未处理的 None...", tags=["code", "review", "python"], difficulty="medium", ), FewShotExample( example_id="ex2", description="技术文档生成", user_input="为这份 API 生成使用文档", assistant_output="## API 文档\n### 接口说明\n...", tags=["document", "api", "writing"], difficulty="easy", ), FewShotExample( example_id="ex3", description="性能优化分析", user_input="这段代码的数据库查询很慢，帮我优化", assistant_output="建议：1. 为 user_id 添加索引...", tags=["performance", "database", "optimization"], difficulty="hard", ), FewShotExample( example_id="ex4", description="周报与工作总结生成", user_input="我是后端工程师，帮我生成本周工作周报", assistant_output="本周工作总结：\n1. 完成了订单模块重构...", tags=["report", "writing", "weekly"], difficulty="easy", ), ] for ex in examples: await selector.add_example(ex) print(f"示例库共有 {selector.example_count} 个示例\n") # 测试1：代码相关 query query1 = "这段 Python 代码有没有安全问题？" selected1 = await selector.select(query1, k=2) print(f"Query: {query1}") print("选中示例:") for ex in selected1: print(f" - {ex.description}") # 测试2：文档相关 query query2 = "写一份项目周报总结" selected2 = await selector.select(query2, k=2) 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())