基于 LlamaIndex 实现RAG向量检索入门

本章将通过多个案例实现私有数据集检索功能，虽然 LangChain 也可以实现向量检索，但在企业级项目中，通常会选用 LlamaIndex 这类专门面向检索场景的框架。LlamaIndex 是 MIT 协议开源、以 RAG 为核心的大模型应用开发框架。它充当大模型与私有数据之间的中间层，能够接入 PDF、Word 等各类本地或内网文档，对原始数据构建索引，以此解决大模型无法读取私有资料、容易产生幻觉的问题，可用于搭建知识问答、AI 智能体等应用。

• 项目地址：https://github.com/run-llama/llama_index

一、环境部署

本项目需要部署两个轻量化GGUF量化模型，分别承担对话交互与语义向量检索能力，适配本地低资源部署场景。其中对话大模型负责承接用户提问、逻辑推理、文本生成等通用对话任务；Embedding向量嵌入模型负责将文本、语句等非结构化数据转化为低维稠密向量，让计算机能够量化计算文本间的语义相似度，实现精准检索、内容匹配、知识库召回等功能。

• 对话模型：https://www.modelscope.cn/models/LoveSeaW/Qwen2.5-1.5b-instruct-gguf
• 向量模型：https://www.modelscope.cn/models/guweigui/Qwen3-Embedding-0.6B-Q8_0-GGUF

我们通过llama-server分别启动两个模型服务，绑定不同本地端口，独立提供对话推理和向量生成能力。

加载通义千问对话量化模型，绑定127.0.0.1本地11433端口，设置4096上下文窗口，满足日常对话、文本生成、问答推理需求。

CMD> llama-server.exe -m qwen2.5-1.5b-instruct-q4_k_m.gguf --host 127.0.0.1 --port 11433 -c 4096

load_model: loading model 'qwen2.5-1.5b-instruct-q4_k_m.gguf'
llama_server: model loaded
llama_server: listening on http://127.0.0.1:11433

加载向量量化模型，需额外配置专属参数开启向量生成能力 --embeddings 开启向量嵌入功能，--pooling mean 采用均值池化策略聚合文本向量，保证语义表征精准；同时限制批次参数提升本地部署稳定性。

CMD> llama-server.exe -m Qwen3-Embedding-0.6B-Q8_0.gguf --host 127.0.0.1 --port 11434 -c 4096 --embeddings --pooling mean -b 512 -ub 512

load_model: loading model 'Qwen3-Embedding-0.6B-Q8_0.gguf'
llama_server: model loaded
llama_server: listening on http://127.0.0.1:11434

批量安装RAG项目所需核心依赖，包含框架、向量数据库、文档解析、接口适配等全量包。

# 配置国内PIP源
CMD> pip config set global.index-url https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple
CMD> pip config set global.trusted-host mirrors.tuna.tsinghua.edu.cn
CMD> pip config list

# 安装核心包
CMD> pip install llama-index

# 或安装完整版本（包含所有依赖）
CMD> pip install llama-index[complete]

# 向量数据库支持
CMD> pip install chromadb
CMD> pip install pinecone-client
CMD> pip install weaviate-client

# 文档解析支持
CMD> pip install pypdf
CMD> pip install docx2txt
CMD> pip install python-pptx

# 检索接口通信包
CMD> pip install llama-index-core llama-index-llms-openai-like
CMD> pip install llama-index-embeddings-openai

CMD> pip list
Package                                  Version
---------------------------------------- -----------
llama-index                              0.14.24
llama-index-core                         0.14.24
llama-index-embeddings-openai            0.6.0
llama-index-instrumentation              0.6.0
llama-index-llms-openai                  0.7.10
llama-index-llms-openai-like             0.8.0
llama-index-workflows                    2.23.3

测试Chat对话接口

通过LlamaIndex的OpenAI兼容接口，对接本地11433端口对话服务，实现基础大模型问答调用，验证对话服务可用性。本地llama-server兼容OpenAI接口格式，可直接通过通用类适配调用，无需额外改造服务端。

