LlamaIndex实现(基于PDF|CSV文件)RAG检索增强生成:NaiveRAG

什么是 RAG？

RAG（Retrieval-Augmented Generation，检索增强生成） 是一种结合 信息检索（Retrieval） 和 文本生成（Generation） 的AI技术，用于提升大语言模型（LLM）生成答案的准确性和可靠性。

核心思想：

1. 检索（Retrieval）：从外部知识库（如数据库、文档、网页）中查找与问题相关的信息。
2. 增强（Augmented）：将检索到的信息作为上下文输入给生成模型。
3. 生成（Generation）：模型基于检索到的信息生成更精准、更可信的回答。
   

---

为什么需要 RAG？

传统大语言模型（如ChatGPT）存在以下问题，而RAG能有效解决：

1. 知识局限性（静态知识）

• 问题：LLM 的训练数据是固定的，无法实时更新（例如，GPT-4 的知识截止到 2023 年）。
• RAG 的解决方式：动态检索最新数据，确保答案的时效性。

2. 幻觉问题（Hallucination）

• 问题：LLM 可能编造看似合理但错误的信息（如虚构事实、错误引用）。
• RAG 的解决方式：基于检索到的真实数据生成答案，减少“瞎猜”。

3. 可解释性 & 可验证性

• 问题：传统 LLM 的回答像“黑箱”，用户无法验证来源。
• RAG 的解决方式：提供引用来源（如文档片段、网页链接），让用户检查可信度。

4. 定制化知识库

• 问题：通用 LLM 无法访问企业私有数据（如内部文档、产品手册）。
• RAG 的解决方式：连接企业数据库，让模型基于特定数据生成答案（如客服机器人）。

---

RAG vs 传统 LLM

对比项	传统 LLM（如ChatGPT）	RAG 增强的 LLM
知识更新	依赖训练数据（静态）	可实时检索最新数据
准确性	可能产生幻觉	基于检索结果生成，更可靠
可解释性	无来源引用	可提供参考文档
适用场景	通用问答、创意写作	事实查询、专业领域问答（如医疗、法律）

---

LlamaIndex实现RAG

基于PDF文件的RAG

RAG 是当前最受欢迎的AI技术之一，被广泛应用于企业级AI助手和专业问答系统！ 🚀
LlamaIndex 是一个编排框架，可简化将私有数据与公共数据集成以使用大型语言模型（LLM）构建应用程序的过程。它提供了数据摄取、索引编制和查询工具，使其成为满足生成式人工智能需求的多功能解决方案。
本文带大家利用LlamaIndex实现一个基础的检索增强生成（RAG）系统，用于处理并查询PDF文档。该系统采用流水线设计，通过文档编码和节点创建构建向量索引，从而实现相关信息检索。

流程如图

核心组件

1. PDF文档处理与文本提取
2. 文本分块处理
3. 基于FAISS向量库和OpenAI嵌入的流水线构建
4. 文档检索器配置
5. 系统性能评估模块

技术细节

文档预处理

1. 使用SimpleDirectoryReader加载PDF文档
2. 通过SentenceSplitter将文本分割为节点/块，支持自定义块大小和重叠量

文本清洗

应用自定义转换器TextCleaner处理PDF中的特殊格式问题

流水线构建

1. 采用OpenAI嵌入生成文本向量表示
2. 基于FAISS向量库实现高效相似度搜索

检索器配置

检索器设置为返回与查询最相关的2个文本块

特色功能

1. 模块化设计：封装为可复用的独立函数
2. 可配置分块：支持调整块大小和重叠量
3. 高效检索：采用FAISS实现快速相似度搜索
4. 评估体系：包含系统性能评估功能

使用示例

代码包含测试查询：“气候变化的主要原因是什么？”，展示如何从处理后的文档中检索相关上下文

系统评估

通过evaluate_rag函数评估检索器性能（具体评估指标未在代码中体现）

方案优势

1. 可扩展性：支持大文档分块处理
2. 灵活性：参数可自由调整
3. 高效率：FAISS实现高维空间快速搜索
4. 先进NLP集成：采用OpenAI嵌入技术

结论

