LLM RAG 检索增强生成原理与应用详解

2. 按语义分割

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai.embeddings import OpenAIEmbeddings

# This is a long document we can split up.
with open("../../state_of_the_union.txt") as f:
    state_of_the_union = f.read()
    
text_splitter = SemanticChunker(OpenAIEmbeddings())

docs = text_splitter.create_documents([state_of_the_union])  
print(docs[0].page_content)

2.2.2 数据信息（Metadata）

在进行文本分割的同时，我们还可以给分割的文本添加一下 Metadata 的数据，方便记录该文本段的一些基本信息，如文章来源、作者信息等。 一个是能在进行文本召回时可以作为过滤搜索，另一方面还在作为发给 LLM 的补充数据，让 LLM 生成的内容更为丰富。

metadatas = [{"document": 1}, {"document": 2}]  
documents = text_splitter.create_documents(  
[state_of_the_union, state_of_the_union], metadatas=metadatas  
)  
print(documents[0])

2.2.3 分割参数

在进行文本分割时，我们还需要重点关注两个参数 chunk_size 和 chunk_overlap，这两个参数分别表示分割长度和两段分割文本重合长度。 在实际 RAG 应用中，chunk_size 需要结合向量数据库的来选择合适大小，比如腾讯云的向量数据库，一次只支持单块 512token（400 左右字符）的大小写入，那 chunk_size 就应该设置 400 多。chunk_overlap 的大小建议设置在 chunk_size 的 1/5 左右，在召回多段文本时，可以增加数据的丰富度。 实际情况请结合具体项目进行设置和测试验证。

2.3 数据向量化（Text embedding models）

在进行数据分割后，需要对文本数据段进行向量化。目前主流的中文向量化模型有：

使用 OpenAIEmbeddings 向量化处理：

from langchain_openai import OpenAIEmbeddings  

embeddings_model = OpenAIEmbeddings(openai_api_key="...")

embeddings = embeddings_model.embed_documents(
    [
        "Hi there!",
        "Oh, hello!",
        "What's your name?",
        "My friends call me World",
        "Hello World!"
    ]
)
len(embeddings), len(embeddings[0])

目前 LangChain 支持 37 种 embedding model，这些向量化模型核心功能就是将文本向量化，提供给向量数据库进行存储。

2.4 向量数据库（Vector stores）

[图示：向量数据库存储示意图]

数据向量化后，就需要将向量数据存储进向量数据库。目前有很多开源向量数据库，如 ChromaDB、Faiss-CPU、LanceDB。云服务厂商也陆续推出了向量数据库服务，包括腾讯云、阿里云的向量数据库。

LanceDB 向量数据库使用：

from langchain_community.document_loaders import TextLoader
from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter
from langchain_community.vectorstores import LanceDB

import lancedb

db = lancedb.connect("/tmp/lancedb")
table = db.create_table(
    "my_table",
    data=[
        {
            "vector": embeddings.embed_query("Hello World"),
            "text": "Hello World",
            "id": "1",
        }
    ],
    mode="overwrite",
)

# Load the document, split it into chunks, embed each chunk and load it into the vector store.
raw_documents = TextLoader('../../../state_of_the_union.txt').load()
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
documents = text_splitter.split_documents(raw_documents)
db = LanceDB.from_documents(documents, OpenAIEmbeddings(), connection=table)

如以 LangChain VectorStore 为基础类，实现的支持腾讯云的向量数据库服务。 目前 LangChain 支持 47 种向量数据库接入，开发者也可以自行实现 VectorStore，定义自己的向量数据库。 主要实现以下三个抽象方法：

@abstractmethod
def add_texts(
    self,
    texts: Iterable[str],
    metadatas: Optional[List[dict]] = None,
    **kwargs: Any,
) -> List[str]:
    """Run more texts through the embeddings and add to the vectorstore."""
    
