基于 LlamaIndex+PostgreSQL 实现RAG持久化
前文《基于 LlamaIndex 实现 RAG 向量检索入门》已完成本地大模型与 Embedding 向量检索的基础搭建，但仅支持临时向量存储，无法持久化复用。本文将在此基础上实现 RAG 持久化存储方案，借助 PostgreSQL + pgvector 向量插件对接 LlamaIndex。环境采用 CentOS Stream 10，演示数据库编译安装与插件配置，基于 VectorStoreRetriever 实现持久化向量检索，搭建可稳定复用的本地 RAG 服务。
一、编译数据库
PostgreSQL（简称 PG）是一款成熟开源的对象关系型数据库，具备强大的可扩展能力。借助 pgvector 插件，它可以新增向量数据类型，实现向量存储与相似度检索，非常适合用来给 RAG 系统做向量持久化。本节内容基于 CentOS Stream 10 环境，采用源码编译方式安装 PostgreSQL 18.6，并部署 pgvector 向量插件，完成向量数据库基础环境搭建。
1、系统默认未开启编译所需的 CRB 软件源，首先开启 CRB 源并更新缓存，随后安装数据库编译、运行及插件部署所需的全套依赖包，为后续源码编译提供环境。
# 开启CRB源
[root@localhost ~]# dnf config-manager --enable crb
[root@localhost ~]# dnf clean all && dnf makecache
# 开发工具 + PG编译依赖
[root@localhost ~]# dnf groupinstall -y "Development Tools"
[root@localhost ~]# dnf install -y readline-devel zlib-devel openssl-devel libxml2-devel libxslt-devel bison flex git wget libicu-devel systemd-devel perl-core perl-FindBin
Last metadata expiration check: 0:01:56 ago on Thu 17 Sep 2026 06:14:31 PM CST.
Package readline-devel-8.2-11.el10.x86_64 is already installed.
Package zlib-ng-compat-devel-2.2.3-3.el10.x86_64 is already installed.
Package openssl-devel-1:3.5.8-1.el10.x86_64 is already installed.
Package libxml2-devel-2.12.5-15.el10.x86_64 is already installed.
Package libxslt-devel-1.1.39-10.el10.x86_64 is already installed.
Package bison-3.8.2-9.el10.x86_64 is already installed.
Package flex-2.6.4-19.el10.x86_64 is already installed.
Package git-2.52.0-1.el10.x86_64 is already installed.
Package wget-1.24.5-8.el10.x86_64 is already installed.
Package libicu-devel-74.2-5.el10.x86_64 is already installed.
Package systemd-devel-257-33.el10.x86_64 is already installed.
Package perl-4:5.40.2-515.el10.x86_64 is already installed.
Package perl-FindBin-1.54-515.el10.noarch is already installed.
Dependencies resolved.
Nothing to do.
Complete!
2、本文采用官方稳定版 PostgreSQL 18.6 源码包进行编译安装，自定义安装路径至 /usr/local/pgsql，同时编译内置扩展组件，最大化数据库原生能力，编译全程约五分钟。
[root@localhost ~]# wget https://ftp.postgresql.org/pub/source/v18.6/postgresql-18.6.tar.gz
[root@localhost ~]# tar -zxvf postgresql-18.6.tar.gz
[root@localhost ~]# cd postgresql-18.6
# 编译配置
[root@localhost ~]# ./configure --prefix=/usr/local/pgsql --with-openssl --with-libxml --with-systemd --without-icu
# 编译
[root@localhost ~]# make -j$(nproc)
[root@localhost ~]# make install
# 编译安装contrib扩展
[root@localhost ~]# cd contrib
[root@localhost ~]# make -j$(nproc)
[root@localhost ~]# make install
3、PostgreSQL 禁止 root 用户直接运行服务，需创建专用普通用户 postgres 用于进程托管，同时创建独立数据存储目录，并配置严格权限，保障数据库数据安全。
[root@localhost ~]# useradd -M -s /usr/sbin/nologin postgres
[root@localhost ~]# mkdir -p /var/lib/pgsql
[root@localhost ~]# chown postgres:postgres /var/lib/pgsql
[root@localhost ~]# chmod 700 /var/lib/pgsql
4、为方便后续数据库命令全局调用、识别数据目录路径，需为 postgres 用户配置环境变量，刷新后即可正常使用 pg_config、initdb 等核心命令。
[root@localhost ~]# su - postgres
Last login: Thu Sep 17 18:31:18 CST 2026 on pts/1
[postgres@localhost ~]$ vi ~/.bashrc
export PATH=/usr/local/pgsql/bin:$PATH
export PGDATA=/usr/local/pgsql/data
[postgres@localhost ~]$ source ~/.bashrc
[postgres@localhost ~]$ pg_config
[postgres@wintcp ~]$ pg_config
5、通过 initdb 命令初始化数据库核心数据目录、系统表与配置文件，同时设置超级管理员 postgres 密码，完成数据库基础初始化。
[postgres@localhost ~]$ initdb -D /usr/local/pgsql/data -U postgres -W
The files belonging to this database system will be owned by user "postgres".
This user must also own the server process.
The database cluster will be initialized with locale "en_US.UTF-8".
The default database encoding has accordingly been set to "UTF8".
The default text search configuration will be set to "english".
Data page checksums are enabled.
Enter new superuser password: 1233
Enter it again: 1233
fixing permissions on existing directory /usr/local/pgsql/data ... ok
creating subdirectories ... ok
selecting dynamic shared memory implementation ... posix
selecting default "max_connections" ... 100
selecting default "shared_buffers" ... 128MB
selecting default time zone ... Asia/Shanghai
creating configuration files ... ok
running bootstrap script ... ok
performing post-bootstrap initialization ... ok
syncing data to disk ... ok
[postgres@localhost ~]$ exit
logout
6、为实现 PostgreSQL 开机自启、进程统一管理，编写 systemd 服务配置文件，支持 start/stop/restart/reload 标准运维命令。
[root@localhost ~]# vi /etc/systemd/system/postgresql.service
[Unit]
Description=PostgreSQL 18.6 database server
Documentation=https://www.postgresql.org/docs/
After=network.target
[Service]
Type=simple
User=postgres
Group=postgres
Environment=PGDATA=/usr/local/pgsql/data
ExecStart=/usr/local/pgsql/bin/postgres -D ${PGDATA}
ExecReload=/usr/local/pgsql/bin/pg_ctl reload -D ${PGDATA}
ExecStop=/usr/local/pgsql/bin/pg_ctl stop -D ${PGDATA}
TimeoutSec=300
[Install]
WantedBy=multi-user.target
7、加载系统服务、启动数据库进程，配置开机自启，并查看服务运行状态，确认数据库正常启动。
[root@localhost ~]# ln -s /usr/local/pgsql/bin/postgres /usr/local/pgsql/bin/postmaster
[root@localhost ~]# systemctl daemon-reload
[root@localhost ~]# systemctl start postgresql
[root@localhost ~]# systemctl enable postgresql
[root@localhost ~]# systemctl status postgresql
● postgresql.service - PostgreSQL 18.6 database server
Loaded: loaded (/etc/systemd/system/postgresql.service; enabled; preset: disabled)
Active: active (running) since Thu 2026-09-17 18:29:22 CST; 21s ago
Invocation: f6131d3c463748a5a5a4dd3ca09407c0
Docs: https://www.postgresql.org/docs/
Main PID: 16676 (postgres)
Tasks: 9 (limit: 10318)
Memory: 20.8M (peak: 20.8M)
CPU: 35ms
CGroup: /system.slice/postgresql.service
├─16676 /usr/local/pgsql/bin/postgres -D /usr/local/pgsql/data
├─16677 "postgres: io worker 1"
├─16678 "postgres: io worker 0"
├─16679 "postgres: io worker 2"
├─16680 "postgres: checkpointer "
├─16681 "postgres: background writer "
├─16683 "postgres: walwriter "
├─16684 "postgres: autovacuum launcher "
└─16685 "postgres: logical replication launcher "
Sep 17 18:29:22 wintcp systemd[1]: Started postgresql.service - PostgreSQL 18.6 database server.
Sep 17 18:29:22 wintcp postgres[16676]: 2026-09-17 18:29:22.565 CST [16676] LOG: starting PostgreSQL >
Sep 17 18:29:22 wintcp postgres[16676]: 2026-09-17 18:29:22.566 CST [16676] LOG: listening on IPv6 ad>
Sep 17 18:29:22 wintcp postgres[16676]: 2026-09-17 18:29:22.566 CST [16676] LOG: listening on IPv4 ad>
Sep 17 18:29:22 wintcp postgres[16676]: 2026-09-17 18:29:22.569 CST [16676] LOG: listening on Unix so>
Sep 17 18:29:22 wintcp postgres[16682]: 2026-09-17 18:29:22.575 CST [16682] LOG: database system was >
Sep 17 18:29:22 wintcp postgres[16676]: 2026-09-17 18:29:22.578 CST [16676] LOG: database system is r>
8、pgvector 是 PostgreSQL 专用向量检索插件，可拓展向量数据类型、相似度计算能力。通过源码编译方式安装，适配当前编译版 PostgreSQL 18.6，保证版本兼容性。
[root@localhost ~]# git clone https://github.com/pgvector/pgvector.git
[root@localhost ~]# cd pgvector
[root@localhost ~]# make PG_CONFIG=/usr/local/pgsql/bin/pg_config -j$(nproc)
[root@localhost ~]# make install PG_CONFIG=/usr/local/pgsql/bin/pg_config
9、登录数据库启用 vector 扩展，验证插件安装成功，同时创建专属业务数据库 storage_db 和业务用户 storage_user，并授予完整权限，用于后续 LlamaIndex 对接存储向量数据。
[root@localhost ~]# su - postgres
[postgres@localhost ~]$ psql -h 127.0.0.1
psql (16.14, server 18.6)
WARNING: psql major version 16, server major version 18.
Some psql features might not work.
Type "help" for help.
postgres=# CREATE EXTENSION IF NOT EXISTS vector;
CREATE EXTENSION
postgres=# SELECT extname, extversion FROM pg_extension WHERE extname='vector';
extname | extversion
---------+------------
vector | 0.8.6
(1 row)
postgres=# CREATE DATABASE storage_db;
CREATE DATABASE
postgres=# CREATE USER storage_user WITH PASSWORD '1233';
CREATE ROLE
postgres=# GRANT ALL PRIVILEGES ON DATABASE storage_db TO storage_user;
GRANT
postgres=# \c storage_db
psql (16.14, server 18.6)
You are now connected to database "storage_db" as user "postgres".
storage_db=# GRANT ALL ON SCHEMA public TO storage_user;
GRANT
postgres=# exit
[postgres@localhost ~]$ exit
logout
10、默认 PostgreSQL 仅支持本地访问，本文修改配置开启全网远程访问，适配外部程序、服务器对接向量数据库，配置完成后重启服务并验证公网连通性。
# 启动远程访问权限
[root@localhost ~]# vi /usr/local/pgsql/data/postgresql.conf
listen_addresses = '*'
# 添加一行访问控制，按需修改网段
[root@localhost ~]# vi /usr/local/pgsql/data/pg_hba.conf
host all all 0.0.0.0/0 scram-sha-256
11、重启服务并验证公网连接，可正常登录即代表 PostgreSQL + pgvector 向量持久化数据库环境搭建完成，可直接对接 LlamaIndex 实现 RAG 向量持久化存储与检索。
[root@localhost ~]# systemctl restart postgresql
[root@localhost ~]# psql -h 8.122.231.178 -p 5432 -U postgres -d postgres
Password for user postgres: 1233
psql (16.14, server 18.6)
Type "help" for help.
postgres=#
二、使用数据库
在企业落地场景中，一般会搭配元数据过滤实现文档权限隔离，检索结果再接入 Rerank 重排优化召回精度，是企业知识库最常用的基线方案。下面我们通过 LlamaIndex 对接前面部署好的 PostgreSQL+pgvector 环境，演示基础 RAG 示例，再封装成可复用的 RAG 服务类，实现文档增量更新、按文档 ID 删除、元数据过滤查询等实用能力。
- 向量检索的完整链路：文档分块 → Embedding 向量化 → 向量存入向量库 → 生成查询向量返回
PGVector 作为 PostgreSQL 扩展，可直接在关系型数据库中承载向量数据，对比 Qdrant、Milvus 等专用向量库，优势是无需额外维护独立向量服务，同时原生支持元数据过滤，可基于部门、文档类型、权限标签实现企业多租户场景。
安装所需要的依赖包如下所示：
pip install llama-index llama-index-vector-stores-postgres pgvector psycopg2-binary llama-index-embeddings-openai llama-index-llms-openai-like -i https://pypi.org/simple
最小化示例
本示例为最小可用 Demo，基于 LlamaIndex 对接 PGVector。代码中自定义 Embedding 实现类调用本地 Embedding 服务，使用 OpenAILike 接入本地大模型；配置 PGVector 数据库连接参数，创建向量存储对象。首次执行读取 data 目录下文档，自动分块、生成向量并持久化存入 PostgreSQL；
后续运行可直接从数据库加载向量索引，无需重复向量化。最后构建查询引擎，执行向量相似度检索，将召回的上下文交给大模型，完成文档问答。该示例适合验证整套 RAG 链路连通性，仅实现基础入库与问答，没有封装增量更新、文档删除、元数据过滤等生产级能力。
import os
import requests
from typing import List
from llama_index.core import Settings, SimpleDirectoryReader, VectorStoreIndex, StorageContext
from llama_index.core.embeddings import BaseEmbedding
from llama_index.llms.openai_like import OpenAILike
from llama_index.vector_stores.postgres import PGVectorStore
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")
# PGVector 数据库配置
db_name = "storage_db"
host = "8.122.231.178"
password = "1233"
port = "5432"
user = "storage_user"
vector_table_name = "llama_rag_vector"
# Qwen3-Embedding-0.6B 维度 1024
vector_store = PGVectorStore.from_params(
database=db_name,
host=host,
password=password,
port=port,
user=user,
table_name=vector_table_name,
embed_dim=1024,
)
storage_context = StorageContext.from_defaults(vector_store=vector_store)
# 从PG加载索引函数
def load_index_from_pg():
storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex.from_vector_store(
vector_store,
storage_context=storage_context
)
return index
if __name__ == "__main__":
# 第一次运行：构建索引，写入PG向量库
documents = SimpleDirectoryReader(
"./data/",
required_exts=[".pdf", ".docx", ".txt"]
).load_data()
index = VectorStoreIndex.from_documents(
documents,
storage_context=storage_context,
show_progress=True
)
# 第二次及以后运行：直接从PG加载
# index = load_index_from_pg()
query_engine = index.as_query_engine(similarity_top_k=3)
response = query_engine.query("请总结文档里面的核心内容，使用汉语回复")
print("回答：")
print(response)
代码运行输出提示信息：
Applying transformations: 100%|███████████████████████████| 1/1 [00:00<00:00, 797.55it/s]
Generating embeddings: 100%|██████████████████████████████
Generating embeddings: 100%|██████████████████████████████| 1/1 [00:01<00:00, 1.46s/it]
回答：
文档的核心内容是关于人工智能大模型如何理解和处理自然语言，以及如何通过向量数据库和向量检索技术来处理和查找文本内容。
封装示例
本案例将 RAG 业务逻辑封装为独立 RAGService 类，基于 LlamaIndex 与 PGVector。类内部封装模型初始化、数据库连接、文档加载、向量新增 / 删除、问答检索等功能。支持增量添加文档、按文档 ID 删除向量、元数据条件过滤检索，并增加请求异常重试机制。
主函数演示完整调用流程：初始化 RAG 实例、清空历史向量、加载文档入库、执行带元数据过滤的问答查询。相比前面的基础示例，代码模块化，支持文档动态维护，具备生产环境所需的基础容错与权限过滤能力。
import os
import time
import psycopg2
import requests
from pathlib import Path
from typing import List
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from llama_index.core import Settings, SimpleDirectoryReader, VectorStoreIndex, StorageContext, Document
from llama_index.core.embeddings import BaseEmbedding
from llama_index.core.node_parser import SentenceSplitter
from llama_index.llms.openai_like import OpenAILike
from llama_index.vector_stores.postgres import PGVectorStore
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.vector_stores import MetadataFilter, MetadataFilters, FilterOperator
class LocalLlamaServerEmbedding(BaseEmbedding):
api_base: str
embed_model_name: str
api_key: str = "dummy"
def _get_embedding(self, text: str) -> List[float]:
max_text_len = 2048
text = text[:max_text_len]
url = f"{self.api_base}/embeddings"
payload = {
"input": text,
"model": self.embed_model_name
}
headers = {"Authorization": f"Bearer {self.api_key}"}
try:
resp = requests.post(url, json=payload, headers=headers, timeout=30)
resp.raise_for_status()
except requests.exceptions.RequestException as e:
raise RuntimeError(f"Embedding服务调用失败: {e}")
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)
# RAG检索服务类
class RAGService:
def __init__(
self,
db_config: dict,
embed_api_base: str,
llm_base_url: str,
llm_model: str,
embed_model_name: str,
chunk_size: int = 512,
chunk_overlap: int = 50,
batch_size: int = 10
):
self.db_config = db_config
self.embed_api_base = embed_api_base
self.llm_base_url = llm_base_url
self.llm_model = llm_model
self.embed_model_name = embed_model_name
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.batch_size = batch_size
self._vector_store = None
self.splitter = SentenceSplitter(chunk_size=self.chunk_size, chunk_overlap=self.chunk_overlap)
self._init_settings()
def _init_settings(self):
"""初始化LLM与Embedding全局配置"""
os.environ["OPENAI_API_KEY"] = "dummy"
os.environ["OPENAI_BASE_URL"] = self.llm_base_url
llm = OpenAILike(
model=self.llm_model,
api_base=os.environ["OPENAI_BASE_URL"],
api_key=os.environ["OPENAI_API_KEY"],
is_chat_model=True,
context_window=4096,
temperature=0.1
)
Settings.llm = llm
Settings.embed_model = LocalLlamaServerEmbedding(
api_base=self.embed_api_base,
embed_model_name=self.embed_model_name
)
print("[+] LLM与Embedding模型初始化完成")
def _get_vector_store(self) -> PGVectorStore:
"""单例获取PGVectorStore"""
if self._vector_store is None:
print("[+] 初始化PGVectorStore连接")
self._vector_store = PGVectorStore.from_params(
database=self.db_config["database"],
host=self.db_config["host"],
password=self.db_config["password"],
port=self.db_config["port"],
user=self.db_config["user"],
table_name=self.db_config["table_name"],
embed_dim=self.db_config["embed_dim"],
hnsw_kwargs={
"hnsw_m": 16,
"hnsw_ef_construction": 64,
"hnsw_ef_search": 40,
"hnsw_dist_method": "vector_cosine_ops",
},
)
return self._vector_store
def load_index_from_pg(self) -> VectorStoreIndex:
"""从PG加载已有索引"""
vector_store = self._get_vector_store()
storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex.from_vector_store(
vector_store,
storage_context=storage_context
)
return index
def add_or_update_knowledge(self, docs: List[Document]) -> VectorStoreIndex:
"""增量新增/更新文档：存在则删除旧chunk，再写入新文档"""
vector_store = self._get_vector_store()
doc_ids = [doc.metadata["doc_id"] for doc in docs]
print(f"待处理文档doc_ids: {doc_ids}")
filters = MetadataFilters(
filters=[
MetadataFilter(
key="doc_id",
value=doc_ids,
operator=FilterOperator.IN
)
]
)
exist_nodes = vector_store.get_nodes(filters=filters)
exist_doc_ids = {n.metadata["doc_id"] for n in exist_nodes}
print(f"数据库中已存在的doc_ids: {exist_doc_ids}")
new_docs = []
update_doc_ids = []
for d in docs:
if d.metadata["doc_id"] in exist_doc_ids:
update_doc_ids.append(d.metadata["doc_id"])
else:
new_docs.append(d)
if update_doc_ids:
print(f"删除旧文档向量，doc_ids={update_doc_ids}")
del_filters = MetadataFilters(
filters=[
MetadataFilter(
key="doc_id",
value=update_doc_ids,
operator=FilterOperator.IN
)
]
)
vector_store.delete_nodes(filters=del_filters)
if len(docs) > 0:
storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex.from_documents(
docs,
storage_context=storage_context,
transformations=[self.splitter],
show_progress=True
)
print("[+] 知识库写入完成")
return index
else:
print("[-] 没有待处理文档")
return self.load_index_from_pg()
def delete_knowledge(self, doc_id: str):
"""根据doc_id删除文档全部向量片段"""
vector_store = self._get_vector_store()
del_filters = MetadataFilters(
filters=[
MetadataFilter(key="doc_id", value=doc_id, operator=FilterOperator.EQ)
]
)
vector_store.delete_nodes(filters=del_filters)
print(f"[+] 已删除 doc_id={doc_id} 的所有向量片段")
def clear_all_vector(self) -> None:
"""清空整张向量表，如果表不存在则直接跳过"""
vector_store = self._get_vector_store()
table_name = vector_store.table_name
print(f"[-] 准备清空向量表 [{table_name}] 全部数据")
try:
conn = psycopg2.connect(
database=self.db_config["database"],
host=self.db_config["host"],
password=self.db_config["password"],
port=self.db_config["port"],
user=self.db_config["user"]
)
cur = conn.cursor()
# 判断主表是否存在
cur.execute("""
SELECT EXISTS (
SELECT FROM information_schema.tables
WHERE table_name = %s
);
""", (table_name,))
exists = cur.fetchone()[0]
if exists:
cur.execute(f"TRUNCATE TABLE {table_name};")
conn.commit()
print(f"[+] 向量表 {table_name} 已全部清空")
else:
print(f"[*] 表 {table_name} 不存在，无需清空")
cur.close()
conn.close()
except Exception as e:
print(f"清空向量表失败: {str(e)}")
raise
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=5),
retry=retry_if_exception_type((psycopg2.OperationalError, requests.exceptions.RequestException, RuntimeError))
)
def rag_query(self, query_str: str, filter_meta: dict = None, top_k: int = 3):
"""RAG问答查询，支持元数据过滤，带重试"""
start_time = time.time()
index = self.load_index_from_pg()
filters = None
if filter_meta:
filter_list = []
for k, v in filter_meta.items():
if isinstance(v, list):
op = FilterOperator.IN
else:
op = FilterOperator.EQ
filter_list.append(MetadataFilter(key=k, value=v, operator=op))
filters = MetadataFilters(filters=filter_list)
retriever = VectorIndexRetriever(
index=index,
similarity_top_k=top_k,
filters=filters
)
query_engine = RetrieverQueryEngine.from_args(retriever)
response = query_engine.query(query_str)
cost = time.time() - start_time
print(f"Query: {query_str}, cost={cost:.2f}s, hit_chunk_count={len(response.source_nodes)}")
return response
# -------------------------- 全局配置 --------------------------
DB_CONFIG = {
"database": "storage_db",
"host": "8.122.231.178",
"password": "1233",
"port": "5432",
"user": "storage_user",
"table_name": "llama_rag_vector",
"embed_dim": 1024
}
EMBEDDING_API_BASE = "http://127.0.0.1:11434/v1"
LLM_BASE_URL = "http://127.0.0.1:11433/v1"
LLM_MODEL = "qwen2.5-1.5b-instruct-q4_k_m.gguf"
EMBED_MODEL_NAME = "Qwen3-Embedding-0.6B-Q8_0.gguf"
CHUNK_SIZE = 512
CHUNK_OVERLAP = 50
BATCH_SIZE = 10
# -------------------------- 主程序入口示例 --------------------------
if __name__ == "__main__":
# 实例化RAG服务
rag_service = RAGService(
db_config=DB_CONFIG,
embed_api_base=EMBEDDING_API_BASE,
llm_base_url=LLM_BASE_URL,
llm_model=LLM_MODEL,
embed_model_name=EMBED_MODEL_NAME,
chunk_size=CHUNK_SIZE,
chunk_overlap=CHUNK_OVERLAP,
batch_size=BATCH_SIZE
)
# 清空向量表
rag_service.clear_all_vector()
# 读取本地文档
docs = SimpleDirectoryReader(
"./data/",
required_exts=[".pdf", ".docx", ".txt"]
).load_data()
# 同一个文件所有分片共用同一个doc_id，方便按文件整体删除
file_to_docid = {}
for doc in docs:
fname = Path(doc.metadata["file_path"]).name
if fname not in file_to_docid:
file_to_docid[fname] = f"file_{len(file_to_docid)}"
doc.metadata["doc_id"] = file_to_docid[fname]
doc.metadata["source"] = "./data/"
doc.metadata["upload_time"] = time.strftime("%Y-%m-%d %H:%M:%S")
# 增量入库
index = rag_service.add_or_update_knowledge(docs)
print(f"[+] 文档 {len(docs)} 条已成功入库")
print(index)
# 测试问答 并过滤出前Top1个
resp = rag_service.rag_query("概括文档内容，并返回中文。", filter_meta={"source": "./data/"}, top_k=1)
print("---- LLM回答 ----")
print(resp.response)
# 调用index实例删除指定文件的所有分片
rag_service.delete_knowledge("file_0")
# 检索删除后的分片
print("---- 检索到的源片段 ----")
for node in resp.source_nodes:
print(f"相似度分数：{node.score:.4f}")
print(f"元数据：{node.metadata}")
代码运行输出提示信息：
[+] LLM与Embedding模型初始化完成
[+] 初始化PGVectorStore连接
[-] 准备清空向量表 [llama_rag_vector] 全部数据
[*] 表 llama_rag_vector 不存在，无需清空
待处理文档doc_ids: ['file_0', 'file_1']
数据库中已存在的doc_ids: set()
Applying transformations: 100%|█████████████████████████████| 1/1 [00:00<00:00, 590.00it/s]
Generating embeddings: 100%|████████████████████████████████
Generating embeddings: 100%|███████████████████████████████| 2/2 [00:01<00:00, 1.73it/s]
[+] 知识库写入完成
[+] 文档 2 条已成功入库
Query: 概括文档内容，并返回中文。, cost=2.24s, hit_chunk_count=1
---- LLM回答 ----
你好，世界。
[+] 已删除 doc_id=file_0 的所有向量片段
---- 检索到的源片段 ----
相似度分数：0.6165
元数据：
{
'file_path': 'C: \\Users\\Admin\\Documents\\data\\post2.txt',
'file_name': 'post2.txt',
'file_type': 'text/plain',
'file_size': 18,
'creation_date': '2026-09-18',
'last_modified_date': '2026-09-18',
'doc_id': 'file_1',
'source': './data/',
'upload_time': '2026-09-1812: 02: 43'
}
原文地址: https://www.cveoy.top/t/topic/qHAg 著作权归作者所有。请勿转载和采集!