import os
from llama_index.core import Settings
from llama_index.llms.openai_like import OpenAILike

os.environ["OPENAI_API_KEY"] = "dummy"
os.environ["OPENAI_BASE_URL"] = "http://127.0.0.1:11433/v1"

llm = OpenAILike(
    model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
    api_base=os.environ["OPENAI_BASE_URL"],
    api_key=os.environ["OPENAI_API_KEY"],
    is_chat_model=True,
    context_window=4096
)

if __name__ == "__main__":
    Settings.llm = llm
    resp = llm.complete("你好，大模型！")
    print(resp)

代码运行输出提示信息：

你好！我是阿里云的大规模语言模型，我叫通义千问。

测试Embedding向量接口

LlamaIndex无默认的本地llama-server向量模型适配接口，因此需要自定义Embedding调用类，对接11434端口向量服务，搭建完整的「文档加载-分块向量化-索引存储-语义检索-大模型生成」RAG流程。

在项目根目录新建 data 文件夹，创建 test.txt 测试文档，写入RAG相关测试语料，用于后续检索验证

1. 人工智能大模型能够理解人类自然语言
2. 大模型基于海量文本数据进行预训练
3. 向量数据库用来存储文本对应的向量嵌入
4. 向量检索通过计算向量相似度找到相似内容
5. 文本嵌入模型可以把句子转为多维向量
6. 语义检索不依赖关键词匹配，理解句子含义
7. 关键词检索只匹配字面文字，无法理解语义
8. RAG检索增强生成依靠检索外部知识库
9. RAG可以降低大模型产生幻觉的概率
10. 分块策略会影响知识库检索的效果

使用向量检索代码，先自定义对接本地 embedding 服务LocalLlamaServerEmbedding实现文本向量化，配置OpenAILike接入本地大模型作为LLM，设置全局 Settings 指定模型，运行时读取文档，加载文档后构建内存向量索引，生成查询引擎，最后发起查询让大模型基于检索到的文档片段总结文档核心内容。

• 运行逻辑：自定义本地向量模型调用类 → 绑定本地对话LLM与向量Embedding模型 → 加载本地文档构建向量索引 → 语义检索+大模型总结生成答案

import os
import requests
from typing import List
from llama_index.core import Settings, SimpleDirectoryReader, VectorStoreIndex
from llama_index.core.embeddings import BaseEmbedding
from llama_index.llms.openai_like import OpenAILike

# 自定义Embedding模型
class LocalLlamaServerEmbedding(BaseEmbedding):
    api_base: str
    api_key: str = "dummy"

    def _get_embedding(self, text: str) -> List[float]:
        url = f"{self.api_base}/embeddings"
        payload = {
            "input": text,
            "model": "Qwen3-Embedding-0.6B-Q8_0.gguf"
        }
        headers = {"Authorization": f"Bearer {self.api_key}"}
        resp = requests.post(url, json=payload, headers=headers)
        resp.raise_for_status()
        return resp.json()["data"][0]["embedding"]

    def _get_text_embedding(self, text: str) -> List[float]:
        return self._get_embedding(text)

    def _get_query_embedding(self, query: str) -> List[float]:
        return self._get_embedding(query)

    async def _aget_query_embedding(self, query: str) -> List[float]:
        return self._get_embedding(query)

    async def _aget_text_embedding(self, text: str) -> List[float]:
            return self._get_embedding(text)

# 对话模型地址
os.environ["OPENAI_API_KEY"] = "dummy"
os.environ["OPENAI_BASE_URL"] = "http://127.0.0.1:11433/v1"

llm = OpenAILike(
    model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
    api_base=os.environ["OPENAI_BASE_URL"],
    api_key=os.environ["OPENAI_API_KEY"],
    is_chat_model=True,
    context_window=1024
)

