这次我们直接看一套完整的大模型应用落地链路LangChain、RAG、Agent、Embedding、向量数据库、提示词工程到底怎么从零开始搭起来而不是停留在概念层面。这个问题的核心不在于某一个框架有多强而在于怎么把“文档解析 - 向量化 - 检索 - 生成 - 工具调用 - 智能体决策”这条链路真正跑通。实践里大量项目卡在检索不精确、向量库选型错、Agent 调用不稳定这几个环节而不是模型本身不够好。这篇文章会从技术栈选型开始逐步拆解 RAG 知识库的完整落地流程、Embedding 模型和向量数据库的搭配方式、LangChain 与 LangGraph 的各自定位以及 Agent 工具调用链路的稳定性设计。涉及代码全部给出可以直接测试的 Python 示例硬件部分会说明如何观察资源占用但实际显存数据必须按你自己的运行环境为准。适合正在做知识库问答、自动化 Agent、企业级 RAG 落地的开发者也适合准备系统学习大模型应用开发但不知道从哪条技术路线入手的读者。下面是完整拆解。1. 核心能力速览能力项说明技术栈LangChain / LangGraph、Embedding 模型、向量数据库、Agent 框架主要功能RAG 知识库问答、文档检索、Agent 工具调用、多轮对话、提示词工程核心组件Qwen Embedding 或其他开源 Embedding、Chroma / Milvus / pgvector / Qdrant、LangChain 表达式语言、LangGraph 状态流启动方式Python 脚本启动、Jupyter 分步调试、Docker 部署向量库、服务化 API接口能力LangServe 或 FastAPI 封装检索与问答接口批量任务支持文档批量入库、批量检索测试、批处理问答队列推荐硬件纯 API 方案任意 CPU 可跑本地 Embedding 与重排序建议 4G 以上显存资源占用实际显存以模型版本和推理参数为准文本向量化通常远低于大模型生成适合读者后端开发、算法工程师、AI 应用开发者、大模型技术学习者从表格可以看出来这套链路不是只能跑在高端显卡上的东西。如果全部走云端 APICPU 就能完成整套 RAG 流程如果本地化部署 Embedding 模型普通消费级显卡也能处理。真正消耗资源的通常是生成式大模型本身而这一部分可以按需接 API 或本地推理服务。2. 技术组件全景拆解2.1 LangChain 和 LangGraph 是什么关系LangChain 是一个开发框架核心目标是标准化“调用大模型、拼接提示词、解析输出、管理记忆、连接外部工具”这些操作。它最实用的部分是各种封装好的组件文档加载器、文本分割器、向量存储封装、模型接口统一、输出解析器。适合做 RAG 流程编排和工具调用的初学者。LangGraph 是 LangChain 团队推出的状态化编排框架定位是处理有状态、有分支、需要循环的 Agent 流程。普通 RAG 是一条直线检索 - 组装上下文 - 生成而 Agent 是循环结构规划 - 调用工具 - 观察结果 - 再决策。LangGraph 把这种循环用图节点的方式表达出来可控性比 LangChain 自带的老式 AgentExecutor 要强。如果你只是做知识库问答LangChain 的 LCELLangChain Expression Language就够了如果做多步骤 Agent 或需要人工介入确认关键操作优先用 LangGraph。大多数生产项目会同时依赖两者用 LangChain 处理组件兼容用 LangGraph 控制核心流程。2.2 Embedding 模型怎么选Embedding 的作用是文本向量化它的效果直接决定“检索是否精准”。好的向量空间里语义相近的文本距离更近工程上应该用同一个 Embedding 模型做入库和检索查询端不要混用不同模型。可选的 Embedding 模型大概有这几类类型代表特点通用商用 APIOpenAI text-embedding-3 系列、通义 embedding 接口效果稳定、调用简单、按量付费开源本地部署Qwen Embedding 系列、BGE 系列、GTE 系列数据不出内网、可批量处理、兼容性好评测多云端免费限量部分平台提供免费 embedding 调用适合学习测试不适合生产依赖实操建议先跑通链路用在线 API确认效果后要降成本或保障数据隐私再换本地开源 Embedding。很多团队一上来就追求本地化结果 Embedding 质量不如商用 API导致检索精度整体下跌最后反而耗时最长。Embedding 模型选型最关键的一条是向量空间一致性。入库和检索必须使用同一个模型否则余弦相似度没有意义。2.3 向量数据库选型对比向量数据库负责存储向量并提供相似度检索。选型要看数据量、并发量、部署复杂度和现有技术栈。数据库定位优势适合场景ChromaDB轻量级向量库部署简单、依赖少、适合原型本地开发、小规模知识库Milvus分布式向量库支持百亿级向量、功能完善生产环境、大规模检索pgvectorPostgreSQL 扩展复用关系型数据库、事务能力强已有 PostgreSQL 基础设施Qdrant独立向量数据库过滤条件丰富、Rust 性能好需要复杂过滤条件的搜索原型阶段直接用 ChromaDB一个 pip install 就能跑生产环境如果数据量到百万级向量或需要水平扩展优先考虑 Milvus如果公司本来就重度使用 PostgreSQLpgvector 是最顺手的方案避免多维护一套系统。向量数据库的核心指标不是“支持多少种索引”而是真实数据量下的召回率与延迟。2.4 RAG 和 Agent 的边界RAG检索增强生成解决的问题是“模型不知道私有知识时通过检索给模型补充上下文”。它本身不规划不调用外部工具不做多轮决策。核心链路是文档入库 - 用户提问 - 向量检索 - 拼装提示词 - 模型生成答案Agent 解决的是“任务需要多步骤、需要操作外部系统、需要根据中间结果调整计划”的问题。核心链路是理解目标 - 拆分步骤 - 选择工具 - 调用并接收结果 - 判断是否完成 - 输出或继续现代应用的趋势是 Agentic RAG也就是把 RAG 放进 Agent 的工具列表中。Agent 判断当前问题是否需要检索资料如果需要就调用知识库检索工具拿到结果后再生成回答。这种设计比“每次对话都检索”更智能也能减少无效检索对噪声的引入。3. RAG 本地部署环境准备3.1 操作系统与运行时RAG 链路的核心依赖是 Python。建议使用 Python 3.10 到 3.12 版本。Linux 和 Windows 都能跑生产环境优先 Linux。如果本机 Python 环境比较乱建议先创建独立的虚拟环境避免依赖冲突。Windows 用户注意部分向量数据库或深度学习依赖在 Windows 上的编译链不如 Linux 顺滑遇到错误优先查“Microsoft C Build Tools”是否安装。检查 Python 版本python --version建议创建独立虚拟环境python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate3.2 GPU 与本地推理准备如果计划在本地部署 Embedding 模型或重排序模型需要准备NVIDIA 显卡建议 4G 及以上显存CUDA 环境已配置PyTorch GPU 版本检查 PyTorch 是否能调用 GPUimport torch print(torch.cuda.is_available()) print(torch.cuda.get_device_name(0) if torch.cuda.is_available() else CPU only)如果torch.cuda.is_available()返回 False说明 PyTorch 装成了 CPU 版本或 CUDA 驱动不匹配。重新安装 GPU 版 PyTorch 的方式参考 PyTorch 官网的安装命令生成器输入你的 CUDA 版本即可生成对应命令。需要注意某些国产加速卡环境存在兼容性差异。例如昇腾 910B 系列服务器上如果通过 vLLM 启动 Embedding 和重排序模型失败不一定代表 GPU 算力不足更可能是框架适配问题或模型服务入口不支持该任务类型。遇到类似情况建议先确认加速卡驱动、推理框架版本、以及该框架是否声明支持 Embedding/Reranker 任务的官方说明再决定换框架还是换模型服务方案。3.3 依赖安装基础依赖包括 LangChain、LangChain Community、ChromaDB、OpenAI SDK或对应服务商的 SDK以及文档解析相关库。pip install langchain langchain-community langchain-openai pip install chromadb pip install pypdf如果使用国产模型服务商接口一般兼容 OpenAI SDK 协议只需要修改base_url和api_key即可接入。推荐这种兼容方式因为它把切换成本降到最低后续换供应商几乎不用改代码逻辑。3.4 向量数据库选型安装如果使用 ChromaDB依赖已经通过pip install chromadb安装完毕零配置即可启动。如果使用 Milvus官方推荐 Docker 方式启动docker run -d --name milvus \ -p 19530:19530 \ -p 9091:9091 \ milvusdb/milvus:latestMilvus 启动后通过 19530 端口gRPC提供服务。首次启动需要拉取镜像磁盘占用和时间都要预留充足。如果对 Milvus 版本特性有要求比如支持某种索引类型或特定距离计算方式建议先查官方 release notes再确定镜像 tag。日常学习和中小规模项目不盲目追求多副本、分布式等特性注意规避不必要的运维复杂度。4. RAG 全链路实现从文档到问答4.1 项目结构规划从 0 到 1 落地工程上建议先按目录规划好代码、数据、模型的归属位置避免后面批量处理时“数据散落、配置到处改”的问题。一个可参考的分层结构如下rag_project/ ├── config.py # 全局配置模型名、API地址、数据库路径等 ├── ingest.py # 文档加载、分割、向量化、入库 ├── query.py # 检索问答主流程 ├── agent_demo.py # Agent 工具调用示例 ├── docs/ # 原始文档目录 ├── chroma_db/ # 向量数据库持久化目录 └── requirements.txt # 依赖清单模块分层是 RAG 项目维护成本的分水岭。一次性脚本可以直接打开查询能力但要做批量任务或二次开发把“入库”和“检索”拆分清楚会省很多时间。4.2 配置中心好的配置能让项目在不同模型服务商之间快速切换。把这些内容集中放到 config 中API Base URL、API Key、Embedding 模型名、向量库路径、检索 TopK、文本块大小等。# config.py import os # 建议使用环境变量管理密钥不要写死在代码里提交到仓库 API_KEY os.getenv(LLM_API_KEY, your-api-key) BASE_URL os.getenv(LLM_BASE_URL, https://api.example.com/v1) EMBEDDING_MODEL os.getenv(EMBEDDING_MODEL, your-embedding-model-name) LLM_MODEL os.getenv(LLM_MODEL, your-llm-model-name) # 向量数据库配置 CHROMA_PERSIST_DIR ./chroma_db COLLECTION_NAME knowledge_base # 检索参数 TOP_K 4 CHUNK_SIZE 500 CHUNK_OVERLAP 50这种配置方式在本地跑通之后迁移到 Docker 或 Kubernetes 时只需要通过环境变量覆盖配置代码不需要改。4.3 文档加载与解析RAG 精确度的第一道关卡很多 RAG 项目检索效果差问题往往不是目标模型不够强而是文档加载与分块阶段做得太粗糙。比如 PDF 表格、多栏排版、图像型扫描件这些场景如果直接整篇加载后按固定字符切分切出来的内容往往是断裂的表格、错行的文本和无效的页码页眉。文档加载阶段首先判断文件类型如果是纯文本 PDF用PyPDFLoader或PyMuPDFLoader直接抽取文字如果是扫描件或图片型 PDF需要 OCR光学字符识别能力先把文字识别出来再做文本切分如果是 HTML、Markdown 或 Word直接用不同的加载器分别处理。LangChain 的文档加载器统一封装了这些差异核心是拿到干净的纯文本。文档清洗这一步很重要要移除无意义字符、压缩多余换行、处理乱码。文本分块时优先考虑语义边界而不建议只用固定字符数硬切。最常见的效果提升方案是先按标题层级做结构性切块再对过长块做二次切分。# ingest.py 片段文档加载与分块 from langchain_community.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter loader PyPDFLoader(docs/产品手册.pdf) documents loader.load() text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , , ] ) chunks text_splitter.split_documents(documents) print(f文档共切分为 {len(chunks)} 个文本块)固定长度切块的痛点是切断句子语义你需要在配置里设置合适的重叠长度来缓解。而 RecursiveCharacterTextSplitter 比单字符切分更有优势的原因是它优先按完整段落、句子边界切分只有在内容过长时才继续向下拆。4.4 Embedding 向量化入库文本块准备好后用 Embedding 模型把每段文本转成向量然后写入向量数据库。# ingest.py 片段向量化入库 from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma import config # 如果服务商兼容 OpenAI 接口协议直接使用 OpenAIEmbeddings 并传入 base_url embeddings OpenAIEmbeddings( modelconfig.EMBEDDING_MODEL, api_keyconfig.API_KEY, base_urlconfig.BASE_URL ) vector_store Chroma.from_documents( documentschunks, embeddingembeddings, persist_directoryconfig.CHROMA_PERSIST_DIR, collection_nameconfig.COLLECTION_NAME ) print(向量化入库完成)这里有一个容易踩的坑不要为每个新文档重新执行Chroma.from_documents应该复用已有的向量库实例否则库会被整体覆盖或者每次都要重写全部内容。需要增量入库时先加载已有库再添加新的向量。# 增量入库示例 vector_store Chroma( embedding_functionembeddings, persist_directoryconfig.CHROMA_PERSIST_DIR, collection_nameconfig.COLLECTION_NAME ) vector_store.add_documents(new_chunks)4.5 检索问答主流程入库完成后查询阶段的核心步骤是生成查询向量 - 相似度检索 - 拼接上下文 - 构建提示词 - 调用大模型生成。# query.py 片段RAG 问答 from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough import config embeddings OpenAIEmbeddings( modelconfig.EMBEDDING_MODEL, api_keyconfig.API_KEY, base_urlconfig.BASE_URL ) vector_store Chroma( embedding_functionembeddings, persist_directoryconfig.CHROMA_PERSIST_DIR, collection_nameconfig.COLLECTION_NAME ) retriever vector_store.as_retriever(search_kwargs{k: config.TOP_K}) prompt ChatPromptTemplate.from_template( 你是一个严谨的 AI 助手。请基于以下资料回答问题。 资料 {context} 问题{question} 要求如果资料中没有明确答案请如实说明“资料中未找到相关信息”。不要编造内容。 ) llm ChatOpenAI( modelconfig.LLM_MODEL, api_keyconfig.API_KEY, base_urlconfig.BASE_URL ) def format_docs(docs): return \n\n.join([f[来源{d.metadata.get(source, 未知)}] {d.page_content} for d in docs]) # LCEL 链查询 - 检索 - 格式化上下文 - 提示词 - LLM - 字符串输出 rag_chain ( {context: retriever | format_docs, question: RunnablePassthrough()} | prompt | llm | StrOutputParser() ) result rag_chain.invoke(这个产品的退货政策是什么) print(result)这段代码是完整的 RAG 最小可运行链路跑通后你只需要替换文档、模型名和提示词就可以拓展出各种垂直场景。第一次建议在 Jupyter 里分步执行方便随时看中间结果检索出来的文档是什么、拼装后的提示词对不对、模型输出是否引用正确。5. Agent 工具调用与 LangGraph 流程编排5.1 从 RAG 到 Agent工具调用怎么设计RAG 解决“从知识库找答案”Agent 解决“调用工具完成任务”。两者的核心差异在于RAG 是一步到位的检索生成Agent 是多轮决策循环。想让大模型能够调用工具关键是给模型提供结构化函数定义并解析模型返回的工具调用参数。下面的示例演示了一个“既能回答知识库问题又能执行简单工具操作”的 Agent 雏形。工具定义使用tool装饰器LangChain 会自动提取函数名、描述和参数结构。# agent_demo.py 片段工具定义与调用 from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor import config tool def get_server_status(server_name: str) - str: 查询服务器状态模拟工具。 status_map {web-01: running, db-01: stopped, cache-01: degraded} return status_map.get(server_name, unknown server) tool def calculate_disk_usage(path: str) - str: 计算指定路径的磁盘占用百分比模拟工具。 return f{path} disk usage: 72% tools [get_server_status, calculate_disk_usage] llm ChatOpenAI( modelconfig.LLM_MODEL, api_keyconfig.API_KEY, base_urlconfig.BASE_URL ) prompt ChatPromptTemplate.from_messages([ (system, 你是一个可调用工具的 AI 助手。), (human, {input}), (placeholder, {agent_scratchpad}) ]) agent create_tool_calling_agent(llm, tools, prompt) executor AgentExecutor(agentagent, toolstools, verboseTrue) result executor.invoke({input: 帮我查一下 db-01 的状态顺便看下 /data 的磁盘占用}) print(result[output])这段代码的关键是tool装饰器。模型会根据函数名和 docstring 决定什么时候调用、传入什么参数LangChain 负责把模型的调用意图转成 Python 函数调用再把结果回传给模型继续决策。这种循环会一直持续到模型认为任务完成。5.2 为什么需要 LangGraphcreate_tool_calling_agent加AgentExecutor的方式适合快速验证但生产级 Agent 有两个问题一是循环控制和中间状态不够透明二是出错重试、人工审批、条件分支难以插入。LangGraph 解决这两点它把 Agent 流程建模为“节点 边”的状态图。# langgraph_demo.py 片段LangGraph 最小 Agent 循环 from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage class AgentState(TypedDict): messages: list next_step: str def call_llm(state: AgentState): # 实际项目里这里调用大模型解析是否要调用工具 last_message state[messages][-1].content if 查 in last_message: return {next_step: tool_execute, messages: [(assistant, 需要调用工具查询)]} return {next_step: finish, messages: [(assistant, 无需调用工具直接回答)]} def tool_execute(state: AgentState): return {next_step: finish, messages: [(assistant, 工具执行结果状态正常)]} graph StateGraph(AgentState) graph.add_node(call_llm, call_llm) graph.add_node(tool_execute, tool_execute) graph.add_edge(call_llm, tool_execute) graph.add_edge(tool_execute, END) graph.add_edge(call_llm, END) app graph.compile()LangGraph 的核心优势是可以显式定义“什么时候调用工具、什么时候结束、什么时候回到大模型”每个节点都能独立测试和日志观察。真实项目中经常用 LangGraph 实现“先检索 - 不满意则改写查询 - 二次检索 - 最终生成”这种带分支的 RAG 流程能做到的精细控制是普通线式链难以做到的。5.3 Agent 稳定性设计Agent 工具调用的不稳定因素主要有三类模型没听懂任务、工具参数传错、工具结果太复杂导致模型分析混乱。工程上应对策略如下工具描述要具体让模型知道“什么时候该用这个工具不该用哪个工具”工具入参要做类型校验和默认值兜底模型传错参数时返回清晰的错误信息而不是让进程崩溃工具结果要做摘要或裁剪超长 JSON 结果先抽关键字段再回传给模型避免上下文被无关字段撑爆添加最大迭代次数限制避免 Agent 陷入循环调用设置超时时间工具调用没有响应时暂停任务并给出明确提示很多项目遇到过 Agent 执行器在等待第三方服务响应时长时间挂起的问题建议统一走超时和重试策略。6. 检索增强效果评估与指标RAG 项目上线前最容易被忽视的是指标评估。很多人凭感觉看“回答得对不对”但检索质量、上下文相关性和生成忠实度完全可以量化。常见指标如下指标度量对象含义RecallK检索器前 K 个结果中有多少命中了真正相关的文档MRR检索器第一个相关结果出现在第几位Context Relevance检索器 生成器检索出的上下文与问题是否相关Faithfulness生成器生成的答案是否忠于检索到的上下文Answer Relevance生成器生成的答案是否针对问题而不是答非所问工程上简易评估方式准备 30 到 50 组“问题 - 标准答案或相关文档 ID”测试集批量跑通检索和问答记录正确率。这个动作在项目初期就可以做不用追求完整框架先把基线建立起来再优化分块方式、Embedding 模型和提示词。批量评估的伪代码如下# eval_demo.py 片段批量测试检索质量 test_questions [ 公司年假制度是怎样的, 报销流程需要哪些材料, 服务器故障如何处理 ] results [] for q in test_questions: docs retriever.invoke(q) top_source docs[0].metadata.get(source, unknown) results.append({question: q, top_source: top_source, score: docs[0].metadata.get(score, 0)}) for r in results: print(r)把检索的命中文档和人工标注的正确文档做对比就能算出早期正确率。这种测试会直接暴露分块过碎导致的信息不全、Embedding 模型导致的语义偏差、文档解析阶段表格错位等问题。7. 批量任务与 API 服务化7.1 批量文档入库实际项目中知识库文档往往成百上千不可能一个一个手动入库。批量入库的工程框架包括遍历文件目录、按扩展名选择加载器、入库程序记录每条文档的处理状态、失败重试。# batch_ingest.py 片段批量入库 from pathlib import Path from langchain_community.document_loaders import PyPDFLoader, TextLoader, UnstructuredMarkdownLoader SUPPORTED_EXT { .pdf: PyPDFLoader, .txt: TextLoader, .md: UnstructuredMarkdownLoader, } def load_document(file_path: Path): ext file_path.suffix.lower() if ext not in SUPPORTED_EXT: return [] loader_cls SUPPORTED_EXT[ext] loader loader_cls(str(file_path)) return loader.load() data_dir Path(docs) all_chunks [] for file_path in sorted(data_dir.iterdir()): try: docs load_document(file_path) chunks text_splitter.split_documents(docs) all_chunks.extend(chunks) print(f已处理: {file_path.name}, 分块数: {len(chunks)}) except Exception as e: print(f处理失败: {file_path.name}, 错误: {e}) vector_store.add_documents(all_chunks) print(f批量入库完成共入库 {len(all_chunks)} 个文本块)批量入库建议加入断点续传能力。简单方案是每处理一个文档后打印日志失败文档单独记录到 error.log全部跑完后统一排查重试。7.2 检索问答 API 封装把 RAG 检索问答封装成接口核心目的是把内部实现与外部调用解耦。外部系统只需要传问题接收答案和来源引用即可。下面给出基于 FastAPI 的最小示例按需扩展鉴权、限流、日志等能力。# api_service.py 片段FastAPI 封装 RAG 问答 from fastapi import FastAPI from pydantic import BaseModel app FastAPI(titleRAG API Service) class QueryRequest(BaseModel): question: str top_k: int 4 class QueryResponse(BaseModel): answer: str sources: list[str] app.post(/query, response_modelQueryResponse) def query_rag(req: QueryRequest): docs retriever.invoke(req.question, search_kwargs{k: req.top_k}) context format_docs(docs) answer rag_chain.invoke(req.question) sources list({d.metadata.get(source, unknown) for d in docs}) return QueryResponse(answeranswer, sourcessources) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)启动服务uvicorn api_service:app --host 0.0.0.0 --port 8000服务启动后可以用 curl 测试curl -X POST http://127.0.0.1:8000/query \ -H Content-Type: application/json \ -d {question: 产品保修期是多久, top_k: 3}需要注意FastAPI 服务启动在0.0.0.0会监听所有网卡接口生产环境必须加鉴权或只监听内网地址避免未授权调用。安全边界不能省至少要加一层 API Key 校验或网关代理。7.3 批量问答任务设计批量问答任务适合离线评测、大规模内容审核、批量报告生成。推荐队列结构# batch_query.py 片段批量问答逻辑 questions [ {id: 1, question: A 产品的价格是多少}, {id: 2, question: B 产品的保修政策}, ] results [] for item in questions: try: answer rag_chain.invoke(item[question]) results.append({id: item[id], question: item[question], answer: answer}) except Exception as e: results.append({id: item[id], question: item[question], answer: fERROR: {e}}) # 批量任务建议加间隔避免触发服务端限流 time.sleep(0.5)批量任务还要考虑失败重试和中间结果落盘避免跑到一半进程崩溃后全部重来。可以把每一条结果都实时写入 JSONL 文件下次启动时跳过已处理的问题。8. 资源占用与性能观察8.1 显存与内存观察方法无论使用本地模型还是纯 API都应该掌握资源占用观察方法。最简单的方式是使用nvidia-smi实时查看 GPU 显存watch -n 1 nvidia-smi如果是纯 API 模式的 RAG 应用本机基本不消耗模型推理显存主要开销在文档解析、向量化入库和检索计算。如果本地跑 Embedding 模型显存占用通常在 1G 到 4G 范围具体取决于模型参数量和量化方式需要实测。跑本地大模型生成时显存占用是最高的这也是为什么 RAG 系统通常放在服务端而客户端只做轻量展示。8.2 性能瓶颈分析RAG 链路中的瓶颈通常出现在三个位置第一文档解析。PDF 转纯文本时如果是扫描件需要 OCR速度会很慢建议提前转换成可复制文本的 PDF。第二Embedding 向量化。用 CPU 跑大批量入库会比较耗时有 GPU 可以大幅改善没有 GPU 就想办法压缩文档量和分块数量。第三检索阶段。数据量变大后暴力检索会变慢需要为向量库创建合适的索引不同数据库对应的索引类型和命令行参数各有差异。8.3 降低资源占用的方法使用量化后的开源 Embedding 模型显存占用明显下降控制向量库集合大小按业务域分库而不是所有数据塞进一个集合生成模型优先考虑 API 而不是本地部署把资源消耗集中到检索服务上对入库文档做去重避免重复文本块重复计算向量批量任务尽量串行加小批量并发避免同时触发多路生成导致内存峰值过高。9. 常见问题与排查方法问题现象可能原因排查方式解决方案ChromaDB 已存在但检索不到新文档重复调用 from_documents 覆盖了集合检查集合中向量数量改用 Chroma() 加载已有库后 add_documents中文检索效果差分块把句子切碎或 Embedding 模型不适合中文打印检索出的原始文本片段检查语义完整性调整分隔符或切换为中文优化的 Embedding 模型模型回答在编造内容提示词未限制或检索内容不相关查看最终发给模型的上下文中是否有正确答案强化提示词约束提高 TopK或加 Reranker批量入库时进程中断未做文档级容错查看日志定位具体失败文件逐文件 try/except失败记录到日志后继续API 调用报超时模型服务商接口响应慢或未设置合理超时先 curl 接口测试延迟客户端增加 timeout 值服务端做异步任务或重试Agent 反复调用同一个工具模型没有拿到完整的工具结果开启 verbose 查看工具返回内容对工具返回做摘要限制最大迭代次数向量化入库很慢CPU 推理且文档量大用 nvidia-smi 查看 GPU 是否生效切 GPU 或缩小任务批次检索出的文档来源混乱不同批次使用不同 Embedding 模型检查向量库配置与当前模型是否一致重建向量库并统一 Embedding 模型LangGraph 编译报错节点函数签名或状态定义不匹配查看 traceback 中节点名称检查状态字段类型和节点返回值结构昇腾等特定硬件上 vLLM 启动 Embedding/Reranker 失败框架对 Embedding 任务的支持不完整查看 vLLM 官方说明与加速卡适配文档改用专门的 Embedding 服务或推理框架Agent 执行等待第三方工具响应时长时间挂起工具调用无超时控制观察日志停留在工具调用节点增加超时与失败自动重试机制提示词生效不明显提示词过长或指令被上下文淹没简化提示词关键指令前置按检索模块调整提示词结构与顺序10. 最佳实践与合规提醒10.1 工程化建议RAG 和 Agent 项目要进入生产不只是把链路跑通那么简单。建议从几个维度做工程加固。目录管理要清晰原始文档、处理后的中间文本、向量库、日志、批量结果分别放到不同目录方便问题排查和数据回溯。版本管理要覆盖代码和配置模型名、API 版本、向量库版本、提示词版本都建议纳入版本管理尤其提示词改动直接影响输出效果要有 diff 记录。可以把提示词模板单独拆成 YAML 或 JSON 文件而不是硬编码在业务代码里。提示词上线前做基线评估每改一次提示词跑一遍已有的测试问题集对比答案质量和格式稳定性防止“改好了 A 类问题却破坏了 B 类问题”。服务化接口必须做访问控制FastAPI 服务监听内网地址或在网关层加认证避免知识库内容和模型调用被未授权使用。如果服务暴露到公网至少要求 API Key 和请求频率限制。Agent 涉及外部操作时必须加人工审批节点。特别是能够发消息、改配置、执行命令的工具要把“执行”改成“生成待执行指令人工确认后执行”。Agent 的自主性需要控制在一定范围内不能把关键操作完全交给模型自动决策。10.2 数据合规与授权提醒RAG 知识库可能涉及企业敏感数据和个人信息。入库前确认文档的使用授权范围对敏感信息做脱敏处理。涉及他人肖像、声音、版权内容时必须取得合法授权仅供测试环境验证不直接用于公开或商业场景。涉及企业私有数据时要遵循公司的数据安全规范明确数据存储位置和访问权限。11. 总结与下一步RAG 到 Agent 的完整链路可以拆成四个阶段文档处理与分块、Embedding 与向量检索、提示词拼装与大模型生成、Agent 工具调用与流程编排。绝大多数项目的问题在第一个阶段和第二个阶段而不是模型不够聪明。新手最先要验证的是一条完整链路能否跑通一份已知答案的文档能否被正确检索并回答。做到这一步RAG 的核心能力就已经掌握了。最容易踩的坑有两个一是入库时混用 Embedding 模型导致后续检索全部无效二是把 Agent 的工具调用想得太简单没有超人力和容错机制结果模型反复调用失败或陷入循环。前者靠统一模型解决后者靠 LangGraph 的状态控制和超时重试解决。接下来可以扩展的方向包括领域知识库优化企业内网数据接入、复杂 Agent 工作流多工具协作 人工审批、RAG 效果评测体系检索指标 生成指标 回归测试、以及用 Docker 把 RAG 服务完整容器化部署到服务器。每一步都有清晰的验证标准建议按顺序推进而不是一把抓。建议先收藏这篇文章把示例代码跑通一遍然后用自己的文档替换测试数据很快就能建立一套自己的 RAG 和 Agent 原型。 LangChain+RAG+Agent实战：从向量化到知识库问答全链路搭建
这次我们直接看一套完整的大模型应用落地链路LangChain、RAG、Agent、Embedding、向量数据库、提示词工程到底怎么从零开始搭起来而不是停留在概念层面。这个问题的核心不在于某一个框架有多强而在于怎么把“文档解析 - 向量化 - 检索 - 生成 - 工具调用 - 智能体决策”这条链路真正跑通。实践里大量项目卡在检索不精确、向量库选型错、Agent 调用不稳定这几个环节而不是模型本身不够好。这篇文章会从技术栈选型开始逐步拆解 RAG 知识库的完整落地流程、Embedding 模型和向量数据库的搭配方式、LangChain 与 LangGraph 的各自定位以及 Agent 工具调用链路的稳定性设计。涉及代码全部给出可以直接测试的 Python 示例硬件部分会说明如何观察资源占用但实际显存数据必须按你自己的运行环境为准。适合正在做知识库问答、自动化 Agent、企业级 RAG 落地的开发者也适合准备系统学习大模型应用开发但不知道从哪条技术路线入手的读者。下面是完整拆解。1. 核心能力速览能力项说明技术栈LangChain / LangGraph、Embedding 模型、向量数据库、Agent 框架主要功能RAG 知识库问答、文档检索、Agent 工具调用、多轮对话、提示词工程核心组件Qwen Embedding 或其他开源 Embedding、Chroma / Milvus / pgvector / Qdrant、LangChain 表达式语言、LangGraph 状态流启动方式Python 脚本启动、Jupyter 分步调试、Docker 部署向量库、服务化 API接口能力LangServe 或 FastAPI 封装检索与问答接口批量任务支持文档批量入库、批量检索测试、批处理问答队列推荐硬件纯 API 方案任意 CPU 可跑本地 Embedding 与重排序建议 4G 以上显存资源占用实际显存以模型版本和推理参数为准文本向量化通常远低于大模型生成适合读者后端开发、算法工程师、AI 应用开发者、大模型技术学习者从表格可以看出来这套链路不是只能跑在高端显卡上的东西。如果全部走云端 APICPU 就能完成整套 RAG 流程如果本地化部署 Embedding 模型普通消费级显卡也能处理。真正消耗资源的通常是生成式大模型本身而这一部分可以按需接 API 或本地推理服务。2. 技术组件全景拆解2.1 LangChain 和 LangGraph 是什么关系LangChain 是一个开发框架核心目标是标准化“调用大模型、拼接提示词、解析输出、管理记忆、连接外部工具”这些操作。它最实用的部分是各种封装好的组件文档加载器、文本分割器、向量存储封装、模型接口统一、输出解析器。适合做 RAG 流程编排和工具调用的初学者。LangGraph 是 LangChain 团队推出的状态化编排框架定位是处理有状态、有分支、需要循环的 Agent 流程。普通 RAG 是一条直线检索 - 组装上下文 - 生成而 Agent 是循环结构规划 - 调用工具 - 观察结果 - 再决策。LangGraph 把这种循环用图节点的方式表达出来可控性比 LangChain 自带的老式 AgentExecutor 要强。如果你只是做知识库问答LangChain 的 LCELLangChain Expression Language就够了如果做多步骤 Agent 或需要人工介入确认关键操作优先用 LangGraph。大多数生产项目会同时依赖两者用 LangChain 处理组件兼容用 LangGraph 控制核心流程。2.2 Embedding 模型怎么选Embedding 的作用是文本向量化它的效果直接决定“检索是否精准”。好的向量空间里语义相近的文本距离更近工程上应该用同一个 Embedding 模型做入库和检索查询端不要混用不同模型。可选的 Embedding 模型大概有这几类类型代表特点通用商用 APIOpenAI text-embedding-3 系列、通义 embedding 接口效果稳定、调用简单、按量付费开源本地部署Qwen Embedding 系列、BGE 系列、GTE 系列数据不出内网、可批量处理、兼容性好评测多云端免费限量部分平台提供免费 embedding 调用适合学习测试不适合生产依赖实操建议先跑通链路用在线 API确认效果后要降成本或保障数据隐私再换本地开源 Embedding。很多团队一上来就追求本地化结果 Embedding 质量不如商用 API导致检索精度整体下跌最后反而耗时最长。Embedding 模型选型最关键的一条是向量空间一致性。入库和检索必须使用同一个模型否则余弦相似度没有意义。2.3 向量数据库选型对比向量数据库负责存储向量并提供相似度检索。选型要看数据量、并发量、部署复杂度和现有技术栈。数据库定位优势适合场景ChromaDB轻量级向量库部署简单、依赖少、适合原型本地开发、小规模知识库Milvus分布式向量库支持百亿级向量、功能完善生产环境、大规模检索pgvectorPostgreSQL 扩展复用关系型数据库、事务能力强已有 PostgreSQL 基础设施Qdrant独立向量数据库过滤条件丰富、Rust 性能好需要复杂过滤条件的搜索原型阶段直接用 ChromaDB一个 pip install 就能跑生产环境如果数据量到百万级向量或需要水平扩展优先考虑 Milvus如果公司本来就重度使用 PostgreSQLpgvector 是最顺手的方案避免多维护一套系统。向量数据库的核心指标不是“支持多少种索引”而是真实数据量下的召回率与延迟。2.4 RAG 和 Agent 的边界RAG检索增强生成解决的问题是“模型不知道私有知识时通过检索给模型补充上下文”。它本身不规划不调用外部工具不做多轮决策。核心链路是文档入库 - 用户提问 - 向量检索 - 拼装提示词 - 模型生成答案Agent 解决的是“任务需要多步骤、需要操作外部系统、需要根据中间结果调整计划”的问题。核心链路是理解目标 - 拆分步骤 - 选择工具 - 调用并接收结果 - 判断是否完成 - 输出或继续现代应用的趋势是 Agentic RAG也就是把 RAG 放进 Agent 的工具列表中。Agent 判断当前问题是否需要检索资料如果需要就调用知识库检索工具拿到结果后再生成回答。这种设计比“每次对话都检索”更智能也能减少无效检索对噪声的引入。3. RAG 本地部署环境准备3.1 操作系统与运行时RAG 链路的核心依赖是 Python。建议使用 Python 3.10 到 3.12 版本。Linux 和 Windows 都能跑生产环境优先 Linux。如果本机 Python 环境比较乱建议先创建独立的虚拟环境避免依赖冲突。Windows 用户注意部分向量数据库或深度学习依赖在 Windows 上的编译链不如 Linux 顺滑遇到错误优先查“Microsoft C Build Tools”是否安装。检查 Python 版本python --version建议创建独立虚拟环境python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate3.2 GPU 与本地推理准备如果计划在本地部署 Embedding 模型或重排序模型需要准备NVIDIA 显卡建议 4G 及以上显存CUDA 环境已配置PyTorch GPU 版本检查 PyTorch 是否能调用 GPUimport torch print(torch.cuda.is_available()) print(torch.cuda.get_device_name(0) if torch.cuda.is_available() else CPU only)如果torch.cuda.is_available()返回 False说明 PyTorch 装成了 CPU 版本或 CUDA 驱动不匹配。重新安装 GPU 版 PyTorch 的方式参考 PyTorch 官网的安装命令生成器输入你的 CUDA 版本即可生成对应命令。需要注意某些国产加速卡环境存在兼容性差异。例如昇腾 910B 系列服务器上如果通过 vLLM 启动 Embedding 和重排序模型失败不一定代表 GPU 算力不足更可能是框架适配问题或模型服务入口不支持该任务类型。遇到类似情况建议先确认加速卡驱动、推理框架版本、以及该框架是否声明支持 Embedding/Reranker 任务的官方说明再决定换框架还是换模型服务方案。3.3 依赖安装基础依赖包括 LangChain、LangChain Community、ChromaDB、OpenAI SDK或对应服务商的 SDK以及文档解析相关库。pip install langchain langchain-community langchain-openai pip install chromadb pip install pypdf如果使用国产模型服务商接口一般兼容 OpenAI SDK 协议只需要修改base_url和api_key即可接入。推荐这种兼容方式因为它把切换成本降到最低后续换供应商几乎不用改代码逻辑。3.4 向量数据库选型安装如果使用 ChromaDB依赖已经通过pip install chromadb安装完毕零配置即可启动。如果使用 Milvus官方推荐 Docker 方式启动docker run -d --name milvus \ -p 19530:19530 \ -p 9091:9091 \ milvusdb/milvus:latestMilvus 启动后通过 19530 端口gRPC提供服务。首次启动需要拉取镜像磁盘占用和时间都要预留充足。如果对 Milvus 版本特性有要求比如支持某种索引类型或特定距离计算方式建议先查官方 release notes再确定镜像 tag。日常学习和中小规模项目不盲目追求多副本、分布式等特性注意规避不必要的运维复杂度。4. RAG 全链路实现从文档到问答4.1 项目结构规划从 0 到 1 落地工程上建议先按目录规划好代码、数据、模型的归属位置避免后面批量处理时“数据散落、配置到处改”的问题。一个可参考的分层结构如下rag_project/ ├── config.py # 全局配置模型名、API地址、数据库路径等 ├── ingest.py # 文档加载、分割、向量化、入库 ├── query.py # 检索问答主流程 ├── agent_demo.py # Agent 工具调用示例 ├── docs/ # 原始文档目录 ├── chroma_db/ # 向量数据库持久化目录 └── requirements.txt # 依赖清单模块分层是 RAG 项目维护成本的分水岭。一次性脚本可以直接打开查询能力但要做批量任务或二次开发把“入库”和“检索”拆分清楚会省很多时间。4.2 配置中心好的配置能让项目在不同模型服务商之间快速切换。把这些内容集中放到 config 中API Base URL、API Key、Embedding 模型名、向量库路径、检索 TopK、文本块大小等。# config.py import os # 建议使用环境变量管理密钥不要写死在代码里提交到仓库 API_KEY os.getenv(LLM_API_KEY, your-api-key) BASE_URL os.getenv(LLM_BASE_URL, https://api.example.com/v1) EMBEDDING_MODEL os.getenv(EMBEDDING_MODEL, your-embedding-model-name) LLM_MODEL os.getenv(LLM_MODEL, your-llm-model-name) # 向量数据库配置 CHROMA_PERSIST_DIR ./chroma_db COLLECTION_NAME knowledge_base # 检索参数 TOP_K 4 CHUNK_SIZE 500 CHUNK_OVERLAP 50这种配置方式在本地跑通之后迁移到 Docker 或 Kubernetes 时只需要通过环境变量覆盖配置代码不需要改。4.3 文档加载与解析RAG 精确度的第一道关卡很多 RAG 项目检索效果差问题往往不是目标模型不够强而是文档加载与分块阶段做得太粗糙。比如 PDF 表格、多栏排版、图像型扫描件这些场景如果直接整篇加载后按固定字符切分切出来的内容往往是断裂的表格、错行的文本和无效的页码页眉。文档加载阶段首先判断文件类型如果是纯文本 PDF用PyPDFLoader或PyMuPDFLoader直接抽取文字如果是扫描件或图片型 PDF需要 OCR光学字符识别能力先把文字识别出来再做文本切分如果是 HTML、Markdown 或 Word直接用不同的加载器分别处理。LangChain 的文档加载器统一封装了这些差异核心是拿到干净的纯文本。文档清洗这一步很重要要移除无意义字符、压缩多余换行、处理乱码。文本分块时优先考虑语义边界而不建议只用固定字符数硬切。最常见的效果提升方案是先按标题层级做结构性切块再对过长块做二次切分。# ingest.py 片段文档加载与分块 from langchain_community.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter loader PyPDFLoader(docs/产品手册.pdf) documents loader.load() text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , , ] ) chunks text_splitter.split_documents(documents) print(f文档共切分为 {len(chunks)} 个文本块)固定长度切块的痛点是切断句子语义你需要在配置里设置合适的重叠长度来缓解。而 RecursiveCharacterTextSplitter 比单字符切分更有优势的原因是它优先按完整段落、句子边界切分只有在内容过长时才继续向下拆。4.4 Embedding 向量化入库文本块准备好后用 Embedding 模型把每段文本转成向量然后写入向量数据库。# ingest.py 片段向量化入库 from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma import config # 如果服务商兼容 OpenAI 接口协议直接使用 OpenAIEmbeddings 并传入 base_url embeddings OpenAIEmbeddings( modelconfig.EMBEDDING_MODEL, api_keyconfig.API_KEY, base_urlconfig.BASE_URL ) vector_store Chroma.from_documents( documentschunks, embeddingembeddings, persist_directoryconfig.CHROMA_PERSIST_DIR, collection_nameconfig.COLLECTION_NAME ) print(向量化入库完成)这里有一个容易踩的坑不要为每个新文档重新执行Chroma.from_documents应该复用已有的向量库实例否则库会被整体覆盖或者每次都要重写全部内容。需要增量入库时先加载已有库再添加新的向量。# 增量入库示例 vector_store Chroma( embedding_functionembeddings, persist_directoryconfig.CHROMA_PERSIST_DIR, collection_nameconfig.COLLECTION_NAME ) vector_store.add_documents(new_chunks)4.5 检索问答主流程入库完成后查询阶段的核心步骤是生成查询向量 - 相似度检索 - 拼接上下文 - 构建提示词 - 调用大模型生成。# query.py 片段RAG 问答 from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough import config embeddings OpenAIEmbeddings( modelconfig.EMBEDDING_MODEL, api_keyconfig.API_KEY, base_urlconfig.BASE_URL ) vector_store Chroma( embedding_functionembeddings, persist_directoryconfig.CHROMA_PERSIST_DIR, collection_nameconfig.COLLECTION_NAME ) retriever vector_store.as_retriever(search_kwargs{k: config.TOP_K}) prompt ChatPromptTemplate.from_template( 你是一个严谨的 AI 助手。请基于以下资料回答问题。 资料 {context} 问题{question} 要求如果资料中没有明确答案请如实说明“资料中未找到相关信息”。不要编造内容。 ) llm ChatOpenAI( modelconfig.LLM_MODEL, api_keyconfig.API_KEY, base_urlconfig.BASE_URL ) def format_docs(docs): return \n\n.join([f[来源{d.metadata.get(source, 未知)}] {d.page_content} for d in docs]) # LCEL 链查询 - 检索 - 格式化上下文 - 提示词 - LLM - 字符串输出 rag_chain ( {context: retriever | format_docs, question: RunnablePassthrough()} | prompt | llm | StrOutputParser() ) result rag_chain.invoke(这个产品的退货政策是什么) print(result)这段代码是完整的 RAG 最小可运行链路跑通后你只需要替换文档、模型名和提示词就可以拓展出各种垂直场景。第一次建议在 Jupyter 里分步执行方便随时看中间结果检索出来的文档是什么、拼装后的提示词对不对、模型输出是否引用正确。5. Agent 工具调用与 LangGraph 流程编排5.1 从 RAG 到 Agent工具调用怎么设计RAG 解决“从知识库找答案”Agent 解决“调用工具完成任务”。两者的核心差异在于RAG 是一步到位的检索生成Agent 是多轮决策循环。想让大模型能够调用工具关键是给模型提供结构化函数定义并解析模型返回的工具调用参数。下面的示例演示了一个“既能回答知识库问题又能执行简单工具操作”的 Agent 雏形。工具定义使用tool装饰器LangChain 会自动提取函数名、描述和参数结构。# agent_demo.py 片段工具定义与调用 from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor import config tool def get_server_status(server_name: str) - str: 查询服务器状态模拟工具。 status_map {web-01: running, db-01: stopped, cache-01: degraded} return status_map.get(server_name, unknown server) tool def calculate_disk_usage(path: str) - str: 计算指定路径的磁盘占用百分比模拟工具。 return f{path} disk usage: 72% tools [get_server_status, calculate_disk_usage] llm ChatOpenAI( modelconfig.LLM_MODEL, api_keyconfig.API_KEY, base_urlconfig.BASE_URL ) prompt ChatPromptTemplate.from_messages([ (system, 你是一个可调用工具的 AI 助手。), (human, {input}), (placeholder, {agent_scratchpad}) ]) agent create_tool_calling_agent(llm, tools, prompt) executor AgentExecutor(agentagent, toolstools, verboseTrue) result executor.invoke({input: 帮我查一下 db-01 的状态顺便看下 /data 的磁盘占用}) print(result[output])这段代码的关键是tool装饰器。模型会根据函数名和 docstring 决定什么时候调用、传入什么参数LangChain 负责把模型的调用意图转成 Python 函数调用再把结果回传给模型继续决策。这种循环会一直持续到模型认为任务完成。5.2 为什么需要 LangGraphcreate_tool_calling_agent加AgentExecutor的方式适合快速验证但生产级 Agent 有两个问题一是循环控制和中间状态不够透明二是出错重试、人工审批、条件分支难以插入。LangGraph 解决这两点它把 Agent 流程建模为“节点 边”的状态图。# langgraph_demo.py 片段LangGraph 最小 Agent 循环 from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage class AgentState(TypedDict): messages: list next_step: str def call_llm(state: AgentState): # 实际项目里这里调用大模型解析是否要调用工具 last_message state[messages][-1].content if 查 in last_message: return {next_step: tool_execute, messages: [(assistant, 需要调用工具查询)]} return {next_step: finish, messages: [(assistant, 无需调用工具直接回答)]} def tool_execute(state: AgentState): return {next_step: finish, messages: [(assistant, 工具执行结果状态正常)]} graph StateGraph(AgentState) graph.add_node(call_llm, call_llm) graph.add_node(tool_execute, tool_execute) graph.add_edge(call_llm, tool_execute) graph.add_edge(tool_execute, END) graph.add_edge(call_llm, END) app graph.compile()LangGraph 的核心优势是可以显式定义“什么时候调用工具、什么时候结束、什么时候回到大模型”每个节点都能独立测试和日志观察。真实项目中经常用 LangGraph 实现“先检索 - 不满意则改写查询 - 二次检索 - 最终生成”这种带分支的 RAG 流程能做到的精细控制是普通线式链难以做到的。5.3 Agent 稳定性设计Agent 工具调用的不稳定因素主要有三类模型没听懂任务、工具参数传错、工具结果太复杂导致模型分析混乱。工程上应对策略如下工具描述要具体让模型知道“什么时候该用这个工具不该用哪个工具”工具入参要做类型校验和默认值兜底模型传错参数时返回清晰的错误信息而不是让进程崩溃工具结果要做摘要或裁剪超长 JSON 结果先抽关键字段再回传给模型避免上下文被无关字段撑爆添加最大迭代次数限制避免 Agent 陷入循环调用设置超时时间工具调用没有响应时暂停任务并给出明确提示很多项目遇到过 Agent 执行器在等待第三方服务响应时长时间挂起的问题建议统一走超时和重试策略。6. 检索增强效果评估与指标RAG 项目上线前最容易被忽视的是指标评估。很多人凭感觉看“回答得对不对”但检索质量、上下文相关性和生成忠实度完全可以量化。常见指标如下指标度量对象含义RecallK检索器前 K 个结果中有多少命中了真正相关的文档MRR检索器第一个相关结果出现在第几位Context Relevance检索器 生成器检索出的上下文与问题是否相关Faithfulness生成器生成的答案是否忠于检索到的上下文Answer Relevance生成器生成的答案是否针对问题而不是答非所问工程上简易评估方式准备 30 到 50 组“问题 - 标准答案或相关文档 ID”测试集批量跑通检索和问答记录正确率。这个动作在项目初期就可以做不用追求完整框架先把基线建立起来再优化分块方式、Embedding 模型和提示词。批量评估的伪代码如下# eval_demo.py 片段批量测试检索质量 test_questions [ 公司年假制度是怎样的, 报销流程需要哪些材料, 服务器故障如何处理 ] results [] for q in test_questions: docs retriever.invoke(q) top_source docs[0].metadata.get(source, unknown) results.append({question: q, top_source: top_source, score: docs[0].metadata.get(score, 0)}) for r in results: print(r)把检索的命中文档和人工标注的正确文档做对比就能算出早期正确率。这种测试会直接暴露分块过碎导致的信息不全、Embedding 模型导致的语义偏差、文档解析阶段表格错位等问题。7. 批量任务与 API 服务化7.1 批量文档入库实际项目中知识库文档往往成百上千不可能一个一个手动入库。批量入库的工程框架包括遍历文件目录、按扩展名选择加载器、入库程序记录每条文档的处理状态、失败重试。# batch_ingest.py 片段批量入库 from pathlib import Path from langchain_community.document_loaders import PyPDFLoader, TextLoader, UnstructuredMarkdownLoader SUPPORTED_EXT { .pdf: PyPDFLoader, .txt: TextLoader, .md: UnstructuredMarkdownLoader, } def load_document(file_path: Path): ext file_path.suffix.lower() if ext not in SUPPORTED_EXT: return [] loader_cls SUPPORTED_EXT[ext] loader loader_cls(str(file_path)) return loader.load() data_dir Path(docs) all_chunks [] for file_path in sorted(data_dir.iterdir()): try: docs load_document(file_path) chunks text_splitter.split_documents(docs) all_chunks.extend(chunks) print(f已处理: {file_path.name}, 分块数: {len(chunks)}) except Exception as e: print(f处理失败: {file_path.name}, 错误: {e}) vector_store.add_documents(all_chunks) print(f批量入库完成共入库 {len(all_chunks)} 个文本块)批量入库建议加入断点续传能力。简单方案是每处理一个文档后打印日志失败文档单独记录到 error.log全部跑完后统一排查重试。7.2 检索问答 API 封装把 RAG 检索问答封装成接口核心目的是把内部实现与外部调用解耦。外部系统只需要传问题接收答案和来源引用即可。下面给出基于 FastAPI 的最小示例按需扩展鉴权、限流、日志等能力。# api_service.py 片段FastAPI 封装 RAG 问答 from fastapi import FastAPI from pydantic import BaseModel app FastAPI(titleRAG API Service) class QueryRequest(BaseModel): question: str top_k: int 4 class QueryResponse(BaseModel): answer: str sources: list[str] app.post(/query, response_modelQueryResponse) def query_rag(req: QueryRequest): docs retriever.invoke(req.question, search_kwargs{k: req.top_k}) context format_docs(docs) answer rag_chain.invoke(req.question) sources list({d.metadata.get(source, unknown) for d in docs}) return QueryResponse(answeranswer, sourcessources) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)启动服务uvicorn api_service:app --host 0.0.0.0 --port 8000服务启动后可以用 curl 测试curl -X POST http://127.0.0.1:8000/query \ -H Content-Type: application/json \ -d {question: 产品保修期是多久, top_k: 3}需要注意FastAPI 服务启动在0.0.0.0会监听所有网卡接口生产环境必须加鉴权或只监听内网地址避免未授权调用。安全边界不能省至少要加一层 API Key 校验或网关代理。7.3 批量问答任务设计批量问答任务适合离线评测、大规模内容审核、批量报告生成。推荐队列结构# batch_query.py 片段批量问答逻辑 questions [ {id: 1, question: A 产品的价格是多少}, {id: 2, question: B 产品的保修政策}, ] results [] for item in questions: try: answer rag_chain.invoke(item[question]) results.append({id: item[id], question: item[question], answer: answer}) except Exception as e: results.append({id: item[id], question: item[question], answer: fERROR: {e}}) # 批量任务建议加间隔避免触发服务端限流 time.sleep(0.5)批量任务还要考虑失败重试和中间结果落盘避免跑到一半进程崩溃后全部重来。可以把每一条结果都实时写入 JSONL 文件下次启动时跳过已处理的问题。8. 资源占用与性能观察8.1 显存与内存观察方法无论使用本地模型还是纯 API都应该掌握资源占用观察方法。最简单的方式是使用nvidia-smi实时查看 GPU 显存watch -n 1 nvidia-smi如果是纯 API 模式的 RAG 应用本机基本不消耗模型推理显存主要开销在文档解析、向量化入库和检索计算。如果本地跑 Embedding 模型显存占用通常在 1G 到 4G 范围具体取决于模型参数量和量化方式需要实测。跑本地大模型生成时显存占用是最高的这也是为什么 RAG 系统通常放在服务端而客户端只做轻量展示。8.2 性能瓶颈分析RAG 链路中的瓶颈通常出现在三个位置第一文档解析。PDF 转纯文本时如果是扫描件需要 OCR速度会很慢建议提前转换成可复制文本的 PDF。第二Embedding 向量化。用 CPU 跑大批量入库会比较耗时有 GPU 可以大幅改善没有 GPU 就想办法压缩文档量和分块数量。第三检索阶段。数据量变大后暴力检索会变慢需要为向量库创建合适的索引不同数据库对应的索引类型和命令行参数各有差异。8.3 降低资源占用的方法使用量化后的开源 Embedding 模型显存占用明显下降控制向量库集合大小按业务域分库而不是所有数据塞进一个集合生成模型优先考虑 API 而不是本地部署把资源消耗集中到检索服务上对入库文档做去重避免重复文本块重复计算向量批量任务尽量串行加小批量并发避免同时触发多路生成导致内存峰值过高。9. 常见问题与排查方法问题现象可能原因排查方式解决方案ChromaDB 已存在但检索不到新文档重复调用 from_documents 覆盖了集合检查集合中向量数量改用 Chroma() 加载已有库后 add_documents中文检索效果差分块把句子切碎或 Embedding 模型不适合中文打印检索出的原始文本片段检查语义完整性调整分隔符或切换为中文优化的 Embedding 模型模型回答在编造内容提示词未限制或检索内容不相关查看最终发给模型的上下文中是否有正确答案强化提示词约束提高 TopK或加 Reranker批量入库时进程中断未做文档级容错查看日志定位具体失败文件逐文件 try/except失败记录到日志后继续API 调用报超时模型服务商接口响应慢或未设置合理超时先 curl 接口测试延迟客户端增加 timeout 值服务端做异步任务或重试Agent 反复调用同一个工具模型没有拿到完整的工具结果开启 verbose 查看工具返回内容对工具返回做摘要限制最大迭代次数向量化入库很慢CPU 推理且文档量大用 nvidia-smi 查看 GPU 是否生效切 GPU 或缩小任务批次检索出的文档来源混乱不同批次使用不同 Embedding 模型检查向量库配置与当前模型是否一致重建向量库并统一 Embedding 模型LangGraph 编译报错节点函数签名或状态定义不匹配查看 traceback 中节点名称检查状态字段类型和节点返回值结构昇腾等特定硬件上 vLLM 启动 Embedding/Reranker 失败框架对 Embedding 任务的支持不完整查看 vLLM 官方说明与加速卡适配文档改用专门的 Embedding 服务或推理框架Agent 执行等待第三方工具响应时长时间挂起工具调用无超时控制观察日志停留在工具调用节点增加超时与失败自动重试机制提示词生效不明显提示词过长或指令被上下文淹没简化提示词关键指令前置按检索模块调整提示词结构与顺序10. 最佳实践与合规提醒10.1 工程化建议RAG 和 Agent 项目要进入生产不只是把链路跑通那么简单。建议从几个维度做工程加固。目录管理要清晰原始文档、处理后的中间文本、向量库、日志、批量结果分别放到不同目录方便问题排查和数据回溯。版本管理要覆盖代码和配置模型名、API 版本、向量库版本、提示词版本都建议纳入版本管理尤其提示词改动直接影响输出效果要有 diff 记录。可以把提示词模板单独拆成 YAML 或 JSON 文件而不是硬编码在业务代码里。提示词上线前做基线评估每改一次提示词跑一遍已有的测试问题集对比答案质量和格式稳定性防止“改好了 A 类问题却破坏了 B 类问题”。服务化接口必须做访问控制FastAPI 服务监听内网地址或在网关层加认证避免知识库内容和模型调用被未授权使用。如果服务暴露到公网至少要求 API Key 和请求频率限制。Agent 涉及外部操作时必须加人工审批节点。特别是能够发消息、改配置、执行命令的工具要把“执行”改成“生成待执行指令人工确认后执行”。Agent 的自主性需要控制在一定范围内不能把关键操作完全交给模型自动决策。10.2 数据合规与授权提醒RAG 知识库可能涉及企业敏感数据和个人信息。入库前确认文档的使用授权范围对敏感信息做脱敏处理。涉及他人肖像、声音、版权内容时必须取得合法授权仅供测试环境验证不直接用于公开或商业场景。涉及企业私有数据时要遵循公司的数据安全规范明确数据存储位置和访问权限。11. 总结与下一步RAG 到 Agent 的完整链路可以拆成四个阶段文档处理与分块、Embedding 与向量检索、提示词拼装与大模型生成、Agent 工具调用与流程编排。绝大多数项目的问题在第一个阶段和第二个阶段而不是模型不够聪明。新手最先要验证的是一条完整链路能否跑通一份已知答案的文档能否被正确检索并回答。做到这一步RAG 的核心能力就已经掌握了。最容易踩的坑有两个一是入库时混用 Embedding 模型导致后续检索全部无效二是把 Agent 的工具调用想得太简单没有超人力和容错机制结果模型反复调用失败或陷入循环。前者靠统一模型解决后者靠 LangGraph 的状态控制和超时重试解决。接下来可以扩展的方向包括领域知识库优化企业内网数据接入、复杂 Agent 工作流多工具协作 人工审批、RAG 效果评测体系检索指标 生成指标 回归测试、以及用 Docker 把 RAG 服务完整容器化部署到服务器。每一步都有清晰的验证标准建议按顺序推进而不是一把抓。建议先收藏这篇文章把示例代码跑通一遍然后用自己的文档替换测试数据很快就能建立一套自己的 RAG 和 Agent 原型。