实战：PostgreSQL + pgvector + FastAPI 构建轻量级 RAG
码海寻道 · 大模型、智能体与 RAG 工程组件系列第 56 篇
本篇用一个轻量方案串起 RAG 最小闭环：PostgreSQL 保存 Chunk 和业务元数据，pgvector 保存向量，FastAPI 提供检索接口。示例省略真实模型调用，但保留可替换的 Embedding 边界。
一、依赖和数据库准备
pipinstallfastapi uvicorn psycopg[binary]pgvector pydanticPostgreSQL 启用扩展：
CREATEEXTENSIONIFNOTEXISTSvector;CREATETABLEknowledge_chunks(id bigserialPRIMARYKEY,tenant_idtextNOTNULL,document_idtextNOTNULL,contenttextNOTNULL,embedding vector(768)NOTNULL,statustextNOTNULLDEFAULT'published',created_at timestamptzNOTNULLDEFAULTnow());768只是示例，必须与实际 Embedding 模型维度一致。
二、创建向量索引
开发阶段可以先用精确检索：
SELECTid,document_id,content,1-(embedding<=>'[0.1,0.2,0.3]'::vector)ASsimilarityFROMknowledge_chunksWHEREtenant_id='tenant-a'ANDstatus='published'ORDERBYembedding<=>'[0.1,0.2,0.3]'::vectorLIMIT5;生产环境再根据数据规模和压测选择 HNSW 或 IVFFlat，并注意过滤条件、维护成本和召回率。
索引选择要用固定评测集和真实过滤条件压测，至少比较 Recall@K、P95 延迟、构建时间、内存和更新成本。向量维度、距离度量和归一化方式必须与 Embedding 服务契约一致，改变其中任何一项都应创建新版本索引。
三、Embedding 服务边界
classEmbeddingService:defembed_documents(self,texts:list[str])->list[list[float]]:# 替换为实际模型或供应商 SDKreturn[[0.0]*768for_intexts]defembed_query(self,text:str)->list[float]:returnself.embed_documents([text])[0]服务边界应记录模型名、维度、归一化方式和版本。不要让入库和查询使用不同模型。
四、FastAPI 检索接口
fromfastapiimportFastAPI,HTTPExceptionfrompydanticimportBaseModel,FieldfrompsycopgimportAsyncConnection app=FastAPI()classSearchRequest(BaseModel):query:str=Field(min_length=1,max_length=4000)top_k:int=Field(default=5,ge=1,le=20)@app.post("/api/v1/search")asyncdefsearch(request:SearchRequest,tenant_id:str):vector=embedding_service.embed_query(request.query)asyncwithawaitAsyncConnection.connect(DATABASE_URL)asconn:rows=awaitconn.execute(""" SELECT id, document_id, content, embedding <=> %s::vector AS distance FROM knowledge_chunks WHERE tenant_id = %s AND status = 'published' ORDER BY embedding <=> %s::vector LIMIT %s """,(str(vector),tenant_id,str(vector),request.top_k),)return{"items":[dict(row)forrowinawaitrows.fetchall()]}示例中的tenant_id在生产代码中必须来自认证上下文，不能由普通客户端随意传入。
检索接口还应限制top_k、查询长度、超时和返回字段，并在向量命中后用 PostgreSQL 重新确认文档状态、租户和权限。不要把完整 Chunk 原文无上限返回给前端或模型。
五、从检索到生成
defbuild_context(items:list[dict])->str:return"\n\n".join(f"[来源{i+1}]{item['content']}"fori,iteminenumerate(items))defbuild_prompt(question:str,context:str)->str:returnf"""只根据参考资料回答问题。 如果资料不足，请明确说明。 参考资料：{context}问题：{question}"""实际项目还要加入引用元数据、Reranker、Token 限制和答案持久化。
六、入库接口和幂等
入库时使用document_id + chunk_index + embedding_model作为业务唯一键，重复任务不会产生重复 Chunk。正文发生变化时，应生成新版本而不是直接覆盖旧向量。
入库事务应先写业务事实和幂等记录，再异步生成 Embedding；向量写入成功后更新索引状态，最后切换服务版本。Embedding 失败、维度不匹配和数据库提交成功但向量写入失败，都要进入可重试或补偿状态。
七、轻量方案的边界
适合：
- 原型和小规模生产；
- 业务过滤和 JOIN 很多；
- 团队已有 PostgreSQL 运维能力；
- 查询并发和向量规模可控。
需要评估拆分到 Milvus 的信号：向量搜索明显影响业务 SQL、需要独立扩展、并发和索引规模持续增长。
八、测试清单
Embedding 维度与数据库一致；
空查询和超长查询被拒绝；
查询必须带认证租户范围；
新文档写入后可见性可解释；
删除和版本更新有补偿；
Recall@K 和 P95 延迟有基线；
数据库参数化查询防止注入。
查询超时、结果数量和上下文长度有上限；
权限过滤和删除后的向量清理可验证；
Embedding、Chunk 和索引版本可以回滚。
结语
PostgreSQL + pgvector + FastAPI 是验证轻量 RAG 的好起点：组件少、业务关联方便，也能覆盖上传后的 Chunk 存储、向量检索和回答生成。等指标证明需要独立扩展时，再迁移到专用向量数据库。
下一篇将把 Milvus、Redis 和 MinIO 加入架构，构建更接近企业场景的知识库。
参考资料
- pgvector 官方项目
- PostgreSQL 官方文档
- FastAPI 官方文档
本文为“码海寻道”原创技术文章。SQL、pgvector 和 Embedding 示例需结合实际版本、维度和索引配置验证。
