企业官网建设流程全解析

1. 为什么客服机器人总在胡说八道

做过客服系统的人都有一个共同的痛：你辛辛苦苦搭了一个基于大模型的问答机器人，用户问“你们的退货政策是什么”，它张口就来一段听起来特别合理但完全不是你们公司规定的答案。用户拿着这个答案去找人工客服对质，人工客服一脸懵，最后投诉到你这儿来。

这个问题的根源不在于模型不够聪明，而在于它太聪明了——它太擅长“编”了。大语言模型的本质是一个概率续写机器，它根据上文预测下一个最可能出现的词，而不是根据事实回答问题。当它不知道答案时，它不会说“我不知道”，而是会生成一个“看起来最像正确答案”的文本。这就是所谓的幻觉问题。

RAG（Retrieval-Augmented Generation，检索增强生成）就是来解决这个问题的。它的核心思路特别朴素：既然模型不知道答案，那我就在它回答之前，先把相关的资料找出来塞给它，让它“看着资料回答”。就像开卷考试一样，你不需要把所有知识都背在脑子里，你只需要知道去哪本书的哪一页找答案就行。

我最近用 LangGraph + Milvus + Embedding 这套组合拳，完整搭了一个客服机器人，实测下来效果很稳。这篇文章我会把整个实现过程拆开讲清楚，包括为什么选这些工具、每一步的关键参数怎么定、踩过哪些坑、怎么排查问题。不管你是刚接触 RAG 的新手，还是已经做过一些 demo 想往工程化方向走的老手，应该都能从里面找到有用的东西。

2. RAG 整体架构与方案选型

2.1 RAG 到底在做什么

先把 RAG 的流程用最直白的话说一遍。用户问了一个问题，系统做三件事：

第一，把用户的问题转成一个向量（一串数字），这个过程叫 Embedding。第二，拿着这个向量去一个专门的数据库里找最相似的几段文本，这个数据库叫向量数据库，这个过程叫检索（Retrieval）。第三，把找到的文本和用户的问题拼在一起，交给大模型，让它基于这些文本生成回答，这个过程叫生成（Generation）。

听起来很简单对吧？但工程实现里的坑几乎全在细节里。比如文本怎么切分？切多大？向量用哪个模型转？相似度怎么算？检索出来多少条合适？检索结果怎么排序？这些问题每一个都会直接影响最终效果。

2.2 为什么选 LangGraph 而不是 LangChain 的 Chain

很多人做 RAG 的第一反应是用 LangChain 的 RetrievalQA 链，几行代码就能跑起来。我一开始也是这么做的，但很快就发现不够用。

LangChain 的 Chain 是线性的：检索 → 拼接 → 生成，一条路走到黑。但真实的客服场景远比这复杂。比如用户问“我上周买的那个红色的杯子能退吗”，这个问题需要先判断意图（是问退货政策还是查订单），然后可能需要多轮检索（先查退货政策，再查具体订单信息），最后还要判断检索到的内容是否足够回答问题，不够的话要触发二次检索或者直接告诉用户“我查不到”。

LangGraph 把整个流程建模成一个状态图（State Graph），每个节点是一个处理步骤，节点之间可以有条件分支和循环。这让我可以很自然地实现“检索 → 判断相关性 → 不够就改写查询再检索 → 够了就生成”这样的逻辑。而且 LangGraph 天然支持流式输出和中断恢复，对客服这种需要实时响应的场景很友好。

2.3 为什么选 Milvus 做向量数据库

向量数据库的选择其实挺多的，Milvus、Qdrant、Weaviate、Chroma 都能用。我选 Milvus 主要看中三点：

第一，性能。Milvus 底层是 C++ 写的，索引类型丰富，支持 IVF、HNSW、DiskANN 等多种索引。在百万级向量的场景下，查询延迟可以稳定在毫秒级。客服场景对响应速度要求高，这一点很关键。

第二，部署灵活。Milvus 支持三种部署模式：Milvus Lite（本地轻量版，适合开发和测试）、Standalone（单机版，适合中小规模生产）、Distributed（分布式版，适合大规模生产）。我开发阶段直接用 Milvus Lite，一个milvus_uri: str = "./data/milvus.db"就能在本地跑起来，不需要额外起 Docker 容器。上线的时候再换成 Standalone 或者连远程集群，代码几乎不用改。

第三，生态好。Milvus 对 LangChain 和 LangGraph 的支持很完善，有现成的 VectorStore 封装，省去了很多胶水代码。

