RAG：赋能大模型，打造更智能的AI助手

在人工智能领域，RAG(Retrieval-Augmented Generation，检索增强生成)正逐渐成为提升大模型智能水平的关键技术。传统的大模型依赖于训练数据，往往存在信息滞后和“幻觉”问题。而RAG通过引入外部知识库，赋予大模型实时检索信息的能力，使其能够生成更准确、更可靠的答案。本文将深入探讨RAG的工作原理、优势，以及如何利用现代工具构建自己的RAG系统。

1. RAG 的核心概念与优势

RAG 的核心在于“检索增强”。它不再让大模型完全依赖自身记忆，而是在生成答案之前，先从外部知识库中检索相关信息。想象一下，当你被问到一个复杂问题时，你会先搜索相关资料，然后结合自己的知识进行解答。RAG 正是模仿了这一过程，它将检索(Retrieval)和生成(Generation)两个阶段结合起来，从而提升大模型的性能。

例如，一个保险公司的客户向聊天机器人询问：“我的旅行保险包含哪些医疗紧急情况？” 传统的AI可能会给出模糊或不准确的答案。而RAG加持的聊天机器人可以：

将问题转化为搜索查询。
从客户的具体保单文件中检索相关文本。
利用检索到的上下文生成清晰准确的答案。

RAG 的主要优势在于：

提高准确性： 通过检索最新的信息，避免大模型生成过时或错误的内容。
减少“幻觉”： 基于可靠的外部知识，降低大模型编造信息的可能性。
增强可解释性： 提供答案的来源，方便用户验证信息的真实性。
适应性强： 无需重新训练大模型，即可更新知识库，快速适应新的信息。

2. RAG 的工作流程：一个循序渐进的指南

RAG 的工作流程可以分解为以下几个关键步骤：文本分块、生成嵌入向量、存储向量数据和检索问答。

2.1 文本分块(Chunking)

文本分块是RAG流程的第一步，目的是将大型文档分割成更小、更易于管理的片段，即“块”(Chunks)。这是因为直接对大型文档进行嵌入和检索效率较低。

有多种文本分块策略可供选择，例如：

固定长度分块： 按照固定的字符或Token数量进行分割。原文采用的就是固定长度分块，例如每500个字符分割一次。这种方法简单易用，但可能破坏句子的完整性。
基于句子的分块： 按照句子边界进行分割，保证块的语义完整性。
语义或递归分块： 是一种更高级的分块策略，它会尝试识别文档中的语义结构，例如段落、章节等，然后按照这些结构进行分割。

选择合适的分块策略取决于具体的应用场景。例如，对于技术文档，可以采用语义分块，以保证每个块包含一个完整的概念。

import fitz  # PyMuPDF
import os
import json

def get_pdf_chunks(file_path, chunk_size=500):
    """将PDF文件分割成固定大小的文本块。"""
    doc = fitz.open(file_path)
    full_text = ""
    for page in doc:
        full_text += page.get_text()
    chunks = []
    for i in range(0, len(full_text), chunk_size):
        chunk = full_text[i:i + chunk_size]
        chunks.append({
            "chunk_id": i // chunk_size,
            "text": chunk,
            "char_count": len(chunk),
            "source": os.path.basename(file_path)
        })
    return chunks

def save_chunks_to_markdown(chunks, output_path="output.md"):
    """将文本块保存为Markdown文件。"""
    with open(output_path, "w", encoding="utf-8") as f:
        for chunk in chunks:
            f.write(f"### Chunk {chunk['chunk_id']} - Source: {chunk['source']}\n\n")
            f.write(chunk["text"].strip() + "\n\n")
            f.write("**Embedding (first 10 dimensions shown):**\n\n")
            # 假定的embedding 存在
            if 'embedding' in chunk:
              f.write(f"`{chunk['embedding'][:10]}`\n\n")

def save_chunks_to_json(chunks, output_path="output.json"):
    """将文本块保存为JSON文件。"""
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(chunks, f, indent=2)

if __name__ == "__main__":
    pdf_files = [
        "file1.pdf",
        "file2.pdf",
        "file3.pdf"  #需要自己创建测试文件
    ]
    all_chunks = []
    for pdf_path in pdf_files:
        chunks = get_pdf_chunks(pdf_path)
        all_chunks.extend(chunks)
    save_chunks_to_markdown(all_chunks, "output_all.md")
    save_chunks_to_json(all_chunks, "output_all.json")

2.2 生成嵌入向量(Embedding)

在完成文本分块后，需要将每个文本块转换为嵌入向量。嵌入向量是一种将文本表示为数值向量的技术，它能够捕捉文本的语义信息。语义相似的文本块，其嵌入向量在向量空间中的距离也更近。

常用的嵌入向量模型包括：

