企业官网建设流程全解析

这次我们来看 RAG。如果只看概念，RAG 并不难——把文档切片、向量化、存进向量库，用户提问时先检索，再拼接提示词让大模型生成。但真要把它做成一个能上线的知识库问答系统，问题就会冒出来：切片怎么切才好？检索不召回怎么办？知识库文档格式混乱怎么处理？接口怎么暴露给业务方？批量导入文档时怎么控制内存和速度？这些都是实战里绕不开的坑。

RAG（Retrieval-Augmented Generation，检索增强生成）是目前 AI 应用开发里最常见的技术路线，也是“AI 应用开发工程师”面试和项目里出现频率最高的关键词之一。它解决的核心问题很明确：大模型不知道你私有文档里的内容，但你可以通过“先检索、再生成”的方式，把相关内容塞进上下文中，让模型基于这些资料作答。

本文会从 RAG 的原理出发，逐步讲清楚三件事：一、一个最小可运行的 RAG 流水线怎么写；二、高级检索怎么落地，包括切块策略、混合检索、重排序、元数据过滤以及查询重写；三、知识库服务怎么封装 API、怎么处理批量任务。读完你可以直接拿这套思路去搭自己的知识库问答服务。

1. RAG 核心能力速览

这一套流程本身是标准的，难在细节。同一个 PDF，切片策略不同，检索质量可能差一个量级；同一个问题，向量检索召回不到，换成混合检索加重排序之后效果立刻不一样。

2. 适用场景与使用边界

RAG 适合解决“模型不知道，但文档里写了”的问题。常见场景包括：

• 企业内部制度、产品文档、技术支持知识的问答。
• 针对特定领域的文档解析与检索，例如法律条款、技术规范、设备手册。
• 把线上帮助中心改造成对话式搜索入口。
• 辅助研发人员检索历史设计文档、需求文档、故障报告。
• 面向客服场景构建带引用来源的应答助手。

RAG 不适合解决什么事情？首先，它不适合作为事务型系统的入口，比如查订单、改状态这类强流程操作，RAG 的职责是“检索和生成”，不是“执行”。其次，如果文档内容本身质量差、口径不统一，RAG 只是把“文档混乱”变成了“回答混乱”，不会自动消除错误信息。再者，如果问题高度依赖逻辑推理或多轮复杂任务拆解，单轮 RAG 会吃力，需要走向 Agentic RAG 路线。

使用 RAG 还需要注意合规边界。知识库里的文档必须来源合法，不能把未经授权的他人作品、内部敏感数据随意拖入系统；涉及个人信息和企业机密的资料要做脱敏和权限控制；对外发布或商用前，要对模型输出做人工复核。尤其在企业场景，建议按用户角色控制知识库访问范围，不要把所有文档一股脑开放给所有人。

3. 环境准备与前置条件

搭建 RAG 流水线需要准备以下环境：

• 操作系统：Windows、Linux、macOS 均可，推荐 Linux 做服务部署。
• Python 版本：建议 3.9 或 3.10 以上。
• 依赖库：LangChain 生态组件、向量数据库客户端、嵌入模型库。
• 嵌入模型：可以本地下载 BGE 等开源中文嵌入模型，也可以使用云服务 Embedding 接口。
• 大模型服务：需要一个 OpenAI 兼容的接口地址或本地模型推理服务。
• 硬件要求：纯 CPU 环境可以跑通完整流程，只是嵌入和生成速度会慢；本地大模型推理建议准备 NVIDIA GPU。
• 磁盘空间：嵌入模型一般几百 MB 到 1GB 左右，实际以模型文件大小为准。

安装依赖示例：

pip install langchain langchain-community langchain-openai chromadb sentence-transformers fastapi uvicorn

注意：LangChain 版本迭代较快，不同版本之间的导入路径和方法名可能会有变化。如果你在安装后遇到导入报错，优先去对应版本的官方文档里查一下最新写法。