2.4 Embedding 模型怎么选

Embedding 模型决定了检索的质量上限。我试过好几个模型，包括 OpenAI 的 text-embedding-ada-002、text-embedding-3-small，还有开源的 BGE 系列和 M3E 系列。

如果预算充足且对数据隐私要求不高，OpenAI 的 text-embedding-3-small 是省心之选，1536 维，效果稳定，价格也便宜。但如果你的客服数据涉及敏感信息不能出内网，那就得用开源模型本地部署。我实测下来，BGE-large-zh-v1.5 在中文客服场景下的表现很接近 OpenAI 的模型，而且完全本地跑，不用担心数据泄露。

选 Embedding 模型的时候有一个很容易忽略的点：维度。不同模型的输出维度不一样，ada-002 是 1536 维，BGE-large 是 1024 维，BGE-base 是 768 维。维度越高，表达能力越强，但存储和计算成本也越高。对于客服知识库这种规模（通常几千到几万条文档），768 到 1024 维完全够用，没必要追求最高维度。

3. 核心细节解析与实操要点

3.1 文档切分：RAG 效果的第一道分水岭

文档切分是 RAG 里最容易被低估的环节。很多人直接把整篇文档扔进去做 Embedding，结果检索出来的是一大坨文本，里面只有一小段是相关的，大模型被无关信息干扰，回答质量直线下降。

切分的核心原则是：每一块（chunk）应该是一个语义完整的片段，同时长度要适中。太短了语义不完整，太长了噪声太多。我的经验值是中文 300 到 500 字，英文 200 到 300 个 token。这个范围是经过多次实验得出的：低于 200 字，很多问题的答案会被切断；高于 800 字，检索精度明显下降。

切分策略上，我推荐用递归字符切分（RecursiveCharacterTextSplitter），它会按照段落、句子、逗号的优先级依次尝试切分，尽量保持语义完整性。关键参数是chunk_size和chunk_overlap。chunk_overlap我一般设成chunk_size的 10% 到 20%，比如 chunk_size=500，overlap=80。overlap 的作用是防止一个完整的答案刚好被切在边界上，导致两边都检索不到。

还有一个容易被忽略的细节：元数据。每个 chunk 除了文本内容，还应该带上来源信息，比如文档标题、章节、更新时间。这些元数据在检索后可以用来做过滤和排序，也能在最终回答里标注引用来源，增加可信度。

3.2 Milvus 集合设计与索引选择

Milvus 里的数据组织单位叫集合（Collection），类似于关系数据库里的表。创建集合的时候需要定义字段和索引。

字段方面，至少要有三个：id（主键）、embedding（向量字段）、text（原始文本）。如果要做元数据过滤，再加一个 metadata 字段（JSON 格式）。向量字段的维度必须和 Embedding 模型的输出维度一致，这个在创建集合的时候就要确定，后面改不了。

索引方面，Milvus 支持很多种，我常用的是 HNSW。HNSW 是一种基于图的索引，查询速度快，召回率高，适合大多数场景。关键参数是 M 和 efConstruction。M 控制每个节点的最大连接数，一般设 16 到 64，越大索引越精确但内存占用越高。efConstruction 控制建索引时的搜索范围，一般设 200 到 500。查询的时候还有一个 ef 参数，控制查询时的搜索范围，一般设 64 到 256。

如果数据量特别大（千万级以上），可以考虑 IVF_PQ 索引，它用乘积量化压缩向量，大幅降低内存占用，但会损失一些精度。客服场景一般用不上，HNSW 足够了。

3.3 相似度度量：余弦值还是内积

Milvus 支持三种相似度度量：欧氏距离（L2）、内积（IP）、余弦相似度（COSINE）。很多人搞不清楚什么时候用哪个。

简单来说，如果你用的是归一化后的向量（长度为 1），那么内积和余弦相似度是等价的。大多数 Embedding 模型输出的向量都已经归一化了，所以用 COSINE 或 IP 都行。但如果向量没有归一化，就必须用 COSINE，因为余弦相似度只看向量方向，不看长度。

我实测下来，对于文本 Embedding，COSINE 是最稳妥的选择。Milvus 里设置度量方式是在创建索引的时候指定的，一旦设定就不能改，所以要想清楚。

3.4 LangGraph 状态设计

LangGraph 的核心是状态（State）。状态是一个字典，在所有节点之间传递。每个节点接收当前状态，返回更新后的状态。