本RAG系统为构建复杂信息检索和问答系统提供了基础框架，特别适用于需要快速访问大型文档集中特定信息的应用场景。

---

pip install llama-index
pip install faiss-cpu

库导入与环境配置

from typing import List
from llama_index.core import SimpleDirectoryReader, VectorStoreIndex
from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.schema import BaseNode, TransformComponent
from llama_index.vector_stores.faiss import FaissVectorStore
from llama_index.core.text_splitter import SentenceSplitter
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core import Settings
import faiss
import os
import sys
from dotenv import load_dotenv

sys.path.append(os.path.abspath(os.path.join(os.getcwd(), '..'))) # 添加父目录到路径（适用于notebook环境）

EMBED_DIMENSION = 512  # 嵌入维度

# 分块设置与langchain示例不同：
# langchain按字符串长度计算，llamaindex按token长度计算
CHUNK_SIZE = 200      # 分块大小
CHUNK_OVERLAP = 50    # 分块重叠量

# 加载环境变量
load_dotenv()

# 设置OpenAI API密钥
os.environ["OPENAI_API_KEY"] = os.getenv('OPENAI_API_KEY')

# 配置LlamaIndex全局嵌入模型
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small", dimensions=EMBED_DIMENSION)

文档读取

path = "../data/"
node_parser = SimpleDirectoryReader(input_dir=path, required_exts=['.pdf'])
documents = node_parser.load_data()
print(documents[0])

向量库创建

# 创建FAISS向量库存储嵌入
faiss_index = faiss.IndexFlatL2(EMBED_DIMENSION)
vector_store = FaissVectorStore(faiss_index=faiss_index)

文本清洗转换器

class TextCleaner(TransformComponent):
    """
    流水线文本清洗组件
    功能：清除文本中的杂乱字符
    """
    def __call__(self, nodes, **kwargs) -> List[BaseNode]:
        
        for node in nodes:
            node.text = node.text.replace('\t', ' ')  # 制表符转空格
            node.text = node.text.replace(' \n', ' ') # 段落分隔符转空格
            
        return nodes

流水线构建

text_splitter = SentenceSplitter(chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP)

# 创建包含文档转换和向量库的流水线
pipeline = IngestionPipeline(
    transformations=[
        TextCleaner(),
        text_splitter,
    ],
    vector_store=vector_store, 
)

# 运行流水线并获取生成的节点
nodes = pipeline.run(documents=documents)

检索器创建

vector_store_index = VectorStoreIndex(nodes)
retriever = vector_store_index.as_retriever(similarity_top_k=2)

检索测试

def show_context(context):
    """
    展示检索到的上下文内容
    
    参数：
        context (list): 待展示的上下文列表
    
    按位置序号打印每个上下文内容
    """
    for i, c in enumerate(context):
        print(f"上下文 {i+1}:")
        print(c.text)
        print("\n")

test_query = "气候变化的主要原因是什么？"
context = retriever.retrieve(test_query)
show_context(context)

性能评估

import json
from deepeval import evaluate
from deepeval.metrics import GEval, FaithfulnessMetric, ContextualRelevancyMetric
from deepeval.test_case import LLMTestCaseParams
from evaluation.evalute_rag import create_deep_eval_test_cases

# 设置评估用LLM模型
LLM_MODEL = "gpt-4o"

# 定义评估指标
correctness_metric = GEval(
    name="正确性",
    model=LLM_MODEL,
    evaluation_params=[
        LLMTestCaseParams.EXPECTED_OUTPUT,
        LLMTestCaseParams.ACTUAL_OUTPUT
    ],
    evaluation_steps=[
        "根据预期输出判断实际输出是否事实正确"
    ],
)

faithfulness_metric = FaithfulnessMetric(
    threshold=0.7,
    model=LLM_MODEL,
    include_reason=False
)

relevance_metric = ContextualRelevancyMetric(
    threshold=1,
    model=LLM_MODEL,
    include_reason=True
)