@abstractmethod
def similarity_search(
    self, query: str, k: int = 4, **kwargs: Any
) -> List[Document]:
    """Return docs most similar to query."""
    
@classmethod
@abstractmethod
def from_texts(
    cls: Type[VST],
    texts: List[str],
    embedding: Embeddings,
    metadatas: Optional[List[dict]] = None,
    **kwargs: Any,
) -> VST:
    """Return VectorStore initialized from texts and texts."""

1. add_texts: 将文本数据向量化，添加进向量数据库。
2. similarity_search: 从向量数据库召回数据。
3. from_texts: 类方法，实现将文本数据向量化，添加进向量数据库。

2.5 数据召回（Retrievers）

在讲解完数据加载、数据处理、数据向量化和向量数据库后，我们开始进入数据召回的环节。数据召回是我们向 LLM 提问时，需要根据我们提问的问题向向量数据库召回相关的文档数据，并和问题加载进 Prompt 发送给 LLM。

比如下面这段提示词：

template = """Answer the question based only on the following context:  

{context}  

Question: {question}  
"""

context 就是我们召回的上下文。

数据召回的方法有许多种，应用在不同应用场景。当前 LangChain 主流支持的 Retrievers 有以下 8 种：

这里简述一下不同 Retrievers 的主要应用场景，大家可以具体问题具体分析，再去查阅一下相关文档。

1. 基础的向量召回方法，根据用户问题直接向向量数据库召回数据。
2. 如果你文档数据有许多不同的小块信息，你可以根据问题检索出小块信息，再根据小块信息去索引出原文档或者更大块的数据，将大块数据作为上下文发送给 LLM。
3. 如果你有相关问题的数据集，可以为问题和文档分别存储到不同的向量数据库，在检索时可以根据问题检索出合适文档上下文。
4. 如果你提出的问题可以通过基于元数据 (而不是与文本的相似性) 来获取文档，可以使用这种 Retrievers，利用 LLM 的能力，自动生成对应的检索方法，来召回数据。
5. 如果您发现您检索的文档包含太多不相关的信息，并且分散了 LLM 的注意力，可以利用上下文压缩的方法，将召回的数据利用 LLM 进行数据处理。
6. 如果你的文档数据中包含时间相关的数据，可以考虑用此 Retriever。
7. 用户提出的问题很复杂，需要多个不同的信息来回答，可以使用此 Retriever，利用 LLM 生成多个相关的问题，再分别从向量数据库召回数据。
8. 如果您有多种检索方法，并希望尝试将它们组合起来，可以使用此 Retriever。
9. 当你需要召回多段上下文数据时，但发现 LLM 并没有根据你的上下文来回答问题时，可以考虑使用 Retriever 对你召回的数据进行重新排序，将相似度较高的排在前面，让 LLM 能更好的利用上下文来回答问题。

2.5.1 数据召回算法

在数据召回中，目前业内有两种较为通用的召回算法。

1. 相似度匹配算法（Similarity Search with Euclidean Distance） 这是向量数据库自身具备的特点，通过比较向量之间的距离来判断它们的相似度。

COLLECTION_NAME = "state_of_the_union_test"  

db = PGVector.from_documents(  
embedding=embeddings,  
documents=docs,  
collection_name=COLLECTION_NAME,  
connection_string=CONNECTION_STRING,  
)
query = "What did the president say about Ketanji Brown Jackson"  
docs_with_score = db.similarity_search_with_score(query)

for doc, score in docs_with_score:  
print("-" * 80)  
print("Score: ", score)  
print(doc.page_content)  
print("-" * 80)

2. 最大边界相关算法（Maximal Marginal Relevance） 采用这个算法，会优化召回的数据段之间的相似程度和多样性，对数据重新打分。

docs_with_score = db.max_marginal_relevance_search_with_score(query)

for doc, score in docs_with_score:  
print("-" * 80)  
print("Score: ", score)  
print(doc.page_content)  
print("-" * 80)