对于客服 RAG，我定义的状态包含这几个字段：question（用户问题）、rewritten_question（改写后的问题）、documents（检索到的文档列表）、answer（生成的回答）、need_retry（是否需要重新检索）。

节点方面，我设计了五个：改写查询（rewrite_query）、检索文档（retrieve）、评估相关性（grade_documents）、生成回答（generate）、判断是否需要重试（should_retry）。其中 should_retry 是一个条件边，根据评估结果决定是回到改写查询还是继续生成。

这个设计的精妙之处在于它形成了一个闭环：如果检索到的文档和问题不相关，系统会自动改写查询重新检索，而不是硬着头皮生成一个错误答案。这就是“不会胡说八道”的关键所在。

4. 实操过程与核心环节实现

4.1 环境准备与依赖安装

先把环境搭起来。我假设你用的是 Python 3.10 以上，pip 或者 conda 都行。

pip install langgraph langchain langchain-community pymilvus milvus-lite sentence-transformers fastapi uvicorn

如果你要用 OpenAI 的 Embedding 和 GPT，还需要装 openai 包，并设置 API Key。如果要用本地模型，sentence-transformers 就够了。

Milvus Lite 不需要额外安装服务，pymilvus 包里自带了。但要注意，Milvus Lite 只支持 Linux 和 macOS，Windows 上需要用 Docker 跑 Standalone 版本。在 Mac 上用 Docker 安装 Milvus 也很简单：

docker run -d --name milvus-standalone -p 19530:19530 -p 9091:9091 milvusdb/milvus:latest standalone

启动后用pymilvus连接localhost:19530就行。

4.2 知识库构建：从原始文档到向量

假设你的客服知识库是一堆 Markdown 或者 Word 文档。第一步是把它们读进来，切分，然后转成向量存到 Milvus。

from langchain_community.document_loaders import DirectoryLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_community.vectorstores import Milvus # 加载文档 loader = DirectoryLoader("./knowledge_base", glob="**/*.md") docs = loader.load() # 切分 splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=80, separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""] ) chunks = splitter.split_documents(docs) # Embedding 模型 embeddings = HuggingFaceEmbeddings( model_name="BAAI/bge-large-zh-v1.5", model_kwargs={"device": "cpu"}, encode_kwargs={"normalize_embeddings": True} ) # 存入 Milvus vector_store = Milvus.from_documents( chunks, embeddings, collection_name="customer_service_kb", connection_args={"uri": "./data/milvus.db"}, index_params={ "metric_type": "COSINE", "index_type": "HNSW", "params": {"M": 16, "efConstruction": 200} } )

这段代码里有几个关键点。separators的顺序很重要，中文标点要放在英文标点前面，否则中文句子会被切得很碎。normalize_embeddings=True确保向量归一化，这样 COSINE 和内积的结果一致。index_params里的metric_type一旦设定就不能改，所以要想清楚。

4.3 LangGraph 工作流搭建

接下来是核心部分：用 LangGraph 把整个 RAG 流程串起来。

from typing import TypedDict, List from langgraph.graph import StateGraph, END from langchain_core.documents import Document class RAGState(TypedDict): question: str rewritten_question: str documents: List[Document] answer: str need_retry: bool retry_count: int def rewrite_query(state: RAGState): # 用 LLM 改写查询，提升检索命中率 prompt = f"请将以下用户问题改写为更适合检索的形式，只输出改写后的问题：\n{state['question']}" rewritten = llm.invoke(prompt).content return {"rewritten_question": rewritten, "retry_count": state.get("retry_count", 0) + 1} def retrieve(state: RAGState): query = state.get("rewritten_question") or state["question"] docs = vector_store.similarity_search(query, k=5) return {"documents": docs} def grade_documents(state: RAGState): # 用 LLM 评估检索到的文档是否与问题相关 relevant_docs = [] for doc in state["documents"]: prompt = f"问题：{state['question']}\n文档：{doc.page_content}\n这个文档与问题相关吗？只回答是或否。" result = llm.invoke(prompt).content.strip() if "是" in result: relevant_docs.append(doc) need_retry = len(relevant_docs) == 0 and state.get("retry_count", 0) < 2 return {"documents": relevant_docs, "need_retry": need_retry} def generate(state: RAGState): context = "\n\n".join([doc.page_content for doc in state["documents"]]) prompt = f"""基于以下资料回答问题。如果资料中没有相关信息，请直接说"抱歉，我暂时没有找到相关信息"，不要编造。 资料： {context} 问题：{state['question']} 回答：""" answer = llm.invoke(prompt).content return {"answer": answer} def should_retry(state: RAGState): if state.get("need_retry"): return "rewrite_query" return END # 构建图 workflow = StateGraph(RAGState) workflow.add_node("rewrite_query", rewrite_query) workflow.add_node("retrieve", retrieve) workflow.add_node("grade_documents", grade_documents) workflow.add_node("generate", generate) workflow.set_entry_point("rewrite_query") workflow.add_edge("rewrite_query", "retrieve") workflow.add_edge("retrieve", "grade_documents") workflow.add_conditional_edges("grade_documents", should_retry, { "rewrite_query": "rewrite_query", END: "generate" }) workflow.add_edge("generate", END) app = workflow.compile()