def evaluate_rag(query_engine, num_questions: int = 5) -> None:
    """
    RAG系统评估函数
    
    参数：
        query_engine: 问答引擎
        num_questions (int): 评估问题数量（默认5个）
    """
    
    # 从JSON文件加载问答对
    q_a_file_name = "../data/q_a.json"
    with open(q_a_file_name, "r", encoding="utf-8") as json_file:
        q_a = json.load(json_file)

    questions = [qa["question"] for qa in q_a][:num_questions]
    ground_truth_answers = [qa["answer"] for qa in q_a][:num_questions]
    generated_answers = []
    retrieved_documents = []

    # 生成答案并检索文档
    for question in questions:
        response = query_engine.query(question)
        context = [doc.text for doc in response.source_nodes]
        retrieved_documents.append(context)
        generated_answers.append(response.response)

    # 创建测试用例并评估
    test_cases = create_deep_eval_test_cases(questions, ground_truth_answers, generated_answers, retrieved_documents)
    evaluate(
        test_cases=test_cases,
        metrics=[correctness_metric, faithfulness_metric, relevance_metric]
    )

执行评估

query_engine  = vector_store_index.as_query_engine(similarity_top_k=2)
evaluate_rag(query_engine, num_questions=1)

基于CSV文件的简易RAG（检索增强生成）系统

CSV文件结构与用例

CSV文件包含虚拟客户数据，包括姓名、公司等各种属性。该数据集将用于RAG用例，构建一个客户信息问答系统。

核心组件

1. 加载和分割CSV文件
2. 使用FAISS和OpenAI嵌入创建向量存储
3. 设置查询引擎以处理文档查询
4. 基于CSV数据创建问答功能

方法细节

文档预处理

1. 使用LlamaIndex的PagedCSVReader加载CSV文件
2. 该阅读器将每行数据转换为LlamaIndex文档，并保留相应的列名。不进行进一步的分割。

导入与环境变量设置

from llama_index.core.readers import SimpleDirectoryReader
from llama_index.core import Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.readers.file import PagedCSVReader
from llama_index.vector_stores.faiss import FaissVectorStore
from llama_index.core.ingestion import IngestionPipeline
from llama_index.core import VectorStoreIndex
import faiss
import os
import pandas as pd
from dotenv import load_dotenv


# 从.env文件加载环境变量
load_dotenv()

# 设置OpenAI API密钥环境变量
os.environ["OPENAI_API_KEY"] = os.getenv('OPENAI_API_KEY')


# Llamaindex全局设置，用于LLM和嵌入模型
EMBED_DIMENSION=512
Settings.llm = OpenAI(model="gpt-3.5-turbo")
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small", dimensions=EMBED_DIMENSION)

CSV文件结构与用例

CSV文件包含虚拟客户数据，包括姓名、公司等各种属性。该数据集将用于RAG用例，构建一个客户信息问答系统。

file_path = ('../data/customers-100.csv') # 插入CSV文件路径
data = pd.read_csv(file_path)

# 预览CSV文件
data.head()

向量存储

# 创建FaissVectorStore以存储嵌入
fais_index = faiss.IndexFlatL2(EMBED_DIMENSION)
vector_store = FaissVectorStore(faiss_index=fais_index)

加载并处理CSV数据为文档

csv_reader = PagedCSVReader()

reader = SimpleDirectoryReader( 
    input_files=[file_path],
    file_extractor={".csv": csv_reader}
    )

docs = reader.load_data()

# 查看示例数据块
print(docs[0].text)

Index: 1
Customer Id: DD37Cf93aecA6Dc
First Name: Sheryl
Last Name: Baxter
Company: Rasmussen Group
City: East Leonard
Country: Chile
Phone 1: 229.077.5154
Phone 2: 397.884.0519x718
Email: [e-mail]
Subscription Date: 2020-08-24
Website: http://www.stephenson.com/

数据摄取管道

pipeline = IngestionPipeline(
    vector_store=vector_store,
    documents=docs
)

nodes = pipeline.run()

创建查询引擎

vector_store_index = VectorStoreIndex(nodes)
query_engine = vector_store_index.as_query_engine(similarity_top_k=2)

基于CSV数据向RAG机器人提问

response = query_engine.query("Sheryl Baxter在哪家公司工作？")
response.response

'Rasmussen Group'