算法 Python 实现：

def maximal_marginal_relevance(
    query_embedding: np.ndarray,
    embedding_list: list,
    lambda_mult: float = 0.5,
    k: int = 4,
) -> List[int]:
    """Calculate maximal marginal relevance."""
    if min(k, len(embedding_list)) <= 0:
        return []
    if query_embedding.ndim == 1:
        query_embedding = np.expand_dims(query_embedding, axis=0)
    similarity_to_query = cosine_similarity(query_embedding, embedding_list)[0]
    most_similar = int(np.argmax(similarity_to_query))
    idxs = [most_similar]
    selected = np.array([embedding_list[most_similar]])
    while len(idxs) < min(k, len(embedding_list)):
        best_score = -np.inf
        idx_to_add = -1
        similarity_to_selected = cosine_similarity(embedding_list, selected)
        for i, query_score in enumerate(similarity_to_query):
            if i in idxs:
                continue
            redundant_score = max(similarity_to_selected[i])
            equation_score = (
                lambda_mult * query_score - (1 - lambda_mult) * redundant_score
            )
            if equation_score > best_score:
                best_score = equation_score
                idx_to_add = i
        idxs.append(idx_to_add)
        selected = np.append(selected, [embedding_list[idx_to_add]], axis=0)
    return idxs

我们可以在数据召回实践中，测试不同算法下的效果，来选择合适的算法。

三、RAG 实践挑战与优化

在实际生产环境中部署 RAG 系统时，除了上述基础流程外，还需要关注以下几个关键挑战与优化点：

3.1 检索精度优化

单纯的向量相似度搜索有时无法完全满足业务需求。为了提升检索质量，可以采用混合检索策略（Hybrid Search），结合关键词检索（BM25）和向量检索。关键词检索擅长处理专有名词和精确匹配，而向量检索擅长语义理解。两者加权融合通常能获得更好的召回率。

此外，查询重写（Query Rewriting）也是一种有效手段。当用户提问过于简短或模糊时，可以利用 LLM 将原始问题扩展为更详细的查询语句，再进行向量检索，从而提升上下文的相关性。

3.2 评估体系构建

如何衡量 RAG 系统的效果至关重要。常用的评估指标包括：

1. Recall（召回率）：检索到的相关文档占所有相关文档的比例。
2. Precision（准确率）：检索到的文档中真正相关的比例。
3. Context Precision：上下文信息的精准程度。
4. Faithfulness（忠实度）：生成答案是否严格基于提供的上下文，而非模型幻觉。

可以通过构建测试集（Ground Truth），对比模型输出与标准答案的一致性来进行自动化评估。工具如 RAGAS 提供了端到端的评估框架。

3.3 延迟与成本考量

RAG 涉及多次 API 调用（Embedding、Vector DB 查询、LLM 生成），这会导致响应延迟增加。优化措施包括：

1. 缓存机制：对相同的查询结果进行缓存，减少重复计算。
2. 模型选择：在精度允许范围内，选择更小、更快的 Embedding 模型和 LLM。
3. 异步处理：对于非实时场景，可采用异步队列处理请求。

同时，Token 消耗也是成本大头。通过优化 Prompt 设计，精简不必要的上下文信息，可以有效降低 Token 用量。

四、总结

最后，我们对 RAG 进行一下总结。RAG 底层依赖 LLM 大模型和数据获取、数据存储等相关技术。在 RAG 技术层面基于底层技术，共实现了数据加载、数据处理、数据向量化、向量数据库和数据召回等五种核心技术。可以使用这 5 种技术，完成 RAG 应用实现。

综上所述，RAG 技术有效地解决了大模型的知识时效性和领域专业性不足的问题。通过合理的数据处理、高效的检索策略以及持续的优化评估，企业可以构建出既准确又实用的智能问答系统。随着技术的演进，RAG 将与 Agent 技术进一步融合，成为未来 AI 应用的核心基础设施之一。