这段代码的核心逻辑是：改写查询 → 检索 → 评估相关性 → 如果相关就生成，不相关就回到改写查询重新来。retry_count限制最多重试两次，防止死循环。

4.4 FastAPI 接口封装

最后用 FastAPI 把整个流程包成一个 HTTP 接口，方便前端调用。

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class QueryRequest(BaseModel): question: str class QueryResponse(BaseModel): answer: str sources: List[str] @app.post("/chat", response_model=QueryResponse) async def chat(request: QueryRequest): result = app_graph.invoke({"question": request.question}) sources = list(set([doc.metadata.get("source", "") for doc in result["documents"]])) return QueryResponse(answer=result["answer"], sources=sources)

启动命令：

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

这样前端就可以通过 POST 请求/chat来获取回答了。返回结果里带上sources，让用户知道答案是从哪来的，增加可信度。

5. 常见问题与排查技巧实录

5.1 检索不到相关内容怎么办

这是最常见的问题。用户问了一个问题，检索出来的文档完全不相关。排查思路分三步：

第一步，检查 Embedding 模型是否适合你的语言和领域。如果你用的是英文模型处理中文客服数据，效果肯定差。换 BGE 或者 M3E 这类中文模型试试。

第二步，检查切分粒度。如果 chunk 太大，一个 chunk 里包含多个主题，向量就会变得“模糊”，检索精度下降。试着把 chunk_size 调小到 300 左右看看效果。

第三步，检查查询改写。用户的问题往往很口语化，比如“我买的东西坏了咋整”，直接拿这个去检索可能效果不好。用 LLM 改写成“商品损坏 退换货政策”这样的形式，命中率会高很多。

5.2 大模型还是胡说八道怎么办

即使检索到了相关文档，大模型有时候还是会“自由发挥”。这时候要在 prompt 里加约束。我常用的模板是：

基于以下资料回答问题。如果资料中没有相关信息，请直接说“抱歉，我暂时没有找到相关信息”，不要编造。回答时请引用资料中的原文。

关键是要明确告诉它“不知道就说不知道”，并且给它一个具体的“不知道”的表达方式。另外，把 temperature 设成 0 或者 0.1，降低随机性。

5.3 Milvus 连接失败排查

Milvus Lite 用uri="./data/milvus.db"的时候，如果目录不存在会报错。确保./data目录已经创建。如果用 Docker 版，检查端口 19530 是否被占用，容器是否正常启动。

还有一个坑：Milvus Lite 不支持多进程同时写入。如果你用 uvicorn 起了多个 worker，可能会冲突。开发阶段用单 worker 就行，生产环境换 Standalone 版本。

5.4 响应速度太慢怎么优化

RAG 的延迟主要来自三部分：Embedding 计算、向量检索、LLM 生成。Embedding 和检索通常很快（毫秒级），大头在 LLM 生成。

优化方向有几个：一是用流式输出，让用户先看到部分结果；二是缓存高频问题的答案；三是把grade_documents里的逐条评估改成批量评估，减少 LLM 调用次数；四是如果检索质量已经很好，可以去掉评估环节，直接生成。

5.5 常见问题速查表

6. 工程化落地的几个关键决策

6.1 要不要做查询改写

查询改写会增加一次 LLM 调用，带来额外的延迟和成本。我的建议是：如果你的用户问题比较规范（比如都是“XX政策是什么”这种），可以不做改写。但如果用户问题很口语化、很短、或者包含指代（“那个东西能退吗”），那改写就很有必要。

