动态 Few-shot 构建根据用户历史对话自动填充最相关示例的策略一、深度引言与场景痛点写 Prompt 时最偷懒的做法是塞几个固定示例进去然后祈祷模型能举一反三。刚上线时效果还行时间一长问题就暴露了你的固定示例是关于退换货流程的但用户当前问的是跨境物流时效——模型被不相关的示例干扰反而给出了偏离用户场景的回答。更隐蔽的问题是示例的负迁移。Few-shot 示例本质上是给模型一个输出格式和思维模式的参考。如果示例里的推理解题思路和当前问题需要的思路完全不同模型会错误地沿用示例的解题框架。比如你的示例是事实查询→输出精确答案但用户当前需要的是多个因素权衡→给出建议固定示例反而把模型带歪了。动态 Few-shot 的基本思路很简单别写死示例而是从用户的历史对话中检索和当前问题最相似的交互记录把它们作为上下文示例注入 Prompt。关键是最相似怎么定义——不是简单的关键词匹配而是语义层面的相关性和对话结构层面的适配。二、底层机制与原理深度剖析第一步把用户当前查询向量化。用的是同一个 embedding 模型保证查询向量和历史对话向量的语义空间一致。注意这里不是把整个对话都向量化而是只向量化用户的消息部分——因为用户的消息表达了真实意图而 AI 的回复是执行结果。第二步语义检索。在历史对话的向量索引中做 ANN 检索返回 Top-N 条相似记录。N 可以设大一点比如 50给后续的结构过滤留足候选池。这里有个关键设计历史对话存储时要有结构化的元数据——对话意图、是否成功完成、使用的工具链、最终用户是否给出了正面反馈。第三步结构过滤。光是语义相似不够还要看这些候选对话的质量——只选那些最终被标记为用户满意的对话作为示例。同时做意图去重如果候选池里有 5 条都是查询订单类的对话只保留 2-3 条。多样性比数量更重要相似度过高的示例提供的信息量是边际递减的。第四步精排。语义检索用的是双塔模型的余弦相似度速度够快但精度一般。精排阶段用 cross-encoder 对候选示例做 pairwise 打分取 Top-KK 通常是 3-5。cross-encoder 把查询和候选拼接后送入 transformer做更深度的相关性判断精度高但速度慢——所以只放在候选池缩小到几十条之后才执行。第五步去重与 token 预算控制。精排后的示例仍然可能有重复的模式。去重策略是计算示例之间的文本相似度Jaccard 或 MinHash相似度超过 0.8 的只保留一个。然后把示例按相关性分数从高到低排列逐个计算 token 数直到累计 token 接近预设的上限比如 2000 tokens剩余的全部丢弃。三、生产级代码实现import asyncio import hashlib import logging from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any import numpy as np logger logging.getLogger(__name__) dataclass class ConversationRecord: 历史对话记录 session_id: str user_query: str assistant_response: str intent: str tools_used: list[str] user_satisfied: bool embedding: list[float] | None None timestamp: str field(default_factorylambda: datetime.now(timezone.utc).isoformat()) class DynamicFewShotEngine: 动态 Few-shot 构建引擎 def __init__( self, max_examples: int 5, max_tokens: int 2000, diversity_threshold: float 0.8, ) - None: self._max_examples max_examples self._max_tokens max_tokens self._diversity_threshold diversity_threshold # 生产环境替换为真实的向量存储和 cross-encoder self._history_store: dict[str, ConversationRecord] {} self._cache: dict[str, list[dict[str, str]]] {} async def add_conversation(self, record: ConversationRecord) - None: 添加一条历史对话到存储 # 向量化用户查询生产环境使用真实的 embedding 模型 if record.embedding is None: record.embedding await self._embed(record.user_query) key f{record.session_id}:{hashlib.md5(record.user_query.encode()).hexdigest()[:8]} self._history_store[key] record # 清除相关性缓存新数据可能改变检索结果 self._cache.clear() async def build_examples(self, user_query: str, intent: str ) - list[dict[str, str]]: 构建动态 Few-shot 示例 cache_key hashlib.md5(f{user_query}:{intent}.encode()).hexdigest() if cache_key in self._cache: return self._cache[cache_key] # Step 1: 向量化用户查询 query_embedding await self._embed(user_query) # Step 2: 语义检索 Top-N 候选 candidates await self._semantic_search(query_embedding, top_n50) # Step 3: 结构过滤意图匹配 质量筛选 filtered self._structural_filter(candidates, intent) # Step 4: 精排生产环境使用 cross-encoder ranked await self._rerank(user_query, filtered, top_kself._max_examples * 2) # Step 5: 去重 token 预算控制 examples self._deduplicate_and_trim(ranked, self._max_tokens) # 缓存结果 self._cache[cache_key] examples return examples async def _embed(self, text: str) - list[float]: 向量化文本生产环境接入 embedding API # 这里用简单的 hash 模拟实际使用 OpenAI/本地 embedding 模型 h hashlib.sha256(text.encode()) seed int(h.hexdigest()[:16], 16) rng np.random.RandomState(seed) return rng.randn(1536).tolist() async def _semantic_search( self, query_embedding: list[float], top_n: int ) - list[ConversationRecord]: 向量语义检索候选对话 if not self._history_store: return [] query_vec np.array(query_embedding) scored: list[tuple[float, ConversationRecord]] [] for record in self._history_store.values(): if record.embedding is None: continue sim np.dot(query_vec, np.array(record.embedding)) scored.append((float(sim), record)) scored.sort(keylambda x: -x[0]) return [record for _, record in scored[:top_n]] def _structural_filter( self, candidates: list[ConversationRecord], intent: str ) - list[ConversationRecord]: 结构过滤意图匹配 质量筛选 filtered [ r for r in candidates if r.user_satisfied and (not intent or r.intent intent) ] # 如果没有意图匹配的退回到只看满意度 if not filtered: filtered [r for r in candidates if r.user_satisfied] return filtered async def _rerank( self, query: str, candidates: list[ConversationRecord], top_k: int ) - list[ConversationRecord]: Cross-encoder 精排模拟实现 # 生产环境query candidate.user_query candidate.assistant_response # 送入 cross-encoder 模型做相关性打分 await asyncio.sleep(0) # 模拟异步行为 return candidates[:top_k] def _deduplicate_and_trim( self, examples: list[ConversationRecord], max_tokens: int ) - list[dict[str, str]]: 去重并控制 token 预算 selected: list[dict[str, str]] [] used_tokens 0 for record in examples: # 去重检查与已选示例的文本相似度 if self._is_duplicate(record, selected): continue example_text f用户: {record.user_query}\n助手: {record.assistant_response} # 粗略估计 token 数中文约 1.5 字符/token est_tokens len(example_text) // 2 if used_tokens est_tokens max_tokens: break selected.append({ user: record.user_query, assistant: record.assistant_response, }) used_tokens est_tokens if len(selected) self._max_examples: break return selected def _is_duplicate( self, candidate: ConversationRecord, selected: list[dict[str, str]] ) - bool: 检查候选示例是否与已选示例高度重复 for s in selected: # 简单的 Jaccard 相似度估计 words_a set(candidate.user_query) words_b set(s[user]) if not words_a or not words_b: continue jaccard len(words_a words_b) / len(words_a | words_b) if jaccard self._diversity_threshold: return True return False def inject_examples( self, system_prompt: str, examples: list[dict[str, str]] ) - str: 将示例注入到 System Prompt 中 if not examples: return system_prompt example_lines [\n## 参考示例来自历史成功对话\n] for i, ex in enumerate(examples, 1): example_lines.append(f### 示例 {i}) example_lines.append(f用户问题{ex[user]}) example_lines.append(f正确回答{ex[assistant]}) example_lines.append() return system_prompt \n.join(example_lines) async def main() - None: engine DynamicFewShotEngine(max_examples3, max_tokens1500) # 模拟添加历史对话 history [ ConversationRecord( session_ids1, user_query订单快递进度怎么查, assistant_response您可以在订单详情页点击物流追踪..., intentorder_tracking, tools_used[order_api], user_satisfiedTrue, ), ConversationRecord( session_ids2, user_query退货需要几天到账, assistant_response退款通常在3-5个工作日到账..., intentrefund_inquiry, tools_used[payment_api], user_satisfiedTrue, ), ConversationRecord( session_ids3, user_query怎么修改收货地址, assistant_response您可以在订单详情中修改..., intentaddress_change, tools_used[user_api], user_satisfiedTrue, ), ] for r in history: await engine.add_conversation(r) # 构建动态 Few-shot examples await engine.build_examples(快递到哪了能改地址吗, intentorder_tracking) prompt engine.inject_examples(你是一个客服助手。, examples) print(prompt) if __name__ __main__: logging.basicConfig(levellogging.INFO) asyncio.run(main())_structural_filter中有一个重要的兜底逻辑——如果按意图过滤后候选为空退回到只看满意度。这避免了因意图分类不准导致完全没有示例的情况。_is_duplicate用字符级的 Jaccard 相似度做粗略去重生产环境可以升级为 SimHash 或 MinHash 以支持更大规模。_cache在新增历史对话后自动清空保证检索结果反映最新数据。四、边界分析与架构权衡动态 Few-shot 的主要边界是检索延迟。从向量化到精排总延迟可能在 50-200ms 之间。如果 RAG 链路本来就慢再加 200ms 可能超出 SLO。优化方向小于 1000 条的历史对话可以直接用内存存储 numpy 精确计算相似度省去向量库的网络往返cross-encoder 用 ONNX 部署在本地延迟能降到 10ms 以内。历史对话的膨胀问题也需要解决。用户的对话会越来越多全量存储向量和文本的存储成本线性增长。降维策略是对超过 30 天的对话做聚类后只保留聚类中心的代表性对话对成功对话做抽样而非全量存储每 10 条保留 1 条自动淘汰被标记为不满意的低质量对话。另一个容易被忽视的问题是示例的时效性。30 天前的退换货政策示例可能已经过时了因为政策改了模型学着过时的示例输出错误信息。解决方案是给每条对话记录打上时间戳和来源文档版本号检索时优先选时间近、版本匹配的示例。五、总结动态 Few-shot 的核心价值是把 Prompt 示例从写死的模板变成活的检索结果。管线分为语义检索、结构过滤、精排、去重、注入五步每一步都有明确的职责和优化空间。生产落地的关键考量控制检索延迟不超过链路的 10%-20%对历史对话做降维存储避免成本爆炸优先使用近期的、版本匹配的示例防止知识过期。别把 Few-shot 当成 Prompt 的静态补丁——它是 RAG 检索之外的第二条检索通路值得用同等级别的工程投入来优化。