手搓本地 RAG：向量库 + Qwen3 跑通鸢尾花问答
在本地（完全不出网）搭一套能用的 RAG 到底是什么体感？我的答案是：向量库 + qwen3-embed（向量化）+ 局域网大模型（问答），全部用 OpenAI 兼容 API 串起来，用 LangChain 组装，拿天池上的鸢尾花数据集（150 行经典 CSV）当试验田。 本文选用 pgvector 作为向量库的示例实现，部署见《Docker 部署 pgvector》——本文聚焦 RAG 全链路本身。
这篇文章不是「跑通一个 demo 就完」，而是每一层都讲到你能照着在自己的机器上复现。数据到模型全部局域网内完成，中途不访问任何公网。
我想做什么 / 成功判据
想验证的不是「RAG 是什么」（文档一搜一大把），而是一套完全本地的 RAG 从无到有要经过哪些真实的组件分工：
CSV 数据 → 整理成一句话文档 → qwen3-embed 向量化 → 向量库入库
→ 检索（相似度）→ 拼上下文 → 局域网 LLM 生成答案开工前定成功判据，避免「看着像成了」：
向量库（pgvector）容器起来，
CREATE EXTENSION vector成功；鸢尾花数据集的每一行被整理成一条可读的文本记录，并全部向量化入库（0 条漏）；
给定一个自然语言问题（如「花瓣最长的鸢尾花是什么品种？」），能检索回相关样本片段；
局域网 LLM 能基于检索到的上下文回答，且引用的是库里真实存在的数据，不是凭空编造。
第 4 条是整套系统的底线，也是 RAG 区别于「裸大模型」的价值——答案必须有据可查。
环境基线：Windows（宿主，装有 Docker Desktop + Python 3.13）；局域网内预先跑好 OpenAI 兼容 API 服务（vLLM 或同类网关），暴露
qwen3-embed（embedding）、qwen3（chat）两个模型。两个模型共用同一个 base_url，只是 model 名不同。
过程记录
运行方式：下面所有 Python 代码块按顺序在同一个脚本里运行，变量跨块共享（第 5 步定义的
retriever在第 6 步使用）。不要逐块单独执行，否则跨块的变量会报Unresolved reference。
第 0 步：数据来源——从天池 3494 拿鸢尾花
鸢尾花数据集（Iris，150 行，5 个数值特征 + 品种标签）是每个 ML 入门都见过的经典数据。我从天池数据集平台取：https://tianchi.aliyun.com/dataset/3494
天池的下载需要登录会话，没有公开直链可让脚本自动拉取。所以落地方式是：在已登录的天池页面点下载，把 CSV 存到项目 data/ 目录，再用 pandas 读取。
mkdir data
# 从天池 3494 页面手动下载 iris.csv → data/iris.csvimport pandas as pd
df = pd.read_csv("data/iris.csv")
print(df.shape, df.columns.tolist())
# (150, 5) ['sepal_length', 'sepal_width', 'petal_length', 'petal_width', 'species']150 行、5 列，字段就是鸢尾花最经典的四个测量值（花萼/花瓣的长宽）加品种。数据小到每一条输入输出都可读、可核对——这正是拿它当 RAG 试验田的原因：全链路每一步的输入输出都能看得清清楚楚，而不是丢进一个不可解释的大矩阵。
第 1 步：数据整理——把「行」变成「一句话文档」
RAG 检索的粒度是文档 / 段落，不是结构化表格。所以第一步是把每一行鸢尾花测量值压成一条可检索的文本。关键两点：把英文品种名转成中文，并把数值特征转成带单位、带语义的自然语言：
SPECIES_CN = {
"setosa": "山鸢尾",
"versicolor": "变色鸢尾",
"virginica": "维吉尼亚鸢尾",
}
def row_to_doc(row, ridx):
parts = [f"第 {ridx} 朵鸢尾花"]
parts.append(f"花萼长 {row['sepal_length']:.1f} 厘米")
parts.append(f"花萼宽 {row['sepal_width']:.1f} 厘米")
parts.append(f"花瓣长 {row['petal_length']:.1f} 厘米")
parts.append(f"花瓣宽 {row['petal_width']:.1f} 厘米")
parts.append(f"品种 {SPECIES_CN.get(row['species'], row['species'])}")
return "，".join(parts) + "。"
docs = []
for i, row in df.iterrows():
docs.append({
"text": row_to_doc(row, i),
"metadata": {"rid": int(i), "species": row["species"]},
})
print(docs[0])
# 第 0 朵鸢尾花，花萼长 5.1 厘米，花萼宽 3.5 厘米，花瓣长 1.4 厘米，花瓣宽 0.2 厘米，品种 山鸢尾。这是本次实验最容易踩的地方：如果直接把整行 CSV 原样丢进向量库，中文会话模型问「花萼最宽的山鸢尾」这类问题会非常吃力——因为模型看到的是
5.1,3.5,1.4,0.2,setosa这种没有语义的串。向量化的对象必须是「人话」，不是原始行列。 而且species不转中文，中文问题与英文原文的语义对齐度会变差，检索就抓不准。
第 2 步：向量库（pgvector，部署见专题）
向量库负责存 embedding 并做相似度检索。本文示例用 pgvector（PostgreSQL 的向量扩展），部署细节——Docker 镜像、开 vector 扩展、端口 5433 与本机已有 PostgreSQL 错开、连接与超时配置——不在这里展开，见《Docker 部署 pgvector》
这里只需要确认本地已经有一个能连的向量库，并用 psql 做一次真实验证（注意：vector(3) 是建表列时的类型声明，不能在 SELECT 里当函数调用；构造向量要用类型转换）：
# 假设 pgvector 容器已起（部署见 docker/08-pgvector.md）
docker compose exec pgvector psql -U postgres -d postgres -c "SELECT '[1,2,3]'::vector;"能返回 [1,2,3] 即扩展就绪。后续代码用 postgresql+psycopg://postgres:YourStrong!Passw0rd@localhost:5433/postgres 连接，落地时换成你实际部署的地址、密码和库名。
第 3 步：Python 依赖
uv init # 生成 pyproject.toml 并自动建 .venv
uv add langchain langchain-core langchain-openai langchain-postgres pandas "psycopg[binary]"依赖装进 .venv，代码直接跑 uv run python xxx.py 即可，不用手动激活虚拟环境。
langchain-core提供ChatPromptTemplate/Document；langchain提供LCEL组合能力；langchain-postgres提供PGVector；langchain-openai让你能用OpenAIEmbeddings/ChatOpenAI指向局域网服务。embedding 和 chat 都走 OpenAI 兼容协议，所以复用同一套客户端。注意新版 langchain（1.x）已移除旧的RetrievalQAChain API，问答链路用prompt.format_messages+llm.invoke手动拼（见第 6 步）。
第 4 步：向量化 + 入库
局域网服务暴露在 http://192.168.x.x:8000（换成你实际的地址），用 OpenAI 兼容协议：
from langchain_openai import OpenAIEmbeddings
# ---- 向量库（embedding 服务）----
EMBED_BASE_URL = "http://192.168.x.x:8000/v1" # embedding 服务地址
EMBED_API_KEY = "EMPTY" # vLLM 默认接受任意 key
EMBED_MODEL = "qwen3-embed" # embedding 模型名
embeddings = OpenAIEmbeddings(
model=EMBED_MODEL,
base_url=EMBED_BASE_URL,
api_key=EMBED_API_KEY,
# 不要显式传 dimensions：qwen3-embed 默认即 1024 维，且默认不开
# matryoshka，传了会被 vLLM 以 400 拒绝（见踩坑第 4 条）
check_embedding_ctx_length=False,
)先验证本地服务能被 LangChain 拿到向量：
vec = embeddings.embed_query("花萼很宽的山鸢尾")
print(len(vec), vec[:3]) # (1024, [0.012, -0.03, 0.007, ...])入库（用 from_documents 自动建表）：
from langchain_postgres import PGVector
from langchain_core.documents import Document
CONN = "postgresql+psycopg://postgres:YourStrong!Passw0rd@localhost:5433/postgres"
records = [
Document(
page_content=d["text"],
metadata=d["metadata"],
# 固定 id：让 from_documents 的 upsert 命中同一行，重复执行不追加
id=f"iris-{d['metadata']['rid']}",
)
for d in docs
]
store = PGVector.from_documents(
records,
embeddings,
collection_name="iris_docs",
connection=CONN,
)重复执行不会堆数据：
PGVector.from_documents内部是INSERT ... ON CONFLICT (langchain_document_id) DO UPDATE。Document 没显式id时，LangChain 每次现生成新 uuid 当主键，ON CONFLICT永远不命中 → 每次都是新行。给每条文档固定id（这里用iris-{rid}），upsert 才按 id 覆盖，跑多少遍都是 150 行。若确实想推倒重来，PGVector(connection=CONN, collection_name="iris_docs").delete_collection()后重建。
不是一次性、也不是逐条循环，而是分批：
from_documents内部add_documents按batch_size（默认 512）切片，每批只做一次embed_documents（一次 HTTP 把整批文本发给 embedding 服务）+ 一次批量INSERT。150 条 < 512，一批处理完。这比手写for d in docs: embed_query(d)快得多——后者 N 条 = N 次 HTTP 往返 + N 次单行插入。
数据上规模时：embedding 服务吞吐是瓶颈，可以 ① 调小
batch_size（PGVector.from_documents(..., batch_size=128)，单批 HTTP 体更小、不超时）；② 或自己分批add_documents，每批之间能观察进度、控制并发：store = PGVector.from_documents(records[:64], embeddings, collection_name="iris_docs", connection=CONN) for i in range(64, len(records), 64): store.add_documents(records[i:i+64])百万级再上 HNSW 索引 + 并发 worker，那就不在本文范围了。
这里有个坑：collection_name 会为当前数据建一张固定的向量表；换数据集时，要么换 collection_name，要么先删旧集合。见踩坑部分。
第 5 步：纯向量检索
retriever = store.as_retriever(search_kwargs={"k": 8})
hits = retriever.invoke("花瓣比较大的变色鸢尾")
for h in hits:
print(f"[{h.metadata['rid']}] {h.page_content}")这就是「向量检索」层。150 条的小数据上，纯向量检索已经够用——它把语义最相近的样本排到前面，直接拿 Top K 拼给大模型即可。下面把问题 + 上下文交给局域网 LLM 生成答案。
第 6 步：局域网 LLM 问答（LangChain 组装）
到这里再补一个完整可独立运行的示例：它从 pgvector 里加载已经入库的 iris_docs 集合（不再向量化、不再入库），检索 → 拼上下文 → 生成 → 溯源，单独存成一个 ask.py 就能跑。注：新版 langchain（1.x）已移除旧的 RetrievalQA Chain API，这里用 retriever.invoke + prompt.format_messages + llm.invoke 手动拼链路：
# ask.py —— 完整示例：加载已有向量库 → 检索 → 生成 → 溯源
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_postgres import PGVector
from langchain_core.prompts import ChatPromptTemplate
# ---- 向量库（embedding 服务）----
EMBED_BASE_URL = "http://192.168.x.x:8000/v1" # embedding 服务地址
EMBED_API_KEY = "EMPTY" # vLLM 默认接受任意 key
EMBED_MODEL = "qwen3-embed" # embedding 模型名
# ---- 大语言模型（chat 服务）----
LLM_BASE_URL = "http://192.168.x.x:8000/v1" # chat 服务地址（通常与 embedding 同一个网关）
LLM_API_KEY = "EMPTY" # vLLM 默认接受任意 key
LLM_MODEL = "qwen3" # chat 模型名
# ---- 向量库存储（pgvector）----
CONN = "postgresql+psycopg://postgres:YourStrong!Passw0rd@localhost:5433/postgres"
COLLECTION = "iris_docs"
# -----------------------------------
# 1. 连回已入库的向量库（不再重新向量化/入库）
embeddings = OpenAIEmbeddings(
model=EMBED_MODEL,
base_url=EMBED_BASE_URL,
api_key=EMBED_API_KEY,
# 不要显式传 dimensions：qwen3-embed 默认 1024 维且默认不开 matryoshka（见踩坑第 3 条）
check_embedding_ctx_length=False,
)
store = PGVector(
connection=CONN,
collection_name=COLLECTION,
embeddings=embeddings,
)
retriever = store.as_retriever(search_kwargs={"k": 8})
# 2. 局域网 chat 模型
llm = ChatOpenAI(
model=LLM_MODEL,
base_url=LLM_BASE_URL,
api_key=LLM_API_KEY,
temperature=0.0,
)
prompt = ChatPromptTemplate.from_messages([
("system", "你是严谨的数据库问答助手。只根据给定的上下文中确实出现的信息作答，"
"不要编造数据；若上下文不足以回答，就明说不知道。"),
("human", "上下文：\n{context}\n\n问题：{question}"),
])
# 3. 检索 → 拼上下文 → 生成
question = "花瓣比较长的鸢尾花是哪个品种？"
docs = retriever.invoke(question) # 检索 Top K
context = "\n\n".join(d.page_content for d in docs) # 拼上下文
messages = prompt.format_messages(context=context, question=question)
answer = llm.invoke(messages).content # 生成
print(answer)
print("--- 引用来源 ---")
for d in docs:
print(f"[{d.metadata['rid']}] {d.page_content}")跑通后，你就拿到了一条端到端的本地 RAG：数据在向量库（pgvector）、语义在 qwen3-embed、答案出自局域网大模型，全程不出内网。
结果与数据
这一段在方案验证后回填。这里是计划要采集并展示的判据，避免「看着像成了」。
将要对比的典型问题：
「花瓣最长的鸢尾花是什么品种？」（数值极值 + 品种归类）
「山鸢尾的平均花萼宽度大概是多少？」（品种过滤 + 数值聚合）
「花瓣宽超过 2 厘米的鸢尾花有哪些？」（数值范围过滤）
如果向量检索确实把更对口的样本顶到 Top K，这个对比表格就能给出量化差异。
补充说明：鸢尾花只有 150 行，对真实用的 RAG 是「小到几乎不会检索不到」。选它不是为了证明「检索很强」，而是为了让全链路每一环的输入输出都可读、可核对——向量长什么样、检索怎么排、LLM 怎么拼上下文，在小数据上看得最清楚。跑通后再换几万行的真实文本语料，逻辑不变。
复盘
真正的方法论只有一句：RAG 的复杂度不在「调大模型」，而在数据进入向量库之前和检索之后的「姿势」。
数据整理决定了天花板。 同一份 CSV，直接塞行列和整理成「人话文档」，检索质量天差地别。向量库不会替你理解
5.1,3.5,1.4,0.2,setosa，它只按你给它的语义文本排序。向量检索的 k 值决定上下文质量。 取少了漏掉真对口的样本，取多了把噪音塞进 prompt 稀释注意力。小数据上 k=8 够用，数据上规模时再按命中率调。
OpenAI 兼容协议是真香。 embedding、chat 两个异质模型，因为都走 OpenAI 兼容，LangChain 一套客户端全打通，局域网部署和换云端 API 只是改 base_url 的事。
答案必须能溯源。
return_source_documents=True让你每次都能看到回答引用了哪些片段。没有溯源，RAG 就退化成「看起来一本正经地瞎编」。
还没做、准备接着做的：
把答案与引用来源做成一个可复用的 Python 模块 / skill，暴露「index / ask」两个入口（对应新建知识库与问答）；
换更大的真实文本语料（新闻 / 问答对），量化向量检索 Top-K 命中率；
用
pgvector的 HNSW 索引做一次大规模插入/检索基准（CREATE INDEX ... USING hnsw (embedding vector_cosine_ops)），对比建索引前后耗时；试一次带过滤的元数据检索（如先按
species='setosa'过滤，再向量检索，验证「结构化约束 + 语义检索」的混合检索）。
暂时不碰的：多路召回（BM25 + 向量）、知识图谱 RAG、Agent 记忆、重排序层。先用最小闭环把「数据 → 向量 → 检索 → 生成 → 溯源」跑稳，再上一路复杂度。
踩坑与解决
1. 直接把 CSV 行列塞进向量库，检索一塌糊涂
问题：用原始表格每行做 page_content，中文问题检索不到对口样本。
复现：
page_content = str(row.to_dict())，问「花萼很宽的山鸢尾」返回一堆乱序低相关项。根因：向量化的对象没有语义，
sepal_length=5.1, ...这种键值对不等价于「花萼长 5.1 厘米」这句话。解决：先做「行 → 一句话文档」的整理，数值字段转成带单位的自然语言，英文品种名转中文。
验证：检索「花瓣比较大的变色鸢尾」，Top 命中明显更对口。
2. collection_name 与「换一个数据集」打架
问题：用同一进程再喂另一份数据，报已有集合或数据混在一起。
复现：
PGVector.from_documents用同一个collection_name二次灌入不同数据。根因：每个
collection_name对应 pgvector 里的一张固定向量表，二次from_documents会追加或重建，取决于插入方式，容易串数据。解决：不同数据集用不同
collection_name（如iris_docs/titanic_docs），或删除旧集合：PGVector(connection=CONN, collection_name="iris_docs").delete_collection()。验证：两个集合互不干扰，分别检索各自数据。
3. 给 qwen3-embed 传 dimensions，被 vLLM 以 400 拒绝
问题：
embed_query抛openai.BadRequestError: Model "qwen3-embed" does not support matryoshka representation, changing output dimensions will lead to poor results.复现：
OpenAIEmbeddings(..., dimensions=1024)，而局域网 vLLM 没给 qwen3-embed 开 matryoshka。根因：
dimensions参数会让客户端要求服务端把输出维度截断到指定值（matryoshka / 俄罗斯套娃表示）。qwen3-embed 默认不开这个能力，服务端直接拒绝改维度的请求。解决：不传
dimensions，用模型自身默认维度（qwen3-embed 默认即 1024 维）。pgvector 表列vector(1024)与之对齐即可。只有服务端显式开了 matryoshka 才需要传dimensions。验证：
len(embeddings.embed_query("..."))得到 1024，不再报 400。
4. 重复执行 from_documents，向量库里堆出重复数据
问题：脚本跑一遍入库 150 条，再跑一遍变成 300 条，检索结果重复。
复现：
Document(page_content=..., metadata=...)不带id，反复PGVector.from_documents。根因：
from_documents内部是INSERT ... ON CONFLICT (langchain_document_id) DO UPDATE。Document 没显式id时，LangChain 每次现生成新 uuid 当主键，ON CONFLICT永远不命中，每次都插新行。解决：给每条文档固定
id（本文用iris-{rid}），upsert 才按 id 覆盖；确需推倒重来时delete_collection()重建。验证：连跑三遍，
SELECT count(*) FROM iris_docs始终 150。
参考
天池数据集·鸢尾花数据集（3494）：https://tianchi.aliyun.com/dataset/3494
pgvector：https://github.com/pgvector/pgvector
LangChain：https://python.langchain.com
Qwen3-Embedding：https://huggingface.co/Qwen