最近在优化我们公司的智能客服系统，遇到了一个挺典型的问题：系统采用了RAG（检索增强生成）结合大语言模型的架构，但回答的稳定性让人头疼。同一个用户问题，在不同时间提问，或者刷新几次，得到的答案有时准确，有时却跑偏了，甚至自相矛盾。初步统计，大约有30%的复杂问题存在这种“随机性回答”的情况。这不仅严重影响了用户体验和信任度，也给客服团队的运营带来了额外负担，需要人工频繁介入纠错。

为了解决这个问题，我们调研和尝试了多种方案，最终形成了一套综合性的优化策略，将回答的准确率提升了40%以上。下面就把整个实战过程和思考分享出来。

1. 技术方案对比：为什么传统方法不够用？

在深入我们的方案之前，先看看几种常见的解决思路及其局限性：

1. 答案缓存：这是最直接的想法。将第一次生成的答案缓存起来，后续相同问题直接返回。但问题在于，用户的问题表述千变万化（例如“怎么退货？”和“如何申请售后？”），简单的字符串匹配缓存命中率极低，且无法处理语义相同但表述不同的问题。
2. 答案聚类：对同一问题的多次不同回答进行聚类，选择最大簇的答案作为最终输出。这听起来不错，但存在两个问题：一是聚类本身需要积累一定量的变异答案才能进行，有延迟；二是如果大模型前几次都跑偏了，就会形成一个错误的“主流答案”，导致系统持续犯错。
3. 动态温度系数：通过降低大模型生成时的“温度”（temperature）参数来减少随机性。这确实有效，但代价是可能让回答变得过于死板和模板化，丧失灵活性和创造性，对于需要多角度分析的客服问题并不友好。

综合来看，单一方法很难根治“随机性”问题。我们的思路是：从“检索”和“生成”两个环节同时入手，增加系统的确定性。

2. 核心优化方案：三层架构确保稳定

我们的核心方案围绕三个模块进行：分层检索、一致性校验和动态提示。

2.1 分层检索架构：确保找到最相关的知识

不稳定的答案往往源于检索阶段找到了不相关或次相关的文档片段。我们设计了一个三级过滤的混合检索器。

1. 第一层：业务规则与关键词过滤。首先，利用业务词典和正则规则，对用户问题进行意图分类和关键实体提取。例如，识别出“退货”、“订单12345”、“7天无理由”等。这一步可以快速过滤掉大量无关文档，并为后续检索提供强约束。
2. 第二层：稀疏向量检索（如BM25）。利用传统检索算法，基于关键词匹配快速召回一批相关文档。这一步召回率高，速度快。
3. 第三层：稠密向量检索（如Embedding模型）。使用语义向量模型，将用户问题和文档库转换为向量，进行相似度计算。这一步能捕捉语义相似性，弥补关键词匹配的不足。

最后，我们将第二层和第三层的检索结果进行加权融合重排。权重可以根据业务场景调整，例如，对于事实型问题，加大关键词权重；对于语义理解型问题，加大向量检索权重。

import logging
from typing import List, Dict
from rank_bm25 import BM25Okapi
import numpy as np
from sentence_transformers import SentenceTransformer

logger = logging.getLogger(__name__)

