基于LangChain与PGVector构建RAG应用：从PDF解析到智能问答API部署-开发者社区

1. 项目概述：从零构建一个能与PDF对话的智能应用

如果你手头有一堆PDF文档——可能是行业报告、产品手册、学术论文或者像我们这次要用的“Epic Games诉苹果公司反垄断案”法律文件——然后你希望有一个智能助手，能让你像跟专家聊天一样，随时向它提问并获得基于文档的精准回答，那么这个项目就是为你准备的。这不仅仅是又一个简单的“Hello World”式AI教程，而是一个端到端（End-to-End）的实战项目。我们将从一个完全空白的文件夹开始，一步步搭建起一个完整的、可投入生产环境的聊天应用后端。它的核心是当前AI应用开发中最炙手可热的架构模式之一：检索增强生成（Retrieval-Augmented Generation, RAG）。

简单来说，RAG解决了大语言模型（LLM）的两个核心痛点：知识更新滞后与“幻觉”（即编造信息）。我们不是让LLM凭空回忆它训练数据里可能过时的内容，而是教会它如何从我们提供的、最新的、特定的文档（PDF）中寻找答案。这个过程就像给LLM配了一位超级高效的图书管理员（向量数据库）和一套严格的资料引用规范。

整个项目将分阶段进行，本篇（第一部分）将聚焦于构建坚实的后端引擎。我们会完成从PDF解析、文本智能分块、向量化存储到构建可查询链路的全过程。你将用到的核心“武器库”包括：LangChain（AI应用编排框架）、LangServe（快速部署LangChain应用为API）、PGVector（PostgreSQL的向量扩展，用作向量数据库）、OpenAI的嵌入模型（将文本转化为数学向量），以及Unstructured（强大的文档解析库）。

注意：虽然本教程使用OpenAI的API，但整个架构是模型无关的。一旦你掌握了流程，完全可以替换为其他兼容的嵌入模型和LLM（如开源模型），只需调整相应的LangChain组件即可。

2. 核心架构与工具选型解析

在动手写代码之前，理解我们为什么要选择这些工具以及它们如何协同工作至关重要。这能让你在遇到问题时，知道该从哪个环节入手排查。

2.1 为什么是LangChain和RAG？

LangChain本质上是一个“胶水”框架，它把LLM、各种数据源、工具和记忆模块等，通过“链（Chain）”的方式优雅地连接起来。如果没有它，你需要手动处理API调用、文本处理、上下文管理等一系列繁琐且易错的步骤。LangChain提供了标准化、可复用的组件，让我们能专注于业务逻辑而非底层通信。

而RAG架构，则是我们项目的灵魂。其工作流程可以概括为以下几步：

索引（Indexing）：线下进行。将PDF文档解析为文本，切割成有意义的片段（块），将这些文本块转化为向量（一组数字），并存入向量数据库。
检索（Retrieval）：线上进行。当用户提问时，将问题也转化为向量，然后在向量数据库中搜索与之最相似的几个文本块。
增强（Augmentation）：将检索到的相关文本块和用户问题一起，组合成一个增强的提示（Prompt），发送给LLM。
生成（Generation）：LLM基于这个包含了相关上下文的提示，生成一个准确、有据可依的回答。

这个流程确保了答案来源于你的文档，极大减少了幻觉，并且可以通过更新文档库来更新AI的知识。

2.2 关键组件深度剖析

1. PGVector vs. 其他向量数据库我们选择PGVector，主要基于以下几点考量：

运维简单：如果你的应用已经使用了PostgreSQL，那么PGVector是一个无缝的扩展，无需引入另一个独立的外部服务（如Pinecone, Weaviate），降低了系统复杂度和运维成本。
数据一致性：向量数据和相关的元数据（如原文、来源文件名、页码）都存储在同一数据库中，保证了事务一致性。
成熟度：作为PostgreSQL的扩展，它继承了PG的稳定性、安全性和丰富的查询功能。
本地开发友好：使用Docker可以轻松在本地启动一个带PGVector的PostgreSQL实例，非常适合开发和测试。

