大模型 RAG 技术详解：架构、优势与实战案例

3. 数据安全问题：敏感信息泄露风险

通用 LLM 没有企业内部数据和用户数据。企业想要在保证安全的前提下使用 LLM，最好的方式是将数据放在本地，仅让在线大模型完成归纳功能。这就像你不会把公司的机密文件交给陌生人保管，而是自己妥善保存，只在需要的时候让他人帮忙整理。

银行通常不会将用户的账户信息直接输入到公共 LLM 中处理，而是先在本地脱敏，再将结果发送给 LLM。

from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain

# 本地处理敏感数据
def process_sensitive_data(data):
    return data.replace("敏感信息", "***")

# 初始化 LLM
llm = OpenAI(temperature=0)

# 创建 Prompt 模板
prompt = PromptTemplate(
    input_variables=["data"],
    template="分析以下数据：{data}"
)

# 创建 LLM Chain
chain = LLMChain(llm=llm, prompt=prompt)

# 处理敏感数据
sensitive_data = "用户的敏感信息：1234567890"
processed_data = process_sensitive_data(sensitive_data)

# 调用 LLM 进行分析
result = chain.run(processed_data)
print(result)

二、RAG 技术：检索增强生成

1. 定义与核心思想

RAG（Retrieval-Augmented Generation，检索增强生成）是指 LLM 在回答问题或生成文本时，先从大量文档中检索出相关的信息，然后基于这些信息生成回答。这就像学生考试时可以查阅课本资料，而不是仅凭记忆瞎蒙。

例如，当你问'如何做红烧肉？'，RAG 会先检索相关菜谱文档，再基于文档生成详细做法。

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

# 初始化嵌入模型和向量数据库
embeddings = OpenAIEmbeddings()
db = FAISS.from_texts(["红烧肉的做法：1. 准备食材..."], embeddings)

# 初始化 LLM
llm = OpenAI(temperature=0)

# 创建检索 QA 链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=db.as_retriever()
)

# 提问
result = qa_chain.run("如何做红烧肉？")
print(result)

2. 核心架构

RAG 主要包含两个核心模块：检索器（R）和生成器（G）。

检索器模块

检索器的作用是从知识库中找出最相关的 k 个文档。构建高质量的检索器需要解决三个关键问题：如何获得准确的语义表示？如何协调查询和文档的语义空间？如何对齐检索模型的输出和大语言模型的偏好？

这就像图书馆管理员，能根据读者需求快速找到最相关的书籍。

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings

# 初始化嵌入模型
embeddings = OpenAIEmbeddings()

# 创建向量数据库
documents = ["Python 学习资源 1...", "Python 学习资源 2...", "Python 学习资源 3..."]
db = FAISS.from_texts(documents, embeddings)

# 检索相关文档
query = "如何学习 Python"
docs = db.similarity_search(query, k=2)
for doc in docs:
    print(doc.page_content)

生成器模块

生成器负责将检索到的信息转化为自然流畅的文本。它的输入不仅包括上下文，还有检索到的相关片段，这使得回答更准确。

from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

# 初始化 LLM
llm = OpenAI(temperature=0)

# 创建 Prompt 模板
prompt = PromptTemplate(
    input_variables=["context", "question"],
    template="基于以下上下文回答问题：\n{context}\n\n问题：{question}"
)

# 检索到的上下文
context = "Python 是一种高级编程语言，易于学习和使用。"
question = "Python 是什么？"

# 生成回答
result = llm(prompt.format(context=context, question=question))
print(result)

三、使用 RAG 的八大优势

相比传统微调，RAG 在企业应用中展现出显著优势。

1. 可扩展性与成本控制

开发者不必为每个特定任务重新训练整个大模型，只需外挂知识库即可提供额外信息。就像给普通电脑外接移动硬盘增加容量，无需更换整机。

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

# 加载企业内部知识库
embeddings = OpenAIEmbeddings()
db = FAISS.load_local("enterprise_knowledge_base", embeddings)

llm = OpenAI(temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=db.as_retriever()
)

result = qa_chain.run("公司的请假政策是什么？")
print(result)

2. 准确性与可解释性

通过引用信息来源，用户可以核实答案，增强信任。这就像老师批改作业时，若学生引用了课本内容，会更相信答案的正确性。

from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

llm = OpenAI(temperature=0)
prompt = PromptTemplate(
    input_variables=["context", "question"],
    template="基于以下上下文回答问题，并引用信息来源：\n{context}\n\n问题：{question}"
)

context = "根据《Python 编程从入门到精通》一书，Python 是一种高级编程语言。"
question = "Python 是什么？"

result = llm(prompt.format(context=context, question=question))
print(result)

3. 可控性与安全性

