当前位置：首页 > news >正文

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

news 2026/4/20 11:18:35

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_admin'@'localhost' IDENTIFIED BY 'your_strong_password'; -- 给这个用户授予对新数据库的全部权限 GRANT ALL PRIVILEGES ON smart_customer_service.* TO 'cs_admin'@'localhost'; -- 让权限生效 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 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_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( host='localhost', user='cs_admin', password='your_strong_password', # 换成你的密码 database='smart_customer_service', charset='utf8mb4' ) 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_size=300, chunk_overlap=50) # 将分割后的块存入数据库 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(level=logging.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(title="Janus-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_dtype=torch.float16, # 使用半精度减少内存占用 device_map="auto" # 自动分配到可用的GPU/CPU ) logger.info("模型加载完毕。") # 创建一个文本生成的pipeline generator = pipeline( "text-generation", model=model, tokenizer=tokenizer, device=0 if torch.cuda.is_available() else -1 ) @app.post("/generate", response_model=QueryResponse) async def generate_answer(request: QueryRequest): """ 接收增强后的提示词，生成回答。 """ import time start_time = time.time() try: # 使用模型生成文本 generated_sequences = generator( request.prompt, max_length=request.max_length, do_sample=True, # 启用采样，使生成结果更多样 temperature=0.7, # 控制随机性，0.7是个比较平衡的值 top_p=0.9, # 核采样，进一步控制多样性 num_return_sequences=1 ) 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(answer=answer, processing_time=processing_time) except Exception as e: logger.error(f"生成答案时出错: {e}") raise HTTPException(status_code=500, detail=f"内部服务器错误: {str(e)}") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

把上面的代码保存为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(tokenizer=jieba.lcut, max_features=5000) # 使用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_k=2) print(f"针对问题: '{user_query}'") print(f"检索到 {len(relevant_chunks)} 条相关知识点:") for i, chunk in enumerate(relevant_chunks): print(f"\n--- 相关知识点 {i+1} (相似度: {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"[资料片段 {i+1}]:\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_url="http://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_k=3) 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}, timeout=30 # 设置超时时间 ) 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的魅力和挑战。