# 向量模型地址
Settings.llm = llm
Settings.embed_model = LocalLlamaServerEmbedding(api_base="http://127.0.0.1:11434/v1")

if __name__ == "__main__":
    # 构建RAG索引 过滤出pdf.docx.txt文件
    documents = SimpleDirectoryReader(
        "./data/",
        required_exts=[".pdf", ".docx", ".txt"]
        ).load_data()

    # 创建索引
    index = VectorStoreIndex.from_documents(documents)

    # 创建查询引擎
    query_engine = index.as_query_engine()

    # 进行查询
    response = query_engine.query("请总结文档里面的核心内容，使用汉语回复")
    print("\n回答：")
    print(response)

代码运行输出提示信息：

回答：
文档的核心内容是关于人工智能大模型如何理解和处理自然语言
以及如何通过向量数据库和向量检索技术来处理和查找文本内容。

二、实际应用

使用向量数据库

该代码基于 Llama Index 框架，自定义对接本地 Qwen3 嵌入模型的 Embedding 类，使用本地 OpenAI 兼容接口的 Qwen2.5 大模型，实现向量索引的创建、持久化存储与加载，读取 pdf、docx、txt 文档构建向量库，通过查询引擎检索 top3 相关文本片段并交由大模型回答用户问题。

import os
import requests
from typing import List
from llama_index.core.embeddings import BaseEmbedding
from llama_index.llms.openai_like import OpenAILike
from llama_index.core import (
    VectorStoreIndex,
    SimpleDirectoryReader,
    Settings,
    StorageContext,
    load_index_from_storage
)

# 自定义Embedding模型
class LocalLlamaServerEmbedding(BaseEmbedding):
    api_base: str
    api_key: str = "dummy"

    def _get_embedding(self, text: str) -> List[float]:
        url = f"{self.api_base}/embeddings"
        payload = {
            "input": text,
            "model": "Qwen3-Embedding-0.6B-Q8_0.gguf"
        }
        headers = {"Authorization": f"Bearer {self.api_key}"}
        resp = requests.post(url, json=payload, headers=headers)
        resp.raise_for_status()
        return resp.json()["data"][0]["embedding"]

    def _get_text_embedding(self, text: str) -> List[float]:
        return self._get_embedding(text)

    def _get_query_embedding(self, query: str) -> List[float]:
        return self._get_embedding(query)

    async def _aget_query_embedding(self, query: str) -> List[float]:
        return self._get_embedding(query)

class IndexManager:
    """索引管理器：负责向量索引的创建、持久化与加载"""
    def create_index(self, data_dir="./data/", persist_dir="./storage/"):
        """创建并持久化向量索引"""
        try:
            documents = SimpleDirectoryReader(
                data_dir,
                required_exts=[".pdf", ".docx", ".txt"],
                recursive=False
            ).load_data()
        except Exception as err:
            print(f"[文档读取解析失败]错误信息：{err}")
            raise err

        print(f"成功读取文档片段数量：{len(documents)}")
        index = VectorStoreIndex.from_documents(documents)
        index.storage_context.persist(persist_dir=persist_dir)
        return index

    def load_index(self, persist_dir="storage"):
        """加载磁盘上已持久化的索引"""
        storage_context = StorageContext.from_defaults(persist_dir=persist_dir)
        index = load_index_from_storage(storage_context)
        return index

if __name__ == "__main__":
    # 对话模型地址
    os.environ["OPENAI_API_KEY"] = "dummy"
    os.environ["OPENAI_BASE_URL"] = "http://127.0.0.1:11433/v1"

    llm = OpenAILike(
        model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
        api_base=os.environ["OPENAI_BASE_URL"],
        api_key=os.environ["OPENAI_API_KEY"],
        is_chat_model=True,
        context_window=1024
    )

    # 向量模型地址
    Settings.llm = llm
    Settings.embed_model = LocalLlamaServerEmbedding(api_base="http://127.0.0.1:11434/v1")

    # 初始化管理器
    manager = IndexManager()
    index = None
    try:
        index = manager.load_index()
        print("索引加载成功")
    except Exception as e:
        print(f"索引加载失败：{e}，创建新索引...")
        index = manager.create_index()

    # 查询索引 返回相似度最高的前3个文档片段
    query_engine = index.as_query_engine(
        similarity_top_k=3,
        response_mode="compact"
    )

    resp = query_engine.query("简单概述文档内容，并回答中文简体。")
    print(f"回答：{resp}\n")