允许轻松更新知识库而不必重训模型。就像随时更新手机 APP，无需换机。同时，敏感数据可存储在本地数据库，通过角色权限控制访问。

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()
documents = ["公司的请假政策是：病假需要提供医院证明..."]
db = FAISS.from_texts(documents, embeddings)

# 更新知识库
new_document = "公司的请假政策已更新：病假需要提供三甲医院证明..."
db.add_texts([new_document])

query = "公司的请假政策是什么？"
docs = db.similarity_search(query, k=1)
print(docs[0].page_content)

4. 多功能性与及时性

适用于 QA、摘要、对话等多种任务。且能识别最新信息，不受训练数据截止时间限制。

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
import requests

embeddings = OpenAIEmbeddings()
response = requests.get("https://api.example.com/calendar/2026")
calendar_data = response.json()
db = FAISS.from_texts([str(calendar_data)], embeddings)

llm = OpenAI(temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=db.as_retriever()
)

result = qa_chain.run("2026 年的春节是哪一天？")
print(result)

四、RAG 与 SFT 的对比

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
from transformers import pipeline

# RAG 示例
embeddings = OpenAIEmbeddings()
db = FAISS.from_texts(["最新科技进展：GPT-5 发布..."], embeddings)
llm = OpenAI(temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=db.as_retriever()
)
rag_result = qa_chain.run("最新的科技进展是什么？")
print(f"RAG 回答：{rag_result}")

# SFT 示例
sft_model = pipeline('text-generation', model='fine_tuned_model')
sft_result = sft_model("最新的科技进展是什么？", max_length=100, num_return_sequences=1)
print(f"SFT 回答：{sft_result[0]['generated_text']}")

五、RAG 的典型实现方法

1. 数据索引

这是一个离线过程，将私域数据向量化后构建索引。包括提取、分割、向量化及建索引。

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter
import PyPDF2

def extract_text_from_pdf(pdf_path):
    text = ""
    with open(pdf_path, "rb") as f:
        reader = PyPDF2.PdfReader(f)
        for page in reader.pages:
            text += page.extract_text()
    return text

text_splitter = CharacterTextSplitter()
embeddings = OpenAIEmbeddings()

def create_index(pdf_path):
    text = extract_text_from_pdf(pdf_path)
    texts = text_splitter.split_text(text)
    db = FAISS.from_texts(texts, embeddings)
    db.save_local("pdf_index")

create_index("example.pdf")

2. 数据检索

获取有效信息的关键环节，支持元数据过滤、相似度检索等。

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()
db = FAISS.load_local("pdf_index", embeddings)

query = "Python 学习教程"
docs = db.similarity_search(query, k=3)
for doc in docs:
    print(doc.page_content)

3. 文本生成

将原始 query 和检索文本组合输入模型，本质是 Prompt Engineering。

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

embeddings = OpenAIEmbeddings()
db = FAISS.load_local("pdf_index", embeddings)

llm = OpenAI(temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=db.as_retriever()
)

result = qa_chain.run("如何学习 Python？")
print(result)

六、RAG 的典型案例

1. ChatPDF 及其复刻版

允许用户上传 PDF 并通过聊天界面提问。先将 PDF 转为文本，再用 Embedding API 向量化，最后基于向量回答问题。

import requests
import PyPDF2
from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

def extract_text_from_pdf(pdf_path):
    text = ""
    with open(pdf_path, "rb") as f:
        reader = PyPDF2.PdfReader(f)
        for page in reader.pages:
            text += page.extract_text()
    return text

def create_chat_pdf(pdf_path):
    text = extract_text_from_pdf(pdf_path)
    embeddings = OpenAIEmbeddings()
    db = FAISS.from_texts([text], embeddings)
    db.save_local("chat_pdf_index")

def chat_with_pdf(question):
    embeddings = OpenAIEmbeddings()
    db = FAISS.load_local("chat_pdf_index", embeddings)
    llm = OpenAI(temperature=0)
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=db.as_retriever()
    )
    result = qa_chain.run(question)
    return result

create_chat_pdf("example.pdf")
answer = chat_with_pdf("本文的主要内容是什么？")
print(answer)

2. Baichuan

融合指令意图理解、智能搜索和结果增强，减少模型幻觉。

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

embeddings = OpenAIEmbeddings()
db = FAISS.from_texts(["红烧肉的做法：1. 准备食材..."], embeddings)

llm = OpenAI(temperature=0)
prompt = PromptTemplate(
    input_variables=["context", "question"],
    template="理解用户意图：{question}\n基于以下上下文回答问题：{context}\n\n问题：{question}"
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=db.as_retriever(),
    chain_type_kwargs={"prompt": prompt}
)

result = qa_chain.run("如何做红烧肉？")
print(result)

3. Multi-modal retrieval-based LMs

RA-CM3 等模型结合 CLIP 检索器和 Transformer 生成器，处理图文多模态数据。

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
from PIL import Image
import requests
from io import BytesIO