一个折中方案是用轻量级的规则做预处理，比如把“咋整”“咋办”替换成“怎么办”，把“能退不”替换成“能否退货”。这样不增加 LLM 调用，也能提升检索效果。

6.2 检索多少条文档合适

k值的选择是一个权衡。k 太小，可能漏掉关键信息；k 太大，噪声太多，还会增加 LLM 的输入长度和成本。我的经验值是 3 到 5 条。如果知识库很大、问题很具体，可以设 3；如果知识库较小、问题比较宽泛，可以设 5 到 8。

还有一个技巧是用 MMR（最大边际相关性）检索，它会在相关性和多样性之间做平衡，避免检索出来的文档都是重复内容。LangChain 的 Milvus VectorStore 支持max_marginal_relevance_search方法。

6.3 怎么评估 RAG 效果

评估 RAG 效果不能只看“回答看起来对不对”，要有量化指标。我常用的两个指标是：

检索命中率：对于一组测试问题，人工标注每个问题的正确答案在哪个文档里，然后看检索结果是否包含这个文档。命中率低于 80% 就说明检索环节有问题。

回答准确率：对于一组测试问题，人工判断生成的回答是否正确。这个比较主观，但可以抽样评估。如果准确率低于 90%，就要检查是检索问题还是生成问题。

建议在开发阶段就建一个测试集，至少 50 个问题，覆盖常见场景和边界情况。每次调整参数后跑一遍测试集，看指标变化。

6.4 知识库更新怎么办

客服知识库不是一成不变的，政策会调整，产品会更新。Milvus 支持增量插入和删除。更新流程是：先把旧文档对应的向量删掉（按 id 或 metadata 过滤），然后插入新文档的向量。

如果更新频繁，建议加一个版本号字段，检索的时候只查最新版本。这样不用频繁删除，避免误删。

7. 我踩过的几个坑

第一个坑是 Embedding 模型和向量维度不匹配。我一开始用 BGE-base（768 维）建了集合，后来想换成 BGE-large（1024 维），结果发现 Milvus 集合的维度是固定的，改不了，只能删了重建。所以选模型的时候要想清楚，尽量一步到位。

第二个坑是 chunk_overlap 设得太小。我一开始设了 20，结果很多跨段落的答案被切断了，检索不到。后来调到 80 才解决。overlap 的钱不能省，它直接影响召回率。

第三个坑是忘了归一化。用 COSINE 度量的时候，如果向量没有归一化，结果会不准确。HuggingFaceEmbeddings 的normalize_embeddings=True一定要加上。

第四个坑是 LangGraph 的状态更新。LangGraph 的节点返回的字典是“增量更新”，不是“替换”。如果你返回{"documents": docs}，它会把原来的 documents 覆盖掉。但如果你返回{"answer": answer}，它只会更新 answer 字段，其他字段保持不变。这个机制要理解清楚，否则状态会乱。

第五个坑是 Milvus Lite 的并发限制。开发阶段用单进程没问题，但如果你用uvicorn --workers 4起多个 worker，Milvus Lite 会报锁冲突。生产环境一定要换 Standalone 版本。

8. 后续可以扩展的方向

这套 RAG 系统目前已经能稳定处理大部分客服问答了，但还有几个方向可以继续优化。

一是多路召回。目前只用向量检索，可以再加上关键词检索（BM25），两路结果融合排序。这样对于包含专有名词的问题，召回率会更高。

二是重排序（Rerank）。检索出来 top 20 条，然后用一个交叉编码器（Cross-Encoder）模型对这 20 条做精细排序，取 top 5 给 LLM。这样能显著提升检索精度，代价是增加一点延迟。

三是多模态支持。如果客服知识库里有图片（比如产品图、操作截图），可以考虑用多模态 Embedding 模型（如 CLIP）把图片也向量化，实现图文混合检索。不过这个复杂度比较高，建议先把文本场景做扎实。

四是对话历史管理。目前每次问答都是独立的，没有上下文。如果要支持多轮对话，需要在状态里加上历史消息，并且在检索的时候把历史信息也考虑进去。LangGraph 对多轮对话的支持很好，可以用 checkpointer 来持久化状态。

这套方案我在实际项目里跑了大半年，日均处理几千次咨询，回答准确率稳定在 95% 以上。最关键的经验就是：不要指望一次调优就完美，要建测试集、要量化评估、要持续迭代。RAG 不是一个“配好就完事”的系统，它是一个需要持续运营和优化的工程。