代码运行输出提示信息：

索引加载成功
回答：
文档内容主要介绍了人工智能大模型如何理解和处理自然语言，以及如何通过预训练和向量数据库来存储和检索文本数据。
文档中提到的步骤包括：
1. 人工智能大模型能够理解人类自然语言
2. 大模型基于海量文本数据进行预训练
3. 向量数据库用来存储文本对应的向量嵌入

使用文档检索

这段代码在向量数据库基础上做了封装优化，新增句子文本分片器，在 IndexManager 类中统一完成本地大模型与自定义 Embedding 的初始化，加载或重建向量索引，对外提供问答接口，返回问题答案同时附带检索到的文档源元数据，并且设置系统提示词约束大模型基于文档作答。

import os
import requests
from typing import List
from llama_index.core.embeddings import BaseEmbedding
from llama_index.llms.openai_like import OpenAILike
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core import (
    VectorStoreIndex,
    SimpleDirectoryReader,
    Settings,
    StorageContext,
    load_index_from_storage
)

# 自定义Embedding模型
class LocalLlamaServerEmbedding(BaseEmbedding):
    api_base: str
    api_key: str = "dummy"

    def _get_embedding(self, text: str) -> List[float]:
        url = f"{self.api_base}/embeddings"
        payload = {
            "input": text,
            "model": "Qwen3-Embedding-0.6B-Q8_0.gguf"
        }
        headers = {"Authorization": f"Bearer {self.api_key}"}
        resp = requests.post(url, json=payload, headers=headers)
        resp.raise_for_status()
        return resp.json()["data"][0]["embedding"]

    def _get_text_embedding(self, text: str) -> List[float]:
        return self._get_embedding(text)

    def _get_query_embedding(self, query: str) -> List[float]:
        return self._get_embedding(query)

    async def _aget_query_embedding(self, query: str) -> List[float]:
        return self._get_embedding(query)

    async def _aget_text_embedding(self, text: str) -> List[float]:
        return self._get_embedding(text)

class IndexManager:
    def __init__(self, data_dir="./data/", persist_dir="./storage/"):
        self.data_dir = data_dir
        self.persist_dir = persist_dir
        self.setup_llm()
        self.index = self.load_database()

    def setup_llm(self):
        os.environ["OPENAI_API_KEY"] = "dummy"
        os.environ["OPENAI_BASE_URL"] = "http://127.0.0.1:11433/v1"
        llm = OpenAILike(
            model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
            api_base=os.environ["OPENAI_BASE_URL"],
            api_key=os.environ["OPENAI_API_KEY"],
            is_chat_model=True,
            context_window=1024,
            temperature=0.1,
            system_prompt="""你是一个企业文档助手。
            请基于提供的文档内容回答问题，如果文档中没有相关信息，请明确说明。
            回答要准确、简洁、专业。"""
        )
        # Embedding模型
        embed_model = LocalLlamaServerEmbedding(
            api_base="http://127.0.0.1:11434/v1"
        )
        # 全局设置
        Settings.llm = llm
        Settings.embed_model = embed_model
        # 分段器配置 分片大小 512 重叠 50 个字符
        Settings.node_parser = SentenceSplitter(chunk_size=512, chunk_overlap=50)

    def load_database(self):
        if os.path.exists(self.persist_dir):
            try:
                storage_context = StorageContext.from_defaults(
                    persist_dir=self.persist_dir
                )
                index = load_index_from_storage(storage_context)
                print("[+] 索引加载成功")
                return index
            except Exception as e:
                print(f"[-] 索引加载失败，原因：{e}，重新创建...")
        # 不存在目录 加载失败 则新建索引
        documents = SimpleDirectoryReader(
            self.data_dir,
            required_exts=[".pdf", ".docx", ".txt"],
            recursive=False
        ).load_data()
        print(f"[+] 读取文档片段数量：{len(documents)}")
        index = VectorStoreIndex.from_documents(documents)
        index.storage_context.persist(persist_dir=self.persist_dir)
        print("[+] 索引创建并持久化完成")
        return index

    def get_query_engine(self, similarity_top_k=3):
        """获取默认查询引擎，可后续替换为混合检索引擎"""
        return self.index.as_query_engine(similarity_top_k=similarity_top_k)

    def query(self, question, top_k=3):
        query_engine = self.index.as_query_engine(
            similarity_top_k=top_k,
            response_mode="compact"
        )
        response = query_engine.query(question)
        return {
            "question": question,
            "answer": str(response),
            "sources": [node.metadata for node in response.source_nodes]
        }

