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

Qwen1.5-1.8B GPTQ本地知识库构建实战：从文本清洗到向量检索

news 2026/8/30 23:38:49

Qwen1.5-1.8B GPTQ本地知识库构建实战：从文本清洗到向量检索

你是不是经常觉得，面对自己积累的一大堆文档、笔记或者公司内部资料，想快速找到某个问题的答案特别麻烦？手动搜索效率低，而通用的AI助手又对你的专属知识一无所知。

今天，我们就来解决这个问题。我将带你一步步，用Qwen1.5-1.8B这个轻量级但能力不俗的模型，搭建一个完全运行在你本机上的“专属知识大脑”。这个系统能理解你文档里的内容，并用自然对话的方式回答你的问题。整个过程，从准备文本到最终问答，我都会提供详细的代码，你跟着做就能拥有自己的AI知识库助手。

1. 项目准备：理清思路与备好工具

在开始写代码之前，我们先花几分钟把整个项目的脉络理清楚。这样你就能明白每一步在做什么，而不是机械地复制粘贴。

核心目标：我们要构建一个**检索增强生成（RAG）**系统。简单来说，就是让AI在回答问题时，不是凭空想象，而是先去你的知识库里找到相关的资料，然后基于这些资料来生成答案。这就像是一个配备了“参考资料”的专家，回答更准确、更有依据。

我们的技术路线图：

知识录入：把你的文档（比如TXT、PDF、Word）变成计算机能理解的“记忆碎片”（向量）。
知识存储：把这些“记忆碎片”分门别类地存进一个专门的数据库（向量数据库），方便快速查找。
问答交互：当你提问时，系统先在你的“记忆库”里找到最相关的几个“碎片”，然后交给AI，让它根据这些碎片组织语言来回答你。

为什么选Qwen1.5-1.8B GPTQ？

轻量高效：1.8B的参数规模，在消费级显卡（甚至一些高性能的CPU）上也能流畅运行，非常适合本地部署。
功能全面：它既能将文本转化为高质量的向量（嵌入模型），也能根据上下文生成流畅的回答（生成模型），一套模型解决两个核心问题。
GPTQ量化：GPTQ是一种模型压缩技术，能在几乎不损失精度的情况下，大幅减少模型对显存的需求，让部署门槛更低。

你需要准备好的环境：

Python 3.8+：这是我们的工作语言。
基本的Python包管理知识：会用pip安装库。
约4-6GB的可用磁盘空间：用于存放模型和数据库。
有GPU更好：会快很多。但没有GPU，用CPU也能跑起来，只是速度会慢一些。

接下来，我们就进入实战环节。

2. 搭建环境：安装必要的“施工队”

工欲善其事，必先利其器。我们首先通过命令行来安装所有需要的Python库。打开你的终端或命令提示符，逐行执行下面的命令。

# 安装深度学习框架和模型运行库 pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 # 如果你有NVIDIA GPU，使用此命令 # 如果没有GPU，请使用： pip install torch torchvision torchaudio # 安装Transformers库，它是我们加载和使用Qwen模型的核心 pip install transformers # 安装accelerate，用于优化模型加载和推理 pip install accelerate # 安装sentence-transformers，它提供了便捷的文本嵌入接口 pip install sentence-transformers # 安装ChromaDB，这是我们选用的轻量级、易用的向量数据库 pip install chromadb # 安装langchain，它提供了构建AI应用链的高级抽象，能简化我们的代码 pip install langchain # 安装文档加载器，用于读取不同格式的文件 pip install pypdf2 python-docx # 用于PDF和Word文档 pip install unstructured # 用于更复杂的文档解析

安装过程可能会花点时间，取决于你的网络。全部安装成功后，我们的“工具箱”就齐备了。

3. 第一步：准备你的知识原料（文本处理）

知识库的原料是你的文档。这些文档可能格式杂乱，包含很多无关信息。第一步就是清洗和预处理它们。

假设你的文档放在一个叫my_docs的文件夹里，里面有.txt，.pdf，.docx等文件。我们写一个脚本来处理它们。

创建一个新的Python文件，比如叫做prepare_knowledge.py，然后写入以下代码：