如果打算用 CPU 跑本地嵌入模型，需要安装 PyTorch 的 CPU 版本或自动安装的默认版本都可以，实际看运行环境。

4. 一个最小可运行的 RAG 流水线

下面直接给一套最小可运行的知识库问答流程。这里用 LangChain 作为编排框架，向量数据库用 Chroma，嵌入模型用 BGE 中文模型，大模型接一个 OpenAI 兼容接口。所有代码都是常见写法，具体模型名和接口地址需要按你的实际环境替换。

4.1 文档加载与切片

第一步，把一个文本文件加载进来，切成固定大小的块。

from langchain_community.document_loaders import TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter loader = TextLoader("docs/rag.md", encoding="utf-8") documents = loader.load() splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, separators=["\n\n", "\n", "。", "！", "？", " ", ""] ) chunks = splitter.split_documents(documents) print(f"一共切成了 {len(chunks)} 个文本块")

这里的chunk_size=500是字符数，chunk_overlap=50表示相邻块之间保留 50 个字符的重叠，作用是避免在切分处切断完整语义。separators指定了优先按段落、换行、句号切分，让每个块尽量保持语义完整。

4.2 向量化与写入向量库

第二步，用嵌入模型把每个文本块转成向量，写入 Chroma。

from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_community.vectorstores import Chroma embeddings = HuggingFaceEmbeddings( model_name="BAAI/bge-small-zh-v1.5" ) vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings, persist_directory="./chroma_db" )

首次运行会自动下载嵌入模型文件。BGE 这一系列模型对中文支持比较好，如果网络条件有限，可以提前把模型文件下载好放到本地目录，再用model_path指向本地路径。

4.3 检索

第三步，从向量库里检索与问题最相关的文本块。

retriever = vectorstore.as_retriever(search_kwargs={"k": 5}) query = "如何配置切块重叠参数" docs = retriever.get_relevant_documents(query) for i, doc in enumerate(docs): print(f"[{i + 1}] {doc.page_content[:200]}")

这里k=5表示召回最相似的 5 个文本块。默认走的是向量相似度检索，适合对语义相似问题的召回。

4.4 提示词组装与生成

第四步，把检索结果拼接成上下文，交给大模型生成回答。

from langchain.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI prompt = ChatPromptTemplate.from_template(""" 基于以下资料回答问题。 如果资料中没有相关内容，请你直接说不知道，不要编造。 资料： {context} 问题：{question} 回答： """) llm = ChatOpenAI( model="<你的模型名称>", base_url="<你的OpenAI兼容接口地址>", api_key="<你的API Key>" ) context = "\n\n".join(doc.page_content for doc in docs) chain = prompt | llm answer = chain.invoke({ "context": context, "question": query }) print(answer)

整套流程写下来，一个最小 RAG 应用就通了。你输入一个问题，系统先检索相关文档片段，再让模型基于这些片段作答。这一步跑通之后，后面所有高级检索优化都是在这一套框架上增强。

5. 高级检索实战：从“能召回”到“召得准”

最小流水线能跑通，但线上效果通常不够用。真实知识库里可能有几百个文档，每个文档几十页，问题五花八门，单纯靠向量相似度召回很容易出现“审不准、答不对”。下面按实战顺序讲几种高级检索手段。

5.1 切块策略：先让知识库“切对”

切块是 RAG 质量的第一道关口。切得太碎，语义会断；切得太大，检索噪声会变多，而且大模型上下文窗口也可能放不下。

常用的切块策略有：

• 固定长度切分：按字符或 token 数硬切，简单但容易切断语义。
• 递归字符切分：按段落、句子、标点逐级切分，尽量保持语义完整。
• 按文档结构切分：利用 Markdown 标题、PDF 章节、列表层级切分，适合结构化文档。
• 语义切分：先按句子切，再根据向量相似度判断哪些句子应该合并成一个块，切分更智能，但耗时更高。

