RAG系统构建¶

⚠️ 时效性说明：本章涉及前沿模型/价格/榜单等信息，可能随版本快速变化；请以论文原文、官方发布页和 API 文档为准。

📌 定位说明：本章侧重RAG系统的工程基础与构建实践。 - 📖 RAG的NLP理论基础（检索原理、Embedding技术、评估方法）请参考自然语言处理/14-RAG系统设计 - 📖 RAG研究前沿与长文本技术请参考 LLM学习/04-前沿探索/03-RAG与长文本 - 📖 高级RAG实战技术请参考 LLM应用/18-高级RAG技术

📖 章节导读¶

检索增强生成(Retrieval-Augmented Generation, RAG)是一种结合检索和生成的技术,通过从外部知识库中检索相关信息来增强大模型的生成能力。本章将深入探讨RAG系统的构建方法、优化技巧和实战应用。

🎯 学习目标¶

理解RAG系统的核心原理
掌握RAG系统的构建方法
学会优化RAG系统的性能
了解RAG系统的应用场景
掌握大厂面试中的相关问题

5.1 RAG系统概述¶

5.1.1 什么是RAG¶

定义:RAG(Retrieval-Augmented Generation)是一种结合信息检索和文本生成的技术,通过从外部知识库中检索相关信息来增强大模型的生成能力。

核心思想:

检索阶段:根据用户查询从知识库中检索相关文档
增强阶段:将检索到的文档作为上下文
生成阶段:基于上下文生成回答

为什么需要RAG:

知识更新:大模型的知识截止于训练时间,RAG可以获取最新信息
领域知识:可以集成特定领域的专业知识
减少幻觉:基于检索到的文档生成,减少模型幻觉
可解释性:可以引用检索到的文档,提高可解释性
成本效益:相比微调,RAG的成本更低

5.1.2 RAG vs 微调¶

对比分析:

特性	RAG	微调
知识更新	容易,只需更新知识库	困难,需要重新训练
领域适应	快速,添加领域文档	需要领域数据训练
成本	较低	较高
实现难度	中等	较高
幻觉问题	较少	仍然存在
可解释性	高	低
适用场景	知识密集型任务	特定任务优化

选择建议:

选择RAG:需要最新知识、领域知识、可解释性
选择微调:特定任务优化、风格调整、私有数据

5.1.3 RAG系统架构¶

基本架构:

Text Only

用户查询 → 查询处理 → 向量检索 → 文档排序 → 上下文构建 → 生成回答

详细流程:

查询处理:
查询重写
查询扩展
查询理解
向量检索:
文档向量化
相似度计算
Top-k检索
文档排序:
重排序
相关性评分
多样性选择
上下文构建:
文档拼接
上下文窗口管理
信息筛选
生成回答:
Prompt构建
模型生成
后处理

5.2 RAG系统构建¶

5.2.1 基础RAG实现¶

实现步骤:

准备知识库
文档向量化
构建向量索引
实现检索功能
构建生成Prompt
生成回答

代码实现:

Python

from openai import OpenAI
from sentence_transformers import SentenceTransformer
import numpy as np