class HybridRetriever:
    def __init__(self, docs: List[str], embedding_model_name: str='paraphrase-multilingual-MiniLM-L12-v2'):
        """
        初始化混合检索器
        :param docs: 文档库列表
        :param embedding_model_name: 句子嵌入模型名称
        """
        self.docs = docs
        # 1. 初始化BM25
        tokenized_docs = [doc.split() for doc in docs]
        self.bm25 = BM25Okapi(tokenized_docs)
        # 2. 初始化稠密检索模型
        try:
            self.embedding_model = SentenceTransformer(embedding_model_name)
            self.doc_embeddings = self.embedding_model.encode(docs, show_progress_bar=False)
        except Exception as e:
            logger.error(f"加载嵌入模型失败: {e}")
            raise
        logger.info("混合检索器初始化完成。")

    def retrieve(self, query: str, top_k: int = 10, bm25_weight: float = 0.4, dense_weight: float = 0.6) -> List[Dict]:
        """
        混合检索主函数
        :param query: 用户查询
        :param top_k: 返回结果数量
        :param bm25_weight: BM25分数权重
        :param dense_weight: 向量相似度权重
        :return: 排序后的文档列表，包含文档和分数
        """
        if bm25_weight + dense_weight != 1.0:
            logger.warning("权重之和不为1，已自动归一化。")
            total = bm25_weight + dense_weight
            bm25_weight /= total
            dense_weight /= total

        # BM25检索
        tokenized_query = query.split()
        bm25_scores = self.bm25.get_scores(tokenized_query)
        bm25_scores = self._normalize_scores(bm25_scores)

        # 稠密检索
        try:
            query_embedding = self.embedding_model.encode([query])
            dense_scores = np.dot(query_embedding, self.doc_embeddings.T)[0]
            dense_scores = self._normalize_scores(dense_scores)
        except Exception as e:
            logger.error(f"稠密检索过程出错: {e}", exc_info=True)
            # 降级策略：仅使用BM25
            dense_scores = np.zeros_like(bm25_scores)

        # 加权融合
        combined_scores = bm25_weight * bm25_scores + dense_weight * dense_scores

        # 获取Top-K索引
        top_indices = np.argsort(combined_scores)[-top_k:][::-1]

        results = []
        for idx in top_indices:
            results.append({
                'doc': self.docs[idx],
                'score': combined_scores[idx],
                'bm25_score': bm25_scores[idx],
                'dense_score': dense_scores[idx]
            })
        logger.debug(f"查询 `{query}` 检索完成，返回 {len(results)} 个结果。")
        return results

    @staticmethod
    def _normalize_scores(scores: np.ndarray) -> np.ndarray:
        """归一化分数到[0,1]区间"""
        if np.max(scores) - np.min(scores) == 0:
            return np.ones_like(scores) * 0.5 # 防止除零
        return (scores - np.min(scores)) / (np.max(scores) - np.min(scores))

2.2 回答一致性校验模块：让答案“投票”决定

即使检索到了稳定相关的文档，大模型生成时仍可能产生波动。我们的策略是：让大模型对同一个问题生成多次（例如3次），然后选择一个“共识度”最高的答案。

具体步骤：

1. 利用优化后的检索结果，构造相同的Prompt，让大模型并行或快速串行生成N个候选答案。
2. 使用一个轻量级的句子嵌入模型（如BERT）计算所有候选答案两两之间的语义相似度。
3. 将候选答案视为节点，相似度视为边权重，构建一个图。然后寻找图中最“紧密”的簇。一种简单有效的方法是计算每个答案与其他答案的平均相似度，选择平均相似度最高的那个作为最终答案。这相当于一个“投票”机制，最主流的答案胜出。

from sentence_transformers import util
import numpy as np

def consistency_check(candidate_answers: List[str], threshold: float = 0.85) -> Dict:
    """
    回答一致性校验与选择
    :param candidate_answers: 候选答案列表
    :param threshold: 相似度阈值，用于判断是否达成一致
    :return: 包含最终答案和一致度信息的字典
    """
    if not candidate_answers:
        return {"final_answer": None, "consistency": 0.0, "method": "no_candidates"}

    if len(candidate_answers) == 1:
        return {"final_answer": candidate_answers[0], "consistency": 1.0, "method": "single"}

    # 计算所有候选答案的嵌入向量
    try:
        # 假设有一个全局的 embedding_model
        embeddings = embedding_model.encode(candidate_answers, convert_to_tensor=True)
        # 计算余弦相似度矩阵
        cos_sim_matrix = util.cos_sim(embeddings, embeddings).cpu().numpy()
    except Exception as e:
        logger.error(f"计算答案相似度时出错: {e}", exc_info=True)
        # 降级策略：返回第一个答案
        return {"final_answer": candidate_answers[0], "consistency": 0.0, "method": "fallback"}

    # 计算每个答案与其他答案的平均相似度（排除自身）
    avg_similarities = []
    for i in range(len(candidate_answers)):
        mask = np.ones(len(candidate_answers), dtype=bool)
        mask[i] = False
        avg_sim = cos_sim_matrix[i, mask].mean()
        avg_similarities.append(avg_sim)

    # 选择平均相似度最高的答案
    best_idx = np.argmax(avg_similarities)
    max_avg_sim = avg_similarities[best_idx]

    # 判断是否达成有效一致
    if max_avg_sim >= threshold:
        method = "consensus"
        final_answer = candidate_answers[best_idx]
    else:
        # 如果一致度不够高，可以结合其他策略，例如选择分数最高的检索片段对应的生成答案
        # 这里简化为选择第一个答案，并记录低一致度警告
        logger.warning(f"候选答案一致度过低: {max_avg_sim:.3f}，将返回第一个答案。")
        method = "low_consensus_fallback"
        final_answer = candidate_answers[0]

    return {
        "final_answer": final_answer,
        "consistency": float(max_avg_sim),
        "method": method,
        "all_scores": avg_similarities
    }