实际项目里建议先按文档结构切，没有结构再用递归字符切分。chunk_size不是越大越好，需要根据你的文档类型、模型上下文窗口和问题长度做几组对照实验。一般可以先从 300 到 800 这个范围开始测。

5.2 混合检索：向量检索加 BM25

向量检索擅长语义匹配，但对精确词、编号、型号这类信息不敏感。比如你搜“服务端口 8080”，如果文档里写的是“端口号 8080”，向量检索可能没问题；但如果你搜“产品型号 ABC-123”，向量检索很可能不如“关键词精确匹配”稳定。

解决办法是混合检索，把向量检索和 BM25 关键词检索的结果合并。BM25 是一种经典的关键词排序算法，在文本检索里非常成熟。

from langchain_community.retrievers import BM25Retriever from langchain.retrievers import EnsembleRetriever bm25_retriever = BM25Retriever.from_documents(chunks, k=5) vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5}) ensemble_retriever = EnsembleRetriever( retrievers=[bm25_retriever, vector_retriever], weights=[0.3, 0.7] ) docs = ensemble_retriever.get_relevant_documents("端口 8080 配置")

weights控制两种检索结果的权重。关键词类问题多的场景，可以把 BM25 权重调高；语义类问题多的场景，向量检索权重调高。这套方案在 LangChain 里叫EnsembleRetriever，是很多生产项目的起步选择。

5.3 重排序：让答案顺序更可信

混合检索之后，召回的候选集变多了，但排序可能还不够准。这时候需要用重排序模型再打一次分，把最相关的结果排到最前面。常见做法是用 CrossEncoder 模型，对“查询-文档”的句子对输出一个相关度分数。

from sentence_transformers import CrossEncoder reranker = CrossEncoder("BAAI/bge-reranker-base") pairs = [[query, doc.page_content] for doc in docs] scores = reranker.predict(pairs) ranked = sorted(zip(docs, scores), key=lambda x: x[1], reverse=True) for doc, score in ranked: print(score, doc.page_content[:100])

重排序的优点是精度高，缺点是速度比向量检索慢，所以一般只对召回的前几十个候选结果重排序，而不是对整个知识库打分。重排序模型本身不算大，CPU 上也能跑，但批量请求时要注意推理耗时。

5.4 元数据过滤：RAG 里的“高级检索语法”

用过必应高级检索的人都知道，搜索时可以加site:、filetype:、intitle:这类条件来缩小范围。RAG 里的元数据过滤承担类似职责：给文档打上来源、分类、日期、作者、部门等标签，检索时按标签过滤，能显著减少噪声。

retriever = vectorstore.as_retriever( search_kwargs={ "k": 5, "filter": {"source": "用户手册.pdf"} } )

比如你只想查“XX 产品的故障排查”，可以先按产品线过滤，再按故障类型检索。比全局库检索要精准得多。实际接入时，文档加载阶段就要把元数据一起写入向量库，否则后续没有字段可用。

5.5 查询重写与多轮对话改造

用户在问答场景里的问题往往不够完整，比如先问“RAG 怎么实现”，再问“它的优缺点是什么”，这里“它”指代的是上一轮主题。如果直接把第二句话送去向量检索，效果会很差。

查询重写要做两件事：

• 把指代词替换成明确主题。
• 把短问题扩展成完整问题。

简单实现可以用大模型生成改写后的查询词，也可以用规则做指代消解。LangChain 社区有MultiQueryRetriever这类组件，思路是用大模型从原始问题生成多个相似问法，再分别检索后合并结果。这种“一次检索变多次检索”的做法能明显提高召回率。

5.6 Agentic RAG：从“查一次”到“按需查”

再进一步就是 Agentic RAG。传统 RAG 是“用户提问 -> 查一次 -> 生成”，而 Agentic RAG 让大模型先判断需要什么信息，再决定调用哪个知识库、是否需要再查一次、是否需要多条信息合并后才能回答。