class BasicRAG:
    """基础RAG系统"""

    def __init__(self, documents: list[str], model_name='all-MiniLM-L6-v2'):
        """
        初始化RAG系统

        Args:
            documents: 文档列表
            model_name: 嵌入模型名称
        """
        self.documents = documents
        self.embedding_model = SentenceTransformer(model_name)
        self.client = OpenAI(api_key="your-api-key")

        # 向量化文档
        self.document_embeddings = self.embedding_model.encode(documents)

    def retrieve(self, query: str, top_k: int = 3) -> list[dict]:
        """
        检索相关文档

        Args:
            query: 用户查询
            top_k: 返回的文档数量

        Returns:
            检索到的文档列表
        """
        # 向量化查询
        query_embedding = self.embedding_model.encode([query])

        # 计算相似度
        similarities = np.dot(self.document_embeddings, query_embedding.T).flatten()

        # 获取top-k
        top_indices = np.argsort(similarities)[-top_k:][::-1]

        # 返回结果
        results = []
        for idx in top_indices:
            results.append({
                'document': self.documents[idx],
                'similarity': float(similarities[idx]),
                'index': int(idx)
            })

        return results

    def generate(self, query: str, context: str) -> str:
        """
        生成回答

        Args:
            query: 用户查询
            context: 检索到的上下文

        Returns:
            生成的回答
        """
        prompt = f"""
你是一个专业的问答助手。请根据以下上下文回答用户的问题。

上下文:
{context}

问题: {query}

请基于上下文回答问题,如果上下文中没有相关信息,请明确说明。
"""

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "你是一个专业的问答助手。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3
        )

        return response.choices[0].message.content

    def query(self, query: str, top_k: int = 3) -> dict:
        """
        完整的RAG查询

        Args:
            query: 用户查询
            top_k: 检索的文档数量

        Returns:
            包含检索结果和生成的回答
        """
        # 检索相关文档
        retrieved_docs = self.retrieve(query, top_k)

        # 构建上下文
        context = "\n\n".join([
            f"文档{i+1}:\n{doc['document']}"
            for i, doc in enumerate(retrieved_docs)  # enumerate同时获取索引和元素
        ])

        # 生成回答
        answer = self.generate(query, context)

        return {
            'query': query,
            'retrieved_documents': retrieved_docs,
            'context': context,
            'answer': answer
        }

# 使用示例
documents = [
    "Python是一种高级编程语言,由Guido van Rossum于1991年创建。",
    "机器学习是人工智能的一个分支,它使计算机能够从数据中学习。",
    "深度学习是机器学习的一个子集,使用神经网络进行学习。",
    "Transformer是一种深度学习模型架构,由Google在2017年提出。",
    "大语言模型是基于Transformer架构的预训练模型,如GPT、BERT等。"
]

# 创建RAG系统
rag = BasicRAG(documents)

# 查询
query = "什么是Transformer?"
result = rag.query(query, top_k=2)

print(f"问题: {result['query']}\n")
print("检索到的文档:")
for i, doc in enumerate(result['retrieved_documents'], 1):
    print(f"\n文档{i} (相似度: {doc['similarity']:.4f}):")
    print(doc['document'])

print(f"\n回答:\n{result['answer']}")

5.2.2 高级RAG实现¶

增强功能:

查询重写:优化查询以提高检索质量
混合检索:结合向量检索和关键词检索
重排序:对检索结果进行重排序
上下文压缩:压缩上下文以适应模型窗口

代码实现:

Python

from rank_bm25 import BM25Okapi
import jieba