Sentence Transformers： 这是一个流行的Python库，提供了多种预训练的嵌入向量模型，例如all-MiniLM-L6-v2。
OpenAI Embedding API： OpenAI提供了一系列强大的嵌入向量模型，例如text-embedding-ada-002。
Cohere Embedding API: Cohere也提供了Embedding服务, 有着非常好的多语言支持。

选择合适的嵌入向量模型取决于具体的应用场景和性能要求。例如，对于需要处理大量文本的场景，可以选择轻量级的模型，以提高效率。

from sentence_transformers import SentenceTransformer

def embed_chunks(chunks, model_name="all-MiniLM-L6-v2"):
    """使用SentenceTransformer模型生成文本块的嵌入向量。"""
    model = SentenceTransformer(model_name)
    texts = [chunk["text"] for chunk in chunks]
    embeddings = model.encode(texts, convert_to_numpy=True)
    for i, chunk in enumerate(chunks):
        chunk["embedding"] = embeddings[i].tolist()
    return chunks

# ... (之前的代码)

if __name__ == "__main__":
    pdf_files = [
        "file1.pdf",
        "file2.pdf",
        "file3.pdf"  # 需要自己创建测试文件
    ]
    all_chunks = []
    for pdf_path in pdf_files:
        chunks = get_pdf_chunks(pdf_path)
        all_chunks.extend(chunks)

    all_chunks = embed_chunks(all_chunks) # 添加了 embed_chunks 函数调用
    save_chunks_to_markdown(all_chunks, "output_all.md")
    save_chunks_to_json(all_chunks, "output_all.json")

2.3 存储向量数据(Vector Database)

生成嵌入向量后，需要将它们存储在向量数据库中，以便后续进行快速相似性搜索。

常用的向量数据库包括：

PostgreSQL with pgvector： 这是一个开源的向量数据库，它基于PostgreSQL数据库，通过pgvector扩展提供向量存储和检索功能。
Pinecone： 这是一个云端的向量数据库，提供了高性能的向量搜索服务。
Weaviate： 这是一个开源的向量数据库，支持多种数据类型和查询方式。
Milvus： 这是一个开源的向量数据库，专注于高性能的向量搜索。

选择合适的向量数据库取决于具体的应用场景和性能要求。例如，对于需要高性能搜索的场景，可以选择Pinecone或Milvus。如果希望使用开源解决方案，可以选择PostgreSQL with pgvector或Weaviate。

import psycopg2

db_config = {
    "dbname": "rag_db",
    "user": "yourusername",  # 替换为你的用户名
    "password": "yourpassword",  # 替换为你的密码
    "host": "localhost",
    "port": 5432
}

def create_table_if_not_exists():
    """创建用于存储文本块和嵌入向量的PostgreSQL表。"""
    conn = psycopg2.connect(**db_config)
    cur = conn.cursor()
    cur.execute("""
        CREATE TABLE IF NOT EXISTS pdf_chunks (
            id SERIAL PRIMARY KEY,
            chunk_id INTEGER,
            text TEXT,
            source TEXT,
            embedding VECTOR(384)
        );
    """)
    conn.commit()
    cur.close()
    conn.close()

def store_embeddings_in_postgres(chunks):
    """将文本块和嵌入向量存储到PostgreSQL数据库中。"""
    conn = psycopg2.connect(**db_config)
    cur = conn.cursor()
    for chunk in chunks:
        cur.execute(
            """
            INSERT INTO pdf_chunks (chunk_id, text, source, embedding)
            VALUES (%s, %s, %s, %s)
            """,
            (
                chunk["chunk_id"],
                chunk["text"],
                chunk["source"],
                chunk["embedding"]
            )
        )
    conn.commit()
    cur.close()
    conn.close()

# ... (之前的代码)

if __name__ == "__main__":
    pdf_files = [
        "file1.pdf",
        "file2.pdf",
        "file3.pdf"  # 需要自己创建测试文件
    ]
    all_chunks = []
    for pdf_path in pdf_files:
        chunks = get_pdf_chunks(pdf_path)
        all_chunks.extend(chunks)

    all_chunks = embed_chunks(all_chunks)

    create_table_if_not_exists()  # 确保表已创建
    store_embeddings_in_postgres(all_chunks)  # 将数据存储到数据库
    #save_chunks_to_markdown(all_chunks, "output_all.md")
    #save_chunks_to_json(all_chunks, "output_all.json")

注意： 使用此代码前，确保已经安装了psycopg2 库，并配置了PostgreSQL数据库连接。

2.4 检索问答(Retrieval and Answering)

当用户提出问题时，RAG系统首先将问题转换为嵌入向量，然后在向量数据库中搜索与问题最相关的文本块。最后，RAG系统将检索到的文本块和原始问题一起输入到大模型中，生成最终答案。

import psycopg2
from sentence_transformers import SentenceTransformer
import google.generativeai as genai