import os from pathlib import Path import PyPDF2 from docx import Document from langchain.text_splitter import RecursiveCharacterTextSplitter class DocumentProcessor: def __init__(self, docs_folder_path): self.docs_path = Path(docs_folder_path) self.text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每个文本块的最大字符数 chunk_overlap=50, # 块之间的重叠字符，避免上下文断裂 separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""] # 分割符优先级 ) def read_file(self, file_path): """根据文件后缀名，读取文本内容""" text = "" suffix = file_path.suffix.lower() try: if suffix == '.txt': with open(file_path, 'r', encoding='utf-8') as f: text = f.read() elif suffix == '.pdf': with open(file_path, 'rb') as f: pdf_reader = PyPDF2.PdfReader(f) for page in pdf_reader.pages: text += page.extract_text() + "\n" elif suffix in ['.docx', '.doc']: doc = Document(file_path) for para in doc.paragraphs: text += para.text + "\n" else: print(f"暂不支持的文件格式: {suffix}") except Exception as e: print(f"读取文件 {file_path} 时出错: {e}") return text def clean_text(self, text): """简单的文本清洗""" # 移除过多的换行和空格 import re text = re.sub(r'\n+', '\n', text) text = re.sub(r'[ \t]+', ' ', text) # 这里可以添加更多清洗规则，比如移除特殊字符、URL等 return text.strip() def process_folder(self): """处理整个文件夹，返回清洗并分割后的文本块列表""" all_chunks = [] all_metadatas = [] # 记录每个块的来源信息 if not self.docs_path.exists(): print(f"文件夹 {self.docs_path} 不存在！") return all_chunks, all_metadatas for file_path in self.docs_path.iterdir(): if file_path.is_file(): print(f"正在处理: {file_path.name}") raw_text = self.read_file(file_path) if raw_text: cleaned_text = self.clean_text(raw_text) # 使用LangChain的分割器进行智能分割 chunks = self.text_splitter.split_text(cleaned_text) for i, chunk in enumerate(chunks): all_chunks.append(chunk) # 为每个块记录它来自哪个文件的哪一部分 all_metadatas.append({"source": file_path.name, "chunk_id": i}) print(f" -> 分割为 {len(chunks)} 个文本块。") else: print(f" -> 文件内容为空或读取失败。") print(f"\n总计处理完成，得到 {len(all_chunks)} 个文本块。") return all_chunks, all_metadatas # 使用示例 if __name__ == "__main__": processor = DocumentProcessor("my_docs") # 替换为你的文件夹路径 text_chunks, metadatas = processor.process_folder() # 简单查看一下前两个块 for i in range(min(2, len(text_chunks))): print(f"\n--- 块 {i} (来自 {metadatas[i]['source']}) ---") print(text_chunks[i][:200] + "...") # 只打印前200个字符

运行这个脚本，它会遍历你的文档文件夹，把内容读出来，清洗干净，然后切成一段段大小合适的文本块。这些“块”就是我们后面要喂给模型变成“向量记忆”的基本单位。

4. 第二步：将文本转化为“记忆向量”（嵌入模型）

文本块准备好了，但计算机看不懂文字。我们需要用Qwen1.5的嵌入模型，把每一段文字转换成一个高维度的数字向量（可以理解为一串有特殊意义的数字）。语义相近的文本，它们的向量在空间里的距离也会很近。

创建一个新文件create_vector_store.py：

from sentence_transformers import SentenceTransformer import chromadb from chromadb.config import Settings from prepare_knowledge import DocumentProcessor # 导入上一步的处理器 import time class VectorStoreCreator: def __init__(self, model_name="Qwen/Qwen1.5-1.8B"): # 注意：我们使用sentence-transformers接口，但它底层会调用transformers加载Qwen模型做嵌入 # 由于Qwen1.5原生设计主要用于生成，我们这里演示一种利用其生成能力获取文本表示的方法 # 更严谨的做法是使用专门训练好的嵌入模型，但为了流程完整，我们展示如何用生成模型“适配” print("正在加载模型...") # 实际上，对于1.8B这样的生成模型，直接获取高质量的句子嵌入比较困难。 # 一个实用的替代方案是使用一个轻量且开源的嵌入模型，例如 `all-MiniLM-L6-v2`。 # 这里为了教程连贯性，我们先使用这个轻量嵌入模型。 # 如果你坚持想尝试用Qwen1.5生成特征，需要更复杂的处理（如取最后一层隐状态平均）， # 这超出了入门教程范围，且效果和效率可能不如专用嵌入模型。 self.embed_model = SentenceTransformer('all-MiniLM-L6-v2') # 先使用这个替代 print("嵌入模型加载完毕。") # 初始化Chroma客户端，数据持久化到本地磁盘 self.client = chromadb.PersistentClient(path="./my_chroma_db") def create_embeddings_and_store(self, text_chunks, metadatas, collection_name="my_knowledge_base"): """生成向量并存储到ChromaDB""" print("开始生成文本向量...") start_time = time.time() # 使用嵌入模型批量生成向量 embeddings = self.embed_model.encode(text_chunks, show_progress_bar=True, batch_size=32) print(f"向量生成完成，耗时 {time.time() - start_time:.2f} 秒。") # 创建或获取一个集合（类似于数据库的表） collection = self.client.get_or_create_collection(name=collection_name) print("正在将向量存入数据库...") # 准备数据，每个块需要一个唯一ID ids = [f"chunk_{i}" for i in range(len(text_chunks))] # 添加到集合 collection.add( embeddings=embeddings.tolist(), # ChromaDB接受列表形式的向量 documents=text_chunks, # 原始文本 metadatas=metadatas, # 元数据 ids=ids # ID ) print(f"知识库构建完成！共存储 {collection.count()} 条记录。") return collection # 主流程 if __name__ == "__main__": # 1. 处理文档 print("=== 步骤1: 处理原始文档 ===") processor = DocumentProcessor("my_docs") chunks, metas = processor.process_folder() if not chunks: print("没有处理到任何文本，请检查文档路径和内容。") exit() # 2. 创建向量存储 print("\n=== 步骤2: 创建向量数据库 ===") creator = VectorStoreCreator() collection = creator.create_embeddings_and_store(chunks, metas) # 3. 简单测试一下检索 print("\n=== 步骤3: 简单检索测试 ===") test_query = "什么是机器学习？" # 你可以改成你文档里可能有的内容 test_embedding = creator.embed_model.encode([test_query]).tolist() results = collection.query( query_embeddings=test_embedding, n_results=2 # 返回最相似的2个块 ) print(f"查询: '{test_query}'") print("最相关的文档块：") for i, doc in enumerate(results['documents'][0]): print(f"\n[{i+1}] {doc[:150]}...") # 打印前150字符

