Chatbot RAG检索增强生成实战：从提示词设计到避坑指南

1次阅读

共计 2111 个字符，预计需要花费 6 分钟才能阅读完成。

RAG（Retrieval-Augmented Generation）是结合检索系统和生成模型的技术框架。简单来说，当用户提问时，系统会先从一个知识库中检索相关文档片段，然后将这些片段和问题一起输入生成模型，最终得到回答。这就像你写论文时先查资料再动笔，比凭空编造更靠谱。

提示词在这里扮演着 ” 交通指挥员 ” 的角色：

检索阶段 ：提示词影响搜索关键词的 embedding 生成，决定能召回哪些内容
生成阶段 ：提示词组织检索结果的呈现方式，告诉模型 ” 怎么用这些材料 ”

# 反面教材
prompt = "回答这个问题：" + user_question

后果：模型可能直接忽略检索结果，凭记忆生成错误答案

# 反面教材
prompt = "参考以下文档：" + "\n".join(retrieved_docs[:5])

后果：前 3 篇文档就够用，后 2 篇反而引入噪声

# 反面教材
prompt = "根据资料自由发挥：" + context

后果：可能泄露敏感信息或生成不符合业务要求的内容

def build_prompt(question, contexts):
    return f""" 请严格根据以下信息用中文回答，若资料不足请说明：问题：{question}

    参考材料：{"\n\n".join([f"[{i+1}] {c}" for i,c in enumerate(contexts)])}

    回答要求：- 不超过 3 句话
    - 标注引用来源如 [1][2]
    - 避免主观推测
    """

from sklearn.feature_extraction.text import CountVectorizer

def filter_redundant(docs, max_n=3):
    vectorizer = CountVectorizer().fit(docs)
    X = vectorizer.transform(docs)
    # 计算文档相似度矩阵
    similarity = (X * X.T).A 

    selected_indices = []
    for i in range(len(docs)):
        if not any(similarity[i][j] > 0.7 for j in selected_indices):
            selected_indices.append(i)
            if len(selected_indices) >= max_n:
                break
    return [docs[i] for i in selected_indices]

class ConversationManager:
    def __init__(self):
        self.history = []

    def add_question(self, question, contexts, answer):
        self.history.append({
            "q": question,
            "c": contexts,
            "a": answer
        })

    def current_prompt(self, new_question, new_contexts):
        history_part = "\n".join(f"Q:{item['q']}\nA:{item['a']}" 
            for item in self.history[-2:]  # 只保留最近 2 轮
        )
        return f""" 当前对话历史：{history_part}

        新问题：{new_question}

        参考资料：{"\n".join(new_contexts)}
        """

实体提取法 ：

import spacy

nlp = spacy.load("zh_core_web_sm")
def extract_entities(text):
    return " ".join([ent.text for ent in nlp(text).ents])

摘要生成法 ：

from transformers import pipeline

summarizer = pipeline("summarization", model="Falconsai/zh_summarization")
def zh_summarize(text, max_len=100):
    return summarizer(text, max_length=max_len)[0]["summary_text"]