从 71% 到 92%：动态 Few-shot 召回优化实录

问题背景：为什么固定 Few-shot 在文本分类中会失效

很多团队第一次接入大模型做文本分类时，会在提示词里手写 3～5 个示例。短文本、意图明确时效果不错，但一旦进入真实业务，固定示例很快暴露问题：新输入与示例表层词汇不相似、类别边界样本增多、某些标签在示例集中缺失，模型就会依赖关键词做表面匹配。典型现象是：“退款一直不到账”明明应归为售后进度，却因为出现“退款”被归为退款申请。

静态 Few-shot 的根本问题不是示例数量，而是示例与当前输入的相关性不可控。动态示例选择的目标，是在推理时从候选示例库中召回最能帮助模型判断当前样本的少量案例。它看起来像 RAG，但文本分类的召回目标与普通问答不同：不是找到“语义最接近的文本”，而是找到最能提供分类决策边界的样本。

下面这套方案在一个客服工单分类任务上，把少样本准确率从约 71% 提升到 92%。核心改动包括：标签感知嵌入、难例库、MMR 多样性重排，以及面向结构化输出的分类模板。

方案对比：三种示例选择策略怎么选

策略一：随机或固定示例

实现最简单，每个标签放一两个标准样本。优点是稳定、零延迟；缺点是无法覆盖长尾表达，对标签冲突和口语化输入几乎没有自适应能力。适合标签少、样本高度模板化的冷启动阶段。

策略二：纯向量相似度 Top-K

用 embedding 检索与输入最相似的历史样本。它能解决词汇差异问题，例如“钱怎么还没回来”可以召回“退款未到账”。但纯相似度有两个副作用：一是召回结果高度同质，可能 5 条全是同一类，模型失去对照；二是容易被业务主题词带偏，如“发票”“物流”等强主题词会压过真实意图。

策略三：标签感知召回 + 多样性重排

推荐在生产中使用。先保证候选覆盖当前最可能的标签，再插入“近似但不同类”的对比样本，最后用 MMR 控制冗余。这样每个提示包既包含正例，也包含边界反例，模型能学到“像 A 但其实是 B”的差异。若标签体系稳定，还可以为每个标签维护难例库，成本低于全量向量库，收益却很明显。

实操步骤：构建可运行的动态示例选择器

下面代码不依赖特定厂商 SDK，嵌入函数可替换为 OpenAI、本地 sentence-transformers 或内部 Embedding 服务。示例库建议来自人工确认过的历史工单，并持续把线上误判样本沉淀为难例。

import json
import math
from dataclasses import dataclass
from typing import List, Dict, Callable

# 生产环境可替换为真实 embedding API；此处用哈希向量模拟，保证代码可直接运行
def fake_embedding(text: str, dim: int = 64) -> List[float]:
    vec = [0.0] * dim
    for token in text.lower().split():
        vec[hash(token) % dim] += 1.0
    norm = math.sqrt(sum(x * x for x in vec)) or 1.0
    return [x / norm for x in vec]

@dataclass
class Example:
    text: str
    label: str
    hard: bool = False  # 是否为边界难例

def cosine(a: List[float], b: List[float]) -> float:
    return sum(x * y for x, y in zip(a, b))

class DynamicExampleSelector:
    def __init__(self, examples: List[Example], k: int = 5, lambda_mmr: float = 0.7):
        self.examples = examples
        self.k = k
        self.lambda_mmr = lambda_mmr
        # 关键技巧：把标签语义拼进文本，让嵌入同时感知内容与类别
        self.vectors = {
            id(ex): fake_embedding(f"类别：{ex.label} 内容：{ex.text}")
            for ex in examples
        }

    def select(self, query: str) -> List[Example]:
        qv = fake_embedding(query)
        scored = [
            (cosine(qv, self.vectors[id(ex)]), ex)
            for ex in self.examples
        ]
        scored.sort(key=lambda x: x[0], reverse=True)

        # 先取较大候选池，给 MMR 和难例补充留出空间
        candidates = scored[: max(self.k * 4, 12)]
        selected: List[Example] = []
        chosen_labels = set()

        # 1. 先加入一个最高相似难例，提供决策边界
        hard = [ex for score, ex in candidates if ex.hard]
        if hard:
            selected.append(hard[0])
            chosen_labels.add(hard[0].label)

        # 2. MMR：兼顾与输入的相关性和示例之间的差异性
        pool = candidates[:]
        while len(selected) < self.k and pool:
            best_idx, best_score = None, -1e9
            for i, (rel, ex) in enumerate(pool):
                if not selected:
                    mmr = rel
                else:
                    max_sim = max(
                        cosine(self.vectors[id(ex)], self.vectors[id(s)])
                        for s in selected
                    )
                    # lambda 越高越偏相关性，越低越偏多样性
                    mmr = self.lambda_mmr * rel - (1 - self.lambda_mmr) * max_sim
                if mmr > best_score:
                    best_score, best_idx = mmr, i
            ex = pool.pop(best_idx)[1]
            # 避免首轮全部落在同一标签
            if ex.label in chosen_labels and len(chosen_labels) < 3 and pool:
                continue
            selected.append(ex)
            chosen_labels.add(ex.label)
        return selected