class AdvancedRAG(BasicRAG):
    """高级RAG系统"""

    def __init__(self, documents: list[str], model_name='all-MiniLM-L6-v2'):
        super().__init__(documents, model_name)  # super()调用父类方法

        # 构建BM25索引
        tokenized_docs = [list(jieba.cut(doc)) for doc in documents]
        self.bm25 = BM25Okapi(tokenized_docs)

    def hybrid_retrieve(self, query: str, top_k: int = 3, alpha: float = 0.5) -> list[dict]:
        """
        混合检索(向量检索 + BM25)

        Args:
            query: 用户查询
            top_k: 返回的文档数量
            alpha: 向量检索的权重(0-1)

        Returns:
            检索到的文档列表
        """
        # 向量检索
        query_embedding = self.embedding_model.encode([query])
        vector_similarities = np.dot(self.document_embeddings, query_embedding.T).flatten()

        # BM25检索
        tokenized_query = list(jieba.cut(query))
        bm25_scores = self.bm25.get_scores(tokenized_query)

        # 归一化
        vector_similarities = (vector_similarities - vector_similarities.min()) / \
                            (vector_similarities.max() - vector_similarities.min() + 1e-8)
        bm25_scores = (bm25_scores - bm25_scores.min()) / \
                     (bm25_scores.max() - bm25_scores.min() + 1e-8)

        # 混合评分
        combined_scores = alpha * vector_similarities + (1 - alpha) * bm25_scores

        # 获取top-k
        top_indices = np.argsort(combined_scores)[-top_k:][::-1]

        # 返回结果
        results = []
        for idx in top_indices:
            results.append({
                'document': self.documents[idx],
                'vector_similarity': float(vector_similarities[idx]),
                'bm25_score': float(bm25_scores[idx]),
                'combined_score': float(combined_scores[idx]),
                'index': int(idx)
            })

        return results

    def rerank(self, query: str, documents: list[dict], top_k: int = 3) -> list[dict]:
        """
        重排序检索结果

        Args:
            query: 用户查询
            documents: 检索到的文档
            top_k: 返回的文档数量

        Returns:
            重排序后的文档列表
        """
        # 构建重排序Prompt
        prompt = f"""
请根据以下问题对文档进行相关性评分(1-10分)。

问题: {query}

文档:
"""

        for i, doc in enumerate(documents, 1):
            prompt += f"\n文档{i}:\n{doc['document']}\n"

        prompt += """
请为每个文档打分,并说明理由。格式如下:
文档1: 分数 - 理由
文档2: 分数 - 理由
...
"""

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "你是一个文档相关性评估专家。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3
        )

        # 解析评分
        response_text = response.choices[0].message.content
        import re
        scores = re.findall(r'文档(\d+):\s*(\d+)', response_text)  # re.findall正则查找所有匹配项

        # 更新分数
        for doc_idx, score in scores:
            idx = int(doc_idx) - 1
            if idx < len(documents):
                documents[idx]['rerank_score'] = int(score)

        # 按重排序分数排序
        reranked = sorted(
            documents,
            key=lambda x: x.get('rerank_score', 0),  # lambda匿名函数
            reverse=True
        )

        return reranked[:top_k]

    def query_advanced(self, query: str, top_k: int = 3, use_rerank: bool = True) -> dict:
        """
        高级RAG查询

        Args:
            query: 用户查询
            top_k: 检索的文档数量
            use_rerank: 是否使用重排序

        Returns:
            包含检索结果和生成的回答
        """
        # 混合检索
        retrieved_docs = self.hybrid_retrieve(query, top_k * 2)

        # 重排序
        if use_rerank:
            retrieved_docs = self.rerank(query, retrieved_docs, top_k)

        # 构建上下文
        context = "\n\n".join([
            f"文档{i+1}:\n{doc['document']}"
            for i, doc in enumerate(retrieved_docs)
        ])

        # 生成回答
        answer = self.generate(query, context)

        return {
            'query': query,
            'retrieved_documents': retrieved_docs,
            'context': context,
            'answer': answer
        }

# 使用示例
advanced_rag = AdvancedRAG(documents)

# 查询
query = "什么是Transformer?"
result = advanced_rag.query_advanced(query, top_k=2, use_rerank=True)

print(f"问题: {result['query']}\n")
print("检索到的文档:")
for i, doc in enumerate(result['retrieved_documents'], 1):
    print(f"\n文档{i}:")
    print(f"内容: {doc['document']}")
    print(f"向量相似度: {doc.get('vector_similarity', 0):.4f}")
    print(f"BM25分数: {doc.get('bm25_score', 0):.4f}")
    print(f"重排序分数: {doc.get('rerank_score', 0)}")

print(f"\n回答:\n{result['answer']}")

5.2.3 使用LangChain构建RAG¶

安装依赖:

Bash

pip install langchain langchain-openai langchain-community chromadb

代码实现:

Python

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate

class LangChainRAG:
    """使用LangChain构建的RAG系统"""

    def __init__(self, documents: list[str], api_key: str):
        """
        初始化RAG系统

        Args:
            documents: 文档列表
            api_key: OpenAI API密钥
        """
        self.documents = documents
        self.api_key = api_key

        # 初始化组件
        self.embeddings = OpenAIEmbeddings(openai_api_key=api_key)
        self.llm = ChatOpenAI(
            model_name="gpt-4o",
            temperature=0.3,
            openai_api_key=api_key
        )

        # 构建向量存储
        self.build_vectorstore()

    def build_vectorstore(self):
        """构建向量存储"""
        # 创建文档对象
        from langchain_core.documents import Document
        docs = [Document(page_content=doc) for doc in self.documents]

        # 文本分割
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200
        )
        splits = text_splitter.split_documents(docs)

        # 创建向量存储
        self.vectorstore = Chroma.from_documents(
            documents=splits,
            embedding=self.embeddings
        )

        # 创建检索器
        self.retriever = self.vectorstore.as_retriever(
            search_kwargs={"k": 3}
        )

    def create_chain(self):
        """创建RAG链（LCEL方式，替代已废弃的 RetrievalQA）"""
        system_prompt = """你是一个专业的问答助手。请根据以下上下文回答用户的问题。
如果上下文中没有相关信息，请明确说明。

{context}"""

        prompt = ChatPromptTemplate.from_messages([
            ("system", system_prompt),
            ("human", "{input}")
        ])

        question_answer_chain = create_stuff_documents_chain(self.llm, prompt)
        self.chain = create_retrieval_chain(self.retriever, question_answer_chain)

    def query(self, question: str) -> dict:
        """
        查询

        Args:
            question: 用户问题

        Returns:
            查询结果
        """
        if not hasattr(self, 'chain'):  # hasattr检查对象是否有某属性
            self.create_chain()

        # 执行查询
        result = self.chain.invoke({"input": question})

        return {
            'question': question,
            'answer': result['answer'],
            'source_documents': result['context']
        }

# 使用示例
api_key = "your-api-key"
langchain_rag = LangChainRAG(documents, api_key)

# 查询
question = "什么是Transformer?"
result = langchain_rag.query(question)

print(f"问题: {result['question']}\n")
print(f"回答:\n{result['answer']}\n")
print("来源文档:")
for i, doc in enumerate(result['source_documents'], 1):
    print(f"\n文档{i}:")
    print(doc.page_content)

5.3 RAG系统优化¶

5.3.1 检索优化¶

优化策略:

查询优化:
查询重写
查询扩展
查询理解
索引优化:
选择合适的嵌入模型
优化向量索引
使用混合检索
检索策略:
调整top-k值
使用重排序
多样性选择

代码实现:查询重写

Python

class QueryRewriter:
    """查询重写器"""

    def __init__(self, client):
        self.client = client

    def rewrite(self, query: str, method: str = "expand") -> str:
        """
        重写查询

        Args:
            query: 原始查询
            method: 重写方法(expand/clarify/simplify)

        Returns:
            重写后的查询
        """
        if method == "expand":
            prompt = f"""
请扩展以下查询,使其更加详细和具体,以提高检索质量。

原始查询: {query}

请提供扩展后的查询:
"""
        elif method == "clarify":
            prompt = f"""
请澄清以下查询,使其更加明确和具体。

原始查询: {query}

请提供澄清后的查询:
"""
        else:  # simplify
            prompt = f"""
请简化以下查询,提取核心关键词。

原始查询: {query}

请提供简化后的查询:
"""

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3
        )

        return response.choices[0].message.content

# 使用示例
client = OpenAI(api_key="your-api-key")
rewriter = QueryRewriter(client)

query = "机器学习"
rewritten_query = rewriter.rewrite(query, method="expand")
print(f"原始查询: {query}")
print(f"扩展查询: {rewritten_query}")

5.3.2 生成优化¶

优化策略:

Prompt优化:
优化上下文组织
添加指令引导
使用思维链
模型选择:
选择合适的模型
调整温度参数
使用模型组合
后处理:
答案验证
幻觉检测
答案优化

代码实现:幻觉检测

Python

class HallucinationDetector:
    """幻觉检测器"""

    def __init__(self, client):
        self.client = client

    def detect(self, answer: str, context: str) -> dict:
        """
        检测回答中的幻觉

        Args:
            answer: 生成的回答
            context: 上下文

        Returns:
            检测结果
        """
        prompt = f"""
请检查以下回答是否基于给定的上下文,并指出任何可能的幻觉。

上下文:
{context}

回答:
{answer}

请按照以下格式输出:
1. 回答是否基于上下文: [是/否]
2. 潜在的幻觉: [列出可能的幻觉]
3. 置信度: [高/中/低]
"""

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3
        )

        result_text = response.choices[0].message.content

        # 解析结果
        import re
        is_based = "是" in result_text
        hallucinations = re.findall(r'潜在的幻觉:\s*(.+)', result_text)

        return {
            'is_based_on_context': is_based,
            'hallucinations': hallucinations[0] if hallucinations else None,
            'full_analysis': result_text
        }

# 使用示例
detector = HallucinationDetector(client)

answer = "Transformer是一种深度学习模型架构,由Google在2017年提出,主要用于自然语言处理任务。"
context = "Transformer是一种深度学习模型架构,由Google在2017年提出。"

result = detector.detect(answer, context)
print(result)

5.3.3 性能优化¶

优化策略:

缓存优化:
缓存查询结果
缓存向量嵌入
使用Redis等缓存系统
并行处理:
并行检索
批量处理
异步处理
模型优化:
使用量化模型
模型蒸馏
模型剪枝

代码实现:缓存优化

Python

from functools import lru_cache
import hashlib
import pickle

class CachedRAG(BasicRAG):
    """带缓存的RAG系统"""

    def __init__(self, documents: list[str], model_name='all-MiniLM-L6-v2'):
        super().__init__(documents, model_name)
        self.cache = {}

    def _get_cache_key(self, query: str, top_k: int) -> str:
        """生成缓存键"""
        key_str = f"{query}_{top_k}"
        return hashlib.md5(key_str.encode()).hexdigest()

    def retrieve_with_cache(self, query: str, top_k: int = 3) -> list[dict]:
        """带缓存的检索"""
        cache_key = self._get_cache_key(query, top_k)

        # 检查缓存
        if cache_key in self.cache:
            print("从缓存获取结果")
            return self.cache[cache_key]

        # 执行检索
        results = self.retrieve(query, top_k)

        # 存入缓存
        self.cache[cache_key] = results

        return results

    def query_with_cache(self, query: str, top_k: int = 3) -> dict:
        """带缓存的查询"""
        # 检索相关文档(带缓存)
        retrieved_docs = self.retrieve_with_cache(query, top_k)

        # 构建上下文
        context = "\n\n".join([
            f"文档{i+1}:\n{doc['document']}"
            for i, doc in enumerate(retrieved_docs)
        ])

        # 生成回答
        answer = self.generate(query, context)

        return {
            'query': query,
            'retrieved_documents': retrieved_docs,
            'context': context,
            'answer': answer
        }

# 使用示例
cached_rag = CachedRAG(documents)

# 第一次查询
query = "什么是Transformer?"
result1 = cached_rag.query_with_cache(query)
print(f"第一次查询:\n{result1['answer']}\n")

# 第二次查询(从缓存)
result2 = cached_rag.query_with_cache(query)
print(f"第二次查询:\n{result2['answer']}")

5.4 实战案例¶

5.4.1 案例1:文档问答系统¶

需求:构建一个基于企业文档的问答系统。

实现:

Python