if __name__ == "__main__":
    manager = IndexManager(data_dir="./data/", persist_dir="./storage/")
    res = manager.query("请简单概述文章内容，15个字以内。")
    print("问题：", res["question"])
    print("回答：", res["answer"])
    print("来源：", res["sources"])

    query_engine = manager.get_query_engine()
    response = query_engine.query("请简述文档内容，中文回答。")
    print(f"回答：{response}")

代码运行输出提示信息：

[+] 索引加载成功
问题： 请简单概述文章内容，15个字以内。
回答： RAG检索增强生成依靠检索外部知识库，可以降低大模型产生幻觉的概率，分块策略会影响知识库检索的效果。
来源：
[
    {
        'file_path': 'C: \\Users\\Admin\\data\\test.txt',
        'file_name': 'test.txt',
        'file_type': 'text/plain',
        'file_size': 562,
        'creation_date': '2026-09-17',
        'last_modified_date': '2026-09-17'
    }
]

回答：文档内容描述了人工智能大模型能够理解人类自然语言，大模型基于海量文本数据进行预训练，向量数据库用来存储文本对应的向量嵌入。

使用文档摘要

代码改用 LlamaIndex 的 SummaryIndex 摘要索引，复用本地 Qwen 大模型与自定义嵌入模型，读取本地文档后支持生成简洁摘要、详细摘要以及提取文档关键要点，采用 tree_summarize 模式对文档内容做汇总，适合对文档整体内容进行归纳提取。

import os
import requests
from typing import List
from llama_index.core.embeddings import BaseEmbedding
from llama_index.llms.openai_like import OpenAILike
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core import (
    SummaryIndex,
    SimpleDirectoryReader,
    Settings
)

class LocalLlamaServerEmbedding(BaseEmbedding):
    api_base: str
    api_key: str = "dummy"

    def _get_embedding(self, text: str) -> List[float]:
        url = f"{self.api_base}/embeddings"
        payload = {
            "input": text,
            "model": "Qwen3‑Embedding‑0.6B‑Q8_0.gguf"
        }
        headers = {"Authorization": f"Bearer {self.api_key}"}
        resp = requests.post(url, json=payload, headers=headers)
        resp.raise_for_status()
        return resp.json()["data"][0]["embedding"]

    def _get_text_embedding(self, text: str) -> List[float]:
        return self._get_embedding(text)

    def _get_query_embedding(self, query: str) -> List[float]:
        return self._get_embedding(query)

    async def _aget_query_embedding(self, query: str) -> List[float]:
        return self._get_embedding(query)

