Janus-Pro-7B企业知识库问答实战集成MySQL与向量检索最近跟几个做企业服务的朋友聊天他们都在头疼同一个问题公司内部文档一大堆产品手册、规章制度、技术文档新员工来了找不到老员工也记不全。每次有人问个问题都得翻半天文档效率特别低。这不就是典型的企业知识管理痛点吗信息散落在各处用的时候找不到。传统的搜索只能匹配关键词稍微复杂点的问题就答不上来。现在大模型这么火能不能用它来做个智能问答系统让员工像问同事一样直接问文档呢当然可以。今天我就带你用Janus-Pro-7B这个模型结合MySQL和向量检索技术从头搭建一个企业级的私有知识问答系统。这个方案最大的好处是答案都来自你公司自己的文档准确可靠而且完全私有部署数据安全有保障。1. 为什么需要企业知识库问答系统先说说为什么传统的文档管理方式不够用了。想象一下你们公司新来了一位销售他想了解某个产品的详细技术参数。他可能得先去文件服务器找产品手册PDF打开几十页的文档用CtrlF搜索关键词还不一定能找到最相关的部分。如果问老同事对方可能记得大概但细节也不一定准确。这种模式下信息获取的成本太高了。而一个智能问答系统可以让员工直接用自然语言提问“XX产品支持的最大并发用户数是多少”系统就能从文档里找到准确答案甚至把相关的上下文都给你列出来。更重要的是基于大模型的系统能理解问题的意图。你问“怎么申请年假”和“休年假需要走什么流程”系统知道这是同一个问题都能给你正确的回答。这种理解能力是传统关键词搜索做不到的。2. 系统架构与核心组件整个系统的思路其实挺直观的我画了个简单的流程图帮你理解用户提问 → 向量化查询 → 向量数据库检索 → 获取相关文档片段 → Janus-Pro-7B生成答案 → 返回答案核心就三部分知识存储MySQL存放原始的文档内容比如产品手册的PDF转成文本后存进来。向量检索把文档内容转换成数学向量可以理解成一种“语义指纹”然后存到专门的向量数据库里。用户提问时先把问题也转换成向量然后去数据库里找最相似的文档片段。答案生成Janus-Pro-7B把检索到的相关文档片段和用户问题一起喂给大模型让它生成一个准确、通顺的答案。为什么要用向量检索而不是直接让模型读全部文档因为现在的模型都有上下文长度限制你不可能把公司所有文档都一次性塞给它。向量检索的作用就是先帮模型找到最相关的几段内容大大提高了准确性和效率。Janus-Pro-7B这个模型我选它主要是看中它在中文理解和指令跟随方面的表现不错而且7B的参数量在消费级显卡上就能跑起来部署成本相对友好。3. 环境准备与MySQL配置咱们先从最基础的环境搭建开始。我会尽量把每一步都讲清楚即使你之前没怎么接触过这些工具也能跟着做下来。3.1 基础环境安装首先确保你的机器上有Python环境建议用Python 3.8以上版本。然后安装一些必要的Python包pip install torch transformers sentence-transformers pymysql sqlalchemy这里简单说一下这几个包是干嘛的torchPyTorch深度学习框架运行模型需要transformersHugging Face的模型库加载Janus-Pro-7Bsentence-transformers用来做文本向量化把文字转换成向量pymysql/sqlalchemy操作MySQL数据库3.2 MySQL安装与配置如果你还没有安装MySQL这里提供两种常见的方式。方式一使用Docker推荐最方便如果你有Docker环境一条命令就能启动MySQLdocker run -d \ --name mysql-for-knowledge \ -e MYSQL_ROOT_PASSWORDyour_password \ -e MYSQL_DATABASEknowledge_base \ -p 3306:3306 \ mysql:8.0方式二本地安装MySQL如果你更喜欢本地安装可以去MySQL官网下载对应系统的安装包。安装完成后记得创建一个专门的数据和用户-- 登录MySQL mysql -u root -p -- 创建数据库 CREATE DATABASE knowledge_base CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; -- 创建专用用户更安全 CREATE USER kb_user% IDENTIFIED BY your_password; GRANT ALL PRIVILEGES ON knowledge_base.* TO kb_user%; FLUSH PRIVILEGES;这里用utf8mb4字符集是为了更好地支持中文和各种特殊符号。3.3 数据库表设计接下来设计存储文档的表结构。在我们的知识库系统里至少需要两张表-- 文档表存储原始文档信息 CREATE TABLE documents ( id INT AUTO_INCREMENT PRIMARY KEY, title VARCHAR(500) NOT NULL COMMENT 文档标题, content LONGTEXT NOT NULL COMMENT 文档完整内容, source_type VARCHAR(50) COMMENT 来源类型如manual/policy/guide等, source_path VARCHAR(500) COMMENT 原始文件路径, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_title (title(100)), INDEX idx_source (source_type) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; -- 文档片段表存储拆分后的文本片段和对应的向量 CREATE TABLE document_chunks ( id INT AUTO_INCREMENT PRIMARY KEY, document_id INT NOT NULL COMMENT 关联的文档ID, chunk_index INT NOT NULL COMMENT 片段序号, content TEXT NOT NULL COMMENT 文本片段内容, content_vector LONGTEXT COMMENT 文本向量的JSON字符串, token_count INT COMMENT 文本的token数量, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (document_id) REFERENCES documents(id) ON DELETE CASCADE, INDEX idx_document (document_id, chunk_index) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;为什么要把文档拆分成片段因为很多文档都很长比如一本产品手册可能有上百页。我们检索的时候不需要整篇文档只需要最相关的几个段落。通常每个片段控制在200-500字左右这样检索精度更高。4. 知识处理流程从文档到向量有了数据库接下来就是处理知识的核心环节了。这个过程分为三步文档解析、文本拆分、向量化。4.1 文档解析与导入企业文档格式五花八门有PDF、Word、Excel、PPT甚至还有图片。我们需要先把它们转换成纯文本。这里我写了一个简单的文档处理类支持常见的格式import os from typing import List, Dict import pymysql from sqlalchemy import create_engine, text import PyPDF2 from docx import Document import pandas as pd class DocumentProcessor: def __init__(self, db_config: Dict): 初始化数据库连接 self.engine create_engine( fmysqlpymysql://{db_config[user]}:{db_config[password]} f{db_config[host]}:{db_config[port]}/{db_config[database]} ) def parse_pdf(self, file_path: str) - str: 解析PDF文件为文本 text_content try: with open(file_path, rb) as file: pdf_reader PyPDF2.PdfReader(file) for page_num in range(len(pdf_reader.pages)): page pdf_reader.pages[page_num] text_content page.extract_text() \n except Exception as e: print(f解析PDF失败 {file_path}: {e}) return text_content def parse_docx(self, file_path: str) - str: 解析Word文档为文本 text_content try: doc Document(file_path) for paragraph in doc.paragraphs: text_content paragraph.text \n except Exception as e: print(f解析Word失败 {file_path}: {e}) return text_content def save_to_database(self, title: str, content: str, source_type: str, source_path: str) - int: 将文档保存到数据库返回文档ID with self.engine.connect() as conn: result conn.execute( text( INSERT INTO documents (title, content, source_type, source_path) VALUES (:title, :content, :source_type, :source_path) ), { title: title, content: content, source_type: source_type, source_path: source_path } ) conn.commit() return result.lastrowid # 使用示例 db_config { host: localhost, port: 3306, user: kb_user, password: your_password, database: knowledge_base } processor DocumentProcessor(db_config) # 处理一个PDF文档 pdf_text processor.parse_pdf(产品手册.pdf) doc_id processor.save_to_database( titleXX产品用户手册, contentpdf_text, source_typemanual, source_path产品手册.pdf ) print(f文档导入成功ID: {doc_id})实际项目中你可能还需要处理更多格式比如Excel、PPT甚至图片中的文字需要OCR。核心思路都一样把各种格式转换成纯文本然后存到数据库。4.2 文本拆分与向量化文档导入后我们需要把它拆分成更小的片段然后为每个片段生成向量。import json from sentence_transformers import SentenceTransformer import numpy as np class TextChunker: def __init__(self, chunk_size: int 500, overlap: int 50): 初始化文本拆分器 chunk_size: 每个片段的字符数 overlap: 片段之间的重叠字符数避免信息割裂 self.chunk_size chunk_size self.overlap overlap # 加载向量化模型这里用个轻量级的中文模型 self.embedding_model SentenceTransformer(paraphrase-multilingual-MiniLM-L12-v2) def split_text(self, text: str) - List[str]: 将长文本拆分成片段 chunks [] start 0 while start len(text): # 计算当前片段的结束位置 end start self.chunk_size # 如果还没到文本末尾尝试在句号、换行处截断 if end len(text): # 找最近的句号或换行 for i in range(end, start, -1): if i len(text) and text[i] in [。, ., \n, , ;]: end i 1 break chunk text[start:end].strip() if chunk: # 跳过空片段 chunks.append(chunk) # 移动起始位置考虑重叠 start end - self.overlap return chunks def generate_embeddings(self, texts: List[str]) - List[List[float]]: 为文本列表生成向量 # 转换为向量 embeddings self.embedding_model.encode(texts) # 转换为Python列表格式 return [embedding.tolist() for embedding in embeddings] def process_document(self, document_id: int, content: str): 处理单个文档拆分并生成向量 # 拆分文本 chunks self.split_text(content) # 为每个片段生成向量 embeddings self.generate_embeddings(chunks) # 保存到数据库 with self.engine.connect() as conn: for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)): conn.execute( text( INSERT INTO document_chunks (document_id, chunk_index, content, content_vector, token_count) VALUES (:doc_id, :idx, :content, :vector, :token_count) ), { doc_id: document_id, idx: i, content: chunk, vector: json.dumps(embedding), token_count: len(chunk) // 4 # 粗略估算token数 } ) conn.commit() print(f文档 {document_id} 处理完成生成 {len(chunks)} 个片段) # 使用示例 chunker TextChunker(chunk_size500, overlap50) # 假设我们从数据库读取了一个文档 with processor.engine.connect() as conn: result conn.execute( text(SELECT id, content FROM documents WHERE id :doc_id), {doc_id: doc_id} ) doc result.fetchone() if doc: chunker.process_document(doc[id], doc[content])这里有几个关键点需要注意拆分策略我用了按字符数拆分并在标点处截断。你也可以按段落拆分效果可能更好。向量模型我用了sentence-transformers里的多语言模型它对中文支持不错。如果你对精度要求更高可以考虑专门的中文向量模型。向量存储这里直接把向量以JSON格式存到MySQL的TEXT字段。对于小规模知识库比如几万条记录是可行的。如果数据量很大建议用专门的向量数据库比如Milvus、Qdrant等。5. Janus-Pro-7B模型部署与集成知识处理好了接下来就是核心的模型部分。Janus-Pro-7B是一个7B参数的中英文双语模型在指令跟随和中文理解方面表现不错。5.1 模型加载与初始化import torch from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline class JanusQA: def __init__(self, model_path: str Janus-Pro-7B): 初始化Janus-Pro-7B模型 print(正在加载Janus-Pro-7B模型...) # 加载tokenizer和模型 self.tokenizer AutoTokenizer.from_pretrained(model_path, trust_remote_codeTrue) self.model AutoModelForCausalLM.from_pretrained( model_path, torch_dtypetorch.float16, # 使用半精度减少显存占用 device_mapauto, # 自动分配设备 trust_remote_codeTrue ) # 创建文本生成pipeline self.generator pipeline( text-generation, modelself.model, tokenizerself.tokenizer, device0 if torch.cuda.is_available() else -1 ) print(模型加载完成) def generate_answer(self, question: str, context: str, max_length: int 500) - str: 基于上下文生成答案 # 构建提示词 prompt f基于以下上下文信息请回答问题。如果上下文没有提供足够信息请回答“根据现有信息无法回答”。 上下文 {context} 问题{question} 答案 # 生成答案 result self.generator( prompt, max_lengthmax_length, temperature0.7, # 控制随机性0.7比较平衡 do_sampleTrue, top_p0.9, # 核采样提高多样性 repetition_penalty1.1 # 避免重复 ) # 提取生成的答案 generated_text result[0][generated_text] # 只取“答案”之后的部分 answer_start generated_text.find(答案) 3 answer generated_text[answer_start:].strip() return answer # 初始化模型 qa_system JanusQA()这里有几个实用的技巧显存优化用torch.float16半精度7B模型大概需要14GB显存。如果显存不够可以用torch.bfloat16或者量化版本。提示词设计我加了一句“如果上下文没有提供足够信息请回答...”这样模型在找不到答案时会如实告知而不是胡编乱造。生成参数temperature0.7让回答有一定创造性但不至于太随机top_p0.9保证回答多样性。5.2 向量检索与答案生成现在我们把向量检索和模型生成结合起来import numpy as np from typing import List, Dict class KnowledgeBaseQA: def __init__(self, db_config: Dict, model_path: str Janus-Pro-7B): 初始化知识库问答系统 self.db_engine create_engine( fmysqlpymysql://{db_config[user]}:{db_config[password]} f{db_config[host]}:{db_config[port]}/{db_config[database]} ) self.chunker TextChunker() self.qa_model JanusQA(model_path) def search_similar_chunks(self, query: str, top_k: int 3) - List[Dict]: 检索最相关的文档片段 # 将查询转换为向量 query_vector self.chunker.generate_embeddings([query])[0] # 从数据库获取所有片段实际应用中应该用向量数据库或建立索引 with self.db_engine.connect() as conn: result conn.execute( text(SELECT id, content, content_vector FROM document_chunks) ) all_chunks result.fetchall() # 计算相似度余弦相似度 similarities [] for chunk in all_chunks: if chunk[content_vector]: chunk_vector json.loads(chunk[content_vector]) # 计算余弦相似度 similarity np.dot(query_vector, chunk_vector) / ( np.linalg.norm(query_vector) * np.linalg.norm(chunk_vector) ) similarities.append({ id: chunk[id], content: chunk[content], similarity: similarity }) # 按相似度排序取最相关的top_k个 similarities.sort(keylambda x: x[similarity], reverseTrue) return similarities[:top_k] def answer_question(self, question: str) - Dict: 回答用户问题 print(f正在处理问题: {question}) # 1. 检索相关文档片段 similar_chunks self.search_similar_chunks(question, top_k3) if not similar_chunks: return { answer: 抱歉知识库中没有找到相关信息。, sources: [], confidence: 0.0 } # 2. 构建上下文 context \n\n.join([chunk[content] for chunk in similar_chunks]) # 3. 生成答案 answer self.qa_model.generate_answer(question, context) # 4. 返回结果 return { answer: answer, sources: [ { content: chunk[content][:100] ..., # 截取前100字符 similarity: round(chunk[similarity], 3) } for chunk in similar_chunks ], confidence: round(similar_chunks[0][similarity], 3) } # 使用示例 kb_qa KnowledgeBaseQA(db_config) question 我们产品的退货政策是什么 result kb_qa.answer_question(question) print(f问题: {question}) print(f答案: {result[answer]}) print(f置信度: {result[confidence]}) print(参考来源:) for i, source in enumerate(result[sources], 1): print(f{i}. {source[content]} (相似度: {source[similarity]}))这个完整的流程实现了问题向量化把用户问题转换成向量相似度检索从知识库中找到最相关的文档片段上下文构建把相关片段拼接成模型需要的上下文答案生成让Janus-Pro-7B基于上下文生成答案结果返回返回答案和参考来源方便用户核实6. 实际应用与效果优化系统搭好了怎么用起来效果更好呢我分享几个实际应用中的经验。6.1 不同场景的提问技巧同样的知识库问法不同得到的结果可能差别很大。这里有些实用建议不好的问法“产品信息”太宽泛好的问法“XX型号的产品支持哪些操作系统”不好的问法“怎么用”不具体好的问法“如何设置产品的自动备份功能”复杂的问法“如果客户要求延长退货期限我们应该怎么处理” 这种问题系统也能处理它会从退货政策、客户服务流程等文档中综合信息给出建议。6.2 效果优化技巧在实际使用中你可能会发现一些可以优化的地方1. 检索精度提升# 可以尝试不同的向量模型 # 中文专用模型效果通常更好 from sentence_transformers import SentenceTransformer # 尝试不同的模型 models_to_try [ paraphrase-multilingual-MiniLM-L12-v2, # 多语言通用性好 BAAI/bge-large-zh, # 中文专用效果更好 moka-ai/m3e-base, # 中文嵌入模型 ] # 测试不同模型的检索效果 for model_name in models_to_try: embedding_model SentenceTransformer(model_name) # 测试检索效果...2. 提示词优化提示词的设计对答案质量影响很大。你可以根据业务需求调整def build_prompt_v2(question: str, context: str) - str: 更详细的提示词模板 return f你是一个专业的企业知识库助手。请严格基于提供的上下文信息回答问题。 要求 1. 答案必须基于上下文不要编造信息 2. 如果上下文没有相关信息请明确告知“根据现有信息无法回答” 3. 答案要简洁明了重点突出 4. 如果涉及步骤请分点说明 上下文信息 {context} 用户问题{question} 请根据以上信息回答问题3. 后处理与验证对于重要的业务问题可以增加验证步骤def validate_answer(question: str, answer: str, context: str) - bool: 简单验证答案是否合理 # 检查答案是否包含“无法回答”之类的表述 if 无法回答 in answer or 没有相关信息 in answer: return True # 这是合理的 # 检查答案是否明显偏离上下文 # 这里可以用简单的关键词匹配 important_keywords extract_keywords(question) for keyword in important_keywords: if keyword in context and keyword not in answer: return False # 重要关键词在上下文中但不在答案里 return True6.3 性能考虑与扩展如果知识库很大比如超过10万条记录全量计算相似度会很慢。这时候可以考虑使用向量数据库比如Milvus、Qdrant、Pinecone等它们有专门的向量索引检索速度快。建立混合索引结合关键词索引和向量索引先用关键词过滤再用向量精排。缓存机制对常见问题缓存答案减少重复计算。异步处理文档导入、向量化等耗时操作放到后台异步执行。7. 总结整套方案走下来你会发现搭建一个企业知识库问答系统并没有想象中那么复杂。核心就是三个部分存文档、找相关、生成答案。用MySQL存原始文档用向量检索找相关内容再用Janus-Pro-7B生成答案这个组合在实际应用中效果不错。最大的好处是答案都来自你们公司自己的文档准确可靠而且整个系统可以部署在内网数据安全有保障。实际用起来你会发现一些需要调整的地方。比如检索的准确度可能跟向量模型的选择、文本拆分的策略都有关系。生成答案的质量也跟提示词的设计、模型的参数设置有关。这些都是可以慢慢优化的。如果你刚开始尝试建议从小范围开始比如先拿一个产品手册试试。跑通整个流程后再逐步扩大文档范围。过程中注意收集用户的反馈看看哪些问题回答得好哪些回答得不好然后针对性优化。这种系统真正的价值不是替代人工而是让人工从重复的信息查找中解放出来去做更有价值的工作。员工不用再花时间翻文档系统直接给出答案效率提升是实实在在的。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。