class DocumentQA:
    """文档问答系统"""

    def __init__(self, document_path: str, api_key: str):
        """
        初始化文档问答系统

        Args:
            document_path: 文档路径
            api_key: OpenAI API密钥
        """
        self.api_key = api_key
        self.client = OpenAI(api_key=api_key)

        # 加载文档
        self.documents = self.load_documents(document_path)

        # 构建RAG系统
        self.rag = AdvancedRAG(self.documents)

    def load_documents(self, path: str) -> list[str]:
        """加载文档"""
        # 这里简化处理,实际可以从文件加载
        with open(path, 'r', encoding='utf-8') as f:  # with自动管理文件关闭
            content = f.read()

        # 分割文档
        # 这里简化处理,实际可以使用更复杂的分割策略
        paragraphs = content.split('\n\n')
        return [p.strip() for p in paragraphs if p.strip()]  # 链式调用：strip去除空白

    def ask(self, question: str, top_k: int = 3) -> dict:
        """
        提问

        Args:
            question: 问题
            top_k: 检索的文档数量

        Returns:
            回答和相关信息
        """
        result = self.rag.query_advanced(question, top_k=top_k)

        return {
            'question': question,
            'answer': result['answer'],
            'sources': result['retrieved_documents']
        }

# 使用示例
# 假设有一个文档文件
# document_qa = DocumentQA("company_docs.txt", "your-api-key")
# result = document_qa.ask("公司的核心价值观是什么?")
# print(result['answer'])

5.4.2 案例2:知识库问答¶

需求:构建一个基于知识库的问答系统。

实现:

Python

class KnowledgeBaseQA:
    """知识库问答系统"""

    def __init__(self, knowledge_base: dict[str, str], api_key: str):
        """
        初始化知识库问答系统

        Args:
            knowledge_base: 知识库字典 {主题: 内容}
            api_key: OpenAI API密钥
        """
        self.knowledge_base = knowledge_base
        self.api_key = api_key
        self.client = OpenAI(api_key=api_key)

        # 构建RAG系统
        documents = list(knowledge_base.values())
        self.rag = AdvancedRAG(documents)

    def ask(self, question: str, top_k: int = 3) -> dict:
        """
        提问

        Args:
            question: 问题
            top_k: 检索的文档数量

        Returns:
            回答和相关信息
        """
        result = self.rag.query_advanced(question, top_k=top_k)

        # 查找相关主题
        relevant_topics = []
        for doc in result['retrieved_documents']:
            for topic, content in self.knowledge_base.items():
                if content == doc['document']:
                    relevant_topics.append(topic)
                    break

        return {
            'question': question,
            'answer': result['answer'],
            'relevant_topics': relevant_topics,
            'sources': result['retrieved_documents']
        }

# 使用示例
knowledge_base = {
    "Python": "Python是一种高级编程语言,由Guido van Rossum于1991年创建。",
    "机器学习": "机器学习是人工智能的一个分支,它使计算机能够从数据中学习。",
    "深度学习": "深度学习是机器学习的一个子集,使用神经网络进行学习。",
    "Transformer": "Transformer是一种深度学习模型架构,由Google在2017年提出。",
    "大语言模型": "大语言模型是基于Transformer架构的预训练模型,如GPT、BERT等。"
}

kb_qa = KnowledgeBaseQA(knowledge_base, "your-api-key")
result = kb_qa.ask("什么是Transformer?")
print(f"回答: {result['answer']}")
print(f"相关主题: {result['relevant_topics']}")

5.4.3 案例3:代码助手¶

需求:构建一个基于代码文档的代码助手。

实现:

Python