2.3 动态Prompt模板：给模型更明确的指令

Prompt的微小变化也会导致输出波动。我们引入了“问题类型指纹”到Prompt中。

1. 问题分类：在检索前或检索后，用一个轻量级文本分类模型对用户问题进行快速分类（如：事实查询、操作指南、故障排查、政策咨询等）。
2. 指纹注入：在构造大模型的Prompt时，不仅注入检索到的上下文，还注入分类结果。例如：“你是一个专业的客服助手。当前用户的问题是【操作指南】类问题。请严格根据以下资料，提供清晰、步骤化的操作指引：{context}。问题是：{query}”。
3. 结构化要求：在Prompt中明确要求答案的格式，如“分点列出”、“首先…然后…最后…”，这也能在一定程度上规范模型的输出，减少随机发散。

3. 性能考量与权衡

稳定性提升必然会引入额外计算开销，主要来自：

• N次生成：一致性校验需要生成多个候选答案，耗时线性增加。
• 相似度计算：计算N个答案的两两相似度，复杂度为O(N²)，但N通常很小（3-5），且使用轻量模型，开销可控。
• 混合检索：相比单一检索，需要运行两套算法并进行融合排序。

我们的Trade-off策略如下表所示：

QPS场景	推荐配置	预期延迟增加	稳定性目标
低QPS (<10)	开启三级检索，N=3次生成校验	200-400ms	高稳定性，追求最优答案
中QPS (10-50)	开启混合检索，N=2次生成校验	100-250ms	平衡稳定性与响应速度
高QPS (>50)	优先使用向量检索，缓存一致性结果，N=1（降级）	<100ms	保障可用性，对高频问题保持稳定

建议：对于高频问题，可以将“问题指纹+检索片段”的哈希值作为键，将最终一致性校验后的答案进行缓存，可以极大缓解高并发下的压力。

4. 实践避坑指南

1. 避免过度依赖单一向量库：向量模型可能存在领域不匹配问题。混合检索（关键词+向量）比单一向量检索更鲁棒。定期用业务数据微调嵌入模型也是提升效果的关键。
2. 冷启动阶段的数据预热：系统上线初期，知识库文档少，检索效果差。可以预先构造一个“高频问题-标准答案”对的小型缓存库，在检索结果置信度低时优先使用。同时，建立未命中问题的收集机制，快速扩充知识库。
3. 对话上下文的正确处理：RAG系统常被用于多轮对话。简单的做法是将整个对话历史作为当前查询的上下文。但这会引入大量噪音。更好的做法是：a) 使用查询重写技术，将指代（如“它”、“上面说的”）显式化；b) 仅将最近1-2轮有效对话与当前问题拼接进行检索。

5. 延伸思考：建立持续优化闭环

系统上线不是终点。我们设计了一个基于用户反馈的优化机制：

1. 隐式反馈收集：在客服界面设置“有帮助/无帮助”按钮。将“无帮助”的问答对自动收集到待审核池。
2. 根因分析：对低评分答案进行分析，是检索失败（记录查询和返回的文档片段），还是生成失败（记录查询、上下文和生成的糟糕答案）。
3. 知识库迭代：对于检索失败案例，考虑将用户问题或修正后的标准答案添加到知识库中，或优化相关文档的表述。
4. Prompt模板迭代：对于生成失败案例，分析问题类型，优化对应类别Prompt模板的指令。
5. 模型迭代：定期用积累的高质量（用户认可）的问答对，对大模型进行轻量级的指令微调（Instruction Tuning），让其更贴合客服场景的说话风格和格式。

通过以上这套组合拳，我们成功将智能客服回答的波动性大幅降低，从“随机发挥”变成了“稳定输出”。整个优化过程让我深刻体会到，构建可靠的AI应用，光有强大的基座模型还不够，更需要精心设计围绕它的“确定性工程”架构。希望这些实战经验能给大家带来一些启发。