ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

RAG全链路拆解:从LLM基础到生产级应用实践

RAG全链路拆解:从LLM基础到生产级应用实践 很多人在学习 AI 与 LLM 工程时最头疼的不是某一个概念看不懂而是概念与概念之间、组件与组件之间、课程与实战之间的断层。网上资料虽然多但要么只讲原理不给代码要么给一段代码却说不清它属于 RAG 链路里的哪一环。本文围绕 Udemy 上一套比较完整的 AI LLM Engineering 课程主线来梳理GenAI 应用到底怎么做RAGRetrieval-Augmented Generation检索增强生成在不同业务里怎么落地以及从“能跑通”到“能上生产”之间还缺哪些工程能力。内容会覆盖 LLM 工程的基础概念、开发环境准备、Prompt 工程的核心方法、RAG 完整链路拆解、基于 LangChain 的实战项目、常见问题排查以及工程化最佳实践。适合刚接触大模型应用开发的新手也适合后端、算法、测试同学快速建立完整知识图谱。1. 背景与核心概念1.1 为什么需要 AI EngineeringAI Engineering 指的是把大模型能力稳定、可控、可维护地集成到真实软件系统中的工程实践。它和传统的机器学习工程有一个明显区别传统 ML 工程更多关注训练、调参、模型部署和性能监控而 AI Engineering 面对的是已经训练好的大模型我们更多关注的是上下文管理、工具调用、检索策略、结果校验、成本控制和用户体验。换句话说大模型本身已经给了我们很强的“基础能力”但如何把这种能力变成可靠的产品功能就是 AI Engineering 要解决的问题。1.2 LLM、GenAI、RAG 是什么关系可以把三者理解成三层关系LLMLarge Language Model大语言模型底层模型能力。比如 GPT、Llama、Qwen 这类模型擅长文本生成、理解、推理等。GenAIGenerative AI生成式人工智能基于 LLM 构建的生成式应用。它可以生成文本、图片、代码、音频等本质是利用模型生成新内容。RAG一种应用架构也是 GenAI 应用中最常见的落地方式。它先从外部知识库检索与问题相关的内容再把检索结果作为上下文交给 LLM 生成答案。关系可以理解为LLM 是“大脑”GenAI 是“业务形态”RAG 是让大脑学会“查资料再回答”的方法。1.3 RAG 的基本流程一个最经典的 RAG 流程包含五个步骤文档加载把 PDF、Word、HTML、TXT 等文件读取为纯文本。文本切分把长文本切分成适合向量化的片段。向量化用 Embedding 模型把文本切块转换为向量。检索把用户问题向量化后在向量数据库中做相似度搜索。生成把检索结果和用户问题一起封装成 Prompt交给 LLM 生成回答。理解这五步就理解了 RAG 的主干。后面所有的优化比如改切分策略、换向量库、加 rerank 模型、做混合检索都是在优化某一步。2. 环境准备与工程工具链2.1 开发语言与 IDERAG 应用开发目前最主流、生态最完整的语言是 Python。大部分 LLM SDK、向量数据库客户端、文档解析库都优先支持 Python。建议环境Python 3.10 以上版本。使用 Anaconda 或 venv 创建独立虚拟环境。IDE 推荐 VS Code 或 PyCharm。本文示例以 Python 为主如果你更熟悉 Java 或 TypeScript思路同样适用只是语言绑定不同。2.2 核心依赖库在开始项目前先把常用依赖列出来。不同项目的具体依赖会有差异但下面的库是 RAG 项目中出现频率最高的langchainLLM 应用开发框架封装了模型调用、Prompt、链、Agent、文档处理等能力。langchain-communityLangChain 社区扩展包含大量第三方集成。chromadb轻量级向量数据库适合本地原型开发。openai或对应模型提供方的 SDK用于调用 LLM 和 Embedding 模型接口。pypdf读取 PDF 文件。python-docx读取 Word 文件。tiktokenToken 计算工具。sentence-transformers/text2vec本地 Embedding 模型。fastapi和uvicorn如果需要将 RAG 服务封装成 API。注意以上版本更新很快不要盲目复制网上旧命令。建议在项目根目录准备一个requirements.txt然后逐项安装并验证版本兼容性。示例requirements.txtlangchain0.1.0 langchain-community0.0.10 chromadb0.4.22 openai1.0.0 pypdf3.17.0 python-docx1.1.0 tiktoken0.5.1 fastapi0.109.0 uvicorn0.27.0安装命令pip install -r requirements.txt2.3 模型服务的选择RAG 项目至少需要两类模型服务生成模型负责根据 Prompt 生成最终回答。常见选择有 OpenAI 的 GPT 系列、Anthropic Claude、国产的 Qwen、DeepSeek 等。Embedding 模型负责把文本转换为向量。常见选择有 OpenAI 的text-embedding-3-small、BGE 系列、M3E 等。模型服务可以来自云端 API也可以使用本地模型。云端 API 接入简单、效果稳定本地模型部署成本高、但数据隐私更好。开发阶段建议用云端 API 快速验证生产阶段再根据合规和成本要求做选型。如果你是国内开发者网络访问并不是障碍——很多国产化模型平台提供了 OpenAI 兼容接口只需要把base_url替换为对应平台的地址即可。这也是后面实战例子中我会强调“不把模型地址写死”的原因。3. 核心原理拆解3.1 Token 与上下文窗口LLM 处理文本的基本单位是 Token而不是字符或者单词。Token 可能是一个词的一部分、一个完整单词、甚至一个标点符号。不同模型有不同的 Tokenizer 规则同样一段文字在不同模型下的 Token 数量可能不同。上下文窗口则决定了模型一次能接收的最大 Token 数。常见的窗口大小有 4K、8K、32K、128K 等。RAG 之所以必要恰恰是因为现实知识库的规模远超上下文窗口我们不可能把所有资料都塞进 Prompt。理解 Token 有几点实际意义成本估算API 按 Token 计费Prompt 越长成本越高。回答质量上下文过长可能导致模型“迷失”检索到的无关内容会干扰答案。性能优化减少无效 Token 可以降低延迟和成本。3.2 Embedding 与向量检索Embedding 是把文本映射到一个高维空间中的向量。语义相近的文本其向量在空间中距离也近。比如“怎么修改密码”和“密码重置流程”在向量空间中通常比较接近。RAG 中常用的向量检索算法有余弦相似度计算向量夹角的余弦值值越大越相似。内积适合已归一化的向量。欧式距离值越小越相似。向量数据库负责存储这些向量并快速检索。常见选择有向量数据库部署方式适用场景Chroma本地/嵌入式原型开发和中小规模项目FAISS本地库高性能批量检索Milvus服务端生产级大规模检索Qdrant服务端生产级、支持丰富过滤条件PostgreSQL pgvector服务端已有 PostgreSQL 的场景3.3 Prompt Engineering 的核心方法Prompt Engineering 并不是简单“写提示词”而是一套系统化的上下文设计方法。核心目标有两个让模型理解任务、让模型按照期望格式输出。常见的 Prompt 结构包括角色设定告诉模型“你是一个资深的客服助手”。任务描述明确要完成什么任务。输入数据提供用户问题或检索到的资料。输出约束规定格式、字数、是否可以使用 Markdown。示例给出一个输入输出对让模型模仿格式。下面是一个简单但完整的 Prompt 模板示例prompt_template 你是一个严谨的技术客服助手。 请根据以下资料回答用户问题。 回答要求 1. 只能依据资料内容不要编造事实 2. 如果资料中找不到答案请回复“该问题在现有资料中暂未找到答案” 3. 回答控制在 200 字以内 4. 使用中文回答。 资料内容 {context} 用户问题 {question} 回答 这段模板最大的特点是“约束”清晰。它限制了知识来源、未知情况下的行为、字数、语言。在实际项目中Prompt 调优的优先级往往高于模型切换因为模型不变时Prompt 是唯一可控的变量。3.4 文本切分策略文本切分是 RAG 中容易被低估的环节。切分得太大检索结果不够精确而且浪费 Token切分得太小上下文可能缺少必要逻辑影响生成质量。常用的切分方式固定长度切分按字符数或 Token 数硬切。实现简单但容易切断句子。递归字符切分按段落、句子、标点逐级切分。LangChain 的RecursiveCharacterTextSplitter就是这种思路。按文档结构切分利用 Markdown 标题、HTML 标签、PDF 章节等结构信息切分。语义切分通过 Embedding 判断句子间语义边界。生产环境一般不追求“最完美切分”而是通过检索效果评估来迭代。4. 完整实战案例构建一个本地知识库问答系统下面动手做一个最小可用 RAG 项目。我们以“公司内部 FAQ 文档问答”为场景流程包含文档加载、切分、向量化、检索、生成、API 封装六个部分。4.1 创建项目结构rag-demo/ ├── app/ │ ├── __init__.py │ ├── config.py │ ├── document_loader.py │ ├── text_splitter.py │ ├── vector_store.py │ ├── rag_pipeline.py │ └── api.py ├── data/ │ └── faq.pdf ├── requirements.txt └── README.md4.2 编写配置文件配置文件统一管理模型地址、API Key、向量库路径等参数。项目里建议使用环境变量或配置文件不要把密钥硬编码在代码里。# app/config.py import os from dotenv import load_dotenv load_dotenv() class Config: # 生成模型配置 LLM_BASE_URL os.getenv(LLM_BASE_URL, https://api.openai.com/v1) LLM_API_KEY os.getenv(LLM_API_KEY, ) LLM_MODEL os.getenv(LLM_MODEL, gpt-3.5-turbo) # Embedding 模型配置 EMBEDDING_BASE_URL os.getenv(EMBEDDING_BASE_URL, https://api.openai.com/v1) EMBEDDING_API_KEY os.getenv(EMBEDDING_API_KEY, ) EMBEDDING_MODEL os.getenv(EMBEDDING_MODEL, text-embedding-3-small) # 向量库配置 CHROMA_DB_DIR os.getenv(CHROMA_DB_DIR, ./chroma_db) COLLECTION_NAME os.getenv(COLLECTION_NAME, faq_kb)这里我把BASE_URL单独提出来就是为了方便替换成国内模型平台或其他 OpenAI 兼容服务。4.3 实现文档加载器文档加载的职责是“把文件变成文本”。不同类型的文件需要不同的解析库。# app/document_loader.py from pathlib import Path from typing import List from pypdf import PdfReader def load_pdf(file_path: str) - List[str]: 加载 PDF 文件按页返回文本列表。 reader PdfReader(file_path) pages [] for page in reader.pages: text page.extract_text() if text and text.strip(): pages.append(text.strip()) return pages def load_txt(file_path: str) - List[str]: 加载纯文本文件。 with open(file_path, r, encodingutf-8) as f: content f.read() return [content.strip()] def load_document(file_path: str) - List[str]: 根据文件后缀自动选择加载方式。 suffix Path(file_path).suffix.lower() if suffix .pdf: return load_pdf(file_path) elif suffix .txt: return load_txt(file_path) elif suffix .md: return load_txt(file_path) else: raise ValueError(f暂不支持的文件类型: {suffix})4.4 实现文本切分器切分需要平衡“片段完整”和“长度可控”。这里用 LangChain 的递归切分器同时设置了重叠长度避免关键信息被切断。# app/text_splitter.py from typing import List from langchain.text_splitter import RecursiveCharacterTextSplitter def split_documents(pages: List[str]) - List[str]: 将文档页面切分为适合向量化的文本块。 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每块最大字符数 chunk_overlap50, # 块之间重叠字符数 separators[\n\n, \n, 。, , , ., !, ?, , ,, , ], length_functionlen, ) chunks text_splitter.split_text(\n\n.join(pages)) return chunks注意separators的优先级先尝试按段落分再尝试按句子分最后按字符分。这样可以在保留完整语义的前提下控制长度。4.5 实现向量库模块向量库模块负责把文本块向量化并存储到 Chroma同时提供检索函数。# app/vector_store.py from typing import List from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import Chroma from app.config import Config def build_vector_store(chunks: List[str], collection_name: str Config.COLLECTION_NAME): 构建或重建向量库。 embeddings OpenAIEmbeddings( openai_api_baseConfig.EMBEDDING_BASE_URL, openai_api_keyConfig.EMBEDDING_API_KEY, modelConfig.EMBEDDING_MODEL, ) vector_store Chroma.from_texts( textschunks, embeddingembeddings, collection_namecollection_name, persist_directoryConfig.CHROMA_DB_DIR, ) return vector_store def load_vector_store(collection_name: str Config.COLLECTION_NAME): 加载已有的向量库。 embeddings OpenAIEmbeddings( openai_api_baseConfig.EMBEDDING_BASE_URL, openai_api_keyConfig.EMBEDDING_API_KEY, modelConfig.EMBEDDING_MODEL, ) vector_store Chroma( collection_namecollection_name, embedding_functionembeddings, persist_directoryConfig.CHROMA_DB_DIR, ) return vector_store def search(vector_store, query: str, top_k: int 4) - List[str]: 检索与问题最相关的文本块。 docs vector_store.similarity_search_with_score(query, ktop_k) return [doc.page_content for doc, score in docs]注意Chroma.from_texts传入的是字符串列表而不是 LangChain Document 对象列表。如果你在项目中用的是 Document 对象那么应该使用from_documents方法。4.6 实现 RAG 核心链路RAG 管道把“检索”和“生成”串起来这是整个应用的核心逻辑。# app/rag_pipeline.py from typing import List from langchain.chat_models import ChatOpenAI from app.config import Config from app.vector_store import search PROMPT_TEMPLATE 你是一个严谨的技术客服助手。 请根据以下资料回答用户问题。 回答要求 1. 只能依据资料内容不要编造事实 2. 如果资料中找不到答案请回复“该问题在现有资料中暂未找到答案” 3. 回答控制在 200 字以内 4. 使用中文回答。 资料内容 {context} 用户问题 {question} 回答 def build_prompt(question: str, context: str) - str: 构造大模型输入 Prompt。 return PROMPT_TEMPLATE.format(questionquestion, contextcontext) def generate_answer(question: str, context: List[str]) - str: 调用 LLM 生成最终回答。 llm ChatOpenAI( openai_api_baseConfig.LLM_BASE_URL, openai_api_keyConfig.LLM_API_KEY, modelConfig.LLM_MODEL, temperature0.2, max_tokens500, ) context_text \n\n.join(context) prompt build_prompt(question, context_text) response llm.invoke(prompt) return response.content def rag_query(vector_store, question: str): RAG 完整流程检索 - 构造 Prompt - 生成回答。 # 1. 检索 relevant_docs search(vector_store, question, top_k4) # 2. 生成 answer generate_answer(question, relevant_docs) return { question: question, answer: answer, sources: relevant_docs, }这里有几个工程细节值得注意temperature0.2降低了随机性适合知识问答场景。max_tokens500限制了回答长度避免 Token 浪费。sources字段保留检索到的原文方便后续调试和溯源。4.7 提供 API 服务在真实项目里RAG 应用通常需要对外提供 API。这里用 FastAPI 封装一个接口。# app/api.py from fastapi import FastAPI from pydantic import BaseModel from app.vector_store import load_vector_store from app.rag_pipeline import rag_query app FastAPI(titleRAG Demo API) vector_store load_vector_store() class QueryRequest(BaseModel): question: str class QueryResponse(BaseModel): question: str answer: str sources: list[str] app.post(/query, response_modelQueryResponse) def query(req: QueryRequest): result rag_query(vector_store, req.question) return QueryResponse(**result) app.get(/health) def health(): return {status: ok}启动 API 服务uvicorn app.api:app --reload --port 8000访问http://localhost:8000/docs可以打开 Swagger 文档方便测试接口。4.8 编写索引脚本首次运行项目时需要先建立向量库。单独写一个脚本避免每次查询都重建索引。# scripts/build_index.py from pathlib import Path from app.document_loader import load_document from app.text_splitter import split_documents from app.vector_store import build_vector_store DATA_DIR Path(__file__).parent.parent / data def main(): # 1. 加载文档 all_pages [] for file_path in DATA_DIR.glob(*): if file_path.suffix.lower() in [.pdf, .txt, .md]: print(f加载文档: {file_path}) pages load_document(str(file_path)) all_pages.extend(pages) # 2. 切分文本 chunks split_documents(all_pages) print(f切分得到 {len(chunks)} 个文本块) # 3. 构建向量库 build_vector_store(chunks) print(向量库构建完成) if __name__ __main__: main()运行命令python scripts/build_index.py预期输出加载文档: data/faq.pdf 切分得到 42 个文本块 向量库构建完成5. 常见问题与排查思路RAG 项目虽然跑通容易但在真实场景里问题非常多。下面列几个高频问题。问题现象常见原因解决思路启动时报OpenAIError: Invalid API KeyAPI Key 配置错误或环境变量未加载检查.env文件是否被正确加载查看config.py中读取的 Key 是否有前后空格检索结果为空PDF 是扫描件无法提取文本或文档路径错误先用 PDF 阅读器复制文字验证对扫描件需接入 OCR 能力回答明显脱离资料检索到的 Top-K 内容不相关或 Prompt 没有约束知识来源调大top_k增加相关性阈值过滤优化切分方式并在 Prompt 中强调“只依据资料”Token 消耗过高文本切分过大、检索结果过多、历史对话无限增长调小chunk_size控制检索数量对话场景加滑动窗口向量化速度很慢Embedding 模型过大或使用 CPU 推理使用云端 Embedding API本地模型开启 GPU 加速启动时依赖冲突LangChain 版本升级导致 import 路径变化锁定版本按官方迁移文档调整 import 路径5.1 如何排查“回答质量差”回答质量差是 RAG 项目最头疼的问题。建议按以下顺序排查直接看检索结果。单独打印relevant_docs确认检索到的文本是否和问题相关。检查切分粒度。如果一段文本包含多个知识点检索命中后上下文会混乱。检查 Prompt 约束。模型不是不知道“不能编造”而是 Prompt 里没写清楚。检查温度参数。过高的 temperature 会让模型自由发挥。增加 Rerank 重排序。第一次向量检索 TopK 可以扩大到 20 条再通过 rerank 模型精排到 4 条。5.2 如何排查“检索不到”检索不到通常不是向量库的问题而是文档处理的问题。PDF 是扫描版或图片版提取不到文字。文本编码不是 UTF-8读取后乱码。切分后的文本块太短或太少涵盖不全面。Embedding 模型与查询语言不匹配比如文档是中文但模型偏英文理解。6. 进阶方向从 RAG 到 Agentic RAG 与 GraphRAG基础 RAG 跑通后很多业务会进入进阶阶段。这里介绍两个高频进阶方向。6.1 Agentic RAGAgentic RAG 是在基础 RAG 上引入 Agent 决策能力。它不再是“每来一个问题都按固定流程检索一次”而是让 Agent 自己判断这个问题需不需要检索需要检索哪类知识库检索一次不够要不要多轮检索甚至要不要调用工具或外部 API。典型场景用户问题比较模糊需要 Agent 先拆解成多个子问题。用户问的问题涉及多个知识域需要分别检索不同知识库。上下文相关性强需要多轮追问后才能确定检索条件。Agentic RAG 的优势是更灵活代价是链路更长、调试更复杂、Token 消耗更高。初学者不要一上来就上 Agent而是先确保基础 RAG 效果稳定。6.2 GraphRAGGraphRAG 是把知识图谱和 RAG 结合。它先构建知识图谱保存实体之间的显式关系再基于图结构检索。相比传统向量检索它更适合需要多跳推理的场景。举一个简单例子传统 RAG 能回答“A 公司发布了什么产品”。如果问题是“A 公司产品与 B 公司产品有什么关系”就可能需要图结构来关联多级信息。GraphRAG 的代价是构建复杂、成本高。当前它的定位是解决“对全局性问题理解差”的问题而不是完全替代向量检索。6.3 RAG 评估指标很多同学做完 RAG 后不知道效果好不好。常用指标可以分为两类类别指标说明检索质量RecallK在 TopK 结果中相关文档占全部相关文档的比率检索质量MRR第一个相关结果排得越靠前分值越高生成质量Faithfulness / 忠实度回答是否与检索到的资料一致不编造生成质量Answer Relevancy回答是否切题是否直接回答了问题端到端人工评分结合业务场景做真人打分最可靠但最耗时实际项目中建议两手抓离线阶段用自动指标做回归上线前组织标注人员做一轮人工评测。7. 工程化最佳实践与学习路线7.1 工程化最佳实践从“能跑通的 Demo”到“能上生产的系统”中间需要补齐很多东西。安全与合规不要把 API Key 写在代码里使用环境变量或密钥管理服务。对上传文档做敏感信息识别防止隐私数据被索引。对外 API 必须做认证鉴权和限流防止被刷。生产环境删除、更新知识库前先备份并做好数据变更审批。性能优化文档解析和向量化是离线耗时操作建议做定时任务而不是每次请求触发。向量库连接建议使用连接池避免每请求创建新连接。对高频问题可以用缓存机制减少模型调用次数。检索相关参数TopK、相似度阈值做成配置项方便多环境下调整。可维护性RAG 代码做好模块拆分加载、切分、向量化、检索、生成各管一段。日志里记录检索结果、Prompt 内容和模型输出方便问题复盘。模型版本和 Prompt 版本都要管理建议加版本号。建立质量回测集。每次改 Prompt 或换模型跑同一批测试用例对比结果。7.2 关于模型精度FP16、FP32、BF16在部署本地模型时你可能会遇到 FP16、FP32、BF16 这些概念。它们本质上是数值精度的不同表示方式FP32单精度浮点数精度高占用显存大。FP16半精度浮点数占用显存小但表示范围有限训练时容易出现溢出。BF16Brain Floating Point与 FP16 一样占用 2 个字节但保留了更大的指数范围更适合大模型推理。在推理场景BF16 通常比 FP16 更稳定是很多大模型推理框架的默认选择。如果你在本地部署模型时遇到显存不够可以考虑从 FP32 切到 BF16 或 FP16但要注意输出效果可能轻微下降。这里不建议盲目追求量化。先跑通再用精度换规模。7.3 学习路线建议如果你是从零开始建议按下面路线推进先跑通一个最小 LLM 调用 Demo理解 API 参数和 Token 计算。掌握 Prompt Engineering重点练习角色设定、Few-Shot 示例、输出约束。实现一个本地文档问答完整走一遍 RAG 五步流程。加入对话记忆让机器人支持多轮对话。学习 Agent 基础了解工具调用和 ReAct 模式。深入研究 RAG 优化方向切分策略、Rerank、混合检索、指令微调。了解生产部署API 封装、监控、评测、数据回流。7.4 从 Demo 到生产的关键提醒Demo 和生产之间最大的差距不在于模型的强弱而在于工程体系的完整度。我在实际项目里反复遇到三类线上问题一是知识库更新后向量库没有同步重建导致用户持续看到旧答案二是没有记录检索日志回答错误时完全无法定位是检索问题还是模型问题三是没有对成本和延迟做监控模型调用量上涨后费用失控。所以当你完成后端逻辑后一定要做三件事给知识库加版本管理和重建机制。给每次查询加日志字段至少包括问题、检索结果、最终回答、Token 消耗。给模型调用加监控面板观察每日调用量和费用曲线。这几件事不做RAG 系统上线后大概率会在某个周末接到业务方的紧急电话。最后补充一个自查清单适合在提交代码前快速检查是否能明确说清楚每个模块的输入和输出是否所有密钥都从环境变量读取是否记录了每次问答的检索结果和模型输出知识库更新后向量库能否平滑重建是否有基础的调用限流和异常兜底如果这些问题都能给出明确回答那么你的 RAG 项目已经从“会跑”走到了“可用”。下一步可以继续深入 Agent、微调和多模态方向把这些工程经验复用到更多 GenAI 场景中。
返回列表