这篇文章将从头到尾展示如何通过 RAG 技术 Chat Model 来实现智能客服或者其他能力的AI一步一步拆解每一个核心的实现技术。完成项目代码开源在 GiteeAI-Powered RAG-Based Customer Service System : 这是一个轻量级的 RAGRetrieval-Augmented Generation检索增强生成项目围绕智能客服场景实现了一个最小可运行示例。项目支持将文本知识上传到本地知识库使用向量检索召回相关内容再结合大模型生成回答。https://gitee.com/cce-debug/ai-powered-rag-based-customer-service-system一 需求分析这次的项目就以某商品的衣服智能客服为例进行构建该项目分为两大部分在线流程和离线流程部分。离线流程需要实现相对应本地数据库的构建即构建向量数据库在线部分实现的是人和AI之间的交互AI需要根据用户提问在本地知识库中检索相关文档进行回答用户问题。离线部分的本地数据库需要能够自由更新在线部分的AI需要基于本地知识生成答案。二 离线流程这里需要实现本地数据库的存储不和以往的数据库类似这里采用向量数据库存储好处就是当句子都被向量化之后语义之间的比配检索就只需要计算两个向量之间的余弦相似度就可以完成匹配这样在后期抽取用户相应知识的时候就比较方便了。2.1 余弦余弦相似度向量化的核心是把句子变成一串数字向量数字代表语义意思越像数字越接近。例如忘记密码怎么办→[0.98, 0.02, 0.05]我密码忘了怎么弄→[0.97, 0.03, 0.04]如何重置密码→[0.96, 0.04, 0.03]怎么下单购买→[0.05, 0.99, 0.02]意思一样所转化为的向量数字几乎一样但是意思不同向量数字几乎完全不同。计算公式如下结果范围是 [-1, 1] 。1完全一样0完全无关-1完全相反。向量数据库存储客服知识库其核心优势就在于将自然语言转换为数值化向量通过计算向量间的余弦相似度实现语义级的检索。相较于传统数据库依赖关键词精确匹配的方式向量数据库能够理解用户问句的真实意图能够大幅提升智能客服问答的准确率与用户体验。2.2 数据向量化在上面我们知道了当文本被向量化后可以很好通过语义进行抽取相关资料但是文本如何被向量化呢向量化的时候又怎么保证相同语义的数据被向量化后 “靠的近” 呢关键技术点就在Embedding模型 在这里不详细介绍Embedding模型具体的原理只是较为形象的讲解一下其主要实现的效果。我们可以把词向量模型Embedding 模型理解成一个翻译机输入Embedding 模型一句话Embedding 模型输出一串固定长度的数字比如 384 维、768 维、1024 维这个过程叫文本嵌入Text Embedding具体步骤如下“忘记密码怎么办” → 1 分词忘记 / 密码 / 怎么办 → 2 每个词得到向量 → 3 整句合成一个向量 [0.12, 0.55, 0.89, 0.21, ...] 共384个数字所以数据向量化这个过程我们通过 Embedding 模型来实现那 Embedding 模型又是如何被训练的呢因为Embedding 模型训练的时候就强制让 “意思相近” 的文本向量靠近。该模型的训练目标就是“相似的句子向量距离近不相似的句子距离远”。训练时用这种数据相似句对忘记密码怎么办密码忘了怎么找回→ 向量拉近不相似句对忘记密码怎么办怎么下单支付→ 向量推开模型通过大量数据学习自动把语义信息编码到数字里。具体的 Embedding 模型选择也有很多种不同的 Embedding 模型有不同的输出维度输出维度越高的模型其语义考察的会更加丰富更加准确。Embedding 模型的使用import os from pathlib import Path import requests from sentence_transformers import SentenceTransformer mirror_endpoint https://hf-mirror.com model_name BAAI/bge-small-zh-v1.5 model_dir Path(.hf-local-models) / BAAI__bge-small-zh-v1.5 required_files [ 1_Pooling/config.json, config.json, config_sentence_transformers.json, model.safetensors, modules.json, sentence_bert_config.json, special_tokens_map.json, tokenizer.json, tokenizer_config.json, vocab.txt, ] model_dir.mkdir(parentsTrue, exist_okTrue) for relative_path in required_files: local_path model_dir / relative_path if local_path.exists() and local_path.stat().st_size 0: continue local_path.parent.mkdir(parentsTrue, exist_okTrue) url f{mirror_endpoint}/{model_name}/resolve/main/{relative_path} print(fdownloading: {relative_path}) with requests.get(url, streamTrue, timeout(30, 300)) as response: response.raise_for_status() with open(local_path, wb) as file: for chunk in response.iter_content(chunk_size1024 * 1024): if chunk: file.write(chunk) # 加载本地模型 model SentenceTransformer(str(model_dir)) text 我遇到了404的错误怎么办 embedding model.encode(text) print(embedding) print(f导入成功向量维度{len(embedding)}) # 输出 512 即正常这里通过 huggingface 的镜像网站下载 BAAI/bge-small-zh-v1.5 这个词嵌入模型这个模型是北京智源研究院BAAI推出的轻量级、高性能中文文本嵌入Embedding模型专为中文语义检索、RAG、智能客服等场景设计是目前中文小模型里 “速度与精度” 平衡最好的选择之一。最后 text 我遇到了404的错误怎么办 就被向量化为了上述的 512 维的向量。2.3 向量数据库本地数据通过 Embedding 模型可以将对应文本转化为对应的向量那转化的向量如何存储这里使用Chroma向量数据库。 Chroma 是一款轻量级、开源、AI 原生的向量数据库主打极简开发、开箱即用非常适合 RAG、语义搜索、AI Agent 记忆等场景。下面介绍其使用方法2.3.1 向量数据库创建/连接# 初始化向量数据库创建/连接 vector_store Chroma( collection_namecollection1, # 类似裕数据库的表名 embedding_functionSentenceTransformerEmbeddings(model), # 向量化的模型 persist_directory./chroma_db, # 数据存储路径 )2.3.2 向向量数据库中添加数据① InMemoryVectorStore 实现loader CSVLoader(file_path./data/rag.csv, encodingutf-8) documents loader.load() vector_store InMemoryVectorStore( embeddingSentenceTransformerEmbeddings(model), ) text_splitter RecursiveCharacterTextSplitter( chunk_size100, chunk_overlap0, separators[\n\n, \n, , ], length_functionlen, ) split_texts text_splitter.split_documents(documents) vector_store.add_documents( documentssplit_texts, )上述实现方式是存储在内存中程序结束后存储的向量就会自动的被清楚。主要实现流程向量数据库的创建 vector_store 本地化的文本数据进行分段 text_splitter.split_documents 最后通过 vector_store.add_documents 将分段后的文本数据加入 vector_store在加入的时候由于add_documents 是 Iterator 的所以会一段一段的将分段后的文本数据类型是 document 转化为向量并存储到vector_store 中。每一个被分段的文本数据如下所示Document ( metadata { source : ./data/rag.csv , row : 0 }, page_content user_query : 如何安装Python的pandas库 )② 本地持久化存储# 1 定义向量数据库 vector_store Chroma( collection_name collection1, # 集合名称类似数据库中的表名 embedding_function SentenceTransformerEmbeddings(model), # 嵌入模型 persist_directory ./chroma_db, # 持久化数据库路径 ) # 2 分段 text_splitter RecursiveCharacterTextSplitter( chunk_size100, chunk_overlap0, separators[\n\n, \n, , ], length_functionlen, ) chunks text_splitter.split_documents(text) # 这里的text是本地文本数据 # 3 加入向量数据库 metadata { source: file_name, create_time: datetime.now().strftime(%Y-%m-%d %H:%M:%S), operator: operator_name } vector_store.add_texts( chunks, metadatas [metadata for _ in range(len(chunks))] )本地持久化存储使用 Chroma 来创建较比 InMemoryVectorStore 方式需要多提供 collection_name 类似于表名 和 persist_directory 持久化存储的路径 。2.4 本地数据上传去重防止本地数据重复上传至向量数据库所以再加入向量数据库前我们需要判断一下等待加入数据的数据是否已经存在于向量数据库中通过 MD5技术来实现。MD5 是一种哈希算法作用只有一个把任意长度的文本 / 文件转换成一串固定长度32 位的唯一字符串。特点输入一模一样 → 输出 MD5 一定一模一样输入差一个字 / 一个符号 → MD5 完全不同32 位字符串例如d41d8cd98f00b204e9800998ecf8427e一句话来说MD5 数据的唯一身份证号。这样的话我们再将本地文本存储到向量数据库的时候应该记录下来对应的MD5同时新文件上传的时候同样计算其对应的MD5并查看MD5是否存在于专门记录MD5值的文件中如存在则表示该文件已经被上传如不存在则表示该文件没有被上传。def get_string_md5(input_string:str, encodingutf-8): 获取字符串的md5值 :param input_string: 输入字符串 :return: md5值 str_bytes input_string.encode(encoding encoding) # 将字符串转换为字节 md5_obj hashlib.md5() # 得到md5对象 md5_obj.update(str_bytes) # 更新md5对象 md5_hex md5_obj.hexdigest() # 得到md5的十六进制值 return md5_hex # 返回md5的十六进制值通过上述代码即可获得对应字符串的MD5值。2.5 开发本地知识库文件上传界面import time import streamlit as st from konwledage_base import KnowledgeBaseService st.title(Dataset Upload Module) st.markdown(---) st.subheader(Intelligent customer service, exemplified by apparel product support) st.caption(Cai Cien) st.caption(CSDN blog: https://blog.csdn.net/2401_84080967?typeblog) st.markdown(---) with st.container(): uploader_file st.file_uploader( Please upload a txt file:, type[txt], accept_multiple_filesFalse, ) user_name st.text_input(Please input your name:) btn st.button(Upload Data) if service not in st.session_state: st.session_state[service] KnowledgeBaseService() if btn: if user_name is None or user_name.strip() or uploader_file is None: st.warning(Please add name and upload a txt file) else: file_details { filename: uploader_file.name, filetype: uploader_file.type, filesize: uploader_file.size / 1024 } st.write(file_details) text uploader_file.getvalue().decode(utf-8) with st.spinner(UPLOADING...): time.sleep(2) result st.session_state[service].upload_by_str(text, file_details[filename], user_name) st.write(result)这里使用 streamlit 便捷开发streamlit适合数据科学家、AI 工程师快速做原型、仪表盘、模型 Demo、内部小工具几天内搞定。不适合通用网站、高并发服务、复杂前端、生产级大型应用。三 在线流程在线流程中 AI 需要基于用户提问和本地知识生成答案所以在线流程中的核心技术就是 RAG 技术当接受到用户问题后首先要进行语义理解在本地向量数据库中查找相对应的知识并抽取知识将抽取到的知识和用户提问构建成对应的提示词输入给 AI 进行回答。3.1 RAG 技术的实现RAG 技术的实现也就是要根据用户提问去本地数据库抽取对应的知识其实这一步很简单如下from langchain_chroma import Chroma from sentence_transformers import SentenceTransformer import config_data as config from konwledage_base import SentenceTransformerEmbeddings class VectorStoreService: def __init__(self): self.model_dir config.embedding_model_dir self.model SentenceTransformer(self.model_dir) self.embeddings SentenceTransformerEmbeddings(self.model) self.vector_store Chroma( collection_nameconfig.collection_name, embedding_functionself.embeddings, persist_directoryconfig.persist_directory ) def get_retriever(self): similarity_threshold config.similarity_threshold retriever self.vector_store.as_retriever(search_kwargs{k: similarity_threshold}) # 向量库转换成检索器 return retriever # test if __name__ __main__: service VectorStoreService() retriever service.get_retriever() results retriever.invoke(实施流程中怎么定义学生模型的结构参数空间与部署约束) print(results)在 VectorStoreService 中定义 get_retriever 函数用于得到对应向量数据库的检索器 retriever 向量数据库对象 vector_store 有一个 as_retriever 来定义其对应的检索器 retriever。当创建 VectorStoreService 对象的时候其通过 service.get_retriever() 就得到了对应向量数据库的检索器这时候调用 invoke 方法即可完成对应的检索。输出如下所示这样 RAG 技术就已经实现了可以成功的根据用户提问在向量数据库中抽取出对应的资料。3.2 构建 RAG 后的提示词模板self.prompt_template ChatPromptTemplate.from_messages([ {role: system, content: 以我提供的参考资料为主简洁和专业的回答用户问题不要编造信息。参考资料{reference}}, {role: system, content: 并且提供用户的对话历史记录如下所示}, MessagesPlaceholder(variable_namehistory), {role: user, content: 用户提问{question}}, ])reference 将被补全通过 RAG 得到的资料MessagesPlaceholder(variable_namehistory), 将被补全用户的历史聊天记录question 将被补全用户提问当这三个参数都被补全的话提示词就被构建好了就可以喂给 AI 模型进行回复了。3.3 构建完整服务链class RAGService: def __init__(self): self.vector_store_service VectorStoreService() self.prompt_template ChatPromptTemplate.from_messages([ {role: system, content: 以我提供的参考资料为主简洁和专业的回答用户问题不要编造信息。参考资料{reference}}, {role: system, content: 并且提供用户的对话历史记录如下所示}, MessagesPlaceholder(variable_namehistory), {role: user, content: 用户提问{question}}, ]) self.chat_model ChatModelService().chat self.chain self._get_chain() def _get_chain(self): retriever self.vector_store_service.get_retriever() def format_document(docs: list[Document]): if not docs: return 无相关参考资料 formatted_str for doc in docs: formatted_str f文档内容{doc.page_content}\n文档元数据{doc.metadata}\n\n return formatted_str chain ( { history: itemgetter(history), question: itemgetter(question), reference: itemgetter(question) | retriever | format_document, } | self.prompt_template | self.chat_model | StrOutputParser() ) conversation_chain RunnableWithMessageHistory( chain, get_history, input_messages_keyquestion, history_messages_keyhistory, ) return conversation_chain if __name__ __main__: session_config { configurable: { session_id: user_001 } } print(type(RAGService().chain)) result RAGService().chain.invoke({question: 什么是RAG}, configsession_config) print(result)当 RAGService().chain.invoke({question: 什么是RAG}, configsession_config) 的时候由于对普通链进行了加强 RunnableWithMessageHistory 使得链条具有记忆功能同时当链条加强为带有记忆功能后其 invoke 用户问题的时候需要传入的是字典而非单个 str 。conversation_chain 首先得到列表 ({question: 什么是RAGhistory: [ HumanMessage()AIMessage() ]}然后传到基础链{ history: [HumanMessage(), AIMessage()......] question: 什么是RAG reference: 什么是RAG | retriever | format_document, } | self.prompt_template | self.chat_model | StrOutputParser()reference 中入链的是 retriever 通过构建的检索器在相关的数据库中检索资料将检索到的资料通过format_document整理格式reference 得到的是 Document 对象。当以下{ history: [HumanMessage(), AIMessage()] question: 什么是RAG reference: 什么是RAG | retriever | format_document, }被填充完成后就可以被送入 prompt_template 得到完整提示词模板最后送入模型。invoke的时候填入的是config那如何通过config自动得到history首先RunnableWithMessageHistory类会自动的将config中的 session_id 传递给 get_history函数得到历史记录需要注意的是get_history 是我们自己实习的函数名也可以自己任取RunnableWithMessageHistory类只是默认将 session_id 作为参数传递给get_history函数。3.4 历史记录存储怎么实现conversation_chain RunnableWithMessageHistory( chain, get_history, # 你的本地文件存储 input_messages_keyquestion, history_messages_keyhistory, )这个组件会自动做 3 件事每次 invoke 前 → 自动从文件读取 history 自动将 session_id 传入 get_history 函数把 history 塞进 prompt每次 invoke 后 → 自动把新消息写入文件LangChain 的RunnableWithMessageHistory是自动帮你存、自动帮你取的组件不需要手动调用 add_message。但是 get_history 必须自己实现import json import os from typing import Sequence from langchain_core.chat_history import BaseChatMessageHistory from langchain_core.messages import BaseMessage, message_to_dict, messages_from_dict BASE_DIR os.path.dirname(os.path.abspath(__file__)) DEFAULT_HISTORY_DIR os.path.join(BASE_DIR, chat_history) def get_history(session_id: str) - BaseChatMessageHistory: # 每个 session_id 对应一个本地 JSON 文件 return FileChatMessageHistory(session_idsession_id, storage_pathDEFAULT_HISTORY_DIR) class FileChatMessageHistory(BaseChatMessageHistory): def __init__(self, session_id: str, storage_path: str): self.session_id session_id self.storage_path storage_path os.makedirs(self.storage_path, exist_okTrue) self.file_path os.path.join(self.storage_path, f{self.session_id}.json) property def messages(self) - list[BaseMessage]: try: with open(self.file_path, r, encodingutf-8) as f: messages_dict json.load(f) return messages_from_dict(messages_dict) except (FileNotFoundError, json.JSONDecodeError): return [] def add_messages(self, messages: Sequence[BaseMessage]) - None: all_messages list(self.messages) all_messages.extend(messages) messages_dict [message_to_dict(msg) for msg in all_messages] with open(self.file_path, w, encodingutf-8) as f: json.dump(messages_dict, f, ensure_asciiFalse, indent2) # 兼容单条消息写入 def add_message(self, message: BaseMessage) - None: self.add_messages([message]) def clear(self) - None: with open(self.file_path, w, encodingutf-8) as f: json.dump([], f, ensure_asciiFalse)3.5 用户交互界面实现import streamlit as st import time from rag import RAGService st.title(RAG AI) st.divider() prompt st.chat_input(请输入您的问题) if rag_service not in st.session_state: st.session_state[rag_service] RAGService() if message not in st.session_state: st.session_state[message] [{role: assistant, content: 你好我是一个基于检索增强生成RAG的AI助手请问有什么可以帮助你的}] # 一启动 将历史记录写入聊天界面 for msg in st.session_state[message]: st.chat_message(msg[role]).write(msg[content]) if prompt: cache_list [] st.chat_message(user).write(prompt) st.session_state[message].append({role: user, content: prompt}) with st.spinner(Retrieval-Augmented Generation AI is answering...): time.sleep(2) res_stream st.session_state[rag_service].chain.stream({question: prompt}, config{session_id: user_001}) def capture(generator, cache_list): for item in generator: cache_list.append(item) yield item st.chat_message(assistant).write_stream(capture(res_stream, cache_list)) st.session_state[message].append({role: assistant, content: .join(cache_list)})四 总结到这里一套完整可用的 RAG 智能客服系统就全部实现完成了。从本地知识库搭建、文本向量化、文档分块、MD5 去重防止重复导入到在线查询时的语义检索、提示词构造、大模型回答再到对话历史本地保存和简单的交互界面我们把 RAG 里最核心的流程都走了一遍代码也都完整落地。整个项目用的都是轻量开源工具环境简单、上手快既适合用来入门理解 RAG 工作原理也可以直接改成自己项目里的知识库问答、智能客服等实用功能。希望这篇实战教程能帮你真正动手跑通一套 RAG 系统让大模型不再只会 “凭空回答”而是能基于你自己的业务数据给出更可靠、更贴合场景的答案。在此基础上你还可以继续扩展支持 PDF/Excel 等更多文件格式、更换更强的向量模型、优化流式回复效果、增加用户权限控制等一步步把小 Demo 打磨成更成熟的 AI 应用。本文参考链接RAG项目-01、RAG项目案例介绍_哔哩哔哩_bilibilihttps://www.bilibili.com/video/BV1yjz5BLEoY?spm_id_from333.788.videopod.episodesvd_sourcef7674444cfdd21d11e002b2ea6707abfp46