LangChain RAG实战：用PGVector把你的本地知识库变成智能问答机器人（Python代码详解）
LangChain RAG实战：用PGVector把你的本地知识库变成智能问答机器人（Python代码详解）
你是否曾经面对堆积如山的本地文档感到无从下手？PDF报告、Markdown笔记、TXT日志散落在各个文件夹，每次查找关键信息都像大海捞针。现在，借助LangChain和PGVector，你可以将这些杂乱无章的文件转化为一个能理解自然语言的智能助手。本文将手把手带你实现这个转变，从零开始构建一个真正可用的知识库问答系统。
1. 环境准备与数据基础建设
在开始之前，我们需要搭建好技术栈的基础设施。不同于简单的演示项目，生产级的知识库系统需要考虑数据处理的完整链路。以下是关键组件的部署方案：
核心组件清单：
- PostgreSQL 16+：选择长期支持版本确保稳定性
- pgvector 0.8.0：专门为向量运算优化的扩展
- Python 3.10+：推荐使用虚拟环境隔离依赖
- LangChain 0.1.x：当前最活跃的AI应用框架
安装过程需要注意几个技术细节：
# 创建Python虚拟环境 python -m venv rag_env source rag_env/bin/activate # Linux/Mac # rag_env\Scripts\activate # Windows # 安装核心依赖 pip install langchain psycopg2-binary pgvector langchain-openai tiktoken数据库配置有个容易被忽视的关键点——shared_preload_libraries参数需要包含vector：
-- 在postgresql.conf中添加 shared_preload_libraries = 'vector' # 需要重启服务生效 -- 创建专用数据库 CREATE DATABASE knowledge_base WITH ENCODING='UTF8'; \c knowledge_base CREATE EXTENSION IF NOT EXISTS vector;2. 文档处理流水线设计
原始文档就像未经加工的矿石，需要经过多道工序才能变成可用的知识原料。我们设计了一个工业级处理流水线：
文档加载器矩阵：
| 文件类型 | 加载器类 | 特殊参数 | 处理难点 |
|---|---|---|---|
| PyPDFLoader | password | 格式保留 | |
| Markdown | UnstructuredMarkdownLoader | mode="elements" | 代码块处理 |
| Word | Docx2txtLoader | - | 表格转换 |
| HTML | BSHTMLLoader | - | 标签清理 |
文本分割是影响后续检索质量的关键步骤，这里有个实战验证过的配置方案：
from langchain.text_splitter import RecursiveCharacterTextSplitter text_splitter = RecursiveCharacterTextSplitter( chunk_size=800, # 经验值：中文约400字，英文约800词 chunk_overlap=150, length_function=len, separators=["\n\n", "\n", "。", "！", "？", "\. ", "! ", r"\? ", " "] )3. 向量存储的工程化实现
PGVector的强大之处在于它把向量搜索能力直接集成到关系型数据库中。我们来看一个经过生产验证的存储方案：
连接配置最佳实践：
import psycopg2 from langchain_community.vectorstores import PGVector CONNECTION_STRING = ( "postgresql://user:password@host:5432/knowledge_base" "?connect_timeout=10&keepalives=1&keepalives_idle=30" ) # 带重试机制的连接池 def get_vector_store(embeddings): max_retries = 3 for attempt in range(max_retries): try: return PGVector( embedding_function=embeddings, connection_string=CONNECTION_STRING, collection_name="tech_docs", use_jsonb=True, pre_delete_collection=False # 避免测试时误删数据 ) except psycopg2.OperationalError as e: if attempt == max_retries - 1: raise time.sleep(2 ** attempt)批量插入优化技巧：
# 分批次插入文档，每批500个chunk batch_size = 500 for i in range(0, len(texts), batch_size): batch = texts[i:i + batch_size] vector_store.add_documents(batch) print(f"Inserted batch {i//batch_size + 1}")4. 检索增强生成(RAG)的深度优化
基础版的RAG可能产生无关结果，我们需要实现智能检索增强：
多级检索策略：
- 首轮向量相似度搜索
- 二次筛选：基于元数据过滤（文档类型、更新时间等）
- 最终排序：结合相关度分数和时间衰减因子
实现代码示例：
from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor def create_smart_retriever(vector_store): base_retriever = vector_store.as_retriever( search_type="mmr", # 最大边际相关算法 search_kwargs={"k": 10, "filter": {"source": "official"}} ) compressor = LLMChainExtractor.from_llm(ChatOpenAI(temperature=0)) return ContextualCompressionRetriever( base_compressor=compressor, base_retriever=base_retriever )问答链的进阶配置：
qa_chain = RetrievalQA.from_chain_type( llm=ChatOpenAI(model="gpt-4-1106-preview"), chain_type="refine", # 处理长文档更有效 retriever=retriever, chain_type_kwargs={ "question_prompt": PromptTemplate( template="""...自定义提示模板...""" ), "refine_prompt": PromptTemplate( template="""...精炼阶段提示...""" ), }, return_source_documents=True )5. 系统监控与持续改进
部署后的知识库需要持续优化，这里有几个关键指标需要监控：
健康度检查清单：
- 检索准确率：人工评估前20个结果的精确率
- 响应延迟：P99应控制在3秒以内
- API调用成本：监控OpenAI的token消耗趋势
- 知识新鲜度：文档平均更新时间指标
实现简单的监控日志：
import logging from datetime import datetime rag_logger = logging.getLogger("rag_system") rag_logger.setLevel(logging.INFO) handler = logging.FileHandler('rag_performance.log') handler.setFormatter(logging.Formatter('%(asctime)s - %(message)s')) rag_logger.addHandler(handler) def log_query(query, result, latency): rag_logger.info( f"Query: {query[:50]}... | " f"Latency: {latency:.2f}s | " f"Sources: {len(result['source_documents'])}" )在项目后期，可以考虑实现自动化评估流水线：
from ragas import evaluate from ragas.metrics import faithfulness, answer_relevancy def evaluate_rag(questions, ground_truths): dataset = { "question": questions, "answer": [qa_chain.run(q) for q in questions], "contexts": [[res.page_content for res in qa_chain.run(q)["source_documents"]] for q in questions], "ground_truth": ground_truths } score = evaluate( dataset, metrics=[faithfulness, answer_relevancy], ) return score6. 用户界面与交互设计
技术最终要服务于用户体验，这里实现一个控制台交互界面：
from prompt_toolkit import prompt from prompt_toolkit.history import FileHistory from prompt_toolkit.auto_suggest import AutoSuggestFromHistory def chat_interface(): print("知识库助手已启动(输入/exit退出)") while True: try: user_input = prompt( "用户> ", history=FileHistory('.rag_history'), auto_suggest=AutoSuggestFromHistory(), multiline=False ) if user_input.lower() == '/exit': break start_time = time.time() result = qa_chain({"query": user_input}) latency = time.time() - start_time print(f"\n助手> {result['result']}") print(f"\n[本次响应耗时: {latency:.2f}s]") log_query(user_input, result, latency) except KeyboardInterrupt: continue except Exception as e: print(f"系统错误: {str(e)}")对于需要部署为Web服务的情况，可以使用FastAPI构建REST端点：
from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class QueryRequest(BaseModel): question: str user_id: str = None @app.post("/query") async def handle_query(request: QueryRequest): result = qa_chain({"query": request.question}) return { "answer": result["result"], "sources": [ {"content": doc.page_content, "metadata": doc.metadata} for doc in result["source_documents"] ] }