当然，如果追求极高的检索性能或需要处理海量向量（数十亿级），专业的向量数据库可能有优势。但对于大多数中小型应用和起步阶段，PGVector是一个务实且强大的选择。

2. Semantic Chunker（语义分块器）传统文本分块方法通常是按固定字符数或句子数进行机械切割，这很容易把一个完整的意思从中间切断。比如，一个重要的论点可能被分在两个块里，导致检索时信息不完整。 LangChain的SemanticChunker采用了更智能的方式。它先使用嵌入模型计算句子或小段文本的向量，然后根据向量之间的语义相似度来决定切割点。当它检测到相邻文本段的语义发生较大转折时（比如从“案件背景”切换到“法律争议焦点”），就会在此处进行分块。这样得到的文本块，其内部语义一致性更高，作为检索单元的质量也更好。

3. LangServe的角色LangServe允许我们将精心构建的LangChain链（Chain）快速打包成一个标准的REST API。它自动生成OpenAPI文档，并提供交互式的Playground界面供我们测试。这意味着我们的后端服务可以很容易地被任何前端（Web、移动端）调用，也为后续的部署和集成铺平了道路。

3. 环境准备与项目初始化

现在，让我们打开终端，开始实际的搭建工作。请确保你的系统已安装Python（建议3.10+）和Docker。

3.1 创建项目并安装核心依赖

首先，创建一个全新的项目目录并初始化虚拟环境，这是管理Python项目依赖的最佳实践。

# 创建项目目录并进入 mkdir pdf-rag-chatbot && cd pdf-rag-chatbot # 创建并激活虚拟环境（以venv为例） python -m venv venv # 在Windows上: venv\Scripts\activate # 在macOS/Linux上: source venv/bin/activate # 创建核心依赖文件 requirements.txt

在requirements.txt中，填入以下内容。这里我们锁定了主要库的版本，以确保教程的稳定性。

langchain==0.1.0 langchain-openai==0.0.2 langchain-community==0.0.10 langserve==0.0.34 unstructured[pdf]==0.10.30 openai==1.3.0 psycopg2-binary==2.9.9 pgvector==0.2.0 python-dotenv==1.0.0 fastapi==0.104.1 uvicorn[standard]==0.24.0

安装依赖：

pip install -r requirements.txt

实操心得：使用psycopg2-binary而非psycopg2可以避免本地编译PostgreSQL客户端库可能带来的麻烦，特别适合快速开发和部署。在生产环境，如果对性能和兼容性有极致要求，可以考虑安装完整版。

3.2 使用Docker启动带PGVector的数据库

我们将使用Docker来运行PostgreSQL + PGVector，这是最便捷的方式。确保Docker守护进程正在运行。

# 创建一个用于存储数据库数据的本地目录，防止容器删除后数据丢失 mkdir -p pgdata # 运行PostgreSQL容器 docker run --name pgvector-db -e POSTGRES_USER=raguser -e POSTGRES_PASSWORD=ragpassword -e POSTGRES_DB=ragdb -p 5432:5432 -v $(pwd)/pgdata:/var/lib/postgresql/data -d ankane/pgvector

这条命令做了以下几件事：

--name pgvector-db：给容器起个名字，方便管理。
-e：设置环境变量，这里定义了用户名、密码和数据库名。
-p 5432:5432：将容器的5432端口映射到主机的5432端口，这样我们的应用才能连接。
-v $(pwd)/pgdata:/var/lib/postgresql/data：将主机当前目录下的pgdata文件夹挂载到容器的数据目录，实现数据持久化。
-d：在后台运行容器。
ankane/pgvector：使用了集成了PGVector扩展的PostgreSQL镜像。

运行后，可以使用docker ps检查容器状态，用docker logs pgvector-db查看启动日志。

3.3 配置环境变量

为了安全地管理API密钥和数据库连接信息，我们使用.env文件。在项目根目录创建.env文件：

OPENAI_API_KEY=你的OpenAI_API密钥 DATABASE_URL=postgresql://raguser:ragpassword@localhost:5432/ragdb

