LlamaIndex 多文件 RAG 实现

概述

本项目演示如何使用 LlamaIndex 构建一个基于多文档的智能问答系统，采用 ReAct 智能体框架，结合检索工具进行问答。

技术架构

Pydantic 兼容性处理

def _patch_dashscope_response_for_pydantic():
    """
    兼容 pydantic 的 isinstance/hasattr 检查，
    避免 DashScopeResponse 抛 KeyError
    """
    try:
        from dashscope.api_entities import dashscope_response as dsr
        target_cls = dsr.DashScopeAPIResponse
    except Exception:
        return

    if getattr(target_cls, "_pydantic_safe_getattr", False):
        return

    original_getattr = target_cls.__getattr__

    def __getattr__(self, attr):
        if attr.startswith("__pydantic_"):
            raise AttributeError(attr)
        return original_getattr(self, attr)

    target_cls.__getattr__ = __getattr__
    target_cls._pydantic_safe_getattr = True

这是一个重要的兼容性补丁，解决 DashScope 响应对象与 Pydantic 检查的冲突问题。

LLM 和 Embedding 配置

def setup_llm_and_embedding():
    # 兼容 pydantic 检查
    _patch_dashscope_response_for_pydantic()

    # 从环境变量获取 API 密钥
    api_key = os.getenv('DASHSCOPE_API_KEY')
    if not api_key:
        raise ValueError("请设置环境变量 DASHSCOPE_API_KEY")

    # 配置 DashScope 语言模型
    llm = DashScope(
        model="deepseek-v3",  # 使用 DeepSeek v3 模型
        api_key=api_key,
        temperature=0.7,      # 温度参数，控制生成随机性
        top_p=0.8,           # 核采样参数，控制输出多样性
    )

    # 配置文本嵌入模型
    embed_model = DashScopeEmbedding(
        model_name="text-embedding-v1",
        api_key=api_key,
    )

    return llm, embed_model

配置语言模型和嵌入模型，温度和 top_p 参数用于控制生成的随机性和多样性。

文档加载与索引创建

def load_documents_and_create_index(file_dir: str = './docs'):
    persist_dir = "./storage"

    # 尝试从存储中加载已存在的索引
    if os.path.exists(persist_dir):
        try:
            storage_context = StorageContext.from_defaults(
                persist_dir=persist_dir)
            index = load_index_from_storage(storage_context)
            print("从存储加载索引成功")
            return index
        except Exception as e:
            print(f"加载索引失败: {e}，将重新创建索引")

    # 检查文档目录是否存在
    if not os.path.exists(file_dir):
        print(f"文档目录 {file_dir} 不存在")
        return None

    # 使用 SimpleDirectoryReader 读取所有文档
    # 自动支持多种格式：txt, pdf, docx, html, md 等
    reader = SimpleDirectoryReader(file_dir)
    documents = reader.load_data()

    if not documents:
        print("没有找到任何文档")
        return None

    print(f"加载了 {len(documents)} 个文档")

    # 创建向量存储索引
    index = VectorStoreIndex.from_documents(documents)

    # 持久化索引到本地
    index.storage_context.persist(persist_dir=persist_dir)
    print(f"索引已保存到 {persist_dir}")

    return index

实现索引的持久化存储，首次运行创建索引并保存，后续运行直接加载，提升启动速度。支持多种文档格式的自动识别和加载。

ReAct 智能体创建

def create_agent(index, llm):
    # 创建检索器，返回最相似的 5 个文档片段
    retriever = index.as_retriever(similarity_top_k=5)
    
    # 创建查询引擎
    query_engine = index.as_query_engine(similarity_top_k=5)

    # 定义系统提示词
    system_instruction = '''你是一个乐于助人的AI助手。
你可以从给定的文档中检索相关信息来回答用户的问题。
你总是用中文回复用户。'''

    # 创建检索工具函数
    def retrieve_documents(query: str) -> str:
        """从文档中检索相关信息的工具函数"""
        response = query_engine.query(query)
        return str(response)

    # 将检索函数包装为 LlamaIndex 工具
    retrieve_tool = FunctionTool.from_defaults(fn=retrieve_documents)

    # 创建 ReAct 框架智能体
    agent = ReActAgent(
        tools=[retrieve_tool],          # 可用工具列表
        llm=llm,                        # 使用的 LLM 模型
        verbose=True,                   # 输出详细执行过程
        system_prompt=system_instruction,
        streaming=False,                # 禁用流式输出
    )

    return agent, retriever

ReAct 框架允许智能体通过推理（Reasoning）和行动（Acting）循环解决问题，可以进行多步推理和工具调用。

异步主函数

async def main():
    # 配置 LLM 和 Embedding 模型
    llm, embed_model = setup_llm_and_embedding()
    Settings.llm = llm
    Settings.embed_model = embed_model

    # 加载文档并创建向量索引
    index = load_documents_and_create_index()
    if index is None:
        print("无法创建索引，程序退出")
        return

    # 创建智能体和检索器
    agent, retriever = create_agent(index, llm)

    # 定义查询示例
    query = "介绍下雇主责任险"
    print(f"\n用户查询: {query}\n")

    # 展示检索过程和结果
    print("\n===== 召回的文档内容 =====")
    retrieved_nodes = retriever.retrieve(query)
    if retrieved_nodes:
        for i, node in enumerate(retrieved_nodes):
            print(f"\n文档片段 {i+1}:")
            print(f"内容: {node.text[:200]}...")
            print(f"元数据: {node.metadata}")
            if hasattr(node, 'score'):
                print(f"相似度分数: {node.score}")
    else:
        print("没有召回任何文档内容")
    print("===========================\n")

    # 使用智能体回答问题
    print("\n===== 智能体回复 =====")
    result = await agent.run(query)
    if hasattr(result, "response") and hasattr(result.response, "content"):
        print(result.response.content)
    else:
        print(result)
    print("======================\n")


if __name__ == "__main__":
    asyncio.run(main())

使用 asyncio 实现异步执行，提升性能。展示完整的问答流程，包括文档召回和智能体回复。

核心特点

1. 多格式文档支持

SimpleDirectoryReader 自动支持多种文档格式：txt、pdf、docx、html、md 等，无需手动指定加载器。

2. 索引持久化

首次运行创建索引并保存到 ./storage 目录，后续运行直接加载，避免重复计算。

3. ReAct 智能推理

智能体可以进行多步推理，根据需要调用检索工具，处理复杂查询更加智能。

4. 详细日志输出

展示召回的文档片段、相似度分数和元数据，便于调试和理解检索过程。

应用场景

这种实现方式适合：

• 企业知识库问答系统
• 技术文档智能检索
• 学术论文问答助手
• 法律文档查询系统
• 客服知识库应用

扩展建议

可以进一步优化：

• 添加更多工具（搜索、计算器、代码执行等）
• 实现对话历史管理，支持多轮对话
• 添加文档更新和增量索引功能
• 集成更多检索策略（混合检索、重排序等）
• 添加评估指标，监控系统性能