class DocumentSummarizer:
    def __init__(self):
        self.setup_llm()

    def setup_llm(self):
        os.environ["OPENAI_API_KEY"] = "dummy"
        os.environ["OPENAI_BASE_URL"] = "http://127.0.0.1:11433/v1"

        llm = OpenAILike(
            model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
            api_base=os.environ["OPENAI_BASE_URL"],
            api_key=os.environ["OPENAI_API_KEY"],
            is_chat_model=True,
            context_window=1024,
            temperature=0.3,
            system_prompt="你是一个专业的文档摘要助手，请提供准确、简洁的摘要。"
        )
        embed_model = LocalLlamaServerEmbedding(api_base="http://127.0.0.1:11434/v1")
        Settings.llm = llm
        Settings.embed_model = embed_model
        Settings.node_parser = SentenceSplitter(chunk_size=512, chunk_overlap=50)

    def summarize_documents(self, documents, summary_type="brief"):
        try:
            summary_index = SummaryIndex.from_documents(documents)
        except Exception as e:
            print(f"[构建SummaryIndex失败]{e}")
            raise e

        # 三个选择摘要类型
        if summary_type == "brief":
            prompt = "请为这些文档生成一个简洁的摘要，突出主要观点。"
        elif summary_type == "detailed":
            prompt = "请为这些文档生成一个详细的摘要，包括关键信息和重要细节。"
        else:
            prompt = "请为这些文档生成摘要。"

        query_engine = summary_index.as_query_engine(
            response_mode="tree_summarize"
        )
        response = query_engine.query(prompt)
        return str(response)

    def extract_key_points(self, documents):
        try:
            summary_index = SummaryIndex.from_documents(documents)
        except Exception as e:
            print(f"[构建SummaryIndex失败]{e}")
            raise e

        query_engine = summary_index.as_query_engine()
        response = query_engine.query(
            "请列出这些文档中的关键要点，以项目符号的形式呈现。"
        )
        return str(response)

if __name__ == "__main__":
    summarizer = DocumentSummarizer()
    try:
        documents = SimpleDirectoryReader(
            "./data/",
            required_exts=[".pdf", ".docx", ".txt"],
            recursive=False
        ).load_data()
    except Exception as err:
        print(f"[文档读取解析失败]错误信息：{err}")
        raise err

    print(f"读取文档片段数：{len(documents)}")
    brief_summary = summarizer.summarize_documents(documents, "brief")
    print("[简洁摘要]")
    print(brief_summary)

    detailed_summary = summarizer.summarize_documents(documents, "detailed")
    print("[详细摘要]")
    print(detailed_summary)

    key_points = summarizer.extract_key_points(documents)
    print("[关键要点]")
    print(key_points)

代码运行输出提示信息：

读取文档片段数：1

[简洁摘要]
这些文档的主要观点包括：
1. 人工智能大模型能够理解人类自然语言。
2. 大模型基于海量文本数据进行预训练。
3. 向量数据库用来存储文本对应的向量嵌入。
4. 向量检索通过计算向量相似度找到相似内容。
5. 文本嵌入模型可以把句子转为多维向量。
6. 语义检索不依赖关键词匹配，理解句子含义。
7. 关键词检索只匹配字面文字，无法理解语义。
8. RAG检索增强生成依靠检索外部知识库。
9. RAG可以降低大模型产生幻觉的概率。
10. 分块策略会影响知识库检索的效果。

[详细摘要]
这些文档主要讨论了人工智能大模型的几个关键点，包括：1. 人工智能大模型能够理解人类自然语言；2. 大模型基于海量文本数据进行预训练；3. 向量数据库用来存储文本对应的向量嵌入；4. 向量检索通过计算向量相似度
找到相似内容；5. 文本嵌入模型可以把句子转为多维向量；6. 语义检索不依赖关键词匹配，理解句子含义；7. 关键词检索只匹配字面文字，无法理解语义；8. RAG检索增强生成依靠检索外部知识库；9. RAG可以降低大模型产
生幻觉的概率；10. 分块策略会影响知识库检索的效果。

[关键要点]
1. 分块策略会影响知识库检索的效果