尧图网站建设 
这次我们来看一个 RAG 全流程系统搭建的实战教程。RAG检索增强生成是当前让大模型落地应用的关键技术它能有效解决模型幻觉、知识更新滞后和私有数据利用的问题。但很多教程只讲概念落地时却卡在环境配置、组件选型和流程串联上。这篇文章直接带你从零搭建一个可运行的 RAG 系统涵盖 Embedding 模型、向量数据库、检索优化和 LangGraph 编排并探讨 LoRA 微调如何进一步提升效果。如果你关心如何在自己的机器上跑通一个完整的 RAG 流程如何选择组件如何优化检索效果以及如何用 LangGraph 构建更复杂的智能体工作流这篇文章可以直接收藏。我们将重点关注每个环节的实操步骤、资源占用和效果验证目标是让你看完就能动手搭建。1. 核心能力速览在深入细节前我们先快速了解这个 RAG 系统教程涵盖的核心能力和技术栈。能力项说明项目类型RAG检索增强生成全流程系统搭建与实战教程核心组件Embedding 模型、向量数据库、检索优化、LangGraph 工作流编排涉及技术大模型LLM、向量检索、智能体Agent、参数高效微调LoRA硬件门槛中等。Embedding 和轻量化 LLM 推理可在 CPU 或消费级 GPU如 8G 显存上运行若涉及 LoRA 微调或运行更大模型建议 12G 显存。启动方式基于 Python 脚本和命令行启动涉及多个独立服务向量数据库、API 服务的协调。接口能力支持构建 RESTful API 服务提供文档索引、检索、问答等接口。批量任务支持批量文档解析、向量化入库以及批量问答测试。适合场景构建企业知识库问答、智能客服、个人文档助手、研究文献分析等私有化 AI 应用。2. 适用场景与使用边界2.1 这个系统适合谁AI 应用开发者希望将大模型能力与私有数据结合构建垂直领域应用。技术团队需要评估 RAG 技术栈为项目选型提供技术验证。个人学习者想系统性掌握从数据准备、向量化、检索到智能体编排的完整链路。2.2 能解决什么问题知识实时性大模型的训练数据有截止日期RAG 可以通过检索最新文档来提供实时信息。减少幻觉要求模型严格依据提供的上下文检索到的文档片段生成答案减少胡编乱造。数据隐私与成本私有数据无需上传至云端大模型可在本地或内网环境处理保障安全并控制成本。可解释性生成的答案可以追溯到源文档方便核查和验证。2.3 不适合什么场景对延迟极其敏感的场景RAG 流程涉及文档检索、上下文拼接和 LLM 生成整体延迟高于直接调用 LLM。高度结构化、确定性查询例如精确的 SQL 查询或代码编译传统编程方法更合适。完全没有向量化或检索需求的纯生成任务如创意写作、开放式对话。2.4 版权、隐私与安全边界数据合规确保用于构建知识库的文档拥有合法授权不侵犯版权。隐私保护如果处理包含个人敏感信息PII的文档需在入库前进行脱敏处理。生成内容审核RAG 能减少幻觉但生成内容仍需人工审核特别是在法律、医疗等高风险领域。系统安全对外提供 API 服务时需实施身份认证、速率限制和输入过滤防止恶意攻击。3. 环境准备与前置条件搭建 RAG 系统需要准备一个干净的 Python 环境并安装必要的底层依赖。3.1 基础软件环境操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04 推荐)。本文以 Ubuntu/Linux 环境命令为主Windows 用户可使用 WSL2 或相应调整。Python 版本Python 3.9 或 3.10。推荐使用conda或venv创建虚拟环境。包管理工具pip版本需更新至最新。3.2 关键依赖项整个流程涉及多个库我们将分模块安装。核心依赖包括深度学习框架PyTorch根据你的 CUDA 版本安装。Embedding 模型库sentence-transformers,FlagEmbedding或text2vec。向量数据库客户端chromadb,milvus或qdrant-client。大模型调用与编排langchain,langchain-community,langgraph。大模型本地运行ollama(推荐用于本地运行轻量 LLM) 或openai库 (用于调用云端 API)。文档解析unstructured,pypdf,markdown,python-docx等。3.3 硬件建议检查清单CPU4 核以上用于文档解析和轻量推理。内存16GB 以上处理大量文档时内存消耗较大。GPU可选但推荐Embedding使用bge-m3等模型时GPU 可大幅加速。6G 显存可流畅运行。LLM 推理使用Qwen-7B-Chat、Llama-3-8B等量化模型时8G 显存是起步门槛。若仅用 CPU 推理速度会慢很多。LoRA 微调需要更多显存建议 12G 或以上。磁盘空间至少预留 10GB 空间用于存放模型文件和向量数据库。4. 安装部署与启动方式我们将分步安装各个组件。假设你的项目根目录为rag_project。4.1 创建并激活虚拟环境# 使用 conda conda create -n rag_env python3.10 conda activate rag_env # 或使用 venv python -m venv rag_env source rag_env/bin/activate # Linux/macOS # rag_env\Scripts\activate # Windows4.2 安装 PyTorch访问 PyTorch 官网 获取适合你环境的安装命令。例如对于 CUDA 11.8pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118如果只用 CPU则安装 CPU 版本。4.3 安装核心 Python 包# 1. Embedding 与文本处理 pip install sentence-transformers FlagEmbedding # 2. 向量数据库 (这里以 ChromaDB 为例它轻量且无需单独服务) pip install chromadb # 3. LangChain 生态 pip install langchain langchain-community langchain-chroma langgraph # 4. 文档加载器 pip install unstructured pypdf python-docx markdown # 5. Web 框架 (用于构建 API) pip install fastapi uvicorn # 6. 环境变量管理 pip install python-dotenv4.4 安装并启动 Ollama (用于本地 LLM)Ollama 可以方便地在本地运行各种开源大模型。# Linux/macOS 安装 curl -fsSL https://ollama.com/install.sh | sh # Windows 请从官网下载安装包安装后拉取一个轻量模型例如qwen2.5:7b(约 4.7GB) 或llama3.2:3b(更小)ollama pull qwen2.5:7b启动 Ollama 服务通常安装后自动运行ollama serve服务默认运行在http://localhost:11434。4.5 项目目录结构建议创建如下目录结构便于管理rag_project/ ├── data/ # 存放原始文档 (PDF, TXT, MD等) ├── processed/ # 存放处理后的文本块 ├── vector_db/ # 向量数据库持久化目录 ├── models/ # (可选) 存放本地 Embedding 或 LoRA 模型 ├── src/ │ ├── __init__.py │ ├── document_loader.py # 文档加载与切分 │ ├── embedding.py # Embedding 模型封装 │ ├── vector_store.py # 向量数据库操作 │ ├── retriever.py # 检索器封装 │ ├── chain.py # LangChain 链定义 │ ├── agent.py # LangGraph Agent 定义 │ └── api.py # FastAPI 应用 ├── requirements.txt ├── .env.example └── README.md5. 功能测试与效果验证我们将按照 RAG 的核心流程分步测试每个环节。5.1 步骤一文档加载与预处理目标将不同格式的文档转换为统一的纯文本块。创建src/document_loader.pyfrom langchain_community.document_loaders import DirectoryLoader, PyPDFLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter import os def load_and_split_documents(data_dir./data, chunk_size500, chunk_overlap50): 加载指定目录下的文档并进行切分。 documents [] # 加载 PDF 文件 if os.path.exists(os.path.join(data_dir, *.pdf)): pdf_loader DirectoryLoader(data_dir, glob**/*.pdf, loader_clsPyPDFLoader) documents.extend(pdf_loader.load()) # 加载 TXT 文件 if os.path.exists(os.path.join(data_dir, *.txt)): txt_loader DirectoryLoader(data_dir, glob**/*.txt, loader_clsTextLoader) documents.extend(txt_loader.load()) # 文本切分 text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, length_functionlen, separators[\n\n, \n, 。, , , , , , ] ) splits text_splitter.split_documents(documents) print(f共加载 {len(documents)} 个文档切分为 {len(splits)} 个文本块。) return splits if __name__ __main__: # 测试在 data 目录下放几个测试文档 chunks load_and_split_documents() for i, chunk in enumerate(chunks[:2]): # 打印前两个块 print(f--- Chunk {i} ---) print(chunk.page_content[:200]) # 打印前200字符 print()运行测试cd rag_project python src/document_loader.py预期结果成功读取data/目录下的文档并打印出切分后的文本块预览。如果没有文档可以手动创建几个test.txt或放入 PDF 文件。5.2 步骤二Embedding 与向量入库目标将文本块转化为向量并存入向量数据库。创建src/embedding.py和src/vector_store.pysrc/embedding.py:from sentence_transformers import SentenceTransformer import torch class EmbeddingModel: def __init__(self, model_nameBAAI/bge-m3, deviceNone): 初始化 Embedding 模型。 device: 指定 cuda 或 cpu。若为None则自动检测。 if device is None: device cuda if torch.cuda.is_available() else cpu self.device device print(f正在加载 Embedding 模型: {model_name} 设备: {self.device}) # 首次使用会下载模型国内网络可能较慢可考虑使用镜像源或提前下载 self.model SentenceTransformer(model_name, deviceself.device) def encode(self, texts): 将文本列表编码为向量。 if isinstance(texts, str): texts [texts] embeddings self.model.encode(texts, normalize_embeddingsTrue) return embeddings.tolist()src/vector_store.py:from langchain_chroma import Chroma from langchain.schema import Document from src.embedding import EmbeddingModel import os class VectorStoreManager: def __init__(self, persist_directory./vector_db, embedding_modelNone): self.persist_directory persist_directory if embedding_model is None: # 使用本地封装的 EmbeddingModel self.embedding_model EmbeddingModel() # LangChain 需要的是一个可调用的嵌入函数 from langchain.embeddings import HuggingFaceEmbeddings self.embedding_function HuggingFaceEmbeddings(model_nameBAAI/bge-m3) else: self.embedding_function embedding_model # 初始化或加载向量数据库 self.vector_store Chroma( persist_directoryself.persist_directory, embedding_functionself.embedding_function ) def add_documents(self, documents): 将 Document 对象列表添加到向量数据库。 # 提取文本和元数据 texts [doc.page_content for doc in documents] metadatas [doc.metadata for doc in documents] # 使用 LangChain 的 add_texts 方法 ids self.vector_store.add_texts(textstexts, metadatasmetadatas) print(f已添加 {len(ids)} 个文档块到向量数据库。) return ids def similarity_search(self, query, k4): 相似性检索返回最相关的 k 个文档块。 results self.vector_store.similarity_search(query, kk) return results def persist(self): 持久化向量数据库到磁盘。 self.vector_store.persist() print(f向量数据库已持久化到: {self.persist_directory}) def create_and_populate_vector_store(): 完整的向量化入库流程。 from document_loader import load_and_split_documents # 1. 加载并切分文档 print(步骤1: 加载并切分文档...) chunks load_and_split_documents() # 2. 初始化向量数据库管理器 print(步骤2: 初始化向量数据库...) vs_manager VectorStoreManager() # 3. 添加文档块 print(步骤3: 向量化并入库...) vs_manager.add_documents(chunks) # 4. 持久化 vs_manager.persist() print(向量数据库构建完成) # 5. 简单检索测试 print(\n--- 检索测试 ---) test_query 什么是机器学习 # 根据你的文档内容修改 results vs_manager.similarity_search(test_query, k2) for i, doc in enumerate(results): print(f[结果 {i1}] {doc.page_content[:150]}...) if __name__ __main__: create_and_populate_vector_store()运行测试python src/vector_store.py预期结果成功加载bge-m3Embedding 模型首次运行需下载约 1.2GB。将data/下的文档切块、向量化并存入./vector_db目录。执行一次检索测试并打印出最相关的文档片段。资源观察运行此脚本时观察 GPU 显存占用如果使用 GPU。bge-m3模型加载后推理时显存占用通常在 1-2GB。CPU 模式会占用更多内存。5.3 步骤三检索增强生成RAG链测试目标将检索到的文档作为上下文与大模型结合生成答案。创建src/chain.pyfrom langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate from langchain_community.llms import Ollama from src.vector_store import VectorStoreManager def setup_rag_chain(): 设置一个简单的 RAG 问答链。 # 1. 加载向量数据库 vs_manager VectorStoreManager() retriever vs_manager.vector_store.as_retriever(search_kwargs{k: 4}) # 2. 连接本地 LLM (通过 Ollama) llm Ollama(modelqwen2.5:7b, base_urlhttp://localhost:11434) # 3. 定义提示词模板 prompt_template 请根据以下上下文信息回答问题。如果上下文信息不足以回答问题请直接说“根据提供的信息无法回答此问题”不要编造答案。 上下文 {context} 问题{question} 答案 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) # 4. 创建链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrieverretriever, chain_type_kwargs{prompt: PROMPT}, return_source_documentsTrue ) return qa_chain if __name__ __main__: print(初始化 RAG 链...) qa_chain setup_rag_chain() # 测试问题 test_questions [ RAG 是什么, # 根据你的知识库内容调整问题 向量数据库在 RAG 中起什么作用, ] for question in test_questions: print(f\n 问题: {question} ) result qa_chain.invoke({query: question}) print(f答案: {result[result]}) print(来源文档:) for i, doc in enumerate(result[source_documents][:2]): # 显示前两个来源 print(f [{i1}] {doc.page_content[:100]}...)运行测试确保 Ollama 服务正在运行 (ollama serve)。执行脚本python src/chain.py预期结果脚本会加载向量数据库和 LLM 模型。对每个测试问题会先检索相关文档然后结合上下文生成答案。答案应基于你知识库中的内容并附上来源文档片段。效果验证检查答案是否准确、是否引用了正确的源文档。如果答案出现幻觉或无关可能需要调整检索数量k或提示词模板。5.4 步骤四使用 LangGraph 构建带状态的智能体工作流目标引入 LangGraph构建一个可以管理多轮对话、工具调用等复杂逻辑的智能体。创建src/agent.pyfrom langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, List from langchain_community.tools import DuckDuckGoSearchRun from langchain_community.llms import Ollama from src.vector_store import VectorStoreManager import operator # 1. 定义状态结构 class AgentState(TypedDict): question: str context: Annotated[List[str], operator.add] # 累积的上下文信息 answer: str search_needed: bool # 2. 定义节点函数 def retrieve_node(state: AgentState): 检索节点从向量数据库获取相关信息。 vs_manager VectorStoreManager() docs vs_manager.similarity_search(state[question], k3) retrieved_context [doc.page_content for doc in docs] return {context: retrieved_context, search_needed: False} def search_node(state: AgentState): 搜索节点如果检索结果不足调用网络搜索。 search_tool DuckDuckGoSearchRun() search_result search_tool.run(state[question]) return {context: [search_result], search_needed: False} def decide_node(state: AgentState): 决策节点判断是否需要网络搜索。 # 简单的决策逻辑如果检索到的上下文太短或没有则搜索 total_context_len sum(len(c) for c in state.get(context, [])) if total_context_len 100: # 阈值可调整 return {search_needed: True} else: return {search_needed: False} def generate_node(state: AgentState): 生成节点综合上下文调用 LLM 生成最终答案。 llm Ollama(modelqwen2.5:7b) combined_context \n\n.join(state[context]) prompt f 你是一个有帮助的助手。请根据以下信息回答问题。 相关信息 {combined_context} 问题{state[question]} 请给出准确、简洁的答案。如果信息不足可以说明。 答案 response llm.invoke(prompt) return {answer: response} # 3. 构建图 def create_agent_graph(): workflow StateGraph(AgentState) # 添加节点 workflow.add_node(retrieve, retrieve_node) workflow.add_node(search, search_node) workflow.add_node(decide, decide_node) workflow.add_node(generate, generate_node) # 设置入口点 workflow.set_entry_point(retrieve) # 定义边 workflow.add_edge(retrieve, decide) workflow.add_conditional_edges( decide, lambda x: search if x[search_needed] else generate, {search: search, generate: generate} ) workflow.add_edge(search, generate) workflow.add_edge(generate, END) # 编译图 return workflow.compile() # 4. 运行测试 if __name__ __main__: graph create_agent_graph() test_questions [LangGraph 是什么, 今天北京天气怎么样] for q in test_questions: print(f\n 问题: {q}) initial_state {question: q, context: [], answer: , search_needed: False} result graph.invoke(initial_state) print(f答案: {result[answer]}) print(f使用的上下文来源数: {len(result[context])})运行测试python src/agent.py预期结果对于“LangGraph 是什么”应主要从你的向量数据库如果已存入相关文档中检索并生成答案。对于“今天北京天气怎么样”由于向量库中 likely 没有实时天气数据决策节点会判断需要搜索然后调用 DuckDuckGo 搜索并生成答案。效果验证这个简单的智能体展示了 LangGraph 如何编排不同的节点检索、决策、搜索、生成实现比简单 RAG 链更复杂的逻辑。你可以根据需要增加更多节点如历史记录管理、工具调用等。6. 接口 API 与批量任务6.1 构建 FastAPI 服务将 RAG 能力封装成 API方便集成到其他应用。创建src/api.pyfrom fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Optional import uvicorn from src.chain import setup_rag_chain app FastAPI(titleRAG 问答 API, description基于本地知识库的检索增强生成服务) # 全局加载链简单示例生产环境需考虑并发和重载 qa_chain None app.on_event(startup) async def startup_event(): global qa_chain print(正在加载 RAG 链...) qa_chain setup_rag_chain() print(RAG 链加载完成。) class QueryRequest(BaseModel): question: str top_k: Optional[int] 4 class QueryResponse(BaseModel): answer: str sources: List[str] app.post(/query, response_modelQueryResponse) async def query_knowledge_base(request: QueryRequest): if qa_chain is None: raise HTTPException(status_code503, detail服务未就绪) try: result qa_chain.invoke({query: request.question}) sources [doc.page_content[:200] for doc in result[source_documents][:request.top_k]] return QueryResponse(answerresult[result], sourcessources) except Exception as e: raise HTTPException(status_code500, detailf处理查询时出错: {str(e)}) class BatchQueryRequest(BaseModel): questions: List[str] top_k: Optional[int] 4 class BatchQueryResponse(BaseModel): results: List[QueryResponse] app.post(/batch_query, response_modelBatchQueryResponse) async def batch_query_knowledge_base(request: BatchQueryRequest): if qa_chain is None: raise HTTPException(status_code503, detail服务未就绪) responses [] for q in request.questions: try: result qa_chain.invoke({query: q}) sources [doc.page_content[:200] for doc in result[source_documents][:request.top_k]] responses.append(QueryResponse(answerresult[result], sourcessources)) except Exception as e: responses.append(QueryResponse(answerf处理问题{q}时出错: {str(e)}, sources[])) return BatchQueryResponse(resultsresponses) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)启动服务python src/api.py服务将在http://localhost:8000启动。访问http://localhost:8000/docs可以看到自动生成的 API 文档。6.2 API 调用测试使用curl或 Python 脚本测试接口。单次查询curl -X POST http://localhost:8000/query \ -H Content-Type: application/json \ -d {question: 什么是 Embedding, top_k: 2}批量查询curl -X POST http://localhost:8000/batch_query \ -H Content-Type: application/json \ -d {questions: [什么是 RAG, LangChain 有什么用], top_k: 2}Python 客户端示例import requests import json url http://localhost:8000/query payload {question: 向量数据库如何工作, top_k: 3} response requests.post(url, jsonpayload, timeout60) if response.status_code 200: result response.json() print(f答案: {result[answer]}) print(来源:) for i, source in enumerate(result[sources]): print(f {i1}. {source}) else: print(f请求失败: {response.status_code}, {response.text})6.3 批量任务处理对于大量文档的离线处理可以编写脚本。创建batch_process.pyimport os from src.document_loader import load_and_split_documents from src.vector_store import VectorStoreManager import logging from tqdm import tqdm logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) def batch_index_documents(data_root_dir, batch_size50): 批量索引文档目录。 data_root_dir: 包含多个子目录的根目录每个子目录为一类文档。 vs_manager VectorStoreManager() all_docs [] for category in os.listdir(data_root_dir): category_path os.path.join(data_root_dir, category) if os.path.isdir(category_path): logging.info(f正在处理类别: {category}) try: # 为每个文档块添加类别元数据 chunks load_and_split_documents(category_path) for chunk in chunks: chunk.metadata[category] category all_docs.extend(chunks) logging.info(f 类别 {category} 处理完成共 {len(chunks)} 个块。) except Exception as e: logging.error(f 处理类别 {category} 时出错: {e}) # 分批入库避免内存溢出 logging.info(f开始向量化入库总计 {len(all_docs)} 个文档块...) for i in tqdm(range(0, len(all_docs), batch_size)): batch all_docs[i:ibatch_size] vs_manager.add_documents(batch) vs_manager.persist() logging.info(批量索引完成) if __name__ __main__: # 假设你的文档按类别放在 ./data/ 下的不同子文件夹中 batch_index_documents(./data)7. 资源占用与性能观察7.1 各组件资源占用Embedding 模型 (bge-m3)GPU 模式加载模型约占用 1.2GB 显存。编码时显存占用随批量大小增加单条文本推理很快。CPU 模式内存占用约 2-3GB编码速度慢于 GPU。向量数据库 (ChromaDB)纯内存操作时内存占用与向量数据量成正比。100万条 768 维向量约占用 3GB 内存。开启持久化后会写入磁盘。本地 LLM (Ollama withqwen2.5:7b)4-bit 量化约占用 4-5GB 显存。8-bit 量化约占用 7-8GB 显存。CPU 推理内存占用可能超过 16GB速度较慢。LangGraph 工作流主要是内存开销用于维护状态和中间结果通常不大。7.2 性能优化建议Embedding 批量处理对大量文档进行向量化时使用encode方法的batch_size参数充分利用 GPU 并行能力。向量数据库选型ChromaDB轻量、易用适合中小规模数据百万级以下和原型开发。Milvus/Qdrant专为大规模向量检索设计支持分布式、高性能过滤适合生产环境海量数据。LLM 推理优化使用量化模型如 GGUF 格式。调整 Ollama 的num_ctx上下文长度和num_threadCPU线程数参数。考虑使用vLLM或TGI等高性能推理框架部署 API 服务。检索优化重排序 (Rerank)在初步检索召回后使用更精细的交叉编码器模型对结果重排序提升精度。可集成bge-reranker等模型。混合检索结合关键词检索如 BM25和向量检索提高召回率。元数据过滤利用向量数据库的元数据过滤功能在检索前缩小范围。8. 常见问题与排查方法问题现象可能原因排查方式解决方案ImportError: No module named ‘xxx’依赖包未安装或虚拟环境未激活。检查当前 Python 环境 (which python或pip list)。激活正确的虚拟环境并pip install -r requirements.txt。CUDA out of memoryGPU 显存不足。使用nvidia-smi观察显存占用。1. 减少批量大小 (batch_size)。2. 使用 CPU 模式 (device‘cpu’)。3. 使用更小的模型或量化模型。Ollama 连接失败Ollama 服务未启动或端口被占用。检查ollama serve是否运行端口 11434 是否监听。启动 Ollama 服务或检查防火墙设置。检索结果不相关1. Embedding 模型不适合领域。2. 文本切分不合理。3. 检索参数k太小或太大。1. 检查源文档内容。2. 打印检索到的文本块查看。1. 尝试其他 Embedding 模型如text2vec。2. 调整chunk_size和chunk_overlap。3. 调整k值或引入重排序。LLM 答案质量差1. 提示词设计不佳。2. 检索到的上下文质量低。3. 模型能力有限。1. 打印出传递给 LLM 的完整提示词。2. 检查上下文是否与问题相关。1. 优化提示词模板明确指令。2. 提升检索质量。3. 更换或微调 LLM。向量数据库写入/读取慢1. 数据量过大。2. 未使用持久化每次重启重新计算。观察磁盘 I/O 和 CPU 使用率。1. 对于大规模数据考虑 Milvus/Qdrant。2. 确保正确调用persist()方法。API 服务请求超时1. LLM 生成速度慢。2. 网络问题。3. 服务端处理瓶颈。查看服务日志测量各环节耗时。1. 为 API 设置合理的超时时间。2. 对 LLM 调用进行异步处理或队列管理。3. 考虑使用更快的模型或推理后端。9. 进阶LoRA 微调优化 Embedding 或 LLM如果通用 Embedding 模型或 LLM 在你的垂直领域表现不佳可以考虑使用 LoRA 进行微调。9.1 LoRA 微调 Embedding 模型目标让 Embedding 模型更适应你的专业领域术语和语义。基本步骤准备数据收集query, positive_doc, negative_doc三元组数据。正例是相关文档负例是不相关或弱相关文档。选择框架使用peft(Parameter-Efficient Fine-Tuning) 和transformers库。训练脚本概览from transformers import AutoModel, AutoTokenizer from peft import LoraConfig, get_peft_model import torch from datasets import Dataset # ... 数据加载和预处理代码 ... model_name BAAI/bge-m3 model AutoModel.from_pretrained(model_name) tokenizer AutoTokenizer.from_pretrained(model_name) # 配置 LoRA lora_config LoraConfig( r8, # LoRA 秩 lora_alpha32, target_modules[query, key, value], # 针对 Transformer 的注意力模块 lora_dropout0.1, biasnone, ) model get_peft_model(model, lora_config) model.print_trainable_parameters() # 查看可训练参数量通常只有原模型的 0.1%-1% # 定义对比学习损失函数 (如 InfoNCE) # ... 训练循环代码 ...使用微调后的模型训练完成后保存适配器权重并在推理时加载。9.2 LoRA 微调 LLM目标让 LLM 更遵循你期望的回答格式、风格或领域知识。基本步骤准备数据准备指令微调格式的数据如{instruction: ..., input: ..., output: ...}。选择工具使用trl(Transformer Reinforcement Learning) 库的SFTTrainer或unsloth进行高效微调。快速上手使用 unslothpip install unslothfrom unsloth import FastLanguageModel import torch model, tokenizer FastLanguageModel.from_pretrained( model_name unsloth/llama-3-8b-bnb-4bit, # 或其他支持模型 max_seq_length 2048, dtype None, load_in_4bit True, # 4-bit 量化加载 ) model FastLanguageModel.get_peft_model( model, r 16, target_modules [q_proj, k_proj, v_proj, o_proj, gate_proj, up_proj, down_proj,], lora_alpha 16, lora_dropout 0, bias none, use_gradient_checkpointing unsloth, random_state 3407, use_rslora False, loftq_config None, ) # ... 准备数据并训练 ...集成到 RAG将微调后的 LLM 模型路径或 Ollama 中的新模型名配置到你的 RAG 链或 LangGraph 中。硬件要求LoRA 微调 LLM 通常需要 12GB 以上显存对于 7B 模型。使用unsloth或bitsandbytes的 4-bit 量化可以显著降低需求。10. 总结与下一步通过本教程我们完成了一个从文档处理到智能问答的完整 RAG 系统搭建。这套流程的核心价值在于可落地你可以在自己的机器上使用开源组件构建一个服务于私有数据的 AI 应用。最值得尝试的点模块化清晰每个组件加载、嵌入、存储、检索、生成都可独立替换和升级。本地化部署从 Embedding 模型、向量数据库到 LLM均可运行在本地环境保障数据隐私。灵活的编排能力通过 LangGraph你可以设计复杂的多步骤工作流超越简单的问答链。最先应该验证的功能端到端流程确保从一篇 PDF 文档到最终生成答案的整个链路能跑通。检索质量用几个关键问题测试看返回的文档片段是否真的相关。API 服务尝试通过 HTTP 请求调用你的知识库这是集成到其他应用的基础。最容易踩的坑环境配置Python 版本、CUDA 版本、依赖冲突。务必使用虚拟环境。显存不足同时加载 Embedding 模型和 LLM 可能导致 OOM。建议分开测试或使用 CPU 模式暂替。文档预处理不合理的文本切分是检索效果差的常见原因。多尝试不同的chunk_size和切分策略。后续扩展方向引入重排序 (Rerank)在向量检索后增加一个重排序步骤使用bge-reranker等模型对 Top-K 结果重新打分排序能显著提升精度。实现多路检索结合关键词检索如 BM25和向量检索提升召回率。构建 Web UI使用Gradio或Streamlit快速搭建一个交互式前端界面。接入更多工具在 LangGraph 智能体中集成计算器、代码执行器、网络搜索等工具打造更强大的 AI 助手。探索 Agentic RAG让智能体主动决定何时检索、检索什么、如何迭代优化查询实现更自主的问答。建议将本教程的代码作为起点根据你的具体数据和需求进行调整。在实践中数据质量文档清洗、切分和提示词工程往往比模型本身更重要。 从零搭建RAG系统实战：LangGraph编排与LoRA微调优化
这次我们来看一个 RAG 全流程系统搭建的实战教程。RAG检索增强生成是当前让大模型落地应用的关键技术它能有效解决模型幻觉、知识更新滞后和私有数据利用的问题。但很多教程只讲概念落地时却卡在环境配置、组件选型和流程串联上。这篇文章直接带你从零搭建一个可运行的 RAG 系统涵盖 Embedding 模型、向量数据库、检索优化和 LangGraph 编排并探讨 LoRA 微调如何进一步提升效果。如果你关心如何在自己的机器上跑通一个完整的 RAG 流程如何选择组件如何优化检索效果以及如何用 LangGraph 构建更复杂的智能体工作流这篇文章可以直接收藏。我们将重点关注每个环节的实操步骤、资源占用和效果验证目标是让你看完就能动手搭建。1. 核心能力速览在深入细节前我们先快速了解这个 RAG 系统教程涵盖的核心能力和技术栈。能力项说明项目类型RAG检索增强生成全流程系统搭建与实战教程核心组件Embedding 模型、向量数据库、检索优化、LangGraph 工作流编排涉及技术大模型LLM、向量检索、智能体Agent、参数高效微调LoRA硬件门槛中等。Embedding 和轻量化 LLM 推理可在 CPU 或消费级 GPU如 8G 显存上运行若涉及 LoRA 微调或运行更大模型建议 12G 显存。启动方式基于 Python 脚本和命令行启动涉及多个独立服务向量数据库、API 服务的协调。接口能力支持构建 RESTful API 服务提供文档索引、检索、问答等接口。批量任务支持批量文档解析、向量化入库以及批量问答测试。适合场景构建企业知识库问答、智能客服、个人文档助手、研究文献分析等私有化 AI 应用。2. 适用场景与使用边界2.1 这个系统适合谁AI 应用开发者希望将大模型能力与私有数据结合构建垂直领域应用。技术团队需要评估 RAG 技术栈为项目选型提供技术验证。个人学习者想系统性掌握从数据准备、向量化、检索到智能体编排的完整链路。2.2 能解决什么问题知识实时性大模型的训练数据有截止日期RAG 可以通过检索最新文档来提供实时信息。减少幻觉要求模型严格依据提供的上下文检索到的文档片段生成答案减少胡编乱造。数据隐私与成本私有数据无需上传至云端大模型可在本地或内网环境处理保障安全并控制成本。可解释性生成的答案可以追溯到源文档方便核查和验证。2.3 不适合什么场景对延迟极其敏感的场景RAG 流程涉及文档检索、上下文拼接和 LLM 生成整体延迟高于直接调用 LLM。高度结构化、确定性查询例如精确的 SQL 查询或代码编译传统编程方法更合适。完全没有向量化或检索需求的纯生成任务如创意写作、开放式对话。2.4 版权、隐私与安全边界数据合规确保用于构建知识库的文档拥有合法授权不侵犯版权。隐私保护如果处理包含个人敏感信息PII的文档需在入库前进行脱敏处理。生成内容审核RAG 能减少幻觉但生成内容仍需人工审核特别是在法律、医疗等高风险领域。系统安全对外提供 API 服务时需实施身份认证、速率限制和输入过滤防止恶意攻击。3. 环境准备与前置条件搭建 RAG 系统需要准备一个干净的 Python 环境并安装必要的底层依赖。3.1 基础软件环境操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04 推荐)。本文以 Ubuntu/Linux 环境命令为主Windows 用户可使用 WSL2 或相应调整。Python 版本Python 3.9 或 3.10。推荐使用conda或venv创建虚拟环境。包管理工具pip版本需更新至最新。3.2 关键依赖项整个流程涉及多个库我们将分模块安装。核心依赖包括深度学习框架PyTorch根据你的 CUDA 版本安装。Embedding 模型库sentence-transformers,FlagEmbedding或text2vec。向量数据库客户端chromadb,milvus或qdrant-client。大模型调用与编排langchain,langchain-community,langgraph。大模型本地运行ollama(推荐用于本地运行轻量 LLM) 或openai库 (用于调用云端 API)。文档解析unstructured,pypdf,markdown,python-docx等。3.3 硬件建议检查清单CPU4 核以上用于文档解析和轻量推理。内存16GB 以上处理大量文档时内存消耗较大。GPU可选但推荐Embedding使用bge-m3等模型时GPU 可大幅加速。6G 显存可流畅运行。LLM 推理使用Qwen-7B-Chat、Llama-3-8B等量化模型时8G 显存是起步门槛。若仅用 CPU 推理速度会慢很多。LoRA 微调需要更多显存建议 12G 或以上。磁盘空间至少预留 10GB 空间用于存放模型文件和向量数据库。4. 安装部署与启动方式我们将分步安装各个组件。假设你的项目根目录为rag_project。4.1 创建并激活虚拟环境# 使用 conda conda create -n rag_env python3.10 conda activate rag_env # 或使用 venv python -m venv rag_env source rag_env/bin/activate # Linux/macOS # rag_env\Scripts\activate # Windows4.2 安装 PyTorch访问 PyTorch 官网 获取适合你环境的安装命令。例如对于 CUDA 11.8pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118如果只用 CPU则安装 CPU 版本。4.3 安装核心 Python 包# 1. Embedding 与文本处理 pip install sentence-transformers FlagEmbedding # 2. 向量数据库 (这里以 ChromaDB 为例它轻量且无需单独服务) pip install chromadb # 3. LangChain 生态 pip install langchain langchain-community langchain-chroma langgraph # 4. 文档加载器 pip install unstructured pypdf python-docx markdown # 5. Web 框架 (用于构建 API) pip install fastapi uvicorn # 6. 环境变量管理 pip install python-dotenv4.4 安装并启动 Ollama (用于本地 LLM)Ollama 可以方便地在本地运行各种开源大模型。# Linux/macOS 安装 curl -fsSL https://ollama.com/install.sh | sh # Windows 请从官网下载安装包安装后拉取一个轻量模型例如qwen2.5:7b(约 4.7GB) 或llama3.2:3b(更小)ollama pull qwen2.5:7b启动 Ollama 服务通常安装后自动运行ollama serve服务默认运行在http://localhost:11434。4.5 项目目录结构建议创建如下目录结构便于管理rag_project/ ├── data/ # 存放原始文档 (PDF, TXT, MD等) ├── processed/ # 存放处理后的文本块 ├── vector_db/ # 向量数据库持久化目录 ├── models/ # (可选) 存放本地 Embedding 或 LoRA 模型 ├── src/ │ ├── __init__.py │ ├── document_loader.py # 文档加载与切分 │ ├── embedding.py # Embedding 模型封装 │ ├── vector_store.py # 向量数据库操作 │ ├── retriever.py # 检索器封装 │ ├── chain.py # LangChain 链定义 │ ├── agent.py # LangGraph Agent 定义 │ └── api.py # FastAPI 应用 ├── requirements.txt ├── .env.example └── README.md5. 功能测试与效果验证我们将按照 RAG 的核心流程分步测试每个环节。5.1 步骤一文档加载与预处理目标将不同格式的文档转换为统一的纯文本块。创建src/document_loader.pyfrom langchain_community.document_loaders import DirectoryLoader, PyPDFLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter import os def load_and_split_documents(data_dir./data, chunk_size500, chunk_overlap50): 加载指定目录下的文档并进行切分。 documents [] # 加载 PDF 文件 if os.path.exists(os.path.join(data_dir, *.pdf)): pdf_loader DirectoryLoader(data_dir, glob**/*.pdf, loader_clsPyPDFLoader) documents.extend(pdf_loader.load()) # 加载 TXT 文件 if os.path.exists(os.path.join(data_dir, *.txt)): txt_loader DirectoryLoader(data_dir, glob**/*.txt, loader_clsTextLoader) documents.extend(txt_loader.load()) # 文本切分 text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, length_functionlen, separators[\n\n, \n, 。, , , , , , ] ) splits text_splitter.split_documents(documents) print(f共加载 {len(documents)} 个文档切分为 {len(splits)} 个文本块。) return splits if __name__ __main__: # 测试在 data 目录下放几个测试文档 chunks load_and_split_documents() for i, chunk in enumerate(chunks[:2]): # 打印前两个块 print(f--- Chunk {i} ---) print(chunk.page_content[:200]) # 打印前200字符 print()运行测试cd rag_project python src/document_loader.py预期结果成功读取data/目录下的文档并打印出切分后的文本块预览。如果没有文档可以手动创建几个test.txt或放入 PDF 文件。5.2 步骤二Embedding 与向量入库目标将文本块转化为向量并存入向量数据库。创建src/embedding.py和src/vector_store.pysrc/embedding.py:from sentence_transformers import SentenceTransformer import torch class EmbeddingModel: def __init__(self, model_nameBAAI/bge-m3, deviceNone): 初始化 Embedding 模型。 device: 指定 cuda 或 cpu。若为None则自动检测。 if device is None: device cuda if torch.cuda.is_available() else cpu self.device device print(f正在加载 Embedding 模型: {model_name} 设备: {self.device}) # 首次使用会下载模型国内网络可能较慢可考虑使用镜像源或提前下载 self.model SentenceTransformer(model_name, deviceself.device) def encode(self, texts): 将文本列表编码为向量。 if isinstance(texts, str): texts [texts] embeddings self.model.encode(texts, normalize_embeddingsTrue) return embeddings.tolist()src/vector_store.py:from langchain_chroma import Chroma from langchain.schema import Document from src.embedding import EmbeddingModel import os class VectorStoreManager: def __init__(self, persist_directory./vector_db, embedding_modelNone): self.persist_directory persist_directory if embedding_model is None: # 使用本地封装的 EmbeddingModel self.embedding_model EmbeddingModel() # LangChain 需要的是一个可调用的嵌入函数 from langchain.embeddings import HuggingFaceEmbeddings self.embedding_function HuggingFaceEmbeddings(model_nameBAAI/bge-m3) else: self.embedding_function embedding_model # 初始化或加载向量数据库 self.vector_store Chroma( persist_directoryself.persist_directory, embedding_functionself.embedding_function ) def add_documents(self, documents): 将 Document 对象列表添加到向量数据库。 # 提取文本和元数据 texts [doc.page_content for doc in documents] metadatas [doc.metadata for doc in documents] # 使用 LangChain 的 add_texts 方法 ids self.vector_store.add_texts(textstexts, metadatasmetadatas) print(f已添加 {len(ids)} 个文档块到向量数据库。) return ids def similarity_search(self, query, k4): 相似性检索返回最相关的 k 个文档块。 results self.vector_store.similarity_search(query, kk) return results def persist(self): 持久化向量数据库到磁盘。 self.vector_store.persist() print(f向量数据库已持久化到: {self.persist_directory}) def create_and_populate_vector_store(): 完整的向量化入库流程。 from document_loader import load_and_split_documents # 1. 加载并切分文档 print(步骤1: 加载并切分文档...) chunks load_and_split_documents() # 2. 初始化向量数据库管理器 print(步骤2: 初始化向量数据库...) vs_manager VectorStoreManager() # 3. 添加文档块 print(步骤3: 向量化并入库...) vs_manager.add_documents(chunks) # 4. 持久化 vs_manager.persist() print(向量数据库构建完成) # 5. 简单检索测试 print(\n--- 检索测试 ---) test_query 什么是机器学习 # 根据你的文档内容修改 results vs_manager.similarity_search(test_query, k2) for i, doc in enumerate(results): print(f[结果 {i1}] {doc.page_content[:150]}...) if __name__ __main__: create_and_populate_vector_store()运行测试python src/vector_store.py预期结果成功加载bge-m3Embedding 模型首次运行需下载约 1.2GB。将data/下的文档切块、向量化并存入./vector_db目录。执行一次检索测试并打印出最相关的文档片段。资源观察运行此脚本时观察 GPU 显存占用如果使用 GPU。bge-m3模型加载后推理时显存占用通常在 1-2GB。CPU 模式会占用更多内存。5.3 步骤三检索增强生成RAG链测试目标将检索到的文档作为上下文与大模型结合生成答案。创建src/chain.pyfrom langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate from langchain_community.llms import Ollama from src.vector_store import VectorStoreManager def setup_rag_chain(): 设置一个简单的 RAG 问答链。 # 1. 加载向量数据库 vs_manager VectorStoreManager() retriever vs_manager.vector_store.as_retriever(search_kwargs{k: 4}) # 2. 连接本地 LLM (通过 Ollama) llm Ollama(modelqwen2.5:7b, base_urlhttp://localhost:11434) # 3. 定义提示词模板 prompt_template 请根据以下上下文信息回答问题。如果上下文信息不足以回答问题请直接说“根据提供的信息无法回答此问题”不要编造答案。 上下文 {context} 问题{question} 答案 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) # 4. 创建链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrieverretriever, chain_type_kwargs{prompt: PROMPT}, return_source_documentsTrue ) return qa_chain if __name__ __main__: print(初始化 RAG 链...) qa_chain setup_rag_chain() # 测试问题 test_questions [ RAG 是什么, # 根据你的知识库内容调整问题 向量数据库在 RAG 中起什么作用, ] for question in test_questions: print(f\n 问题: {question} ) result qa_chain.invoke({query: question}) print(f答案: {result[result]}) print(来源文档:) for i, doc in enumerate(result[source_documents][:2]): # 显示前两个来源 print(f [{i1}] {doc.page_content[:100]}...)运行测试确保 Ollama 服务正在运行 (ollama serve)。执行脚本python src/chain.py预期结果脚本会加载向量数据库和 LLM 模型。对每个测试问题会先检索相关文档然后结合上下文生成答案。答案应基于你知识库中的内容并附上来源文档片段。效果验证检查答案是否准确、是否引用了正确的源文档。如果答案出现幻觉或无关可能需要调整检索数量k或提示词模板。5.4 步骤四使用 LangGraph 构建带状态的智能体工作流目标引入 LangGraph构建一个可以管理多轮对话、工具调用等复杂逻辑的智能体。创建src/agent.pyfrom langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, List from langchain_community.tools import DuckDuckGoSearchRun from langchain_community.llms import Ollama from src.vector_store import VectorStoreManager import operator # 1. 定义状态结构 class AgentState(TypedDict): question: str context: Annotated[List[str], operator.add] # 累积的上下文信息 answer: str search_needed: bool # 2. 定义节点函数 def retrieve_node(state: AgentState): 检索节点从向量数据库获取相关信息。 vs_manager VectorStoreManager() docs vs_manager.similarity_search(state[question], k3) retrieved_context [doc.page_content for doc in docs] return {context: retrieved_context, search_needed: False} def search_node(state: AgentState): 搜索节点如果检索结果不足调用网络搜索。 search_tool DuckDuckGoSearchRun() search_result search_tool.run(state[question]) return {context: [search_result], search_needed: False} def decide_node(state: AgentState): 决策节点判断是否需要网络搜索。 # 简单的决策逻辑如果检索到的上下文太短或没有则搜索 total_context_len sum(len(c) for c in state.get(context, [])) if total_context_len 100: # 阈值可调整 return {search_needed: True} else: return {search_needed: False} def generate_node(state: AgentState): 生成节点综合上下文调用 LLM 生成最终答案。 llm Ollama(modelqwen2.5:7b) combined_context \n\n.join(state[context]) prompt f 你是一个有帮助的助手。请根据以下信息回答问题。 相关信息 {combined_context} 问题{state[question]} 请给出准确、简洁的答案。如果信息不足可以说明。 答案 response llm.invoke(prompt) return {answer: response} # 3. 构建图 def create_agent_graph(): workflow StateGraph(AgentState) # 添加节点 workflow.add_node(retrieve, retrieve_node) workflow.add_node(search, search_node) workflow.add_node(decide, decide_node) workflow.add_node(generate, generate_node) # 设置入口点 workflow.set_entry_point(retrieve) # 定义边 workflow.add_edge(retrieve, decide) workflow.add_conditional_edges( decide, lambda x: search if x[search_needed] else generate, {search: search, generate: generate} ) workflow.add_edge(search, generate) workflow.add_edge(generate, END) # 编译图 return workflow.compile() # 4. 运行测试 if __name__ __main__: graph create_agent_graph() test_questions [LangGraph 是什么, 今天北京天气怎么样] for q in test_questions: print(f\n 问题: {q}) initial_state {question: q, context: [], answer: , search_needed: False} result graph.invoke(initial_state) print(f答案: {result[answer]}) print(f使用的上下文来源数: {len(result[context])})运行测试python src/agent.py预期结果对于“LangGraph 是什么”应主要从你的向量数据库如果已存入相关文档中检索并生成答案。对于“今天北京天气怎么样”由于向量库中 likely 没有实时天气数据决策节点会判断需要搜索然后调用 DuckDuckGo 搜索并生成答案。效果验证这个简单的智能体展示了 LangGraph 如何编排不同的节点检索、决策、搜索、生成实现比简单 RAG 链更复杂的逻辑。你可以根据需要增加更多节点如历史记录管理、工具调用等。6. 接口 API 与批量任务6.1 构建 FastAPI 服务将 RAG 能力封装成 API方便集成到其他应用。创建src/api.pyfrom fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Optional import uvicorn from src.chain import setup_rag_chain app FastAPI(titleRAG 问答 API, description基于本地知识库的检索增强生成服务) # 全局加载链简单示例生产环境需考虑并发和重载 qa_chain None app.on_event(startup) async def startup_event(): global qa_chain print(正在加载 RAG 链...) qa_chain setup_rag_chain() print(RAG 链加载完成。) class QueryRequest(BaseModel): question: str top_k: Optional[int] 4 class QueryResponse(BaseModel): answer: str sources: List[str] app.post(/query, response_modelQueryResponse) async def query_knowledge_base(request: QueryRequest): if qa_chain is None: raise HTTPException(status_code503, detail服务未就绪) try: result qa_chain.invoke({query: request.question}) sources [doc.page_content[:200] for doc in result[source_documents][:request.top_k]] return QueryResponse(answerresult[result], sourcessources) except Exception as e: raise HTTPException(status_code500, detailf处理查询时出错: {str(e)}) class BatchQueryRequest(BaseModel): questions: List[str] top_k: Optional[int] 4 class BatchQueryResponse(BaseModel): results: List[QueryResponse] app.post(/batch_query, response_modelBatchQueryResponse) async def batch_query_knowledge_base(request: BatchQueryRequest): if qa_chain is None: raise HTTPException(status_code503, detail服务未就绪) responses [] for q in request.questions: try: result qa_chain.invoke({query: q}) sources [doc.page_content[:200] for doc in result[source_documents][:request.top_k]] responses.append(QueryResponse(answerresult[result], sourcessources)) except Exception as e: responses.append(QueryResponse(answerf处理问题{q}时出错: {str(e)}, sources[])) return BatchQueryResponse(resultsresponses) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)启动服务python src/api.py服务将在http://localhost:8000启动。访问http://localhost:8000/docs可以看到自动生成的 API 文档。6.2 API 调用测试使用curl或 Python 脚本测试接口。单次查询curl -X POST http://localhost:8000/query \ -H Content-Type: application/json \ -d {question: 什么是 Embedding, top_k: 2}批量查询curl -X POST http://localhost:8000/batch_query \ -H Content-Type: application/json \ -d {questions: [什么是 RAG, LangChain 有什么用], top_k: 2}Python 客户端示例import requests import json url http://localhost:8000/query payload {question: 向量数据库如何工作, top_k: 3} response requests.post(url, jsonpayload, timeout60) if response.status_code 200: result response.json() print(f答案: {result[answer]}) print(来源:) for i, source in enumerate(result[sources]): print(f {i1}. {source}) else: print(f请求失败: {response.status_code}, {response.text})6.3 批量任务处理对于大量文档的离线处理可以编写脚本。创建batch_process.pyimport os from src.document_loader import load_and_split_documents from src.vector_store import VectorStoreManager import logging from tqdm import tqdm logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) def batch_index_documents(data_root_dir, batch_size50): 批量索引文档目录。 data_root_dir: 包含多个子目录的根目录每个子目录为一类文档。 vs_manager VectorStoreManager() all_docs [] for category in os.listdir(data_root_dir): category_path os.path.join(data_root_dir, category) if os.path.isdir(category_path): logging.info(f正在处理类别: {category}) try: # 为每个文档块添加类别元数据 chunks load_and_split_documents(category_path) for chunk in chunks: chunk.metadata[category] category all_docs.extend(chunks) logging.info(f 类别 {category} 处理完成共 {len(chunks)} 个块。) except Exception as e: logging.error(f 处理类别 {category} 时出错: {e}) # 分批入库避免内存溢出 logging.info(f开始向量化入库总计 {len(all_docs)} 个文档块...) for i in tqdm(range(0, len(all_docs), batch_size)): batch all_docs[i:ibatch_size] vs_manager.add_documents(batch) vs_manager.persist() logging.info(批量索引完成) if __name__ __main__: # 假设你的文档按类别放在 ./data/ 下的不同子文件夹中 batch_index_documents(./data)7. 资源占用与性能观察7.1 各组件资源占用Embedding 模型 (bge-m3)GPU 模式加载模型约占用 1.2GB 显存。编码时显存占用随批量大小增加单条文本推理很快。CPU 模式内存占用约 2-3GB编码速度慢于 GPU。向量数据库 (ChromaDB)纯内存操作时内存占用与向量数据量成正比。100万条 768 维向量约占用 3GB 内存。开启持久化后会写入磁盘。本地 LLM (Ollama withqwen2.5:7b)4-bit 量化约占用 4-5GB 显存。8-bit 量化约占用 7-8GB 显存。CPU 推理内存占用可能超过 16GB速度较慢。LangGraph 工作流主要是内存开销用于维护状态和中间结果通常不大。7.2 性能优化建议Embedding 批量处理对大量文档进行向量化时使用encode方法的batch_size参数充分利用 GPU 并行能力。向量数据库选型ChromaDB轻量、易用适合中小规模数据百万级以下和原型开发。Milvus/Qdrant专为大规模向量检索设计支持分布式、高性能过滤适合生产环境海量数据。LLM 推理优化使用量化模型如 GGUF 格式。调整 Ollama 的num_ctx上下文长度和num_threadCPU线程数参数。考虑使用vLLM或TGI等高性能推理框架部署 API 服务。检索优化重排序 (Rerank)在初步检索召回后使用更精细的交叉编码器模型对结果重排序提升精度。可集成bge-reranker等模型。混合检索结合关键词检索如 BM25和向量检索提高召回率。元数据过滤利用向量数据库的元数据过滤功能在检索前缩小范围。8. 常见问题与排查方法问题现象可能原因排查方式解决方案ImportError: No module named ‘xxx’依赖包未安装或虚拟环境未激活。检查当前 Python 环境 (which python或pip list)。激活正确的虚拟环境并pip install -r requirements.txt。CUDA out of memoryGPU 显存不足。使用nvidia-smi观察显存占用。1. 减少批量大小 (batch_size)。2. 使用 CPU 模式 (device‘cpu’)。3. 使用更小的模型或量化模型。Ollama 连接失败Ollama 服务未启动或端口被占用。检查ollama serve是否运行端口 11434 是否监听。启动 Ollama 服务或检查防火墙设置。检索结果不相关1. Embedding 模型不适合领域。2. 文本切分不合理。3. 检索参数k太小或太大。1. 检查源文档内容。2. 打印检索到的文本块查看。1. 尝试其他 Embedding 模型如text2vec。2. 调整chunk_size和chunk_overlap。3. 调整k值或引入重排序。LLM 答案质量差1. 提示词设计不佳。2. 检索到的上下文质量低。3. 模型能力有限。1. 打印出传递给 LLM 的完整提示词。2. 检查上下文是否与问题相关。1. 优化提示词模板明确指令。2. 提升检索质量。3. 更换或微调 LLM。向量数据库写入/读取慢1. 数据量过大。2. 未使用持久化每次重启重新计算。观察磁盘 I/O 和 CPU 使用率。1. 对于大规模数据考虑 Milvus/Qdrant。2. 确保正确调用persist()方法。API 服务请求超时1. LLM 生成速度慢。2. 网络问题。3. 服务端处理瓶颈。查看服务日志测量各环节耗时。1. 为 API 设置合理的超时时间。2. 对 LLM 调用进行异步处理或队列管理。3. 考虑使用更快的模型或推理后端。9. 进阶LoRA 微调优化 Embedding 或 LLM如果通用 Embedding 模型或 LLM 在你的垂直领域表现不佳可以考虑使用 LoRA 进行微调。9.1 LoRA 微调 Embedding 模型目标让 Embedding 模型更适应你的专业领域术语和语义。基本步骤准备数据收集query, positive_doc, negative_doc三元组数据。正例是相关文档负例是不相关或弱相关文档。选择框架使用peft(Parameter-Efficient Fine-Tuning) 和transformers库。训练脚本概览from transformers import AutoModel, AutoTokenizer from peft import LoraConfig, get_peft_model import torch from datasets import Dataset # ... 数据加载和预处理代码 ... model_name BAAI/bge-m3 model AutoModel.from_pretrained(model_name) tokenizer AutoTokenizer.from_pretrained(model_name) # 配置 LoRA lora_config LoraConfig( r8, # LoRA 秩 lora_alpha32, target_modules[query, key, value], # 针对 Transformer 的注意力模块 lora_dropout0.1, biasnone, ) model get_peft_model(model, lora_config) model.print_trainable_parameters() # 查看可训练参数量通常只有原模型的 0.1%-1% # 定义对比学习损失函数 (如 InfoNCE) # ... 训练循环代码 ...使用微调后的模型训练完成后保存适配器权重并在推理时加载。9.2 LoRA 微调 LLM目标让 LLM 更遵循你期望的回答格式、风格或领域知识。基本步骤准备数据准备指令微调格式的数据如{instruction: ..., input: ..., output: ...}。选择工具使用trl(Transformer Reinforcement Learning) 库的SFTTrainer或unsloth进行高效微调。快速上手使用 unslothpip install unslothfrom unsloth import FastLanguageModel import torch model, tokenizer FastLanguageModel.from_pretrained( model_name unsloth/llama-3-8b-bnb-4bit, # 或其他支持模型 max_seq_length 2048, dtype None, load_in_4bit True, # 4-bit 量化加载 ) model FastLanguageModel.get_peft_model( model, r 16, target_modules [q_proj, k_proj, v_proj, o_proj, gate_proj, up_proj, down_proj,], lora_alpha 16, lora_dropout 0, bias none, use_gradient_checkpointing unsloth, random_state 3407, use_rslora False, loftq_config None, ) # ... 准备数据并训练 ...集成到 RAG将微调后的 LLM 模型路径或 Ollama 中的新模型名配置到你的 RAG 链或 LangGraph 中。硬件要求LoRA 微调 LLM 通常需要 12GB 以上显存对于 7B 模型。使用unsloth或bitsandbytes的 4-bit 量化可以显著降低需求。10. 总结与下一步通过本教程我们完成了一个从文档处理到智能问答的完整 RAG 系统搭建。这套流程的核心价值在于可落地你可以在自己的机器上使用开源组件构建一个服务于私有数据的 AI 应用。最值得尝试的点模块化清晰每个组件加载、嵌入、存储、检索、生成都可独立替换和升级。本地化部署从 Embedding 模型、向量数据库到 LLM均可运行在本地环境保障数据隐私。灵活的编排能力通过 LangGraph你可以设计复杂的多步骤工作流超越简单的问答链。最先应该验证的功能端到端流程确保从一篇 PDF 文档到最终生成答案的整个链路能跑通。检索质量用几个关键问题测试看返回的文档片段是否真的相关。API 服务尝试通过 HTTP 请求调用你的知识库这是集成到其他应用的基础。最容易踩的坑环境配置Python 版本、CUDA 版本、依赖冲突。务必使用虚拟环境。显存不足同时加载 Embedding 模型和 LLM 可能导致 OOM。建议分开测试或使用 CPU 模式暂替。文档预处理不合理的文本切分是检索效果差的常见原因。多尝试不同的chunk_size和切分策略。后续扩展方向引入重排序 (Rerank)在向量检索后增加一个重排序步骤使用bge-reranker等模型对 Top-K 结果重新打分排序能显著提升精度。实现多路检索结合关键词检索如 BM25和向量检索提升召回率。构建 Web UI使用Gradio或Streamlit快速搭建一个交互式前端界面。接入更多工具在 LangGraph 智能体中集成计算器、代码执行器、网络搜索等工具打造更强大的 AI 助手。探索 Agentic RAG让智能体主动决定何时检索、检索什么、如何迭代优化查询实现更自主的问答。建议将本教程的代码作为起点根据你的具体数据和需求进行调整。在实践中数据质量文档清洗、切分和提示词工程往往比模型本身更重要。