重要安全提示：务必把.env文件添加到.gitignore中，避免将密钥提交到版本控制系统。

4. 构建文档处理流水线（索引阶段）

这是RAG的基石。我们将创建一个模块，专门负责将PDF“消化”成向量数据库里可检索的形式。

4.1 创建文档加载与解析模块

在项目根目录创建ingestion.py文件。

import os from typing import List from langchain_community.document_loaders import DirectoryLoader, UnstructuredPDFLoader from langchain.text_splitter import SemanticChunker from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import PGVector from langchain.schema import Document from dotenv import load_dotenv # 加载环境变量 load_dotenv() class PDFIngestor: def __init__(self, pdf_directory: str = "./docs"): """ 初始化PDF处理器。 :param pdf_directory: 存放PDF文件的目录路径 """ self.pdf_directory = pdf_directory # 确保目录存在 os.makedirs(self.pdf_directory, exist_ok=True) # 初始化OpenAI的嵌入模型，用于后续的语义分块和向量化 self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small") def load_documents(self) -> List[Document]: """ 从指定目录加载所有PDF文档。 使用UnstructuredPDFLoader，它能更好地处理复杂版式的PDF。 """ print(f"正在从 {self.pdf_directory} 加载PDF文档...") # 使用通配符加载所有pdf文件 loader = DirectoryLoader( self.pdf_directory, glob="**/*.pdf", loader_cls=UnstructuredPDFLoader, # 可以传递更多参数给Unstructured，例如使用OCR模式 loader_kwargs={"mode": "elements", "strategy": "fast"} ) documents = loader.load() print(f"成功加载 {len(documents)} 个文档。") return documents def chunk_documents(self, documents: List[Document]) -> List[Document]: """ 使用语义分块器将长文档分割成有意义的文本块。 """ print("正在使用语义分块器处理文档...") # 初始化语义分块器，它内部会使用我们提供的嵌入模型来计算语义边界 text_splitter = SemanticChunker( embeddings=self.embeddings, # breakpoint_threshold_type: 决定何时分块的算法，percentile是一个稳健的选择 breakpoint_threshold_type="percentile", # 这个值需要根据你的文档内容微调，值越小，分块越细 breakpoint_threshold_amount=80 ) chunks = text_splitter.split_documents(documents) print(f"文档被分割成 {len(chunks)} 个语义块。") return chunks def create_and_store_vectors(self, chunks: List[Document], connection_string: str, collection_name: str = "epic_vs_apple"): """ 将文本块向量化并存储到PGVector数据库中。 """ print(f"正在生成向量并存储到集合 '{collection_name}' 中...") # 这一步会完成：1. 为每个chunk调用OpenAI Embeddings API生成向量。 # 2. 在PostgreSQL中创建表（如果不存在）。 # 3. 将所有向量和元数据（文档内容、来源等）插入数据库。 db = PGVector.from_documents( documents=chunks, embedding=self.embeddings, collection_name=collection_name, connection_string=connection_string, # 预删除同名集合，这在开发阶段很有用，生产环境需谨慎 pre_delete_collection=True ) print("向量存储完成！") return db if __name__ == "__main__": # 示例：运行完整的索引流程 ingestor = PDFIngestor("./docs") # 假设你的PDF放在项目根目录的docs文件夹下 raw_docs = ingestor.load_documents() if raw_docs: chunked_docs = ingestor.chunk_documents(raw_docs) # 从环境变量获取数据库连接字符串 conn_str = os.getenv("DATABASE_URL") if conn_str: ingestor.create_and_store_vectors(chunked_docs, conn_str) else: print("错误：请在.env文件中设置DATABASE_URL环境变量。") else: print("未找到PDF文档，请将PDF文件放入./docs目录。")

注意事项：UnstructuredPDFLoader的mode和strategy参数对解析质量影响很大。对于纯文本PDF，"fast"策略足够；如果PDF包含大量图片或复杂排版，可能需要使用"hi_res"策略并安装额外的OCR依赖（如pytesseract）。这会使解析速度变慢，但提取的文本更准确。