embeddings = OpenAIEmbeddings()
db = FAISS.from_texts(["猫在草地上玩耍的图片：https://example.com/cat.jpg"], embeddings)

llm = OpenAI(temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=db.as_retriever()
)

result = qa_chain.run("生成一只猫在草地上玩耍的图片")
print(result)

image_url = result.split("：")[-1].strip()
response = requests.get(image_url)
image = Image.open(BytesIO(response.content))
image.show()

七、RAG 存在的问题

尽管 RAG 优势明显，但在工程落地中仍需注意以下挑战。

1. 检索效果依赖 embedding 和算法

如果 embedding 模型无法准确表示语义，或检索算法不够精准，会导致无关信息干扰输出。

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

embeddings = OpenAIEmbeddings()
db = FAISS.from_texts(["猫是一种哺乳动物...", "狗是一种哺乳动物..."], embeddings)

llm = OpenAI(temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=db.as_retriever()
)

result = qa_chain.run("如何训练一只狗？")
print(result)

2. 信息利用仍是黑盒

大模型如何利用检索到的信息生成回答并不透明，可能存在与检索信息冲突的情况。

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

embeddings = OpenAIEmbeddings()
db = FAISS.from_texts(["猫是一种哺乳动物...", "狗是一种哺乳动物..."], embeddings)

llm = OpenAI(temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=db.as_retriever()
)

result = qa_chain.run("猫是一种什么动物？")
print(result)

3. 效率问题

无差别检索所有任务会增加输入长度，降低生成速度。

import time
from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

embeddings = OpenAIEmbeddings()
db = FAISS.from_texts(["猫是一种哺乳动物...", "狗是一种哺乳动物...", "鸟是一种脊椎动物...", "鱼是一种脊椎动物..."], embeddings)

llm = OpenAI(temperature=0)
qa_chain_2 = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=db.as_retriever(search_kwargs={"k": 2})
)
qa_chain_4 = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=db.as_retriever(search_kwargs={"k": 4})
)

start_time = time.time()
result_2 = qa_chain_2.run("猫是一种什么动物？")
end_time = time.time()
print(f"检索 2 个文档耗时：{end_time - start_time}秒")

start_time = time.time()
result_4 = qa_chain_4.run("猫是一种什么动物？")
end_time = time.time()
print(f"检索 4 个文档耗时：{end_time - start_time}秒")

4. 事实查证困难

无法精准引用来源，检索真实性取决于数据源质量。

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

embeddings = OpenAIEmbeddings()
db = FAISS.from_texts(["猫是一种爬行动物...（来源：不可靠网站）"], embeddings)

llm = OpenAI(temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=db.as_retriever()
)

result = qa_chain.run("猫是一种什么动物？")
print(result)

八、RAG 的拓展方案

1. 多模态 RAG

处理图像、音频、视频等多种模态数据，将其转化为向量进行检索。

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
from PIL import Image
import pytesseract

def extract_text_from_image(image_path):
    image = Image.open(image_path)
    text = pytesseract.image_to_string(image)
    return text

embeddings = OpenAIEmbeddings()
db = FAISS.from_texts(["猫是一种哺乳动物..."], embeddings)

llm = OpenAI(temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=db.as_retriever()
)

image_text = extract_text_from_image("cat.jpg")
result = qa_chain.run(f"这张图片中的动物是什么？图片文本：{image_text}")
print(result)

2. 实时 RAG

集成新闻 API 等实时数据源，确保回答的及时性。

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
import requests

def get_latest_tech_news():
    response = requests.get("https://newsapi.org/v2/top-headlines?country=us&category=technology&apiKey=YOUR_API_KEY")
    news = response.json()
    return news['articles'][0]['content']

embeddings = OpenAIEmbeddings()
db = FAISS.from_texts([get_latest_tech_news()], embeddings)

llm = OpenAI(temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=db.as_retriever()
)

result = qa_chain.run("最新的科技新闻是什么？")
print(result)

3. 个性化 RAG

根据用户历史偏好调整检索策略和生成方式。

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

embeddings = OpenAIEmbeddings()
db = FAISS.from_texts(["《肖申克的救赎》是一部经典的剧情片...", "《阿甘正传》是一部励志的剧情片...", "《星际穿越》是一部科幻片..."], embeddings)

llm = OpenAI(temperature=0)
user_preference = "剧情片"

prompt = PromptTemplate(
    input_variables=["context", "question", "user_preference"],
    template="用户偏好：{user_preference}\n基于以下上下文回答问题：{context}\n\n问题：{question}"
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=db.as_retriever(),
    chain_type_kwargs={"prompt": prompt}
)

result = qa_chain.run({"question": "推荐一些适合我的电影", "user_preference": user_preference})
print(result)