这次我们来看一个完整的 RAG 系统构建实战。RAG检索增强生成技术正成为连接大模型与私有知识的关键桥梁但一个真正可用的系统远不止“向量检索”那么简单。本文将带你从零构建一个包含 Embedding 模型、向量数据库、检索优化策略并最终通过 LangGraph 实现智能体工作流的全流程系统。无论你是想搭建本地知识库还是为业务系统集成智能问答能力这套从组件选型到工程落地的完整方案都值得你深入实践。本文的重点不是空谈概念而是提供一套可执行、可验证的实战指南。我们将重点关注每个组件的选型理由、部署门槛、接口调用方式以及如何将它们串联成一个稳定高效的系统。你会看到如何用 BGE-M3 模型处理文本用 Chroma 或 Qdrant 存储向量用 Rerank 模型优化结果并最终通过 LangGraph 构建一个具备长期记忆和复杂决策能力的 AI 智能体。1. 核心能力速览在深入细节之前我们先通过下表快速了解这个 RAG 系统实战方案的核心能力和技术选型让你对整体架构和资源要求有一个清晰的认识。能力项说明与选型核心目标构建端到端的 RAG 系统实现从文档处理、向量检索到智能生成的完整流水线。Embedding 模型推荐使用BGE-M3。它支持多语言、长文本且在同规模模型中检索效果领先。也可选用Qwen2.5-Embedding等国产优秀模型。向量数据库Chroma轻量、易集成或Qdrant高性能、生产级。本文将以 Chroma 为例进行演示。检索优化引入Rerank重排序模型如 BGE-Reranker对初步检索结果进行精排大幅提升答案相关性。智能体框架使用LangGraph构建可编排、带状态、支持长期记忆的 AI 智能体工作流超越简单的链式调用。大模型基础可对接 OpenAI API、通义千问、DeepSeek 等云端 API或本地部署的Ollama、vLLM等推理框架。硬件门槛Embedding 与 Rerank 模型可在 CPU 或 GPU 上运行。BGE-M3 参数量约 2.8 亿6GB 以上显存可获得更好性能。向量数据库内存占用与文档规模正相关小型知识库 2-4GB RAM 足够。大模型若本地部署需根据模型大小7B、14B等准备相应显存。启动与部署各组件均可通过 Docker 或 Pip 安装提供标准的 API 接口HTTP/gRPC方便独立部署和集成。是否支持 API是。Embedding、向量数据库、大模型、LangGraph 服务均可通过 API 调用便于系统集成。是否支持批量任务是。文档解析、向量化入库、批量问答均可通过脚本实现自动化处理。适合场景企业知识库问答、AI 客服、代码助手、个人学习笔记检索、长文档分析与总结等。2. 适用场景与使用边界这个 RAG 系统方案并非万能明确其适用边界能帮助你更好地决策。它非常适合以下场景私有知识问答你有大量的内部文档、手册、报告需要一个大模型来准确回答相关问题且不希望泄露数据到公网。降低大模型幻觉通过检索到的真实文档作为生成依据能有效约束大模型“胡言乱语”提升回答的准确性和可信度。构建专业领域助手例如法律、医疗、金融等领域通用大模型缺乏专业知识RAG 可以快速为其注入领域知识。实现长期记忆与个性化结合 LangGraph 的状态管理可以让智能体记住与用户的对话历史、个人偏好提供连贯的个性化服务。它可能不适用或需要额外工作的场景实时性要求极高的信息如果知识库更新频率是分钟级甚至秒级传统的向量检索入库流程可能有延迟需要考虑流式处理方案。高度结构化、精确的查询例如“查询2024年3月A产品的全部销售数据”这类问题可能更适合直接查询数据库而非语义检索。对成本极其敏感的超大规模应用Embedding 和重排序模型的调用、向量数据库的存储与计算在文档量极大时会产生可观成本。重要的合规与安全边界数据安全处理企业敏感数据时务必确保整个流水线Embedding、向量数据库、大模型都部署在可控的内网或私有云环境。版权与隐私为 RAG 系统灌入的文档必须确保你拥有合法使用权避免侵犯他人版权或泄露个人隐私信息。生成内容审核即使提供了检索依据大模型的生成内容仍需进行合规性审核特别是面向公众的服务。3. 环境准备与前置条件开始搭建前请确保你的开发环境满足以下基本要求。我们将以 Python 为主要开发语言。操作系统Linux (Ubuntu 20.04)、macOS 或 Windows (WSL2 推荐)。生产环境建议使用 Linux。Python 环境Python 3.9 或 3.10。推荐使用conda或venv创建独立的虚拟环境。# 创建并激活虚拟环境 (以 conda 为例) conda create -n rag_system python3.10 conda activate rag_system包管理工具pip版本需更新至最新。硬件建议CPU现代多核处理器如 Intel i5/i7 或 AMD Ryzen 5/7。内存建议 16GB 或以上确保能同时运行多个服务。GPU可选但推荐对于 Embedding 和 Rerank 模型拥有 NVIDIA GPU如 GTX 1660, RTX 3060 及以上并安装 CUDA 工具包可以显著加速推理。本文演示也会涵盖 CPU 运行方案。磁盘空间预留至少 10GB 空间用于安装依赖、下载模型和存储向量数据。网络能顺畅访问 GitHub、PyPI 和 Hugging Face 以下载代码和模型。4. 安装部署与启动方式我们将分步安装各个核心组件。你可以选择将所有组件集成在一个项目中也可以将它们作为独立微服务部署。4.1 安装基础依赖首先安装通用的 Python 库包括机器学习框架、HTTP 客户端等。pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 # 根据你的CUDA版本调整 pip install transformers sentence-transformers langchain langchain-community langgraph pip install chromadb pypdf python-dotenv tiktoken # 向量数据库、PDF解析、环境变量管理 pip install fastapi uvicorn # 用于构建API服务4.2 部署 Embedding 模型服务我们使用sentence-transformers库来加载和运行 BGE-M3 模型。你可以将其封装为一个简单的 HTTP 服务。创建一个文件embedding_service.pyfrom sentence_transformers import SentenceTransformer import numpy as np import uvicorn from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List app FastAPI(titleBGE-M3 Embedding Service) # 加载模型首次运行会自动从 Hugging Face 下载 # 指定设备如果有GPU则使用cuda model SentenceTransformer(BAAI/bge-m3, devicecuda) class EmbeddingRequest(BaseModel): texts: List[str] normalize_embeddings: bool True # 是否对向量进行归一化有利于余弦相似度计算 class EmbeddingResponse(BaseModel): embeddings: List[List[float]] model: str app.post(/embed, response_modelEmbeddingResponse) async def get_embeddings(request: EmbeddingRequest): try: # 模型推理 embeddings model.encode(request.texts, normalize_embeddingsrequest.normalize_embeddings, convert_to_numpyTrue) # 转换为Python list格式返回 embeddings_list embeddings.tolist() return EmbeddingResponse(embeddingsembeddings_list, modelBGE-M3) except Exception as e: raise HTTPException(status_code500, detailstr(e)) if __name__ __main__: # 启动服务默认监听 8001 端口 uvicorn.run(app, host0.0.0.0, port8001)启动服务python embedding_service.py服务启动后你可以通过http://localhost:8001/docs查看自动生成的 API 文档并进行测试。4.3 部署向量数据库 (Chroma)Chroma 可以以客户端-服务器模式运行也可以嵌入在应用中。这里我们以独立服务模式运行方便其他组件调用。使用 Docker 启动 Chroma 是最简单的方式docker pull chromadb/chroma docker run -d --name chroma_db -p 8000:8000 chromadb/chroma服务将在http://localhost:8000运行。你也可以使用chromadb客户端库在 Python 代码中直接创建内存或持久化集合。4.4 部署 Rerank 模型服务重排序模型用于对检索到的 Top K 个文档进行精排。我们同样将其封装为 API。创建rerank_service.pyfrom transformers import AutoModelForSequenceClassification, AutoTokenizer import torch import uvicorn from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List app FastAPI(titleBGE Reranker Service) # 加载模型和分词器 model_name BAAI/bge-reranker-large # 也可以使用 base 版本 tokenizer AutoTokenizer.from_pretrained(model_name) model AutoModelForSequenceClassification.from_pretrained(model_name) model.eval() device torch.device(cuda if torch.cuda.is_available() else cpu) model.to(device) class RerankRequest(BaseModel): query: str documents: List[str] class RerankResponse(BaseModel): scores: List[float] ranked_doc_indices: List[int] app.post(/rerank, response_modelRerankResponse) async def rerank_documents(request: RerankRequest): try: pairs [[request.query, doc] for doc in request.documents] with torch.no_grad(): inputs tokenizer(pairs, paddingTrue, truncationTrue, return_tensorspt, max_length512) inputs {k: v.to(device) for k, v in inputs.items()} scores model(**inputs, return_dictTrue).logits.view(-1, ).float() scores scores.cpu().tolist() # 按分数从高到低排序 ranked_indices sorted(range(len(scores)), keylambda i: scores[i], reverseTrue) ranked_scores [scores[i] for i in ranked_indices] return RerankResponse(scoresranked_scores, ranked_doc_indicesranked_indices) except Exception as e: raise HTTPException(status_code500, detailstr(e)) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8002)启动服务python rerank_service.py4.5 连接大模型与 LangGraph这里我们以调用 OpenAI 兼容 API例如本地部署的 Ollama为例展示如何将其与 LangGraph 集成。首先确保你有一个可用的 LLM 服务端点。例如使用 Ollama 在本地运行一个模型# 拉取并运行一个模型例如 Qwen2.5:7B ollama pull qwen2.5:7b ollama run qwen2.5:7b # Ollama 默认 API 服务在 http://localhost:11434然后在 Python 中配置 LangChain 和 LangGraphimport os from langchain_openai import ChatOpenAI from langchain.schema import HumanMessage, SystemMessage from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator # 1. 配置 LLM (指向本地 Ollama) llm ChatOpenAI( base_urlhttp://localhost:11434/v1, api_keyollama, # Ollama 不需要真实的 key modelqwen2.5:7b, temperature0.1 ) # 2. 定义状态结构 class AgentState(TypedDict): question: str retrieved_docs: list reranked_docs: list final_answer: str conversation_history: Annotated[list, operator.add] # 用于存储对话历史实现长期记忆 # 3. 定义各个节点函数 def retrieve(state: AgentState): 模拟检索节点这里调用我们之前部署的向量数据库和 Embedding 服务 # 实际项目中这里应调用向量数据库的检索接口 # 示例返回模拟的检索结果 mock_docs [ 文档ARAG 系统包含检索、增强、生成三个核心步骤。, 文档B向量数据库如 Chroma 用于高效存储和查询 Embedding。, 文档CLangGraph 可以构建有状态的、多步骤的 AI 智能体工作流。 ] return {retrieved_docs: mock_docs} def rerank(state: AgentState): 重排序节点调用 Rerank 服务对检索结果排序 # 实际项目中这里应调用 rerank_service 的 API # 假设我们直接使用原始顺序 return {reranked_docs: state[retrieved_docs]} def generate_answer(state: AgentState): 生成节点基于重排序后的文档调用 LLM 生成最终答案 context \n.join(state[reranked_docs]) system_prompt f你是一个专业的助手。请严格根据以下提供的上下文信息来回答问题。如果上下文信息不足以回答问题请如实告知。 上下文 {context} messages [ SystemMessage(contentsystem_prompt), HumanMessage(contentstate[question]) ] response llm.invoke(messages) return {final_answer: response.content} # 4. 构建图 workflow StateGraph(AgentState) workflow.add_node(retrieve, retrieve) workflow.add_node(rerank, rerank) workflow.add_node(generate, generate_answer) # 5. 定义边执行顺序 workflow.set_entry_point(retrieve) workflow.add_edge(retrieve, rerank) workflow.add_edge(rerank, generate) workflow.add_edge(generate, END) # 6. 编译图 app workflow.compile() # 7. 运行图 initial_state {question: 请解释 RAG 系统的基本流程。, conversation_history: []} result app.invoke(initial_state) print(result[final_answer])这个示例展示了 LangGraph 如何将检索、重排序、生成等步骤编排成一个有向图。你可以在此基础上增加条件判断、循环、人工审核等节点构建更复杂的智能体。5. 功能测试与效果验证现在我们来验证各个组件以及整个流水线是否工作正常。5.1 测试 Embedding 服务使用curl或 Pythonrequests库测试 Embedding API。curl -X POST http://localhost:8001/embed \ -H Content-Type: application/json \ -d { texts: [什么是人工智能, 机器学习是人工智能的一个子领域。], normalize_embeddings: true }预期返回一个 JSON包含两个文本对应的 1024 维BGE-M3 的默认维度向量。检查向量是否为浮点数列表且长度正确。5.2 测试向量数据库的写入与检索编写一个简单的脚本将一段文本转化为向量并存入 Chroma然后进行相似性检索。import chromadb from sentence_transformers import SentenceTransformer import uuid # 初始化客户端和模型 chroma_client chromadb.HttpClient(hostlocalhost, port8000) embedding_model SentenceTransformer(BAAI/bge-m3) # 创建或获取一个集合类似于数据库的表 collection chroma_client.get_or_create_collection(nametest_knowledge) # 准备文档 documents [ LangChain 是一个用于开发大语言模型应用的框架。, 向量数据库专门为高维向量数据的快速检索而设计。, Embedding 模型将文本转化为具有语义信息的数值向量。 ] ids [str(uuid.uuid4()) for _ in documents] # 生成向量 embeddings embedding_model.encode(documents).tolist() # 添加到集合 collection.add( documentsdocuments, embeddingsembeddings, idsids ) print(文档已入库。) # 进行检索 query 有什么工具可以处理文本向量 query_embedding embedding_model.encode([query]).tolist() results collection.query( query_embeddingsquery_embedding, n_results2 ) print(检索结果) for i, doc in enumerate(results[documents][0]): print(f{i1}. {doc})运行脚本应能正确检索到与查询语义相近的文档例如“Embedding 模型...”。5.3 测试 Rerank 服务模拟一个检索返回了多个可能相关文档的场景测试 Rerank 服务是否能找出最相关的那一个。import requests rerank_url http://localhost:8002/rerank query 如何选择合适的向量数据库 candidate_docs [ MySQL 是一种流行的关系型数据库。, # 不相关 Chroma 是一个轻量级的开源向量数据库易于集成。, # 最相关 Python 是一种编程语言。, # 不相关 向量数据库的性能指标包括 QPS 和召回率。 # 相关 ] payload { query: query, documents: candidate_docs } response requests.post(rerank_url, jsonpayload) if response.status_code 200: result response.json() print(重排序分数:, result[scores]) print(重排序后的索引:, result[ranked_doc_indices]) print(最相关的文档:, candidate_docs[result[ranked_doc_indices][0]]) else: print(请求失败:, response.text)预期输出中关于 Chroma 的文档应该获得最高分并被排在第一位。5.4 端到端 RAG 流程测试将以上所有服务串联模拟一个完整的用户问答流程。用户提问“用 LangGraph 构建智能体有什么优势”系统流程将问题转化为向量调用 Embedding 服务。在向量数据库中检索出 Top 5 相关文档片段。使用 Rerank 服务对 5 个片段进行精排选出 Top 3。将问题和 Top 3 文档片段组合成提示词发送给大模型。大模型生成最终答案。验证标准最终答案应包含“状态管理”、“多步骤编排”、“长期记忆”、“循环与条件分支”等 LangGraph 的关键优势点并且这些信息应来源于检索到的文档而非大模型的通用知识。你可以通过编写一个集成脚本来自动化这个测试流程并检查最终答案的准确性和相关性。6. 接口 API 与批量任务一个生产可用的 RAG 系统必须提供稳定的 API 并支持批量处理。6.1 构建统一的 RAG API 网关我们可以创建一个 FastAPI 应用作为整个系统的入口内部调用 Embedding、向量数据库、Rerank 和 LLM 服务。# rag_gateway.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List import requests import logging app FastAPI(titleRAG System Gateway) logging.basicConfig(levellogging.INFO) EMBEDDING_URL http://localhost:8001/embed CHROMA_URL http://localhost:8000 # Chroma 客户端直接连接 RERANK_URL http://localhost:8002/rerank LLM_URL http://localhost:11434/v1/chat/completions # Ollama 示例 class QueryRequest(BaseModel): question: str top_k_retrieve: int 5 top_k_rerank: int 3 class QueryResponse(BaseModel): answer: str used_contexts: List[str] process_time: float app.post(/query, response_modelQueryResponse) async def query_knowledge_base(request: QueryRequest): import time start_time time.time() try: # 1. 将问题转换为向量 embed_response requests.post(EMBEDDING_URL, json{texts: [request.question]}) query_vector embed_response.json()[embeddings][0] # 2. 向量检索 (这里简化实际需调用Chroma HTTP API或客户端) # 假设我们有一个函数 chroma_retrieve retrieved_docs chroma_retrieve(query_vector, request.top_k_retrieve) # 3. 重排序 rerank_payload {query: request.question, documents: retrieved_docs} rerank_response requests.post(RERANK_URL, jsonrerank_payload) ranked_indices rerank_response.json()[ranked_doc_indices] top_contexts [retrieved_docs[i] for i in ranked_indices[:request.top_k_rerank]] # 4. 调用 LLM 生成答案 llm_prompt f基于以下上下文回答问题。如果上下文不包含答案请说“根据现有资料无法回答”。 上下文{ .join(top_contexts)} 问题{request.question} 答案 llm_payload { model: qwen2.5:7b, messages: [{role: user, content: llm_prompt}], stream: False } llm_response requests.post(LLM_URL, jsonllm_payload) answer llm_response.json()[choices][0][message][content] process_time time.time() - start_time return QueryResponse(answeranswer, used_contextstop_contexts, process_timeprocess_time) except Exception as e: logging.error(fRAG 流程错误: {e}) raise HTTPException(status_code500, detailf内部处理错误: {e}) def chroma_retrieve(query_vector, top_k): # 此处应实现真正的 Chroma 检索逻辑 # 示例返回模拟数据 return [ 文档1LangGraph 支持创建有状态的图。, 文档2智能体可以根据状态决定下一步行动。, 文档3RAG 结合了检索和生成。 ] if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8003)6.2 批量文档处理与入库对于大量文档我们需要一个离线批处理流程。# batch_processing.py import os from glob import glob from PyPDF2 import PdfReader from sentence_transformers import SentenceTransformer import chromadb from tqdm import tqdm import hashlib def extract_text_from_pdf(pdf_path): 从PDF提取文本 reader PdfReader(pdf_path) text for page in reader.pages: text page.extract_text() \n return text def split_text(text, chunk_size500, chunk_overlap50): 将长文本分割成小块简单的按字符分割生产环境建议用语义分割 chunks [] start 0 while start len(text): end start chunk_size chunk text[start:end] chunks.append(chunk) start end - chunk_overlap return chunks def process_document_folder(folder_path, collection_nameknowledge_base): 批量处理文件夹内的文档支持PDF、TXT # 初始化模型和数据库客户端 embed_model SentenceTransformer(BAAI/bge-m3) chroma_client chromadb.PersistentClient(path./chroma_db) # 持久化到本地 collection chroma_client.get_or_create_collection(namecollection_name) supported_ext [.pdf, .txt] all_chunks [] all_metadatas [] all_ids [] for ext in supported_ext: for file_path in tqdm(glob(os.path.join(folder_path, f*{ext})), descf处理 {ext} 文件): try: if ext .pdf: text extract_text_from_pdf(file_path) else: # .txt with open(file_path, r, encodingutf-8) as f: text f.read() chunks split_text(text) file_id hashlib.md5(file_path.encode()).hexdigest()[:8] for i, chunk in enumerate(chunks): chunk_id f{file_id}_chunk{i} all_chunks.append(chunk) all_metadatas.append({source: os.path.basename(file_path), chunk_index: i}) all_ids.append(chunk_id) except Exception as e: print(f处理文件 {file_path} 时出错: {e}) # 批量生成向量并入库 if all_chunks: print(f正在为 {len(all_chunks)} 个文本块生成向量...) embeddings embed_model.encode(all_chunks, show_progress_barTrue).tolist() print(正在写入向量数据库...) collection.add( documentsall_chunks, embeddingsembeddings, metadatasall_metadatas, idsall_ids ) print(f批量处理完成成功入库 {len(all_chunks)} 个文本块。) else: print(未找到可处理的文档。) if __name__ __main__: # 指定你的文档文件夹路径 doc_folder ./my_documents process_document_folder(doc_folder)这个脚本会遍历指定文件夹下的 PDF 和 TXT 文件进行文本提取、分块、向量化并存入本地的 Chroma 数据库。7. 资源占用与性能观察部署和运行 RAG 系统时需要密切关注资源消耗以便进行容量规划和优化。Embedding 模型推理CPU 模式运行 BGE-M3 这类模型单次推理约 500 字符在 Intel i7 上可能耗时 100-300 毫秒。批处理可以显著提升吞吐量。GPU 模式在 RTX 3060 (12GB) 上推理延迟可降至 10-50 毫秒。使用nvidia-smi观察显存占用加载 BGE-M3 模型约占用 1.5-2GB 显存。建议对于高频调用务必启用 GPU 并采用批处理 API。向量数据库内存Chroma 在内存中存储索引。100 万个 1024 维向量的集合大约占用 4GB 内存向量本身加上额外的索引开销。磁盘如果使用持久化模式数据会保存在磁盘。定期清理无用集合。查询性能查询速度与集合大小和索引类型有关。对于百万级数据HNSW 索引可以在毫秒级返回结果。Rerank 模型重排序通常是计算密集型的。BGE-Reranker-Large 模型比同规模 Embedding 模型更耗资源。在 CPU 上对 10 个候选文档进行重排序可能需要 1-2 秒在 GPU 上可缩短至 200 毫秒以内。策略不要对大量候选文档如 Top 100进行重排序通常先通过向量检索筛选出 Top 10-20再进行精排。大模型推理这是最大的资源消耗者。本地运行 7B 参数模型如 Qwen2.5-7B至少需要 8-10GB GPU 显存进行推理。如果使用 API 服务如 OpenAI, DeepSeek则关注网络延迟和 Token 消耗成本。优化使用量化模型如 GPTQ, AWQ可将显存需求降低 30-50%。对于问答任务合理设置max_tokens以避免生成过长无用内容。综合链路延迟一个完整的 RAG 查询检索重排序生成延迟可能在 2 秒到 10 秒不等取决于文档库大小、模型部署位置和网络状况。性能观测点在 API 网关层记录每个环节的耗时便于定位瓶颈。常见的瓶颈顺序是LLM生成 Rerank Embedding 向量检索。8. 常见问题与排查方法在搭建和运行过程中你可能会遇到以下问题。这里提供排查思路。问题现象可能原因排查方式解决方案Embedding 服务启动失败1. 端口被占用。2. 模型文件下载失败或损坏。3. CUDA 版本与 PyTorch 不匹配。1. 检查端口8001是否被其他程序占用 (netstat -an | grep 8001)。2. 查看服务启动日志确认模型下载进度和错误。3. 运行python -c import torch; print(torch.cuda.is_available())验证 CUDA。1. 更换服务端口。2. 手动从 Hugging Face 下载模型到本地在代码中指定local_files_onlyTrue。3. 重新安装匹配的 PyTorch 和 CUDA 版本。向量数据库连接超时1. Chroma 服务未启动。2. 防火墙阻止了端口访问。3. 客户端配置的主机/端口错误。1. 检查 Docker 容器或 Chroma 进程是否在运行 (docker ps或ps aux | grep chroma)。2. 尝试在服务器本地用curl http://localhost:8000/api/v1/heartbeat测试。1. 启动 Chroma 服务。2. 配置防火墙规则开放对应端口。3. 确认客户端连接的host和port与服务端一致。检索结果不相关1. Embedding 模型不适合当前领域。2. 文本分块策略不合理块太大或太小。3. 没有使用重排序。1. 用一些领域内关键词测试 Embedding 的相似度。2. 检查分块后的文本是否保持了语义完整性。3. 对比使用重排序前后的结果质量。1. 尝试其他 Embedding 模型如text-embedding-3-small,nomic-embed-text。2. 调整分块大小和重叠长度或采用语义分割库。3. 集成 Rerank 模型。大模型回答未基于检索内容1. 提示词Prompt设计不佳未强制模型参考上下文。2. 检索到的上下文本身质量差或无关。1. 检查发送给 LLM 的完整 Prompt确保包含明确的指令如“请仅根据以下上下文回答”。2. 单独测试检索环节看返回的文档是否真的与问题相关。1. 优化 Prompt 工程增加系统指令的约束力。2. 提升检索质量优化 Embedding、分块、重排序。批量入库速度慢1. 单条处理未使用批处理。2. 在 CPU 上进行 Embedding 计算。3. 数据库写入未优化。1. 观察代码是逐条调用encode还是批量调用。2. 检查模型是否运行在 GPU 上。3. 观察磁盘 I/O 或网络延迟。1. 将文档累积到一定数量如 32 或 64 条后批量调用model.encode。2. 确保环境支持 CUDA 并将模型加载到 GPU。3. 对于大规模入库考虑使用向量数据库的批量导入工具。LangGraph 图编译或运行错误1. 状态State定义与节点函数返回值不匹配。2. 图中存在循环依赖或未定义的边。1. 仔细检查AgentState的TypedDict定义与每个节点返回的字典键名是否一致。2. 使用app.get_graph().draw_mermaid()输出图结构检查。1. 确保每个节点返回的字典是AgentState的子集。2. 遵循set_entry_point和add_edge的顺序确保图是连通且无歧义的。系统整体延迟过高1. 某个组件通常是 LLM成为瓶颈。2. 网络调用过多或序列化开销大。1. 在 API 网关记录每个微服务的调用耗时。2. 检查是否在循环中频繁调用服务而未做缓存或批处理。1. 对 LLM 生成进行流式输出以提升用户体验感知。2. 考虑将 Embedding 和 Rerank 服务与主应用放在同机房减少网络延迟。3. 对不变的查询或文档 Embedding 进行缓存。9. 最佳实践与使用建议基于实战经验以下建议能帮助你构建更稳健、高效的 RAG 系统。从简单开始逐步迭代不要一开始就追求完美的多阶段检索和复杂智能体。先搭建一个最基础的“Embedding 向量检索 LLM”管道并跑通确保数据能流起来。然后再逐步加入重排序、查询改写、元数据过滤等优化。重视数据预处理与分块垃圾进垃圾出。文档的清洗、格式统一、分块策略对最终效果影响巨大。对于技术文档按章节或子标题分块效果更好。可以尝试不同的分块大小如 256, 512, 1024 字符和重叠长度通过检索效果评估最佳配置。实施全面的评估体系不要只靠“感觉”判断系统好坏。定义评估指标如检索相关度检索到的 Top K 文档中有多少是真正相关的答案准确性生成的答案与标准答案或事实是否一致答案相关性答案是否直接回应了问题延迟与吞吐量单个查询响应时间、系统能承受的 QPS。为生产环境做好准备服务化与监控将 Embedding、向量数据库、Rerank、LLM 网关等都部署为独立的、可监控的微服务。配置管理使用环境变量或配置文件管理模型路径、API 密钥、服务地址等避免硬编码。错误处理与重试在网络调用、模型推理等环节加入健壮的错误处理和指数退避重试机制。版本控制对模型版本、数据库 Schema、API 接口进行版本控制便于回滚和更新。安全与合规始终优先访问控制为你的 RAG API 网关添加认证如 API Key、JWT。输入输出过滤对用户输入进行必要的清洗和过滤防止 Prompt 注入。对模型输出进行内容安全审核。数据审计记录用户的查询和系统的回答用于效果分析和问题追溯同时注意隐私保护。探索进阶架构当基础 RAG 稳定后可以探索更高级的模式Agentic RAG让智能体主动决定何时检索、检索什么、如何迭代优化查询。Hybrid Search结合关键词搜索BM25和向量搜索取长补短。Query Rewriting/Expansion对用户原始查询进行改写或扩展提升检索召回率。10. 总结与下一步通过本文的实战演练你应该已经掌握了一个现代化 RAG 系统的核心组件和搭建流程从 BGE-M3 生成 Embedding到用 Chroma 存储和检索向量再到用 Rerank 模型提升精度最后用 LangGraph 编排成一个可扩展的智能体工作流。这套组合拳解决了传统 RAG 在准确性、可控性和复杂性上的诸多痛点。最值得你立刻动手尝试的是将你本地的一份文档比如一个项目 README 或一篇技术文章通过我们提供的批量处理脚本灌入系统然后提出几个问题看看它能否准确回答。这个端到端的验证能让你最快地感受到 RAG 的能力和当前局限。最容易踩的坑通常是环境配置CUDA 版本、端口冲突和数据预处理分块不合理导致检索失效。建议严格按照第 3、4 节的步骤准备环境并花时间优化你的文档分块策略。接下来你可以沿着这几个方向深入性能优化尝试量化 Embedding 和 Rerank 模型测试 GPU 推理与批处理带来的提升。效果优化在 LangGraph 中引入“查询理解”和“答案验证”节点构建一个自我修正的循环。工程化用 Docker Compose 或 Kubernetes 编排所有服务并接入 Prometheus 和 Grafana 进行监控。领域适配针对你的专业领域如法律、医疗代码收集领域数据对 Embedding 模型进行微调LoRA这将带来最显著的效果提升。构建 RAG 系统是一个持续迭代的过程希望这份从零到一的实战指南能成为你可靠的起点。建议收藏本文在部署和优化过程中随时参考。 从零构建RAG系统实战：BGE-M3、Chroma与LangGraph全流程指南
这次我们来看一个完整的 RAG 系统构建实战。RAG检索增强生成技术正成为连接大模型与私有知识的关键桥梁但一个真正可用的系统远不止“向量检索”那么简单。本文将带你从零构建一个包含 Embedding 模型、向量数据库、检索优化策略并最终通过 LangGraph 实现智能体工作流的全流程系统。无论你是想搭建本地知识库还是为业务系统集成智能问答能力这套从组件选型到工程落地的完整方案都值得你深入实践。本文的重点不是空谈概念而是提供一套可执行、可验证的实战指南。我们将重点关注每个组件的选型理由、部署门槛、接口调用方式以及如何将它们串联成一个稳定高效的系统。你会看到如何用 BGE-M3 模型处理文本用 Chroma 或 Qdrant 存储向量用 Rerank 模型优化结果并最终通过 LangGraph 构建一个具备长期记忆和复杂决策能力的 AI 智能体。1. 核心能力速览在深入细节之前我们先通过下表快速了解这个 RAG 系统实战方案的核心能力和技术选型让你对整体架构和资源要求有一个清晰的认识。能力项说明与选型核心目标构建端到端的 RAG 系统实现从文档处理、向量检索到智能生成的完整流水线。Embedding 模型推荐使用BGE-M3。它支持多语言、长文本且在同规模模型中检索效果领先。也可选用Qwen2.5-Embedding等国产优秀模型。向量数据库Chroma轻量、易集成或Qdrant高性能、生产级。本文将以 Chroma 为例进行演示。检索优化引入Rerank重排序模型如 BGE-Reranker对初步检索结果进行精排大幅提升答案相关性。智能体框架使用LangGraph构建可编排、带状态、支持长期记忆的 AI 智能体工作流超越简单的链式调用。大模型基础可对接 OpenAI API、通义千问、DeepSeek 等云端 API或本地部署的Ollama、vLLM等推理框架。硬件门槛Embedding 与 Rerank 模型可在 CPU 或 GPU 上运行。BGE-M3 参数量约 2.8 亿6GB 以上显存可获得更好性能。向量数据库内存占用与文档规模正相关小型知识库 2-4GB RAM 足够。大模型若本地部署需根据模型大小7B、14B等准备相应显存。启动与部署各组件均可通过 Docker 或 Pip 安装提供标准的 API 接口HTTP/gRPC方便独立部署和集成。是否支持 API是。Embedding、向量数据库、大模型、LangGraph 服务均可通过 API 调用便于系统集成。是否支持批量任务是。文档解析、向量化入库、批量问答均可通过脚本实现自动化处理。适合场景企业知识库问答、AI 客服、代码助手、个人学习笔记检索、长文档分析与总结等。2. 适用场景与使用边界这个 RAG 系统方案并非万能明确其适用边界能帮助你更好地决策。它非常适合以下场景私有知识问答你有大量的内部文档、手册、报告需要一个大模型来准确回答相关问题且不希望泄露数据到公网。降低大模型幻觉通过检索到的真实文档作为生成依据能有效约束大模型“胡言乱语”提升回答的准确性和可信度。构建专业领域助手例如法律、医疗、金融等领域通用大模型缺乏专业知识RAG 可以快速为其注入领域知识。实现长期记忆与个性化结合 LangGraph 的状态管理可以让智能体记住与用户的对话历史、个人偏好提供连贯的个性化服务。它可能不适用或需要额外工作的场景实时性要求极高的信息如果知识库更新频率是分钟级甚至秒级传统的向量检索入库流程可能有延迟需要考虑流式处理方案。高度结构化、精确的查询例如“查询2024年3月A产品的全部销售数据”这类问题可能更适合直接查询数据库而非语义检索。对成本极其敏感的超大规模应用Embedding 和重排序模型的调用、向量数据库的存储与计算在文档量极大时会产生可观成本。重要的合规与安全边界数据安全处理企业敏感数据时务必确保整个流水线Embedding、向量数据库、大模型都部署在可控的内网或私有云环境。版权与隐私为 RAG 系统灌入的文档必须确保你拥有合法使用权避免侵犯他人版权或泄露个人隐私信息。生成内容审核即使提供了检索依据大模型的生成内容仍需进行合规性审核特别是面向公众的服务。3. 环境准备与前置条件开始搭建前请确保你的开发环境满足以下基本要求。我们将以 Python 为主要开发语言。操作系统Linux (Ubuntu 20.04)、macOS 或 Windows (WSL2 推荐)。生产环境建议使用 Linux。Python 环境Python 3.9 或 3.10。推荐使用conda或venv创建独立的虚拟环境。# 创建并激活虚拟环境 (以 conda 为例) conda create -n rag_system python3.10 conda activate rag_system包管理工具pip版本需更新至最新。硬件建议CPU现代多核处理器如 Intel i5/i7 或 AMD Ryzen 5/7。内存建议 16GB 或以上确保能同时运行多个服务。GPU可选但推荐对于 Embedding 和 Rerank 模型拥有 NVIDIA GPU如 GTX 1660, RTX 3060 及以上并安装 CUDA 工具包可以显著加速推理。本文演示也会涵盖 CPU 运行方案。磁盘空间预留至少 10GB 空间用于安装依赖、下载模型和存储向量数据。网络能顺畅访问 GitHub、PyPI 和 Hugging Face 以下载代码和模型。4. 安装部署与启动方式我们将分步安装各个核心组件。你可以选择将所有组件集成在一个项目中也可以将它们作为独立微服务部署。4.1 安装基础依赖首先安装通用的 Python 库包括机器学习框架、HTTP 客户端等。pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 # 根据你的CUDA版本调整 pip install transformers sentence-transformers langchain langchain-community langgraph pip install chromadb pypdf python-dotenv tiktoken # 向量数据库、PDF解析、环境变量管理 pip install fastapi uvicorn # 用于构建API服务4.2 部署 Embedding 模型服务我们使用sentence-transformers库来加载和运行 BGE-M3 模型。你可以将其封装为一个简单的 HTTP 服务。创建一个文件embedding_service.pyfrom sentence_transformers import SentenceTransformer import numpy as np import uvicorn from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List app FastAPI(titleBGE-M3 Embedding Service) # 加载模型首次运行会自动从 Hugging Face 下载 # 指定设备如果有GPU则使用cuda model SentenceTransformer(BAAI/bge-m3, devicecuda) class EmbeddingRequest(BaseModel): texts: List[str] normalize_embeddings: bool True # 是否对向量进行归一化有利于余弦相似度计算 class EmbeddingResponse(BaseModel): embeddings: List[List[float]] model: str app.post(/embed, response_modelEmbeddingResponse) async def get_embeddings(request: EmbeddingRequest): try: # 模型推理 embeddings model.encode(request.texts, normalize_embeddingsrequest.normalize_embeddings, convert_to_numpyTrue) # 转换为Python list格式返回 embeddings_list embeddings.tolist() return EmbeddingResponse(embeddingsembeddings_list, modelBGE-M3) except Exception as e: raise HTTPException(status_code500, detailstr(e)) if __name__ __main__: # 启动服务默认监听 8001 端口 uvicorn.run(app, host0.0.0.0, port8001)启动服务python embedding_service.py服务启动后你可以通过http://localhost:8001/docs查看自动生成的 API 文档并进行测试。4.3 部署向量数据库 (Chroma)Chroma 可以以客户端-服务器模式运行也可以嵌入在应用中。这里我们以独立服务模式运行方便其他组件调用。使用 Docker 启动 Chroma 是最简单的方式docker pull chromadb/chroma docker run -d --name chroma_db -p 8000:8000 chromadb/chroma服务将在http://localhost:8000运行。你也可以使用chromadb客户端库在 Python 代码中直接创建内存或持久化集合。4.4 部署 Rerank 模型服务重排序模型用于对检索到的 Top K 个文档进行精排。我们同样将其封装为 API。创建rerank_service.pyfrom transformers import AutoModelForSequenceClassification, AutoTokenizer import torch import uvicorn from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List app FastAPI(titleBGE Reranker Service) # 加载模型和分词器 model_name BAAI/bge-reranker-large # 也可以使用 base 版本 tokenizer AutoTokenizer.from_pretrained(model_name) model AutoModelForSequenceClassification.from_pretrained(model_name) model.eval() device torch.device(cuda if torch.cuda.is_available() else cpu) model.to(device) class RerankRequest(BaseModel): query: str documents: List[str] class RerankResponse(BaseModel): scores: List[float] ranked_doc_indices: List[int] app.post(/rerank, response_modelRerankResponse) async def rerank_documents(request: RerankRequest): try: pairs [[request.query, doc] for doc in request.documents] with torch.no_grad(): inputs tokenizer(pairs, paddingTrue, truncationTrue, return_tensorspt, max_length512) inputs {k: v.to(device) for k, v in inputs.items()} scores model(**inputs, return_dictTrue).logits.view(-1, ).float() scores scores.cpu().tolist() # 按分数从高到低排序 ranked_indices sorted(range(len(scores)), keylambda i: scores[i], reverseTrue) ranked_scores [scores[i] for i in ranked_indices] return RerankResponse(scoresranked_scores, ranked_doc_indicesranked_indices) except Exception as e: raise HTTPException(status_code500, detailstr(e)) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8002)启动服务python rerank_service.py4.5 连接大模型与 LangGraph这里我们以调用 OpenAI 兼容 API例如本地部署的 Ollama为例展示如何将其与 LangGraph 集成。首先确保你有一个可用的 LLM 服务端点。例如使用 Ollama 在本地运行一个模型# 拉取并运行一个模型例如 Qwen2.5:7B ollama pull qwen2.5:7b ollama run qwen2.5:7b # Ollama 默认 API 服务在 http://localhost:11434然后在 Python 中配置 LangChain 和 LangGraphimport os from langchain_openai import ChatOpenAI from langchain.schema import HumanMessage, SystemMessage from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator # 1. 配置 LLM (指向本地 Ollama) llm ChatOpenAI( base_urlhttp://localhost:11434/v1, api_keyollama, # Ollama 不需要真实的 key modelqwen2.5:7b, temperature0.1 ) # 2. 定义状态结构 class AgentState(TypedDict): question: str retrieved_docs: list reranked_docs: list final_answer: str conversation_history: Annotated[list, operator.add] # 用于存储对话历史实现长期记忆 # 3. 定义各个节点函数 def retrieve(state: AgentState): 模拟检索节点这里调用我们之前部署的向量数据库和 Embedding 服务 # 实际项目中这里应调用向量数据库的检索接口 # 示例返回模拟的检索结果 mock_docs [ 文档ARAG 系统包含检索、增强、生成三个核心步骤。, 文档B向量数据库如 Chroma 用于高效存储和查询 Embedding。, 文档CLangGraph 可以构建有状态的、多步骤的 AI 智能体工作流。 ] return {retrieved_docs: mock_docs} def rerank(state: AgentState): 重排序节点调用 Rerank 服务对检索结果排序 # 实际项目中这里应调用 rerank_service 的 API # 假设我们直接使用原始顺序 return {reranked_docs: state[retrieved_docs]} def generate_answer(state: AgentState): 生成节点基于重排序后的文档调用 LLM 生成最终答案 context \n.join(state[reranked_docs]) system_prompt f你是一个专业的助手。请严格根据以下提供的上下文信息来回答问题。如果上下文信息不足以回答问题请如实告知。 上下文 {context} messages [ SystemMessage(contentsystem_prompt), HumanMessage(contentstate[question]) ] response llm.invoke(messages) return {final_answer: response.content} # 4. 构建图 workflow StateGraph(AgentState) workflow.add_node(retrieve, retrieve) workflow.add_node(rerank, rerank) workflow.add_node(generate, generate_answer) # 5. 定义边执行顺序 workflow.set_entry_point(retrieve) workflow.add_edge(retrieve, rerank) workflow.add_edge(rerank, generate) workflow.add_edge(generate, END) # 6. 编译图 app workflow.compile() # 7. 运行图 initial_state {question: 请解释 RAG 系统的基本流程。, conversation_history: []} result app.invoke(initial_state) print(result[final_answer])这个示例展示了 LangGraph 如何将检索、重排序、生成等步骤编排成一个有向图。你可以在此基础上增加条件判断、循环、人工审核等节点构建更复杂的智能体。5. 功能测试与效果验证现在我们来验证各个组件以及整个流水线是否工作正常。5.1 测试 Embedding 服务使用curl或 Pythonrequests库测试 Embedding API。curl -X POST http://localhost:8001/embed \ -H Content-Type: application/json \ -d { texts: [什么是人工智能, 机器学习是人工智能的一个子领域。], normalize_embeddings: true }预期返回一个 JSON包含两个文本对应的 1024 维BGE-M3 的默认维度向量。检查向量是否为浮点数列表且长度正确。5.2 测试向量数据库的写入与检索编写一个简单的脚本将一段文本转化为向量并存入 Chroma然后进行相似性检索。import chromadb from sentence_transformers import SentenceTransformer import uuid # 初始化客户端和模型 chroma_client chromadb.HttpClient(hostlocalhost, port8000) embedding_model SentenceTransformer(BAAI/bge-m3) # 创建或获取一个集合类似于数据库的表 collection chroma_client.get_or_create_collection(nametest_knowledge) # 准备文档 documents [ LangChain 是一个用于开发大语言模型应用的框架。, 向量数据库专门为高维向量数据的快速检索而设计。, Embedding 模型将文本转化为具有语义信息的数值向量。 ] ids [str(uuid.uuid4()) for _ in documents] # 生成向量 embeddings embedding_model.encode(documents).tolist() # 添加到集合 collection.add( documentsdocuments, embeddingsembeddings, idsids ) print(文档已入库。) # 进行检索 query 有什么工具可以处理文本向量 query_embedding embedding_model.encode([query]).tolist() results collection.query( query_embeddingsquery_embedding, n_results2 ) print(检索结果) for i, doc in enumerate(results[documents][0]): print(f{i1}. {doc})运行脚本应能正确检索到与查询语义相近的文档例如“Embedding 模型...”。5.3 测试 Rerank 服务模拟一个检索返回了多个可能相关文档的场景测试 Rerank 服务是否能找出最相关的那一个。import requests rerank_url http://localhost:8002/rerank query 如何选择合适的向量数据库 candidate_docs [ MySQL 是一种流行的关系型数据库。, # 不相关 Chroma 是一个轻量级的开源向量数据库易于集成。, # 最相关 Python 是一种编程语言。, # 不相关 向量数据库的性能指标包括 QPS 和召回率。 # 相关 ] payload { query: query, documents: candidate_docs } response requests.post(rerank_url, jsonpayload) if response.status_code 200: result response.json() print(重排序分数:, result[scores]) print(重排序后的索引:, result[ranked_doc_indices]) print(最相关的文档:, candidate_docs[result[ranked_doc_indices][0]]) else: print(请求失败:, response.text)预期输出中关于 Chroma 的文档应该获得最高分并被排在第一位。5.4 端到端 RAG 流程测试将以上所有服务串联模拟一个完整的用户问答流程。用户提问“用 LangGraph 构建智能体有什么优势”系统流程将问题转化为向量调用 Embedding 服务。在向量数据库中检索出 Top 5 相关文档片段。使用 Rerank 服务对 5 个片段进行精排选出 Top 3。将问题和 Top 3 文档片段组合成提示词发送给大模型。大模型生成最终答案。验证标准最终答案应包含“状态管理”、“多步骤编排”、“长期记忆”、“循环与条件分支”等 LangGraph 的关键优势点并且这些信息应来源于检索到的文档而非大模型的通用知识。你可以通过编写一个集成脚本来自动化这个测试流程并检查最终答案的准确性和相关性。6. 接口 API 与批量任务一个生产可用的 RAG 系统必须提供稳定的 API 并支持批量处理。6.1 构建统一的 RAG API 网关我们可以创建一个 FastAPI 应用作为整个系统的入口内部调用 Embedding、向量数据库、Rerank 和 LLM 服务。# rag_gateway.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List import requests import logging app FastAPI(titleRAG System Gateway) logging.basicConfig(levellogging.INFO) EMBEDDING_URL http://localhost:8001/embed CHROMA_URL http://localhost:8000 # Chroma 客户端直接连接 RERANK_URL http://localhost:8002/rerank LLM_URL http://localhost:11434/v1/chat/completions # Ollama 示例 class QueryRequest(BaseModel): question: str top_k_retrieve: int 5 top_k_rerank: int 3 class QueryResponse(BaseModel): answer: str used_contexts: List[str] process_time: float app.post(/query, response_modelQueryResponse) async def query_knowledge_base(request: QueryRequest): import time start_time time.time() try: # 1. 将问题转换为向量 embed_response requests.post(EMBEDDING_URL, json{texts: [request.question]}) query_vector embed_response.json()[embeddings][0] # 2. 向量检索 (这里简化实际需调用Chroma HTTP API或客户端) # 假设我们有一个函数 chroma_retrieve retrieved_docs chroma_retrieve(query_vector, request.top_k_retrieve) # 3. 重排序 rerank_payload {query: request.question, documents: retrieved_docs} rerank_response requests.post(RERANK_URL, jsonrerank_payload) ranked_indices rerank_response.json()[ranked_doc_indices] top_contexts [retrieved_docs[i] for i in ranked_indices[:request.top_k_rerank]] # 4. 调用 LLM 生成答案 llm_prompt f基于以下上下文回答问题。如果上下文不包含答案请说“根据现有资料无法回答”。 上下文{ .join(top_contexts)} 问题{request.question} 答案 llm_payload { model: qwen2.5:7b, messages: [{role: user, content: llm_prompt}], stream: False } llm_response requests.post(LLM_URL, jsonllm_payload) answer llm_response.json()[choices][0][message][content] process_time time.time() - start_time return QueryResponse(answeranswer, used_contextstop_contexts, process_timeprocess_time) except Exception as e: logging.error(fRAG 流程错误: {e}) raise HTTPException(status_code500, detailf内部处理错误: {e}) def chroma_retrieve(query_vector, top_k): # 此处应实现真正的 Chroma 检索逻辑 # 示例返回模拟数据 return [ 文档1LangGraph 支持创建有状态的图。, 文档2智能体可以根据状态决定下一步行动。, 文档3RAG 结合了检索和生成。 ] if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8003)6.2 批量文档处理与入库对于大量文档我们需要一个离线批处理流程。# batch_processing.py import os from glob import glob from PyPDF2 import PdfReader from sentence_transformers import SentenceTransformer import chromadb from tqdm import tqdm import hashlib def extract_text_from_pdf(pdf_path): 从PDF提取文本 reader PdfReader(pdf_path) text for page in reader.pages: text page.extract_text() \n return text def split_text(text, chunk_size500, chunk_overlap50): 将长文本分割成小块简单的按字符分割生产环境建议用语义分割 chunks [] start 0 while start len(text): end start chunk_size chunk text[start:end] chunks.append(chunk) start end - chunk_overlap return chunks def process_document_folder(folder_path, collection_nameknowledge_base): 批量处理文件夹内的文档支持PDF、TXT # 初始化模型和数据库客户端 embed_model SentenceTransformer(BAAI/bge-m3) chroma_client chromadb.PersistentClient(path./chroma_db) # 持久化到本地 collection chroma_client.get_or_create_collection(namecollection_name) supported_ext [.pdf, .txt] all_chunks [] all_metadatas [] all_ids [] for ext in supported_ext: for file_path in tqdm(glob(os.path.join(folder_path, f*{ext})), descf处理 {ext} 文件): try: if ext .pdf: text extract_text_from_pdf(file_path) else: # .txt with open(file_path, r, encodingutf-8) as f: text f.read() chunks split_text(text) file_id hashlib.md5(file_path.encode()).hexdigest()[:8] for i, chunk in enumerate(chunks): chunk_id f{file_id}_chunk{i} all_chunks.append(chunk) all_metadatas.append({source: os.path.basename(file_path), chunk_index: i}) all_ids.append(chunk_id) except Exception as e: print(f处理文件 {file_path} 时出错: {e}) # 批量生成向量并入库 if all_chunks: print(f正在为 {len(all_chunks)} 个文本块生成向量...) embeddings embed_model.encode(all_chunks, show_progress_barTrue).tolist() print(正在写入向量数据库...) collection.add( documentsall_chunks, embeddingsembeddings, metadatasall_metadatas, idsall_ids ) print(f批量处理完成成功入库 {len(all_chunks)} 个文本块。) else: print(未找到可处理的文档。) if __name__ __main__: # 指定你的文档文件夹路径 doc_folder ./my_documents process_document_folder(doc_folder)这个脚本会遍历指定文件夹下的 PDF 和 TXT 文件进行文本提取、分块、向量化并存入本地的 Chroma 数据库。7. 资源占用与性能观察部署和运行 RAG 系统时需要密切关注资源消耗以便进行容量规划和优化。Embedding 模型推理CPU 模式运行 BGE-M3 这类模型单次推理约 500 字符在 Intel i7 上可能耗时 100-300 毫秒。批处理可以显著提升吞吐量。GPU 模式在 RTX 3060 (12GB) 上推理延迟可降至 10-50 毫秒。使用nvidia-smi观察显存占用加载 BGE-M3 模型约占用 1.5-2GB 显存。建议对于高频调用务必启用 GPU 并采用批处理 API。向量数据库内存Chroma 在内存中存储索引。100 万个 1024 维向量的集合大约占用 4GB 内存向量本身加上额外的索引开销。磁盘如果使用持久化模式数据会保存在磁盘。定期清理无用集合。查询性能查询速度与集合大小和索引类型有关。对于百万级数据HNSW 索引可以在毫秒级返回结果。Rerank 模型重排序通常是计算密集型的。BGE-Reranker-Large 模型比同规模 Embedding 模型更耗资源。在 CPU 上对 10 个候选文档进行重排序可能需要 1-2 秒在 GPU 上可缩短至 200 毫秒以内。策略不要对大量候选文档如 Top 100进行重排序通常先通过向量检索筛选出 Top 10-20再进行精排。大模型推理这是最大的资源消耗者。本地运行 7B 参数模型如 Qwen2.5-7B至少需要 8-10GB GPU 显存进行推理。如果使用 API 服务如 OpenAI, DeepSeek则关注网络延迟和 Token 消耗成本。优化使用量化模型如 GPTQ, AWQ可将显存需求降低 30-50%。对于问答任务合理设置max_tokens以避免生成过长无用内容。综合链路延迟一个完整的 RAG 查询检索重排序生成延迟可能在 2 秒到 10 秒不等取决于文档库大小、模型部署位置和网络状况。性能观测点在 API 网关层记录每个环节的耗时便于定位瓶颈。常见的瓶颈顺序是LLM生成 Rerank Embedding 向量检索。8. 常见问题与排查方法在搭建和运行过程中你可能会遇到以下问题。这里提供排查思路。问题现象可能原因排查方式解决方案Embedding 服务启动失败1. 端口被占用。2. 模型文件下载失败或损坏。3. CUDA 版本与 PyTorch 不匹配。1. 检查端口8001是否被其他程序占用 (netstat -an | grep 8001)。2. 查看服务启动日志确认模型下载进度和错误。3. 运行python -c import torch; print(torch.cuda.is_available())验证 CUDA。1. 更换服务端口。2. 手动从 Hugging Face 下载模型到本地在代码中指定local_files_onlyTrue。3. 重新安装匹配的 PyTorch 和 CUDA 版本。向量数据库连接超时1. Chroma 服务未启动。2. 防火墙阻止了端口访问。3. 客户端配置的主机/端口错误。1. 检查 Docker 容器或 Chroma 进程是否在运行 (docker ps或ps aux | grep chroma)。2. 尝试在服务器本地用curl http://localhost:8000/api/v1/heartbeat测试。1. 启动 Chroma 服务。2. 配置防火墙规则开放对应端口。3. 确认客户端连接的host和port与服务端一致。检索结果不相关1. Embedding 模型不适合当前领域。2. 文本分块策略不合理块太大或太小。3. 没有使用重排序。1. 用一些领域内关键词测试 Embedding 的相似度。2. 检查分块后的文本是否保持了语义完整性。3. 对比使用重排序前后的结果质量。1. 尝试其他 Embedding 模型如text-embedding-3-small,nomic-embed-text。2. 调整分块大小和重叠长度或采用语义分割库。3. 集成 Rerank 模型。大模型回答未基于检索内容1. 提示词Prompt设计不佳未强制模型参考上下文。2. 检索到的上下文本身质量差或无关。1. 检查发送给 LLM 的完整 Prompt确保包含明确的指令如“请仅根据以下上下文回答”。2. 单独测试检索环节看返回的文档是否真的与问题相关。1. 优化 Prompt 工程增加系统指令的约束力。2. 提升检索质量优化 Embedding、分块、重排序。批量入库速度慢1. 单条处理未使用批处理。2. 在 CPU 上进行 Embedding 计算。3. 数据库写入未优化。1. 观察代码是逐条调用encode还是批量调用。2. 检查模型是否运行在 GPU 上。3. 观察磁盘 I/O 或网络延迟。1. 将文档累积到一定数量如 32 或 64 条后批量调用model.encode。2. 确保环境支持 CUDA 并将模型加载到 GPU。3. 对于大规模入库考虑使用向量数据库的批量导入工具。LangGraph 图编译或运行错误1. 状态State定义与节点函数返回值不匹配。2. 图中存在循环依赖或未定义的边。1. 仔细检查AgentState的TypedDict定义与每个节点返回的字典键名是否一致。2. 使用app.get_graph().draw_mermaid()输出图结构检查。1. 确保每个节点返回的字典是AgentState的子集。2. 遵循set_entry_point和add_edge的顺序确保图是连通且无歧义的。系统整体延迟过高1. 某个组件通常是 LLM成为瓶颈。2. 网络调用过多或序列化开销大。1. 在 API 网关记录每个微服务的调用耗时。2. 检查是否在循环中频繁调用服务而未做缓存或批处理。1. 对 LLM 生成进行流式输出以提升用户体验感知。2. 考虑将 Embedding 和 Rerank 服务与主应用放在同机房减少网络延迟。3. 对不变的查询或文档 Embedding 进行缓存。9. 最佳实践与使用建议基于实战经验以下建议能帮助你构建更稳健、高效的 RAG 系统。从简单开始逐步迭代不要一开始就追求完美的多阶段检索和复杂智能体。先搭建一个最基础的“Embedding 向量检索 LLM”管道并跑通确保数据能流起来。然后再逐步加入重排序、查询改写、元数据过滤等优化。重视数据预处理与分块垃圾进垃圾出。文档的清洗、格式统一、分块策略对最终效果影响巨大。对于技术文档按章节或子标题分块效果更好。可以尝试不同的分块大小如 256, 512, 1024 字符和重叠长度通过检索效果评估最佳配置。实施全面的评估体系不要只靠“感觉”判断系统好坏。定义评估指标如检索相关度检索到的 Top K 文档中有多少是真正相关的答案准确性生成的答案与标准答案或事实是否一致答案相关性答案是否直接回应了问题延迟与吞吐量单个查询响应时间、系统能承受的 QPS。为生产环境做好准备服务化与监控将 Embedding、向量数据库、Rerank、LLM 网关等都部署为独立的、可监控的微服务。配置管理使用环境变量或配置文件管理模型路径、API 密钥、服务地址等避免硬编码。错误处理与重试在网络调用、模型推理等环节加入健壮的错误处理和指数退避重试机制。版本控制对模型版本、数据库 Schema、API 接口进行版本控制便于回滚和更新。安全与合规始终优先访问控制为你的 RAG API 网关添加认证如 API Key、JWT。输入输出过滤对用户输入进行必要的清洗和过滤防止 Prompt 注入。对模型输出进行内容安全审核。数据审计记录用户的查询和系统的回答用于效果分析和问题追溯同时注意隐私保护。探索进阶架构当基础 RAG 稳定后可以探索更高级的模式Agentic RAG让智能体主动决定何时检索、检索什么、如何迭代优化查询。Hybrid Search结合关键词搜索BM25和向量搜索取长补短。Query Rewriting/Expansion对用户原始查询进行改写或扩展提升检索召回率。10. 总结与下一步通过本文的实战演练你应该已经掌握了一个现代化 RAG 系统的核心组件和搭建流程从 BGE-M3 生成 Embedding到用 Chroma 存储和检索向量再到用 Rerank 模型提升精度最后用 LangGraph 编排成一个可扩展的智能体工作流。这套组合拳解决了传统 RAG 在准确性、可控性和复杂性上的诸多痛点。最值得你立刻动手尝试的是将你本地的一份文档比如一个项目 README 或一篇技术文章通过我们提供的批量处理脚本灌入系统然后提出几个问题看看它能否准确回答。这个端到端的验证能让你最快地感受到 RAG 的能力和当前局限。最容易踩的坑通常是环境配置CUDA 版本、端口冲突和数据预处理分块不合理导致检索失效。建议严格按照第 3、4 节的步骤准备环境并花时间优化你的文档分块策略。接下来你可以沿着这几个方向深入性能优化尝试量化 Embedding 和 Rerank 模型测试 GPU 推理与批处理带来的提升。效果优化在 LangGraph 中引入“查询理解”和“答案验证”节点构建一个自我修正的循环。工程化用 Docker Compose 或 Kubernetes 编排所有服务并接入 Prometheus 和 Grafana 进行监控。领域适配针对你的专业领域如法律、医疗代码收集领域数据对 Embedding 模型进行微调LoRA这将带来最显著的效果提升。构建 RAG 系统是一个持续迭代的过程希望这份从零到一的实战指南能成为你可靠的起点。建议收藏本文在部署和优化过程中随时参考。