Janus-Pro-7B构建智能客服：基于MySQL知识库的精准问答

Janus-Pro-7B构建智能客服基于MySQL知识库的精准问答你是不是也遇到过这种情况公司产品更新了客服团队需要花大量时间重新学习FAQ和手册客户问个稍微复杂点的问题客服就得翻半天文档效率低不说体验还不好。传统的客服系统要么是简单的关键词匹配答非所问要么就是全靠人工成本高响应慢。现在我们可以换个思路了。今天要聊的就是用Janus-Pro-7B这个大语言模型结合我们最熟悉的MySQL数据库搭建一个能“读懂”问题、从知识库里“找”答案、再“组织”成自然回复的智能客服。这背后的核心就是检索增强生成简称RAG。简单说就是不让模型凭空想象而是让它先查资料再根据查到的资料来回答。这样既能利用大模型强大的语言理解能力又能保证回答的准确性和时效性特别适合客服、技术支持这类对准确性要求高的场景。下面我就带你一步步看看怎么把这个想法落地。1. 为什么选择Janus-Pro-7B和MySQL在动手之前我们得先搞清楚手里的“工具”到底好不好用。选Janus-Pro-7B和MySQL这个组合不是拍脑袋决定的而是它们各自的特点正好能补上智能客服的几个关键短板。1.1 Janus-Pro-7B不只是个“聊天机器人”Janus-Pro-7B是一个7B参数规模的开源大语言模型。你可能觉得现在动辄几百B的模型那么多为啥选个7B的原因很简单够用、好用、省资源。对于智能客服场景我们最看重的不是模型能不能写诗或者编故事而是它能不能准确理解用户的意图并且严格按照我们给的信息来回答。Janus-Pro-7B在指令遵循和上下文理解方面表现不错这意味着它能比较好地执行“根据以下资料回答问题”这样的任务。而且7B的规模对于大多数企业来说部署和推理的成本是相对友好的。你不需要准备一堆昂贵的显卡在普通的云服务器上就能跑起来响应速度也足够快不会让用户等得不耐烦。1.2 MySQL老伙计的新角色说到MySQL搞技术的朋友太熟悉了不就是个存数据的数据库嘛。但在我们的RAG方案里它扮演的是“知识大脑”的角色。产品说明书、FAQ问答、故障处理流程、历史工单……这些结构化和非结构化的文本都可以整理好存进MySQL。它的优势很明显稳定可靠经过无数项目验证不用担心数据丢了或者服务挂了。查询高效配合索引能快速从海量知识条目中检索出相关的内容。生态成熟各种语言都能方便地连接和操作工具链完善。成本可控开源免费对于初创公司或预算有限的团队非常友好。把MySQL作为知识库意味着你可以用最熟悉的方式去管理你的客服知识增删改查都和以前一样学习成本几乎为零。1.3 RAG让准确性和可控性兼得最后说说把这两者粘合在一起的“胶水”——检索增强生成。 大模型有个通病容易“胡言乱语”或者知识过时。RAG的思路就是先检索再生成。检索当用户提出一个问题系统不是直接把问题扔给模型而是先用这个问题作为“线索”去MySQL知识库里搜索最相关的几段资料。增强把搜索到的这些资料和用户的原始问题一起打包形成一个“增强版”的提示交给模型。生成模型收到的指令类似于“请根据下面提供的产品资料回答用户的问题。”这样模型生成答案时就有了依据和边界大大提高了回答的准确性和可信度。这个流程完美解决了智能客服既要“智能”理解自然语言又要“靠谱”回答有据可查的核心矛盾。2. 搭建你的智能客服知识库巧妇难为无米之炊再聪明的模型没有高质量的知识库也是白搭。这一步的目标就是把散落在各处的产品知识变成模型能方便“查阅”的格式。2.1 知识原材料从哪里来别把这事想复杂了知识就在你身边产品帮助文档/用户手册这是最权威、最系统的知识来源。FAQ常见问题列表市场、客服部门整理的客户常问问题及答案。历史客服对话记录脱敏后这里面包含了用户真实的提问方式和客服的最佳实践回答。社区论坛精华帖用户之间的讨论往往能覆盖一些意想不到的角落。产品更新日志新功能、已修复的问题这些信息对客服至关重要。把这些文档收集起来保存成.txt或.md这样的纯文本格式准备下一步处理。2.2 安装和配置MySQL如果你的服务器上还没有MySQL安装过程很简单。这里以Ubuntu系统为例其他系统也大同小异。首先更新软件包列表并安装MySQL服务器sudo apt update sudo apt install mysql-server -y安装完成后运行安全安装脚本设置root密码并移除一些不安全默认设置sudo mysql_secure_installation接着登录MySQL为我们的智能客服项目创建一个专用的数据库和用户-- 登录MySQL这里用root或者你有其他有权限的用户也行 sudo mysql -- 创建一个叫smart_customer_service的数据库 CREATE DATABASE smart_customer_service CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; -- 创建一个新用户比如叫cs_admin并设置密码请把your_strong_password换成你自己的密码 CREATE USER cs_adminlocalhost IDENTIFIED BY your_strong_password; -- 给这个用户授予对新数据库的全部权限 GRANT ALL PRIVILEGES ON smart_customer_service.* TO cs_adminlocalhost; -- 让权限生效 FLUSH PRIVILEGES; -- 退出 EXIT;现在你的知识库“房子”就建好了。2.3 设计知识存储表知识库怎么建表核心思想是把大段的文档“切碎”变成一小段一小段有意义的文本块这样检索起来更精准。我们创建一个简单的表-- 使用我们刚创建的数据库 USE smart_customer_service; -- 创建知识片段表 CREATE TABLE knowledge_chunks ( id INT AUTO_INCREMENT PRIMARY KEY, -- 文本内容本身 content TEXT NOT NULL, -- 可以存一个向量用于相似度搜索初期我们可以先用文本匹配所以这列可以先留着 embedding_vector BLOB, -- 这段内容来自哪个文档方便溯源 source_document VARCHAR(255), -- 其他元数据比如文档类型、产品版本等 metadata JSON, -- 记录创建时间 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COLLATEutf8mb4_unicode_ci;这个knowledge_chunks表就是我们知识库的核心。content字段存放切分后的文本块source_document记录来源metadata可以用JSON格式存一些额外信息比如{product: 旗舰版, version: 2.1}。2.4 知识的预处理与入库有了表下一步就是把收集的文本“喂”进去。你不能把一整本100页的手册直接存成一条记录那样检索效率极低。我们需要进行文本分割。这里介绍一个简单实用的方法按固定长度重叠分割。用Python可以轻松实现import pymysql from typing import List import json # 连接到MySQL conn pymysql.connect( hostlocalhost, usercs_admin, passwordyour_strong_password, # 换成你的密码 databasesmart_customer_service, charsetutf8mb4 ) cursor conn.cursor() def split_text_into_chunks(text: str, chunk_size: int 500, chunk_overlap: int 100) - List[str]: 将长文本分割成固定大小的块块之间有重叠避免语义被切断。 :param text: 输入文本 :param chunk_size: 每个块的最大字符数 :param chunk_overlap: 块之间重叠的字符数 :return: 文本块列表 chunks [] start 0 text_length len(text) while start text_length: end start chunk_size chunk text[start:end] chunks.append(chunk) # 移动起始位置减去重叠部分确保上下文连贯 start chunk_size - chunk_overlap return chunks # 假设我们有一个产品手册的文本 product_manual_text 【产品X使用手册 V2.0】 第一章快速入门 1.1 开箱与安装 打开包装盒内含主机、电源适配器、数据线和使用说明书。将电源适配器连接至主机背部电源接口然后接通电源。长按顶部电源键3秒指示灯亮起即表示开机成功。 1.2 设备连接 首次开机后设备将进入配网模式。请打开手机App扫描设备底部的二维码按照App指引完成Wi-Fi网络配置。配置成功后设备指示灯将变为常亮蓝色。 ... # 分割文本 text_chunks split_text_into_chunks(product_manual_text, chunk_size300, chunk_overlap50) # 将分割后的块存入数据库 for i, chunk in enumerate(text_chunks): # 准备元数据 meta json.dumps({ document_type: user_manual, product: Product X, version: 2.0, section: Quick Start }) sql INSERT INTO knowledge_chunks (content, source_document, metadata) VALUES (%s, %s, %s) cursor.execute(sql, (chunk, Product_X_Manual_V2.0.txt, meta)) conn.commit() cursor.close() conn.close() print(f成功导入 {len(text_chunks)} 个知识块。)这样你的非结构化文本知识就变成了数据库里一条条结构化的、易于检索的记录。这是整个智能客服系统“靠谱”的基石。3. 核心引擎检索与生成的联动知识库准备好了现在我们来打造系统的“大脑”——也就是把用户问题、知识检索和答案生成串联起来的核心流程。这部分代码会稍微多一点但我会一步步拆开讲。3.1 搭建Janus-Pro-7B服务环境首先我们需要让Janus-Pro-7B模型跑起来并提供一个API接口供我们的系统调用。这里我们用比较流行的FastAPI和Transformers库来实现一个简单的模型服务。# model_server.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline import torch import logging # 配置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) # 定义请求和响应的数据格式 class QueryRequest(BaseModel): prompt: str # 用户的问题已经和检索到的知识拼接好了 max_length: int 512 class QueryResponse(BaseModel): answer: str processing_time: float # 初始化FastAPI应用 app FastAPI(titleJanus-Pro-7B智能客服API) # 加载模型和分词器这里假设模型已经下载到本地路径 ./janus-pro-7b MODEL_PATH ./janus-pro-7b logger.info(f正在加载模型: {MODEL_PATH}) tokenizer AutoTokenizer.from_pretrained(MODEL_PATH) model AutoModelForCausalLM.from_pretrained( MODEL_PATH, torch_dtypetorch.float16, # 使用半精度减少内存占用 device_mapauto # 自动分配到可用的GPU/CPU ) logger.info(模型加载完毕。) # 创建一个文本生成的pipeline generator pipeline( text-generation, modelmodel, tokenizertokenizer, device0 if torch.cuda.is_available() else -1 ) app.post(/generate, response_modelQueryResponse) async def generate_answer(request: QueryRequest): 接收增强后的提示词生成回答。 import time start_time time.time() try: # 使用模型生成文本 generated_sequences generator( request.prompt, max_lengthrequest.max_length, do_sampleTrue, # 启用采样使生成结果更多样 temperature0.7, # 控制随机性0.7是个比较平衡的值 top_p0.9, # 核采样进一步控制多样性 num_return_sequences1 ) generated_text generated_sequences[0][generated_text] # 生成的文本包含了我们的提示词需要将其剥离只取模型新生成的部分作为答案 # 这里简单处理如果生成的文本以提示词开头则去掉提示词部分。 if generated_text.startswith(request.prompt): answer generated_text[len(request.prompt):].strip() else: answer generated_text.strip() # 如果模型没完全复现提示词则取全部 processing_time time.time() - start_time logger.info(f请求处理完成耗时: {processing_time:.2f}秒) return QueryResponse(answeranswer, processing_timeprocessing_time) except Exception as e: logger.error(f生成答案时出错: {e}) raise HTTPException(status_code500, detailf内部服务器错误: {str(e)}) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)把上面的代码保存为model_server.py然后在你的服务器上运行python model_server.py。现在你的模型就有了一个HTTP接口地址是http://你的服务器IP:8000后面我们会调用它。3.2 从知识库中检索相关信息当用户提出一个问题比如“我的设备连不上Wi-Fi怎么办”我们首先要做的就是去MySQL知识库里找相关的资料。单纯的文本匹配比如用LIKE ‘%Wi-Fi%’效果很差。我们需要更智能的检索。这里我们可以先用一个简单但有效的方法基于TF-IDF和余弦相似度的文本检索。虽然不如向量检索高级但对于很多场景已经够用且实现简单。# retriever.py import pymysql import jieba # 用于中文分词如果是英文知识库可以用nltk from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity import numpy as np import json class KnowledgeRetriever: def __init__(self, db_config): self.conn pymysql.connect(**db_config) self.cursor self.conn.cursor(pymysql.cursors.DictCursor) self.vectorizer TfidfVectorizer(tokenizerjieba.lcut, max_features5000) # 使用jieba分词 self.knowledge_data None self.tfidf_matrix None def load_knowledge(self): 从数据库加载所有知识块 sql SELECT id, content, metadata FROM knowledge_chunks self.cursor.execute(sql) self.knowledge_data self.cursor.fetchall() print(f已加载 {len(self.knowledge_data)} 条知识记录。) # 准备文本用于训练TF-IDF corpus [item[content] for item in self.knowledge_data] self.tfidf_matrix self.vectorizer.fit_transform(corpus) def retrieve(self, query: str, top_k: int 3): 检索与查询最相关的top_k个知识块。 :param query: 用户查询 :param top_k: 返回最相关的条数 :return: 相关度最高的知识块列表 if self.knowledge_data is None: self.load_knowledge() # 将用户查询转换为TF-IDF向量 query_vec self.vectorizer.transform([query]) # 计算查询向量与所有知识块向量的余弦相似度 similarities cosine_similarity(query_vec, self.tfidf_matrix).flatten() # 获取相似度最高的top_k个索引 top_indices similarities.argsort()[-top_k:][::-1] # 返回对应的知识块 results [] for idx in top_indices: results.append({ content: self.knowledge_data[idx][content], similarity: float(similarities[idx]), metadata: json.loads(self.knowledge_data[idx][metadata]) if self.knowledge_data[idx][metadata] else {} }) return results def close(self): self.cursor.close() self.conn.close() # 使用示例 if __name__ __main__: db_config { host: localhost, user: cs_admin, password: your_strong_password, database: smart_customer_service, charset: utf8mb4 } retriever KnowledgeRetriever(db_config) retriever.load_knowledge() user_query 设备无法连接Wi-Fi指示灯一直在闪烁黄色怎么办 relevant_chunks retriever.retrieve(user_query, top_k2) print(f针对问题: {user_query}) print(f检索到 {len(relevant_chunks)} 条相关知识点:) for i, chunk in enumerate(relevant_chunks): print(f\n--- 相关知识点 {i1} (相似度: {chunk[similarity]:.3f}) ---) print(chunk[content][:200] ...) # 打印前200个字符 retriever.close()这段代码建立了一个检索器它会计算用户问题与知识库中每段文本的相似度然后把最相关的几条找出来。这样我们就完成了“检索”这一步。3.3 构建提示词让模型“有据可依”检索到相关资料后我们不能直接把资料和问题扔给模型。需要精心构造一个“提示词”告诉模型应该怎么做。这是决定生成质量的关键一步。# prompt_builder.py def build_rag_prompt(user_query: str, retrieved_chunks: list) - str: 构建RAG提示词。 :param user_query: 用户原始问题 :param retrieved_chunks: 检索到的相关知识块列表 :return: 构造好的完整提示词 # 1. 系统指令设定模型角色和回答规则 system_instruction 你是一个专业、友好的智能客服助手。请严格根据用户提供的“参考资料”来回答问题。 如果参考资料中的信息足以回答问题请基于资料组织语言给出准确、清晰、有帮助的回答。 如果参考资料中的信息不足以完全回答问题你可以结合自己的知识进行补充但必须明确指出哪些信息来自资料哪些是你的补充。 如果参考资料完全无法回答用户的问题请礼貌地告知用户你暂时无法回答这个问题并建议其通过其他渠道如人工客服获取帮助。 请用中文回答语气亲切自然。 # 2. 整合检索到的资料 context_str \n\n.join([f[资料片段 {i1}]:\n{chunk[content]} for i, chunk in enumerate(retrieved_chunks)]) # 3. 组合成最终提示词 final_prompt f{system_instruction} 以下是回答用户问题所需的参考资料 {context_str} 用户的问题 {user_query} 请根据以上资料用中文回答用户的问题 return final_prompt # 使用示例 if __name__ __main__: # 假设这是检索到的结果 mock_chunks [ {content: 当设备指示灯闪烁黄色时表示设备处于配网模式或网络连接异常。请检查路由器是否正常工作并确保输入的Wi-Fi密码正确。}, {content: 重置网络配置的方法长按设备背部的复位键10秒直到指示灯变为红色后松开。设备将重启并清除网络配置请重新使用App进行配网。} ] user_question 我的设备黄灯一直闪连不上网咋弄 prompt build_rag_prompt(user_question, mock_chunks) print(构造的提示词\n) print(- * 50) print(prompt) print(- * 50)看看生成的提示词它明确告诉模型你的角色是客服必须根据我给的材料来回答材料是什么用户问题是什么。这样就把模型“框”在了正确的轨道上。3.4 组装完整问答流程最后我们把检索、提示词构建、模型调用这三个模块串起来形成一个完整的智能客服问答函数。# customer_service_agent.py import requests import json from retriever import KnowledgeRetriever from prompt_builder import build_rag_prompt class CustomerServiceAgent: def __init__(self, db_config, model_api_urlhttp://localhost:8000/generate): self.retriever KnowledgeRetriever(db_config) self.model_api_url model_api_url def answer_question(self, user_query: str): 智能客服问答主函数。 1. 检索相关知识 2. 构建提示词 3. 调用模型生成答案 print(f用户提问: {user_query}) # 1. 检索 print(正在从知识库检索相关信息...) relevant_chunks self.retriever.retrieve(user_query, top_k3) print(f检索到 {len(relevant_chunks)} 条相关信息。) if not relevant_chunks: return 抱歉我暂时没有找到相关问题的解决方案。建议您查看产品手册或联系人工客服获取帮助。 # 2. 构建提示词 prompt build_rag_prompt(user_query, relevant_chunks) # 3. 调用模型API print(正在生成回答...) try: response requests.post( self.model_api_url, json{prompt: prompt, max_length: 600}, timeout30 # 设置超时时间 ) response.raise_for_status() # 检查HTTP错误 result response.json() answer result[answer] print(f回答生成完成耗时: {result[processing_time]:.2f}秒) return answer except requests.exceptions.RequestException as e: print(f调用模型API失败: {e}) # 降级方案如果模型服务挂了可以返回检索到的最相关的一条内容 return f根据知识库相关信息如下\n{relevant_chunks[0][content][:300]}... def close(self): self.retriever.close() # 主程序入口 if __name__ __main__: # 配置你的数据库连接信息 db_config { host: localhost, user: cs_admin, password: your_strong_password, database: smart_customer_service, charset: utf8mb4 } agent CustomerServiceAgent(db_config) # 示例问题 questions [ 产品怎么开机, 忘记管理员密码了怎么办, 设备指示灯红色常亮是什么意思 ] for q in questions: print(\n *50) answer agent.answer_question(q) print(f\n智能客服回答: {answer}) agent.close()运行这个customer_service_agent.py你的智能客服就活过来了它会自动完成从理解问题、查找知识到生成回答的全过程。4. 让系统更智能优化与实践建议基本的流程跑通了但要让这个智能客服真正好用能扛起线上服务的担子还需要在一些细节上打磨。这里分享几个关键的优化方向和实践中的经验。4.1 提升检索精度从文本到向量我们之前用的TF-IDF检索对于简单的场景没问题但它理解不了语义。比如用户问“怎么开机”知识库里写的是“长按电源键启动”TF-IDF可能因为词汇不匹配而检索不到。更高级的方法是使用文本向量模型。你可以把知识库里的每段文本以及用户的问题都通过一个模型比如BGE、text2vec等转换成高维向量一堆数字。检索时不再计算文字匹配度而是计算向量之间的相似度比如余弦相似度。这种方法能更好地理解“开机”和“启动”是同一个意思。实现起来需要多一步在知识入库时预先用向量模型把所有content字段转换成向量存到knowledge_chunks表的embedding_vector字段里可以用BLOB类型存储。检索时将用户问题也转换成向量然后在数据库里进行向量相似度搜索。MySQL 8.0以上版本支持向量相似度搜索或者你也可以用专门的向量数据库如Milvus、Qdrant但对于刚开始的项目用MySQL存向量并计算也是可行的。4.2 设计更聪明的提示词提示词是引导模型的关键。除了我们之前写的基本指令还可以优化指定回答格式比如“请先概括问题再分步骤解答最后给出总结”。控制语气和风格比如“请用亲切、耐心、专业的客服口吻回答”。处理未知问题明确指令“如果资料不足请直接说‘根据现有资料我无法完全确认建议您……’切勿编造信息”。多轮对话在提示词中加入历史对话记录让模型能理解上下文。比如“用户之前问过XX问题我们回答了YYY。现在用户接着问ZZZ。”4.3 系统的部署与监控一个可以上线的系统不能只靠一个Python脚本。服务化将我们写的CustomerServiceAgent封装成一个Web服务比如用FastAPI提供/ask接口方便其他系统如网站、APP调用。异步处理如果问题复杂检索和生成耗时较长可以考虑使用消息队列如RabbitMQ、Celery进行异步处理避免HTTP请求超时。日志与监控记录每一个用户问题和生成的答案这非常重要。一方面可以监控模型有没有“胡说八道”及时发现bad case另一方面这些日志本身就是优化知识库和模型的宝贵数据。知识库更新建立定期或触发式的知识库更新机制。当产品文档更新后能自动或半自动地重新处理文本、分割、生成向量并更新数据库。4.4 效果评估与迭代系统上线后怎么知道它好不好人工抽检定期随机抽取一批问答记录由资深客服评估回答的准确性和友好度。关键指标可以统计“直接回答率”模型根据知识库给出答案的比例、“转人工率”模型无法回答请求转人工的比例、“用户满意度”如果有评价按钮的话。Bad Case分析专门分析回答错误或用户不满意的案例。是检索没找到资料还是提示词没设计好或者是知识库本身缺失根据分析结果有针对性地去优化检索器、提示词或补充知识库。整个项目做下来感觉就像搭积木把成熟的数据库、强大的语言模型和清晰的业务流程组合在一起就能创造出一个实用的智能工具。基于MySQL和Janus-Pro-7B的这套方案最大的优点就是落地快、成本低、可控性强。知识在你自己的数据库里回答的逻辑由你设计的提示词控制不用担心数据泄露也不用为过高的API调用费用发愁。当然它也不是万能的。面对非常复杂的、需要深度推理的客服问题或者知识库覆盖不到的领域它的能力就有边界了。这时候一个设计良好的“转人工”通道就必不可少。把它定位成处理常见、标准问题的“第一道防线”能有效提升客服团队的整体效率。如果你正准备尝试我的建议是从一个小而具体的场景开始比如“产品售后常见问题解答”。用几十条高质量的QA把知识库建起来把流程跑通看到实际效果。然后再逐步扩展知识范围优化检索和提示策略。在这个过程中你会更深刻地理解RAG的魅力和挑战。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。