这段代码做了几件事：

加载一个轻量级的嵌入模型（作为演示，生产环境可选择更匹配的模型）。
读取上一步处理好的文本块。
调用模型，把所有文本块转换成向量。
把这些向量和对应的文本、元数据一起，存进ChromaDB数据库。

运行后，你会看到一个my_chroma_db文件夹，里面就是你的向量化知识库了。

5. 第三步：让AI基于知识库回答问题（RAG流程）

最后一步，也是最激动人心的一步：整合检索和生成。我们创建一个问答链：用户提问 -> 检索相关文档 -> 组合成提示词 -> Qwen1.5生成答案。

创建一个新文件ask_question.py：

from transformers import AutoModelForCausalLM, AutoTokenizer from create_vector_store import VectorStoreCreator import torch class KnowledgeBaseQA: def __init__(self, vector_db_path="./my_chroma_db", collection_name="my_knowledge_base"): # 1. 加载嵌入模型和向量数据库 print("加载检索组件...") self.vec_creator = VectorStoreCreator() self.client = chromadb.PersistentClient(path=vector_db_path) self.collection = self.client.get_collection(name=collection_name) # 2. 加载Qwen1.5-1.8B GPTQ作为生成模型 print("加载生成模型...") model_name = "Qwen/Qwen1.5-1.8B" self.tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True) # 使用量化版本以节省显存 (如果设备支持) self.model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=torch.float16, # 使用半精度浮点数 device_map="auto", # 自动分配模型层到GPU/CPU trust_remote_code=True ) print("模型加载完毕，系统就绪！\n") def retrieve_context(self, query, top_k=3): """从知识库中检索与问题最相关的上下文""" query_embedding = self.vec_creator.embed_model.encode([query]).tolist() results = self.collection.query( query_embeddings=query_embedding, n_results=top_k ) # 将检索到的文档块合并为一段上下文 context = "\n\n".join(results['documents'][0]) return context, results['metadatas'][0] def build_prompt(self, query, context): """构建给生成模型的提示词模板""" prompt_template = f"""你是一个专业的助手，请严格根据以下提供的上下文信息来回答问题。如果上下文没有提供相关信息，请直接说“根据提供的资料，我无法回答这个问题。”，不要编造信息。 上下文信息： {context} 问题：{query} 请根据上下文回答：""" return prompt_template def generate_answer(self, prompt): """使用Qwen模型生成答案""" # 将提示词转换为模型输入的token inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device) # 生成回答 with torch.no_grad(): # 推理阶段，不需要计算梯度 outputs = self.model.generate( **inputs, max_new_tokens=512, # 生成答案的最大长度 do_sample=True, # 启用采样，使生成更有创造性 temperature=0.7, # 采样温度，控制随机性 top_p=0.9, # 核采样参数，控制词汇选择范围 repetition_penalty=1.1 # 重复惩罚，避免重复用词 ) # 解码生成的token，得到文本 answer = self.tokenizer.decode(outputs[0][inputs['input_ids'].shape[1]:], skip_special_tokens=True) return answer.strip() def ask(self, question): """主问答函数""" print(f"问题: {question}") print("正在检索相关知识...") context, source_metas = self.retrieve_context(question) print("正在生成回答...") prompt = self.build_prompt(question, context) answer = self.generate_answer(prompt) print("\n" + "="*50) print("回答：") print(answer) print("="*50) # 显示答案来源 print("\n答案参考自以下文档片段：") for meta in source_metas: print(f" - 来源文件: {meta['source']}, 块ID: {meta['chunk_id']}") return answer # 交互式问答循环 if __name__ == "__main__": qa_system = KnowledgeBaseQA() print("本地知识库问答系统已启动！") print("输入你的问题（输入 'quit' 或 '退出' 结束）") print("-" * 30) while True: user_input = input("\n你的问题: ").strip() if user_input.lower() in ['quit', '退出', 'exit']: print("再见！") break if user_input: qa_system.ask(user_input)

现在，运行这个文件。你会看到模型加载的提示，然后就可以在命令行里用自然语言向你的知识库提问了！系统会先检索，再生成，并告诉你答案参考了哪些原始文档。