4.2 执行文档索引

现在，将你想要让AI学习的PDF文件（例如，关于Epic Games诉苹果案的法律文件摘要）放入项目根目录下的docs文件夹中。然后运行：

python ingestion.py

你将在终端看到一系列日志，描述加载、分块和存储的过程。这个过程可能会花费一些时间，取决于PDF的数量和大小，以及OpenAI Embeddings API的调用。

5. 构建问答链（检索与生成阶段）

索引完成后，我们就可以构建处理用户查询的核心逻辑了。在根目录创建chain.py。

import os from typing import List, Dict, Any from langchain_openai import ChatOpenAI from langchain_community.vectorstores import PGVector from langchain_openai import OpenAIEmbeddings from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough, RunnableLambda from dotenv import load_dotenv load_dotenv() class RAGChainBuilder: def __init__(self, connection_string: str, collection_name: str = "epic_vs_apple"): """ 初始化RAG链构建器。 """ self.connection_string = connection_string self.collection_name = collection_name self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small") # 初始化LLM，我们使用gpt-3.5-turbo，性价比高且响应快 self.llm = ChatOpenAI(model="gpt-3.5-turbo-1106", temperature=0, streaming=True) def _format_docs(self, docs: List[Dict]) -> str: """ 将检索到的文档列表格式化为一个单一的上下文字符串。 这是构建Prompt的关键一步。 """ formatted = [] for i, doc in enumerate(docs): # 提取元数据，如来源文件名 source = doc.metadata.get('source', '未知来源') # 限制上下文长度，避免超出LLM的Token限制 content = doc.page_content[:1000] + "..." if len(doc.page_content) > 1000 else doc.page_content formatted.append(f"【文档片段 {i+1} - 来自: {source}】\n{content}\n") return "\n".join(formatted) def connect_to_vectorstore(self): """ 连接到已存在的PGVector向量存储。 注意：这里使用from_existing_index，因为我们假设索引已经由ingestion.py创建好了。 """ print(f"连接到向量集合: {self.collection_name}") self.vectorstore = PGVector.from_existing_index( embedding=self.embeddings, collection_name=self.collection_name, connection_string=self.connection_string, ) # 创建检索器，设置相似度搜索返回前4个最相关的块 self.retriever = self.vectorstore.as_retriever(search_kwargs={"k": 4}) return self.retriever def build_chain(self): """ 使用LangChain Expression Language (LCEL) 构建RAG链。 LCEL使得链的定义非常清晰和模块化。 """ print("正在构建RAG链...") # 1. 定义Prompt模板 # 这个模板指导LLM如何利用上下文和问题来生成答案，并强调基于文档回答。 prompt_template = ChatPromptTemplate.from_messages([ ("system", """你是一个专业的法律文档分析助手。请严格根据用户提供的上下文信息来回答问题。如果上下文中的信息不足以回答问题，请直接说明“根据提供的资料，我无法回答这个问题”，不要编造信息。 上下文： {context} """), ("human", "{question}") ]) # 2. 使用LCEL组合链 # LCEL的语法 `|` 类似于Unix的管道，表示数据流。 rag_chain = ( # 第一个RunnablePassthrough传递用户的问题 {"context": self.retriever | RunnableLambda(self._format_docs), "question": RunnablePassthrough()} | prompt_template # 将context和question填入模板 | self.llm # 将填充好的Prompt发送给LLM | StrOutputParser() # 将LLM的响应解析为字符串 ) print("RAG链构建完成。") return rag_chain def query(self, question: str): """ 一个便捷的查询方法，用于测试链。 """ if not hasattr(self, 'rag_chain'): self.connect_to_vectorstore() self.rag_chain = self.build_chain() print(f"问题: {question}") print("正在生成回答...") # 使用streaming方式获取回答，可以看到逐词输出的效果 for chunk in self.rag_chain.stream(question): print(chunk, end="", flush=True) print("\n---") if __name__ == "__main__": # 测试链的功能 conn_str = os.getenv("DATABASE_URL") if conn_str: builder = RAGChainBuilder(conn_str) # 测试几个问题 test_questions = [ "Epic Games起诉苹果的主要争议点是什么？", "法院对此案做出了什么初步裁决？", ] for q in test_questions: builder.query(q) else: print("错误：请设置DATABASE_URL环境变量。")

运行python chain.py，你应该能看到AI基于PDF内容生成的流式回答。这证明我们的核心RAG引擎已经可以工作了。

6. 使用LangServe部署为API服务

为了让前端或其他服务能够调用，我们需要将链包装成一个HTTP API。创建app.py。

from fastapi import FastAPI from fastapi.responses import RedirectResponse from langchain_community.vectorstores import PGVector from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough, RunnableLambda from langserve import add_routes import os from dotenv import load_dotenv from chain import RAGChainBuilder # 导入我们刚才写的链构建器 load_dotenv() # 1. 创建FastAPI应用实例 app = FastAPI( title="PDF RAG Chatbot API", version="1.0.0", description="一个基于LangChain和PGVector的RAG聊天机器人API，用于查询特定PDF文档内容。" ) # 2. 首页重定向到LangServe自动生成的文档页面 @app.get("/") async def redirect_root_to_docs(): return RedirectResponse(url="/docs") # 3. 初始化我们的RAG链 conn_str = os.getenv("DATABASE_URL") if not conn_str: raise ValueError("请在.env文件中设置DATABASE_URL") chain_builder = RAGChainBuilder(conn_str) chain_builder.connect_to_vectorstore() rag_chain = chain_builder.build_chain() # 4. 使用LangServe的add_routes将我们的链添加为API端点 # 这会自动处理输入输出schema，并生成OpenAPI文档。 add_routes( app, rag_chain, path="/chat", # API端点路径 # 启用流式输出，这对聊天体验很重要 enable_feedback_endpoint=True, enable_public_trace_link_endpoint=True, ) # 5. 可以添加一个简单的健康检查端点 @app.get("/health") async def health_check(): return {"status": "healthy"} if __name__ == "__main__": import uvicorn # 启动服务器，默认在 http://localhost:8000 uvicorn.run(app, host="0.0.0.0", port=8000)

现在，运行python app.py启动服务。打开浏览器，访问http://localhost:8000，你会被重定向到自动生成的交互式API文档页面（Swagger UI）。在这里，你可以找到/chat端点，并直接通过Playground界面测试你的RAG链，无需编写任何前端代码。

实操心得：LangServe的Playground是一个极其强大的调试工具。你不仅可以测试请求，还能看到LangChain链的详细执行轨迹（trace），这对于理解为什么某个问题没有得到预期答案、检索到了哪些文档片段至关重要。一定要善用这个功能来优化你的Prompt和检索策略。

7. 核心环节：流式输出与源文档追溯

在聊天场景中，让答案逐字逐句地流式（Streaming）返回，能极大提升用户体验。同时，显示答案所引用的源文档（Source）是建立信任的关键。我们在之前的chain.py中已经通过streaming=True和_format_docs函数埋下了伏笔。现在，让我们看看如何在API层面完善它。

7.1 增强链以返回源信息

我们需要修改链，使其在返回答案的同时，也返回检索到的源文档信息。这需要定义一个更复杂的输出类型。更新chain.py中的build_chain方法及相关部分：

from langchain_core.runnables import RunnableParallel from pydantic import BaseModel, Field from typing import List # 定义我们API的响应模型 class ChainOutput(BaseModel): """链的响应模型，包含答案和源文档信息。""" answer: str = Field(description="AI生成的最终答案") source_documents: List[Dict[str, Any]] = Field(description="检索到的源文档列表，包含内容和元数据") class EnhancedRAGChainBuilder(RAGChainBuilder): """增强的RAG链构建器，支持返回源文档。""" def build_chain_with_sources(self): """ 构建一个返回答案和源文档的链。 """ print("正在构建带源文档返回的RAG链...") # 定义检索和格式化的子链 retrieve_and_format = RunnableParallel({ "docs": self.retriever, # 直接检索文档对象 "question": RunnablePassthrough() }) # 主处理函数 def process_retrieved_data(input_dict: Dict) -> Dict: question = input_dict["question"] docs = input_dict["docs"] # 格式化文档内容用于生成答案 formatted_context = self._format_docs(docs) # 准备源文档信息（简化元数据） source_info = [] for doc in docs: source_info.append({ "content_preview": doc.page_content[:200] + "..." if len(doc.page_content) > 200 else doc.page_content, "source": doc.metadata.get('source', 'N/A'), "page": doc.metadata.get('page', 'N/A') }) # 这里是生成答案的核心逻辑（可以复用之前的Prompt和LLM） prompt = ChatPromptTemplate.from_messages([ ("system", f"请根据以下上下文回答问题：\n\n{formatted_context}\n\n如果上下文不相关，请说明无法回答。"), ("human", "{question}") ]) llm_chain = prompt | self.llm | StrOutputParser() answer = llm_chain.invoke({"question": question, "context": formatted_context}) return { "answer": answer, "source_documents": source_info } # 组合最终链 final_chain = retrieve_and_format | RunnableLambda(process_retrieved_data) return final_chain

然后在app.py中，使用这个新的链构建器，并让LangServe知道我们的输出模型：

# 在app.py中更新链的注册部分 from chain import EnhancedRAGChainBuilder from chain import ChainOutput # 导入输出模型 chain_builder = EnhancedRAGChainBuilder(conn_str) chain_builder.connect_to_vectorstore() rag_chain_with_sources = chain_builder.build_chain_with_sources() # 使用output_schema告诉LangServe输出的结构 add_routes( app, rag_chain_with_sources, path="/chat-with-sources", output_schema=ChainOutput, # 关键：指定输出模型 )

现在，访问http://localhost:8000/docs，你会看到新的/chat-with-sources端点，它的响应将清晰地包含answer和source_documents字段。

7.2 实现纯流式文本输出

对于只需要流式文本答案的前端，我们可以创建一个更简单的端点。在app.py中添加：

from fastapi.responses import StreamingResponse from langserve.serialization import WellKnownLCSerializer import asyncio @app.post("/chat-stream") async def chat_stream(question: str): """ 一个简单的流式聊天端点，只返回文本token流。 """ async def event_generator(): # 这里我们直接使用最初构建的简单rag_chain，它本身支持.stream() for chunk in rag_chain.stream(question): yield f"data: {chunk}\n\n" await asyncio.sleep(0.01) # 微小延迟，控制流的速度 yield "data: [DONE]\n\n" return StreamingResponse( event_generator(), media_type="text/event-stream", # Server-Sent Events (SSE) 媒体类型 headers={ 'Cache-Control': 'no-cache', 'Connection': 'keep-alive', 'X-Accel-Buffering': 'no' # 针对Nginx代理的重要设置 } )

这个端点使用了Server-Sent Events (SSE) 协议，这是一种简单的HTTP流式传输技术，非常适合从服务器向客户端推送文本流。前端可以很容易地使用EventSourceAPI 来接收数据。

8. 常见问题、调试技巧与性能优化

在实际开发和运行中，你几乎一定会遇到各种问题。这里记录了一些典型的坑和解决方案。

8.1 安装与依赖问题

pgvector安装失败：如果你没有使用Docker镜像ankane/pgvector，而是需要在本地或服务器安装PGVector扩展，请确保在PostgreSQL中执行CREATE EXTENSION vector;。我们的Python库pgvector只是客户端驱动。
unstructured依赖缺失：对于PDF处理，unstructured可能需要额外的系统依赖。如果遇到错误，请参考其官方文档安装poppler-utils或tesseract（如需OCR）。
- macOS:brew install poppler tesseract
- Ubuntu/Debian:sudo apt-get install poppler-utils tesseract-ocr
OpenAI API 连接错误：检查.env文件中的OPENAI_API_KEY是否正确，以及网络连接。可以尝试在Python中直接运行openai.Embedding.create(...)进行测试。

8.2 内容检索与答案质量问题

检索不到相关内容：
- 检查分块大小：SemanticChunker的breakpoint_threshold_amount参数可能不合适。尝试调整（例如从80改为70或90），或者暂时换用RecursiveCharacterTextSplitter按字符/段落分割，以排除是否是分块算法的问题。
- 检查嵌入模型：确保使用的嵌入模型（如text-embedding-3-small）与索引时使用的模型一致。
- 检查检索数量：search_kwargs={"k": 4}中的k值可能太小。尝试增加到 6 或 8。
- 检查向量数据库连接：确认连接字符串和集合名称正确，并且数据已成功写入。可以写一个简单的脚本查询向量数据库中的记录数。
答案质量差或出现幻觉：
- 强化Prompt：系统提示词（System Prompt）是控制LLM行为的关键。明确指令，如“严格根据上下文”、“引用上下文中的具体语句”、“如果上下文没有提到，就说不知道”。
- 调整LLM温度：我们将temperature设为0，是为了得到更确定、更少随机性的答案。如果答案过于死板，可以微调到0.1。
- 实现“重排序”（Re-ranking）：简单的向量相似度搜索可能返回一些相关但不精确的片段。可以在检索后加入一个重排序模型（如Cohere的Rerank API或开源的BGE Reranker），对Top K的结果进行精排，将最相关的片段放在前面，能显著提升最终答案的准确性。这是一个进阶优化点。
- 检查上下文长度：_format_docs函数中我们限制了每个文档片段的长度。如果限制得太短（比如200字），可能丢失关键信息；如果太长，可能超出LLM的上下文窗口。需要根据你的文档和模型窗口（如gpt-3.5-turbo的16K）做平衡。

8.3 性能优化建议

批量处理嵌入：在索引大量文档时，OpenAIEmbeddings默认可能逐条调用API，速度慢且可能触及速率限制。可以查阅LangChain文档，看是否支持批量嵌入（batch embedding）。
缓存嵌入结果：对于不变的文档库，可以考虑将生成的向量本地缓存，避免重复调用昂贵的Embedding API。LangChain支持与SQLiteCache或RedisSemanticCache集成。
数据库索引：PGVector使用IVFFlat或HNSW索引来加速向量搜索。在数据量较大（>1万条）后，应考虑创建索引。例如：CREATE INDEX ON embeddings USING ivfflat (vector vector_cosine_ops) WITH (lists = 100);。lists参数需要根据数据量调整。
异步处理：在app.py中，如果链的处理较慢，可以考虑使用async端点，并使用LangChain的异步接口（如ainvoke,astream）来避免阻塞事件循环，提高API的并发能力。

8.4 使用LangSmith进行链路追踪与调试（进阶）

LangChain提供了一个强大的可观测性平台——LangSmith。它能记录每一次链执行的详细步骤、输入输出、耗时和Token使用情况。

注册并获取API Key：访问LangSmith官网注册。
配置环境变量：在.env文件中添加LANGCHAIN_TRACING_V2=true和LANGCHAIN_API_KEY=你的langsmith_api_key。
（可选）设置项目名：LANGCHAIN_PROJECT=你的项目名。

配置完成后，再次通过API或脚本调用你的链，所有执行轨迹都会被记录到LangSmith的仪表盘中。你可以清晰地看到：

用户问题是什么。
检索器返回了哪4个文档片段（及其相似度分数）。
最终发送给LLM的完整Prompt是什么。
LLM的完整响应。
每一步的耗时。

这对于调试复杂问题、优化Prompt和检索策略来说，是无可替代的神器。

至此，一个功能完整、具备生产环境潜力的RAG聊天应用后端已经构建完毕。我们完成了从PDF处理、向量存储、智能检索到API部署的全流程，并加入了流式输出和源追溯等增强功能。在下一部分，我们将为这个强大的后端打造一个现代化的TypeScript + React前端，实现真正的全栈应用。

基于LangChain与PGVector构建RAG应用：从PDF解析到智能问答API部署