class CodeAssistant:
    """代码助手"""

    def __init__(self, code_docs: dict[str, str], api_key: str):
        """
        初始化代码助手

        Args:
            code_docs: 代码文档字典 {函数名: 文档}
            api_key: OpenAI API密钥
        """
        self.code_docs = code_docs
        self.api_key = api_key
        self.client = OpenAI(api_key=api_key)

        # 构建RAG系统
        documents = list(code_docs.values())
        self.rag = AdvancedRAG(documents)

    def ask(self, question: str, top_k: int = 3) -> dict:
        """
        提问

        Args:
            question: 问题
            top_k: 检索的文档数量

        Returns:
            回答和相关信息
        """
        result = self.rag.query_advanced(question, top_k=top_k)

        # 查找相关函数
        relevant_functions = []
        for doc in result['retrieved_documents']:
            for func_name, doc_content in self.code_docs.items():
                if doc_content == doc['document']:
                    relevant_functions.append(func_name)
                    break

        return {
            'question': question,
            'answer': result['answer'],
            'relevant_functions': relevant_functions,
            'sources': result['retrieved_documents']
        }

# 使用示例
code_docs = {
    "factorial": """factorial函数用于计算阶乘。
参数: n - 一个非负整数
返回: n的阶乘
示例: factorial(5) 返回 120
""",
    "fibonacci": """fibonacci函数用于计算斐波那契数列的第n项。
参数: n - 一个正整数
返回: 斐波那契数列的第n项
示例: fibonacci(10) 返回 55
""",
    "is_prime": """is_prime函数用于判断一个数是否为质数。
参数: n - 一个正整数
返回: 如果是质数返回True,否则返回False
示例: is_prime(17) 返回 True
"""
}

code_assistant = CodeAssistant(code_docs, "your-api-key")
result = code_assistant.ask("如何计算阶乘?")
print(f"回答: {result['answer']}")
print(f"相关函数: {result['relevant_functions']}")

5.5 练习题¶

练习题1:基础RAG¶

题目:实现一个基础的RAG系统,能够从文档中检索相关信息并回答问题。

参考答案:

Python

class BasicRAG:
    """基础RAG系统"""

    def __init__(self, documents: list[str]):
        self.documents = documents
        self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
        self.document_embeddings = self.embedding_model.encode(documents)

    def retrieve(self, query: str, top_k: int = 3) -> list[dict]:
        query_embedding = self.embedding_model.encode([query])
        similarities = np.dot(self.document_embeddings, query_embedding.T).flatten()
        top_indices = np.argsort(similarities)[-top_k:][::-1]

        results = []
        for idx in top_indices:
            results.append({
                'document': self.documents[idx],
                'similarity': float(similarities[idx])
            })
        return results

练习题2:混合检索¶

题目:实现混合检索,结合向量检索和BM25检索。

参考答案:

Python

def hybrid_retrieve(self, query: str, top_k: int = 3, alpha: float = 0.5) -> list[dict]:
    # 向量检索
    query_embedding = self.embedding_model.encode([query])
    vector_similarities = np.dot(self.document_embeddings, query_embedding.T).flatten()

    # BM25检索
    tokenized_query = list(jieba.cut(query))
    bm25_scores = self.bm25.get_scores(tokenized_query)

    # 归一化
    vector_similarities = (vector_similarities - vector_similarities.min()) / \
                        (vector_similarities.max() - vector_similarities.min() + 1e-8)
    bm25_scores = (bm25_scores - bm25_scores.min()) / \
                 (bm25_scores.max() - bm25_scores.min() + 1e-8)

    # 混合评分
    combined_scores = alpha * vector_similarities + (1 - alpha) * bm25_scores

    # 获取top-k
    top_indices = np.argsort(combined_scores)[-top_k:][::-1]

    return [{'document': self.documents[idx], 'score': float(combined_scores[idx])}
            for idx in top_indices]

5.6 面试准备¶

5.6.1 大厂面试题¶

字节跳动面试题:

问题:什么是RAG?它有什么优势?

参考答案: - RAG是检索增强生成,结合检索和生成 - 优势: - 知识更新容易 - 可以集成领域知识 - 减少幻觉 - 可解释性高 - 成本效益好

问题:RAG和微调有什么区别?

参考答案: - RAG:检索外部知识,无需训练 - 微调:训练模型适应特定任务 - 区别: - 知识更新:RAG容易,微调困难 - 成本:RAG低,微调高 - 可解释性:RAG高,微调低