典型做法是把知识库检索工具暴露给大模型 Agent。模型先规划，再调用检索工具，如果信息不够就调整关键词再查一次，最后把多轮检索结果汇总生成答案。这种设计适合复杂问题、跨文档问题、对比型问题。缺点是链路更长、延迟更高，也更依赖模型本身的能力。

6. 接口 API 与批量任务

知识库问答不能只活在脚本里，要接给业务系统用，需要封装成 API。下面用一个 FastAPI 示例说明怎么把 RAG 链路暴露为服务。

6.1 API 服务封装

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class QueryRequest(BaseModel): question: str class QueryResponse(BaseModel): answer: str source_docs: list[str] @app.post("/api/rag/query", response_model=QueryResponse) def query(req: QueryRequest): docs = retriever.get_relevant_documents(req.question) context = "\n\n".join(doc.page_content for doc in docs) answer = llm.invoke(...) return QueryResponse( answer=answer, source_docs=[doc.page_content for doc in docs] )

启动服务：

uvicorn app:app --host 0.0.0.0 --port 8000

注意：retriever和llm需要在启动时初始化成全局对象，不要每次请求都重新加载模型和向量库，否则服务会非常慢。

调用示例：

curl -X POST http://127.0.0.1:8000/api/rag/query \ -H "Content-Type: application/json" \ -d '{"question": "什么是 RAG？"}'

返回结果示例：

{ "answer": "RAG 是检索增强生成……", "source_docs": [ "文档片段一", "文档片段二" ] }

6.2 批量文档导入

知识库上线前需要把存量文档批量导入。批量导入的第一步是遍历目录，按文件类型选择合适的加载器，然后切片、向量化、写入向量库。

import os from langchain_community.document_loaders import TextLoader, PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter input_dir = "./docs" splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50 ) for file_name in os.listdir(input_dir): file_path = os.path.join(input_dir, file_name) if not os.path.isfile(file_path): continue if file_name.endswith(".txt"): docs = TextLoader(file_path, encoding="utf-8").load() elif file_name.endswith(".pdf"): docs = PyPDFLoader(file_path).load() else: continue chunks = splitter.split_documents(docs) vectorstore.add_documents(chunks) print(f"已导入 {file_name}，切片数量 {len(chunks)}")

批量导入容易踩的坑是内存占用。大量 PDF 一次性加载可能把内存吃满。建议分批处理，每批处理 10 到 20 个文件就写入一次向量库，然后释放引用。如果向量库最终数据量很大，要考虑从 Chroma 迁移到 Milvus、Qdrant 这类更适合服务化部署的向量数据库。

6.3 批量问答任务

批量问答常见于“用知识库给一批历史问题自动打答案草稿”的场景。可以循环读取问题列表，逐条调用检索和生成接口，记录结果。注意两点：控制并发，避免瞬间打满显存或 API 配额；给每条任务加日志和失败重试。

import pandas as pd import requests df = pd.read_csv("questions.csv") api_url = "http://127.0.0.1:8000/api/rag/query" results = [] for question in df["question"]: try: resp = requests.post(api_url, json={"question": question}, timeout=30) data = resp.json() results.append({"question": question, "answer": data["answer"]}) except Exception as exc: results.append({"question": question, "answer": f"ERROR: {exc}"}) pd.DataFrame(results).to_csv("answers.csv", index=False, encoding="utf-8-sig")

批量任务失败时不要盲目重跑整个队列，建议把失败的问题单独记录，重试只针对失败项。如果接口一次性处理太多请求超时，可以在请求侧加timeout，在服务端加异步任务队列。

7. 资源占用与性能观察

RAG 的资源占用分三块：嵌入模型、向量数据库、大模型推理。

嵌入模型用于把文本转成向量，总体算力要求不高。小尺寸的 BGE 嵌入模型在 CPU 上可以运行，一篇文档的向量化耗时取决于文本长度和机器性能。向量数据库的索引在数据量小的时候内存占用很低，数据量大时主要看索引方式和向量维度。