# --- Config ---
db_config = {
    "dbname": "rag_db",
    "user": "yourusername",  # 替换为你的用户名
    "password": "yourpassword",  # 替换为你的密码
    "host": "localhost",
    "port": 5432
}
GOOGLE_API_KEY = "YourAPIKey"  # 替换为你的Google API Key
GENAI_MODEL = "gemini-2.0-flash-exp"

# --- Init ---
genai.configure(api_key=GOOGLE_API_KEY)
model = SentenceTransformer("all-MiniLM-L6-v2")
gemini = genai.GenerativeModel(model_name=GENAI_MODEL)

def get_query_embedding(query):
    """生成用户查询的嵌入向量。"""
    embedding = model.encode([query], convert_to_numpy=True)[0]
    return embedding.tolist()

def search_similar_chunks(query_embedding, top_k=5):
    """在向量数据库中搜索与查询最相似的文本块。"""
    conn = psycopg2.connect(**db_config)
    cur = conn.cursor()
    cur.execute(
        """
        SELECT chunk_id, text, embedding <-> %s::vector AS distance
        FROM pdf_chunks
        ORDER BY embedding <-> %s::vector
        LIMIT %s
        """,
        (query_embedding, query_embedding, top_k)
    )
    results = cur.fetchall()
    cur.close()
    conn.close()
    return results

def build_prompt(user_query, chunks):
    """构建用于生成答案的提示语。"""
    context = "\n\n".join([f"[Chunk {cid}]\n{text.strip()}" for cid, text, _ in chunks])
    prompt = f"""You are a helpful assistant. Answer the question using only the following context. using the context as main make up answers.
    ### Context:{context}
    ### Question:{user_query}
    ### Answer:"""
    return prompt.strip()

def get_answer_from_gemini(prompt):
    """使用Gemini模型生成答案。"""
    response = gemini.generate_content(prompt)
    return response.text.strip()

def main():
    user_query = input("Enter your question: ")
    query_embedding = get_query_embedding(user_query)
    top_chunks = search_similar_chunks(query_embedding)
    prompt = build_prompt(user_query, top_chunks)
    print("\n[Prompt Preview]\n" + prompt[:1000] + ("..." if len(prompt) > 1000 else ""))
    answer = get_answer_from_gemini(prompt)
    print("\n[Answer from Gemini]\n")
    print(answer)

if __name__ == "__main__":
    main()

注意： 使用此代码前，请确保已经安装了google-generativeai 库，并配置了Google API Key。

3. RAG 的优化方向

RAG 系统并非一蹴而就，需要不断优化才能达到最佳性能。以下是一些常见的优化方向：

更智能的分块策略： 探索语义分块、递归分块等更高级的分块策略，以获得更自然的上下文边界。
重新排序检索到的文本块： 使用额外的模型对检索到的文本块进行重新排序，以提高相关性。这可以通过交叉编码器模型（Cross-Encoder）实现，此类模型可以对问题和上下文进行联合建模，并输出一个相关性得分。
微调提示语(Prompt)： 尝试不同的提示语格式、少样本示例或系统指令，以更有效地引导大模型生成答案。好的Prompt能够显著提高大模型的输出质量。
查询路由： 确定用户查询的最佳知识源。如果查询是关于数据库架构的，则路由到数据库架构；如果查询是关于特定文档的，则路由到这些文档。
查询转换： 重写用户查询以提高检索性能。可以使用LLM来执行像LoRA这样的查询转换。例如，可以将一个过于模糊的问题转化为更具体的问题。
增加知识图谱： 将知识图谱纳入RAG流程，可以增强模型对实体和关系的理解，从而提高答案的准确性。

4. RAG 的应用前景

RAG 作为一种强大的技术，正在各个领域展现出巨大的应用潜力：

智能客服： RAG 可以为智能客服提供实时的产品信息、政策法规等知识，提高客户服务的质量和效率。
知识管理： RAG 可以帮助企业构建智能知识库，方便员工快速查找和利用信息。
教育领域： RAG 可以为学生提供个性化的学习辅导，解答问题，提供资料。
金融分析： RAG 可以帮助分析师快速分析大量的财务报告、市场数据，发现投资机会。

5. 总结

RAG 将结构化文档检索与强大的语言生成能力相结合，为大模型提供了实时获取外部知识的能力。通过文本分块、生成嵌入向量、存储向量数据和检索问答等步骤，RAG 能够让大模型根据真实、可靠的上下文信息回答问题，避免了“幻觉”现象的发生。随着技术的不断发展，RAG 将在更多领域发挥重要作用，推动人工智能的进步。通过不断优化分块策略、提示语和检索方法，我们可以构建出更智能、更可靠的RAG系统，为各行各业带来更高的效率和价值。现在就开始探索 RAG 的奇妙世界吧！

RAG：赋能大模型，打造更智能的AI助手