接着把筛选结果注入提示词。文本分类务必要求模型输出结构化 JSON，避免“我认为可能是……”这类不可解析答案。

CLASSIFICATION_PROMPT = """你是严谨的文本分类器。只能从候选标签中选择，不要生成标签外名称。

候选标签：{labels}

判断原则：
1. 优先识别用户当前诉求，而不是文本中提到的业务名词。
2. 如果用户在催促已有处理进度，优先选择“进度查询”。
3. 输出必须是 JSON，不要输出解释性正文。

参考示例：
{examples}

待分类文本：{query}

输出格式：
{{"label": "候选标签之一", "confidence": 0.0到1.0的数字, "reason": "不超过30字的依据"}}
"""

def build_prompt(query: str, selector: DynamicExampleSelector, labels: List[str]) -> str:
    shots = selector.select(query)
    example_text = "\n".join(
        json.dumps({"text": s.text, "label": s.label}, ensure_ascii=False)
        for s in shots
    )
    return CLASSIFICATION_PROMPT.format(
        labels="、".join(labels),
        examples=example_text,
        query=query
    )

# 可直接运行的演示
examples = [
    Example("我要退货，商品不想要了", "退款申请"),
    Example("三天前申请的退款现在还没到", "进度查询", hard=True),
    Example("客服能不能帮我催一下退货审核", "进度查询", hard=True),
    Example("发票信息填错了怎么修改", "资料修改"),
    Example("物流显示签收但我没收到", "物流异常"),
    Example("商品破损，我要申请赔偿", "售后投诉"),
]
selector = DynamicExampleSelector(examples, k=5)
print(build_prompt("我的退款怎么还没处理完", selector,
                   ["退款申请", "进度查询", "资料修改", "物流异常", "售后投诉"]))

评测方法：不要只凭单条样本调提示词

动态选择器必须离线评测。建议固定一个 300～1000 条的人工标注集，比较随机示例、Top-K、标签感知 + MMR 三组策略。除整体准确率外，还要看每个标签的召回率，以及 Top-2 候选标签命中率。后者能反映召回阶段是否把正确类别相关样本放进了提示包。

一个常见误区是只优化端到端分类准确率，却不评估召回质量。如果正确标签对应的示例从未进入候选池，提示词写得再好也很难救回。可以记录每次实际入选的示例 ID、相似度、标签分布和模型输出，形成回放数据集，后续调 k、lambda_mmr 与难例阈值时就能快速对比。

避坑总结：4 个高频错误与修正方法

第一，不要把原始文本直接向量化。 错误现象是同一业务词的样本互相召回，分类意图被主题词淹没。修正方法是把标签名、关键判别规则或归一化后的意图描述拼入嵌入文本，例如“类别：进度查询；语义：催促已有售后流程”。

第二，不要盲目增大 K。 示例从 5 条增加到 12 条后，准确率可能反而下降，还会推高延迟和费用。文本分类通常 3～6 条足够；更重要的是标签覆盖和对照关系。若输入很短，2 个正例加 1 个近似反例往往优于 10 条同质案例。

第三，不要让难例污染示例库。 错误现象是模型被历史误判带偏，频繁选择稀有标签。难例必须经过人工确认，并限制每次最多注入 1～2 条；同时难例文本中不要包含“容易误判为某标签”这类元信息，否则模型可能反向关注错误标签。

第四，不要忽略输出约束与解析兜底。 即使要求 JSON，部分模型仍可能输出 Markdown 代码块或多余解释。生产中应先提取 JSON，再校验 label 是否在白名单内、confidence 是否为数值；解析失败或标签非法时应触发重试或降级到规则分类，而不是写入默认标签。

动态 Few-shot 不是一次性技巧，而是一个小型数据闭环：线上误判进入审核台，人工确认后沉淀为难例，离线评测通过后再进入示例库。把这个闭环跑起来，文本分类系统才能在标签演化和表达变化中持续保持高准确率。

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。