大模型推理是资源大头。如果调用云端 API，本机只负责请求网络，资源占用很小；如果本地部署开源大模型，显存占用由模型参数量、量化精度、上下文长度共同决定。实际项目里，小团队建议先接云端大模型 API，验证业务效果，再考虑本地化部署。

性能观察的几个关键指标：

• 检索耗时：单次向量检索应该在几十到几百毫秒内，如果超过秒级，考虑索引问题或数据量过大。
• 重排序耗时：CrossEncoder 重排会额外增加耗时，候选集越大越慢。
• 生成耗时：大模型生成时间与输出长度强相关。
• 批处理吞吐量：批量文档导入时统计“每分钟切片数”和“每分钟向量化条数”。

降低资源占用的常用手段：

• 嵌入模型选小尺寸变体。
• 向量检索只取 Top K 小一点，重排序候选集控制在 20 到 50 条。
• 本地大模型使用量化版本，减少显存占用。
• 控制并发请求数，避免多个大模型推理任务同时打满显存。
• 批量导入时限制单批文件数量，防止内存溢出。

启动 API 服务后，建议观察端口占用。如果 8000 被占用，换一个端口：

uvicorn app:app --host 0.0.0.0 --port 8001

进程残留也会占用端口，排查时可以查看端口监听情况，找到残留进程后结束。

8. 常见问题与排查方法

9. 最佳实践与使用建议

第一，先小规模验证，再全量铺开。第一次测试只放 10 个文档进知识库，手工验证几十个问题，确认切块和检索效果后再导全量数据。一上来就导几千个文档，出了问题很难定位是切块问题还是文档本身问题。

第二，保留一套最小可运行配置。把文档目录、切块参数、嵌入模型、向量库路径、大模型接口配置全部参数化，放到配置文件里。换项目时只改配置，不改代码。

第三，知识库目录结构要清晰。建议这样组织文件：

rag_project/ ├── docs/ # 原始文档 ├── chunks/ # 切片中间结果，便于排查 ├── chroma_db/ # 向量库持久化目录 ├── output/ # 问答结果与日志 ├── app.py # API 服务 ├── ingest.py # 批量导入脚本 └── config.yaml # 配置文件

切片中间结果建议落盘一份，这样检索结果不对时可以直接查看切片内容，判断是切块问题还是检索问题。

第四，批量任务必须加日志和失败重试。每条导入或问答任务记录文件名、切片数、耗时、成功状态。失败任务单独保存，不要混在成功结果里。

第五，接口服务要做好访问控制。RAG API 接业务系统时，增加鉴权机制、限流策略和日志审计。尤其涉及企业内部资料时，不能把未授权文档暴露给所有调用方。

第六，涉及人脸、声音、版权素材等内容时，必须在导入知识库前确认授权。RAG 本身只是检索和生成，但知识库里的内容责任在搭建者，这一点不能省。

第七，发布前做一轮人工复核。用固定测试集跑一遍问答，检查格式、事实准确性、引用来源是否匹配。RAG 的答案质量是检索和生成共同决定的结果，不是模型单方面负责的。

10. 总结与下一步

RAG 是 AI 应用开发里确定性较高的技术路线。你不需要训练模型，核心工作量在数据处理、检索调优和工程化封装。从最小可用流水线起步，先跑通“文档切片、向量化、检索、生成”这条主链路，再逐步加混合检索、重排序、元数据过滤和查询重写，检索精度会明显改善。

最容易踩的坑有三个：第一是切块参数拍脑袋定，没有对照实验；第二是只靠向量检索，遇到型号、编号类问题召回失败；第三是批量导入时没有做内存控制，大文档直接撑爆进程。这三个问题在真实项目里出现频率很高，建议优先处理。

如果这篇对你有帮助，建议收藏备用。下一步可以做的事：把自己常用的技术文档或产品手册整理成一个小知识库，按本文第 4 节搭出最小流程，然后从第 5 节里挑两个高级检索手段加进去，对比加之前和加之后的答案质量差异。RAG 这个方向不复杂，但每一项细节都值得动手验证一遍。