1. 为什么客服机器人总在胡说八道做过客服系统的人都有一个共同的痛你辛辛苦苦搭了一个基于大模型的问答机器人用户问“你们的退货政策是什么”它张口就来一段听起来特别合理但完全不是你们公司规定的答案。用户拿着这个答案去找人工客服对质人工客服一脸懵最后投诉到你这儿来。这个问题的根源不在于模型不够聪明而在于它太聪明了——它太擅长“编”了。大语言模型的本质是一个概率续写机器它根据上文预测下一个最可能出现的词而不是根据事实回答问题。当它不知道答案时它不会说“我不知道”而是会生成一个“看起来最像正确答案”的文本。这就是所谓的幻觉问题。RAGRetrieval-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_size500overlap80。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因为余弦相似度只看向量方向不看长度。我实测下来对于文本 EmbeddingCOSINE 是最稳妥的选择。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 和 macOSWindows 上需要用 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_size500, chunk_overlap80, separators[\n\n, \n, 。, , , , , , ] ) chunks splitter.split_documents(docs) # Embedding 模型 embeddings HuggingFaceEmbeddings( model_nameBAAI/bge-large-zh-v1.5, model_kwargs{device: cpu}, encode_kwargs{normalize_embeddings: True} ) # 存入 Milvus vector_store Milvus.from_documents( chunks, embeddings, collection_namecustomer_service_kb, connection_args{uri: ./data/milvus.db}, index_params{ metric_type: COSINE, index_type: HNSW, params: {M: 16, efConstruction: 200} } )这段代码里有几个关键点。separators的顺序很重要中文标点要放在英文标点前面否则中文句子会被切得很碎。normalize_embeddingsTrue确保向量归一化这样 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, k5) 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_modelQueryResponse) 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(answerresult[answer], sourcessources)启动命令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 常见问题速查表问题现象可能原因排查方法解决方案检索结果不相关Embedding 模型不匹配换模型测试改用中文模型如 BGE检索结果不相关chunk 过大检查 chunk_size调小到 300-500回答编造信息prompt 约束不足检查 prompt加“不知道就说不知道”回答编造信息temperature 过高检查 LLM 参数设为 0 或 0.1Milvus 连接失败目录不存在检查路径创建 data 目录Milvus 连接失败端口占用检查端口换端口或停冲突服务响应慢LLM 生成慢分段计时用流式输出或缓存响应慢评估环节耗时检查 grade_documents批量评估或去掉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-base768 维建了集合后来想换成 BGE-large1024 维结果发现 Milvus 集合的维度是固定的改不了只能删了重建。所以选模型的时候要想清楚尽量一步到位。第二个坑是 chunk_overlap 设得太小。我一开始设了 20结果很多跨段落的答案被切断了检索不到。后来调到 80 才解决。overlap 的钱不能省它直接影响召回率。第三个坑是忘了归一化。用 COSINE 度量的时候如果向量没有归一化结果会不准确。HuggingFaceEmbeddings 的normalize_embeddingsTrue一定要加上。第四个坑是 LangGraph 的状态更新。LangGraph 的节点返回的字典是“增量更新”不是“替换”。如果你返回{documents: docs}它会把原来的 documents 覆盖掉。但如果你返回{answer: answer}它只会更新 answer 字段其他字段保持不变。这个机制要理解清楚否则状态会乱。第五个坑是 Milvus Lite 的并发限制。开发阶段用单进程没问题但如果你用uvicorn --workers 4起多个 workerMilvus Lite 会报锁冲突。生产环境一定要换 Standalone 版本。8. 后续可以扩展的方向这套 RAG 系统目前已经能稳定处理大部分客服问答了但还有几个方向可以继续优化。一是多路召回。目前只用向量检索可以再加上关键词检索BM25两路结果融合排序。这样对于包含专有名词的问题召回率会更高。二是重排序Rerank。检索出来 top 20 条然后用一个交叉编码器Cross-Encoder模型对这 20 条做精细排序取 top 5 给 LLM。这样能显著提升检索精度代价是增加一点延迟。三是多模态支持。如果客服知识库里有图片比如产品图、操作截图可以考虑用多模态 Embedding 模型如 CLIP把图片也向量化实现图文混合检索。不过这个复杂度比较高建议先把文本场景做扎实。四是对话历史管理。目前每次问答都是独立的没有上下文。如果要支持多轮对话需要在状态里加上历史消息并且在检索的时候把历史信息也考虑进去。LangGraph 对多轮对话的支持很好可以用 checkpointer 来持久化状态。这套方案我在实际项目里跑了大半年日均处理几千次咨询回答准确率稳定在 95% 以上。最关键的经验就是不要指望一次调优就完美要建测试集、要量化评估、要持续迭代。RAG 不是一个“配好就完事”的系统它是一个需要持续运营和优化的工程。