GLM-4-9B-Chat-1M实战教程:Chainlit中集成RAG实现私有知识库问答
GLM-4-9B-Chat-1M实战教程:Chainlit中集成RAG实现私有知识库问答
你是不是经常遇到这样的问题:手头有一堆技术文档、公司资料或者个人笔记,想快速从中找到某个问题的答案,却要花大量时间翻阅?或者想让AI帮你分析这些文档,但它总是回答得不够准确,因为它根本不了解你的私有知识。
今天,我就带你用GLM-4-9B-Chat-1M这个支持百万字上下文的强大模型,结合Chainlit这个轻量级前端,再配上RAG技术,搭建一个真正属于你自己的私有知识库问答系统。整个过程就像搭积木一样简单,不需要复杂的代码,跟着我做,半小时内你就能拥有一个能“读懂”你文档的智能助手。
1. 为什么需要私有知识库问答?
在开始动手之前,我们先聊聊为什么这个方案值得你花时间。
传统搜索的痛点 想象一下,你刚加入一个新项目,手头有50份技术文档、20个会议纪要、还有一堆产品需求文档。当你想了解“用户登录模块的具体实现逻辑”时,你需要:
- 打开每个文档,用Ctrl+F搜索关键词
- 在不同文档间来回切换,拼凑信息
- 可能还要问同事,打断他们的工作
这个过程既低效又容易遗漏关键信息。
通用AI的局限性 你可能会想:“直接用ChatGPT不就行了?”但实际情况是:
- 通用模型不了解你的内部文档、公司规范、项目细节
- 涉及敏感信息时,你不敢把文档上传到公开服务
- 模型可能“胡编乱造”,给出看似合理但实际错误的信息
我们的解决方案 这就是RAG(检索增强生成)技术的用武之地。简单来说,它的工作原理是这样的:
- 检索:当用户提问时,系统从你的文档库中快速找到最相关的片段
- 增强:把这些相关片段和问题一起送给AI模型
- 生成:AI基于这些“证据”生成准确、可靠的回答
而GLM-4-9B-Chat-1M的百万字上下文能力,让它能同时处理大量检索到的文档片段,给出更全面、更准确的答案。
2. 环境准备与快速部署
2.1 系统要求检查
首先确认你的环境满足以下要求:
- 操作系统:Linux(Ubuntu 20.04+推荐)或 macOS
- Python版本:3.8 或更高版本
- 内存:至少16GB RAM(处理大文档时需要更多)
- 存储空间:20GB 以上可用空间
- 网络:能正常访问模型下载源
2.2 一键部署GLM-4-9B-Chat-1M
如果你使用的是CSDN星图镜像,部署已经完成了。只需要验证服务是否正常运行:
# 查看模型服务日志
cat /root/workspace/llm.log
如果看到类似下面的输出,说明模型已经成功加载:
INFO: Started server process [1234]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000
2.3 安装必要的Python包
打开终端,创建一个新的项目目录并安装依赖:
# 创建项目目录
mkdir private-knowledge-qa && cd private-knowledge-qa
# 创建虚拟环境(可选但推荐)
python -m venv venv
source venv/bin/activate # Linux/macOS
# 或 venv\Scripts\activate # Windows
# 安装核心依赖
pip install chainlit langchain langchain-community chromadb pypdf sentence-transformers
这里简单说明一下每个包的作用:
- chainlit:构建聊天界面的轻量级框架
- langchain:AI应用开发框架,提供RAG等工具
- chromadb:向量数据库,用于存储和检索文档
- pypdf:处理PDF文档
- sentence-transformers:文本嵌入模型,将文档转换为向量
3. 构建私有知识库的核心步骤
现在进入最核心的部分:如何让你的文档“活”起来,变成AI能理解的知识。
3.1 准备你的文档
首先,在项目目录下创建一个documents文件夹,把你的文档放进去:
mkdir documents
支持的文档格式包括:
- PDF文件(.pdf):技术手册、论文、报告
- Word文档(.docx):产品说明、会议纪要
- 文本文件(.txt):代码片段、配置说明
- Markdown文件(.md):项目文档、笔记
- PowerPoint(.pptx):演示文稿
你可以把各种格式的文档混合放在一起,系统会自动处理。
3.2 文档加载与分割
创建一个Python脚本document_processor.py:
import os
from langchain_community.document_loaders import (
PyPDFLoader,
TextLoader,
UnstructuredWordDocumentLoader,
UnstructuredMarkdownLoader,
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
class DocumentProcessor:
def __init__(self, docs_folder="documents"):
self.docs_folder = docs_folder
# 创建文本分割器,控制每个片段的大小
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # 每个片段约1000字符
chunk_overlap=200, # 片段间重叠200字符,保持上下文连贯
separators=["\n\n", "\n", "。", "!", "?", ";", ",", " ", ""]
)
def load_documents(self):
"""加载所有文档"""
documents = []
# 遍历文档文件夹
for filename in os.listdir(self.docs_folder):
file_path = os.path.join(self.docs_folder, filename)
try:
if filename.endswith('.pdf'):
loader = PyPDFLoader(file_path)
elif filename.endswith('.docx'):
loader = UnstructuredWordDocumentLoader(file_path)
elif filename.endswith('.txt'):
loader = TextLoader(file_path, encoding='utf-8')
elif filename.endswith('.md'):
loader = UnstructuredMarkdownLoader(file_path)
else:
print(f"跳过不支持的文件格式: {filename}")
continue
loaded_docs = loader.load()
documents.extend(loaded_docs)
print(f"成功加载: {filename} ({len(loaded_docs)}页)")
except Exception as e:
print(f"加载文件 {filename} 时出错: {str(e)}")
return documents
def split_documents(self, documents):
"""分割文档为小片段"""
if not documents:
print("没有找到可处理的文档")
return []
chunks = self.text_splitter.split_documents(documents)
print(f"文档分割完成,共生成 {len(chunks)} 个文本片段")
return chunks
# 使用示例
if __name__ == "__main__":
processor = DocumentProcessor()
# 1. 加载文档
print("开始加载文档...")
docs = processor.load_documents()
# 2. 分割文档
print("开始分割文档...")
chunks = processor.split_documents(docs)
# 3. 查看分割结果
if chunks:
print(f"\n第一个文本片段预览:")
print("-" * 50)
print(chunks[0].page_content[:500] + "...")
print("-" * 50)
运行这个脚本测试文档处理:
python document_processor.py
3.3 创建向量数据库
文档分割好后,我们需要把它们转换成向量(一种数学表示),这样AI才能快速找到相关内容。创建vector_store.py:
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
import os
class VectorStoreManager:
def __init__(self, persist_directory="./chroma_db"):
self.persist_directory = persist_directory
# 使用轻量级的中文嵌入模型
self.embeddings = HuggingFaceEmbeddings(
model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
model_kwargs={'device': 'cpu'}, # 使用CPU,如需GPU可改为'cuda'
encode_kwargs={'normalize_embeddings': True}
)
def create_vector_store(self, chunks):
"""创建向量数据库"""
print("开始创建向量数据库...")
# 创建Chroma向量数据库
vector_store = Chroma.from_documents(
documents=chunks,
embedding=self.embeddings,
persist_directory=self.persist_directory
)
# 持久化保存
vector_store.persist()
print(f"向量数据库已创建并保存到: {self.persist_directory}")
return vector_store
def load_vector_store(self):
"""加载已有的向量数据库"""
if not os.path.exists(self.persist_directory):
print("向量数据库不存在,请先创建")
return None
print("加载向量数据库...")
vector_store = Chroma(
persist_directory=self.persist_directory,
embedding_function=self.embeddings
)
print("向量数据库加载完成")
return vector_store
# 整合文档处理和向量数据库创建
def build_knowledge_base():
"""完整的知识库构建流程"""
from document_processor import DocumentProcessor
# 1. 处理文档
processor = DocumentProcessor()
docs = processor.load_documents()
if not docs:
print("没有找到文档,请将文档放入documents文件夹")
return None
chunks = processor.split_documents(docs)
# 2. 创建向量数据库
vector_manager = VectorStoreManager()
vector_store = vector_manager.create_vector_store(chunks)
return vector_store
if __name__ == "__main__":
# 构建知识库
vector_store = build_knowledge_base()
if vector_store:
# 测试检索功能
test_query = "什么是RAG技术?"
results = vector_store.similarity_search(test_query, k=3)
print(f"\n针对查询 '{test_query}' 检索到的相关文档:")
for i, doc in enumerate(results, 1):
print(f"\n{i}. 相关度: {doc.metadata.get('source', '未知')}")
print(f"内容预览: {doc.page_content[:200]}...")
4. 集成GLM-4-9B-Chat-1M与Chainlit
4.1 配置GLM模型连接
创建model_client.py来连接GLM-4-9B-Chat-1M模型:
import requests
import json
from typing import List, Dict, Any
class GLMClient:
def __init__(self, base_url="http://localhost:8000/v1"):
self.base_url = base_url
self.chat_url = f"{base_url}/chat/completions"
def chat_completion(self, messages: List[Dict], temperature=0.7, max_tokens=2000):
"""调用GLM模型进行对话"""
headers = {
"Content-Type": "application/json"
}
payload = {
"model": "glm-4-9b-chat-1m",
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
"stream": False
}
try:
response = requests.post(
self.chat_url,
headers=headers,
data=json.dumps(payload, ensure_ascii=False),
timeout=60
)
if response.status_code == 200:
result = response.json()
return result["choices"][0]["message"]["content"]
else:
return f"请求失败,状态码: {response.status_code}, 响应: {response.text}"
except Exception as e:
return f"调用模型时出错: {str(e)}"
def test_connection(self):
"""测试模型连接"""
test_message = [{
"role": "user",
"content": "你好,请简单介绍一下你自己。"
}]
print("测试模型连接...")
response = self.chat_completion(test_message, max_tokens=100)
print(f"模型回复: {response}")
return response is not None
if __name__ == "__main__":
client = GLMClient()
if client.test_connection():
print("模型连接测试成功!")
else:
print("模型连接测试失败,请检查服务是否运行")
4.2 实现RAG问答系统
现在把向量检索和模型调用结合起来,创建rag_system.py:
from typing import List, Dict, Any
from model_client import GLMClient
from vector_store import VectorStoreManager
class RAGSystem:
def __init__(self, vector_store_path="./chroma_db"):
self.vector_manager = VectorStoreManager(vector_store_path)
self.vector_store = self.vector_manager.load_vector_store()
self.model_client = GLMClient()
def retrieve_relevant_docs(self, query: str, k: int = 5) -> List[str]:
"""检索相关文档"""
if not self.vector_store:
return ["向量数据库未加载,请先构建知识库"]
results = self.vector_store.similarity_search(query, k=k)
return [doc.page_content for doc in results]
def build_prompt(self, query: str, context_docs: List[str]) -> str:
"""构建包含上下文的提示词"""
context = "\n\n".join([f"[文档片段 {i+1}]: {doc}"
for i, doc in enumerate(context_docs)])
prompt = f"""基于以下上下文信息,回答用户的问题。如果上下文中有相关信息,请基于这些信息回答;如果没有,请根据你的知识回答,并说明信息不在提供的上下文中。
上下文信息:
{context}
用户问题:{query}
请用中文回答,保持专业、准确、简洁。"""
return prompt
def ask(self, query: str, use_context=True) -> str:
"""回答用户问题"""
if not use_context:
# 直接使用模型,不检索上下文
messages = [{"role": "user", "content": query}]
return self.model_client.chat_completion(messages)
# RAG流程:检索 + 生成
print(f"检索相关文档...")
relevant_docs = self.retrieve_relevant_docs(query)
if not relevant_docs or len(relevant_docs[0]) < 10:
print("未找到相关文档,将使用模型自身知识回答")
messages = [{"role": "user", "content": query}]
else:
print(f"找到 {len(relevant_docs)} 个相关文档片段")
prompt = self.build_prompt(query, relevant_docs)
messages = [{"role": "user", "content": prompt}]
print("调用模型生成回答...")
response = self.model_client.chat_completion(messages)
return response
def interactive_test(self):
"""交互式测试"""
print("=" * 60)
print("RAG系统测试模式")
print("输入 'quit' 退出测试")
print("=" * 60)
while True:
query = input("\n请输入问题: ").strip()
if query.lower() in ['quit', 'exit', '退出']:
break
if not query:
continue
print("\n" + "=" * 40)
print("正在处理...")
answer = self.ask(query)
print(f"\n回答: {answer}")
print("=" * 40)
if __name__ == "__main__":
# 初始化RAG系统
print("初始化RAG系统...")
rag_system = RAGSystem()
# 测试查询
test_queries = [
"RAG技术的主要优势是什么?",
"如何构建向量数据库?",
"GLM-4-9B模型有什么特点?"
]
for query in test_queries:
print(f"\n测试问题: {query}")
answer = rag_system.ask(query)
print(f"回答: {answer[:200]}...")
# 启动交互测试
rag_system.interactive_test()
4.3 创建Chainlit聊天界面
最后,创建Chainlit应用app.py:
import chainlit as cl
from rag_system import RAGSystem
import os
# 初始化RAG系统
@cl.on_chat_start
async def start():
"""聊天开始时的初始化"""
await cl.Message(
content="正在加载私有知识库问答系统..."
).send()
try:
# 初始化RAG系统
rag_system = RAGSystem()
# 检查向量数据库是否存在
if not os.path.exists("./chroma_db"):
await cl.Message(
content="检测到尚未构建知识库。请先将文档放入`documents`文件夹,然后运行`python vector_store.py`构建知识库。"
).send()
return
# 存储到用户会话中
cl.user_session.set("rag_system", rag_system)
await cl.Message(
content="系统已就绪!我可以基于您的私有文档回答问题。请开始提问吧!\n\n提示:您也可以上传新的文档文件,我会自动处理并更新知识库。"
).send()
except Exception as e:
await cl.Message(
content=f"系统初始化失败: {str(e)}"
).send()
@cl.on_message
async def main(message: cl.Message):
"""处理用户消息"""
rag_system = cl.user_session.get("rag_system")
if not rag_system:
await cl.Message(
content="系统未正确初始化,请刷新页面重试。"
).send()
return
# 显示思考状态
msg = cl.Message(content="")
await msg.send()
# 处理用户问题
user_query = message.content
# 添加思考动画
await msg.stream_token("正在检索相关文档... ")
# 获取回答
try:
answer = rag_system.ask(user_query)
# 流式输出回答
await msg.stream_token(f"找到答案:\n\n{answer}")
except Exception as e:
await msg.stream_token(f"处理问题时出错: {str(e)}")
@cl.on_file_upload
async def on_file_upload(files: list[cl.File]):
"""处理文件上传"""
# 这里可以扩展功能:自动处理上传的文档并更新知识库
file_names = [file.name for file in files]
await cl.Message(
content=f"收到文件: {', '.join(file_names)}\n\n文件上传功能已收到,文档自动处理功能正在开发中。目前请将文档放入`documents`文件夹后重新构建知识库。"
).send()
# Chainlit配置
if __name__ == "__main__":
# 运行Chainlit应用
from chainlit.cli import run_chainlit
run_chainlit(__file__)
创建Chainlit配置文件chainlit.md:
# 欢迎使用私有知识库问答系统
这是一个基于GLM-4-9B-Chat-1M和RAG技术的智能问答系统,可以基于您的私有文档回答问题。
## 功能特点
- 基于您的私有文档进行问答
- 智能检索最相关信息
- 自然流畅的对话体验
- 数据完全本地处理,安全私密
## 使用方法
1. 将您的文档(PDF、Word、TXT等)放入`documents`文件夹
2. 运行`python vector_store.py`构建知识库
3. 启动本应用开始问答
## 支持的问题类型
- 技术文档查询
- 产品规格咨询
- 项目资料检索
- 任何基于您文档的问题
5. 完整流程演示
现在让我们从头到尾走一遍完整的流程。
5.1 第一步:准备文档
假设我们有一些关于AI技术的文档:
RAG技术详解.pdf- 关于检索增强生成的详细介绍大模型部署指南.docx- 模型部署的最佳实践项目需求.txt- 当前项目的需求文档
把这些文档放入documents文件夹。
5.2 第二步:构建知识库
运行知识库构建脚本:
python vector_store.py
你会看到类似这样的输出:
开始加载文档...
成功加载: RAG技术详解.pdf (15页)
成功加载: 大模型部署指南.docx (8页)
成功加载: 项目需求.txt (1页)
开始分割文档...
文档分割完成,共生成 42 个文本片段
开始创建向量数据库...
向量数据库已创建并保存到: ./chroma_db
5.3 第三步:启动Chainlit应用
启动聊天界面:
chainlit run app.py
在浏览器中打开显示的地址(通常是http://localhost:8000)。
5.4 第四步:开始问答
现在你可以像和朋友聊天一样提问了:
你:我们项目中使用的RAG技术有什么优势?
系统:正在检索相关文档... 找到3个相关文档片段... 基于检索到的信息,RAG技术在我们项目中的主要优势包括:
- 准确性提升:通过检索相关文档片段,确保回答基于实际项目资料
- 信息实时性:可以随时更新知识库,无需重新训练模型
- 可解释性:每个回答都可以追溯到具体的文档来源
- 成本效益:相比微调大模型,RAG的实施成本更低...
你:如何部署GLM-4这样的模型?
系统:根据我们的部署指南,GLM-4模型的部署步骤包括:
- 环境准备:确保有足够的GPU内存和存储空间
- 模型下载:从官方渠道获取模型权重
- 服务部署:使用vLLM等推理框架部署...
- 性能测试:进行压力测试和效果评估...
6. 实用技巧与进阶功能
6.1 优化检索效果
如果你发现检索结果不够准确,可以尝试以下方法:
# 在vector_store.py中调整检索参数
class EnhancedVectorStoreManager(VectorStoreManager):
def __init__(self, persist_directory="./chroma_db"):
super().__init__(persist_directory)
def enhanced_retrieval(self, query: str, k: int = 5, score_threshold: float = 0.7):
"""增强版检索,带分数阈值"""
if not self.vector_store:
return []
# 使用相似度搜索带分数
results = self.vector_store.similarity_search_with_score(query, k=k*2)
# 过滤低分结果
filtered_results = [
doc for doc, score in results
if score >= score_threshold
][:k]
return filtered_results
6.2 添加对话历史
让系统记住之前的对话:
# 在rag_system.py中添加对话历史功能
class ConversationalRAGSystem(RAGSystem):
def __init__(self, vector_store_path="./chroma_db"):
super().__init__(vector_store_path)
self.conversation_history = []
def ask_with_history(self, query: str, max_history=5):
"""考虑对话历史的问答"""
# 添加上下文到查询中
if self.conversation_history:
history_context = "\n".join(
[f"用户: {q}\n助手: {a}"
for q, a in self.conversation_history[-max_history:]]
)
enhanced_query = f"对话历史:\n{history_context}\n\n当前问题:{query}"
else:
enhanced_query = query
# 获取回答
answer = self.ask(enhanced_query)
# 保存到历史
self.conversation_history.append((query, answer))
return answer
6.3 批量处理文档
如果你有大量文档,可以使用批量处理:
# 创建批量处理脚本 batch_process.py
import os
from document_processor import DocumentProcessor
from vector_store import VectorStoreManager
def batch_process_documents(folder_path):
"""批量处理文件夹中的文档"""
processor = DocumentProcessor(folder_path)
vector_manager = VectorStoreManager()
# 分批处理,避免内存不足
batch_size = 10
all_chunks = []
for i in range(0, len(os.listdir(folder_path)), batch_size):
batch_files = os.listdir(folder_path)[i:i+batch_size]
print(f"处理批次 {i//batch_size + 1}: {batch_files}")
# 处理当前批次
processor.docs_folder = folder_path
docs = processor.load_documents()
chunks = processor.split_documents(docs)
all_chunks.extend(chunks)
# 创建向量数据库
vector_store = vector_manager.create_vector_store(all_chunks)
return vector_store
if __name__ == "__main__":
batch_process_documents("./documents")
7. 常见问题解决
7.1 模型服务连接失败
问题:运行时报错,无法连接GLM模型服务。
解决:
# 1. 检查模型服务是否运行
cat /root/workspace/llm.log
# 2. 如果服务未运行,重启服务
cd /root/workspace
python -m vllm.entrypoints.openai.api_server \
--model /root/workspace/glm-4-9b-chat-1m \
--served-model-name glm-4-9b-chat-1m \
--port 8000
# 3. 在model_client.py中调整连接地址
# 如果使用CSDN镜像,地址可能是:
# base_url = "http://localhost:8000/v1"
7.2 内存不足错误
问题:处理大文档时出现内存错误。
解决:
# 调整文本分割参数,减小chunk_size
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # 从1000减小到500
chunk_overlap=100,
)
# 或者分批处理文档
7.3 检索结果不准确
问题:系统检索到的文档不相关。
解决:
- 优化文档结构:确保文档有清晰的标题和段落
- 调整检索参数:增加检索数量k值
- 使用混合检索:结合关键词检索和向量检索
def hybrid_retrieval(query, vector_store, k=5):
"""混合检索:向量检索 + 关键词匹配"""
# 向量检索
vector_results = vector_store.similarity_search(query, k=k)
# 关键词检索(简单实现)
keyword_results = []
for doc in vector_store.get():
if any(keyword in doc.page_content.lower()
for keyword in query.lower().split()):
keyword_results.append(doc)
# 合并结果,去重
all_results = vector_results + keyword_results
unique_results = []
seen_content = set()
for doc in all_results:
if doc.page_content[:100] not in seen_content:
seen_content.add(doc.page_content[:100])
unique_results.append(doc)
return unique_results[:k]
7.4 回答质量不高
问题:模型回答不够准确或详细。
解决:
- 优化提示词:在
build_prompt函数中调整提示词模板 - 增加上下文:检索更多的文档片段(增加k值)
- 后处理回答:对模型回答进行校验和补充
def improved_prompt_template(query, context_docs):
"""改进的提示词模板"""
context = "\n\n".join([
f"【来源 {i+1}】\n{doc}"
for i, doc in enumerate(context_docs)
])
prompt = f"""你是一个专业的助手,需要基于以下参考资料回答问题。
参考资料:
{context}
用户问题:{query}
请按照以下要求回答:
1. 严格基于参考资料,不要编造信息
2. 如果参考资料中有相关信息,请引用具体来源
3. 如果资料不足,请诚实地说明
4. 回答要结构清晰,分点说明
现在请回答:"""
return prompt
8. 总结
通过这个教程,我们完成了一个完整的私有知识库问答系统的搭建。让我们回顾一下关键收获:
技术栈的价值
- GLM-4-9B-Chat-1M:提供了强大的理解和生成能力,百万字上下文让它能处理复杂的文档
- RAG技术:解决了大模型“知识截止”和“胡编乱造”的问题,让回答更准确可靠
- Chainlit:让整个系统有了友好的聊天界面,使用体验大大提升
- 向量数据库:实现了文档的智能检索,毫秒级找到相关信息
实际应用场景 这个系统不仅是一个技术演示,它可以在很多实际场景中发挥作用:
- 企业内部知识库:新员工培训、技术文档查询
- 个人学习助手:整理学习笔记、快速查找资料
- 客户支持系统:基于产品文档回答客户问题
- 研究文献分析:快速从大量论文中提取信息
下一步建议 如果你想让这个系统更强大,可以考虑:
- 支持更多格式:添加Excel、PPT、图片OCR等支持
- 多源知识库:连接数据库、API等外部数据源
- 权限管理:不同用户访问不同文档
- 自动更新:监控文档变化,自动更新向量数据库
最重要的是,现在你已经有了一个完全可控、私有的智能问答系统。你可以根据自己的需求随意调整和扩展,而不用担心数据泄露或服务不稳定。
获取更多AI镜像
想探索更多AI镜像和应用场景?访问 CSDN星图镜像广场,提供丰富的预置镜像,覆盖大模型推理、图像生成、视频生成、模型微调等多个领域,支持一键部署。
更多推荐

所有评论(0)