企业级 RAG 项目里，LangChain 不是唯一选择。做私有数据检索，Llama Index 更顺手——MIT 协议开源，定位就是大模型和私有数据之间的中间层，能把 PDF、Word 等本地或内网文档建成索引，缓解大模型读不到私有资料、张口就编的问题。知识问答、AI 智能体都能拿它搭。
先起两个模型服务
本地低资源场景，走 GGUF 量化路线，起两个模型：一个管对话推理，一个管向量嵌入。
对话模型接用户提问、逻辑推理、文本生成；Embedding 模型把文本转成低维稠密向量，让机器能算语义相似度，支撑检索、匹配、召回。
两个模型分别绑不同本地端口，独立提供服务。
加载通义千问对话量化模型，绑定本地端口，设好上下文窗口，满足日常对话、文本生成、问答推理。
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
向量模型要多配几个参数才能出向量：开启嵌入功能，用均值池化聚合文本向量保证语义表征精准，同时限制批次参数提升本地部署平稳性。
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
批量装 RAG 项目核心依赖，涵盖框架体系、向量数据库、文档解析、API 适配等全套包。
# 配置国内PIP源
CMD> pip config set global.index-url
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
验证对话接口是否通
通过 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"] = "
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)
代码跑通后输出提示信息：
你好！我是阿里云的大规模语言模型，我叫通义千问。
向量 API 要自己写适配类
这里有个坑：Llama Index 没有默认的本地向量模型适配接口，得自定义一个调用类，对接 11434 端口向量服务，才能把「文档加载-分块向量化-索引存储-语义检索-大模型生成」整条 RAG 流程串起来。
在工程根目录新建文件夹，创建测试文档，写入 RAG 相关语料，供后续检索验证。
1. 人工智能大模型能够理解人类自然语言
2. 大模型基于海量文本数据进行预训练
3. 向量数据库用来存储文本对应的向量嵌入
4. 向量检索通过计算向量相似度找到相似内容
5. 文本嵌入模型可以把句子转为多维向量
6. 语义检索不依赖关键词匹配，理解句子含义
7. 关键词检索只匹配字面文字，无法理解语义
8. RAG检索增强生成依靠检索外部知识库
9. RAG可以降低大模型产生幻觉的概率
10. 分块策略会影响知识库检索的效果
向量检索代码的逻辑：先自定义对接本地服务的向量化实现，设定接入本地大模型作为 LLM，设置全局模型，运行时读取说明、加载文档、构建内存向量索引、产出查询引擎，最后发起查询，让大模型基于检索到的文档片段总结核心内容。
整条链路：自定义本地向量模型调用类 → 绑定本地对话 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"] = "
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="
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，自定义对接本地 Qwen 3 嵌入模型的 Embedding 类，用本地 OpenAI 兼容 API 的 Qwen 2.5 大模型，实现向量索引的创建、持久化存储与加载，读 pdf、docx、txt 文档建向量库，查询引擎检索 top 3 相关文本片段交给大模型回答。
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"] = "
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="
# 初始化管理器
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. 向量数据库用来存储文本对应的向量嵌入
封装成文档检索接口
在向量数据库基础上做封装优化，新增句子文本分片器，用 Index Manager 类统一完成本地大模型与自定义 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"] = "
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="
)
# 全局设置
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'
}
]
回答：文档内容描述了人工智能大模型能够理解人类自然语言，大模型基于海量文本数据进行预训练，向量数据库用来存储文本对应的向量嵌入。
换 Summary Index 做摘要
改用 Llama Index 的 Summary Index 摘要索引，复用本地 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"] = "
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="
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. 分块策略会影响知识库检索的效果
一个部署提醒
GGUF 量化模型在 CPU 上跑得动，但向量模型批次参数别开太大，本地内存吃紧时容易 OOM。实测把批次压到个位数更稳，代价是建索引慢一些。生产环境不建议用纯内存索引，进程一重启索引就没了，接向量数据库做持久化更靠谱。