腾讯面试题:

问题:如何优化RAG系统的检索质量?

参考答案: - 查询优化:查询重写、查询扩展 - 索引优化:选择合适的嵌入模型 - 检索策略:混合检索、重排序 - 评估反馈:根据反馈持续优化

问题:RAG系统有哪些局限性?

参考答案: - 依赖检索质量 - 上下文窗口限制 - 检索延迟 - 知识库维护成本

阿里巴巴面试题:

问题:在实际项目中如何应用RAG?

参考答案: - 需求分析:明确应用场景 - 知识库构建:收集和整理文档 - 系统设计:设计RAG架构 - 实现开发:实现检索和生成 - 测试优化:测试和优化性能 - 部署监控:部署并持续监控

问题:如何评估RAG系统的效果?

参考答案: - 检索质量:准确率、召回率 - 生成质量:相关性、准确性 - 系统性能:响应时间、吞吐量 - 用户满意度:用户反馈

5.6.2 面试技巧¶

技巧1:理论联系实际

结合实际项目经验,说明如何应用RAG解决实际问题。

技巧2:对比分析

对比RAG和微调的优缺点,说明选择依据。

技巧3:展示思考过程

说明设计RAG系统的思考过程,展示分析能力。

技巧4:持续优化

说明如何通过迭代优化不断提升RAG系统性能。

📝 本章小结¶

本章系统介绍了RAG系统构建的核心内容:

✅ RAG系统概述:定义、与微调对比、架构
✅ RAG系统构建:基础实现、高级实现、LangChain实现
✅ RAG系统优化:检索优化、生成优化、性能优化
✅ 实战案例:文档问答、知识库问答、代码助手
✅ 练习题:基础RAG、混合检索
✅ 面试准备:大厂面试题和解答技巧

通过本章学习,你应该能够: - 理解RAG系统的核心原理 - 掌握RAG系统的构建方法 - 学会优化RAG系统的性能 - 了解RAG系统的应用场景 - 准备好应对大厂面试

🔗 下一步¶

下一章我们将深入学习向量数据库,掌握向量存储和检索的核心技术。

继续学习: 06-向量数据库.md

💡 思考题¶

RAG系统的核心原理是什么?

检索增强生成：先从外部知识库检索与Query相关的文档片段，再将检索结果作为上下文拼入Prompt，让LLM基于上下文生成回答。流程：Query→Embedding→向量检索→Rerank→拼Context→LLM生成。
RAG和微调各有什么优缺点?

RAG：知识可实时更新、无需训练、可追溯来源、成本低，但受检索质量制约、上下文窗口有限。微调：深度学习领域知识、输出风格可控，但需要训练数据和GPU、知识不易更新。实践中两者常结合使用。
如何优化RAG系统的检索质量?

①语义分块(优于固定长度) ②选好Embedding模型(BGE/GTE) ③混合检索(向量+BM25) ④查询改写/扩展 ⑤Cross-Encoder重排序 ⑥多路召回+融合。核心指标：Recall@K和MRR。
RAG系统有哪些局限性?

①检索失败→回答错误 ②多跳推理能力弱 ③长文档受上下文窗口限制 ④实时性取决于索引更新频率 ⑤复杂查询理解不足 ⑥幻觉无法完全消除。
在实际项目中如何应用RAG?

场景：企业知识库、客服系统、文档助手、代码问答。技术栈：LangChain/LlamaIndex + Milvus/Chroma + BGE Embedding + Reranker。生产关注：分块调优、Embedding监控、缓存机制、异常回退。

📚 参考资料¶

"Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks" - Lewis et al.
"Dense Passage Retrieval for Open-Domain Question Answering" - Karpukhin et al.
LangChain Documentation
Chroma Documentation
Pinecone Documentation

最后更新日期：2026-02-12 适用版本：LLM应用指南 v2026