实战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,statustextNOTNULLDEFAULTpublished,created_at timestamptzNOTNULLDEFAULTnow());768只是示例必须与实际 Embedding 模型维度一致。二、创建向量索引开发阶段可以先用精确检索SELECTid,document_id,content,1-(embedding[0.1,0.2,0.3]::vector)ASsimilarityFROMknowledge_chunksWHEREtenant_idtenant-aANDstatuspublishedORDERBYembedding[0.1,0.2,0.3]::vectorLIMIT5;生产环境再根据数据规模和压测选择 HNSW 或 IVFFlat并注意过滤条件、维护成本和召回率。索引选择要用固定评测集和真实过滤条件压测至少比较 RecallK、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 appFastAPI()classSearchRequest(BaseModel):query:strField(min_length1,max_length4000)top_k:intField(default5,ge1,le20)app.post(/api/v1/search)asyncdefsearch(request:SearchRequest,tenant_id:str):vectorembedding_service.embed_query(request.query)asyncwithawaitAsyncConnection.connect(DATABASE_URL)asconn:rowsawaitconn.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[来源{i1}]{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 维度与数据库一致空查询和超长查询被拒绝查询必须带认证租户范围新文档写入后可见性可解释删除和版本更新有补偿RecallK 和 P95 延迟有基线数据库参数化查询防止注入。查询超时、结果数量和上下文长度有上限权限过滤和删除后的向量清理可验证Embedding、Chunk 和索引版本可以回滚。结语PostgreSQL pgvector FastAPI 是验证轻量 RAG 的好起点组件少、业务关联方便也能覆盖上传后的 Chunk 存储、向量检索和回答生成。等指标证明需要独立扩展时再迁移到专用向量数据库。下一篇将把 Milvus、Redis 和 MinIO 加入架构构建更接近企业场景的知识库。参考资料pgvector 官方项目PostgreSQL 官方文档FastAPI 官方文档本文为“码海寻道”原创技术文章。SQL、pgvector 和 Embedding 示例需结合实际版本、维度和索引配置验证。