1. 项目概述:基于RAG的本地文档智能问答系统
在人工智能技术快速发展的今天,大语言模型(LLM)已经展现出惊人的文本理解和生成能力。然而,这些模型在实际应用中仍面临两个关键挑战:一是知识更新滞后,无法及时获取最新信息;二是容易产生"幻觉",即生成看似合理但实际错误的内容。检索增强生成(RAG)技术正是解决这些问题的有效方案。
本项目将构建一个完整的本地文档智能问答系统,核心功能包括:
- 支持PDF、DOCX、TXT、MD等多种格式文档上传
- 自动解析文档内容并进行向量化存储
- 基于检索到的文档片段生成准确回答
- 支持多轮对话和上下文理解
- 提供简洁的Web交互界面
这个系统特别适合以下场景:
- 企业内部知识库问答
- 个人文档管理和检索
- 教育领域的资料查询
- 法律、医疗等专业领域的文档分析
2. 系统架构与工作流程
2.1 整体架构设计
系统采用三层架构设计,各层职责明确:
-
前端层:
- 基于Streamlit构建的Web界面
- 提供文件上传、聊天交互等功能
- 实现流式输出提升用户体验
-
服务层:
- 文档处理模块:解析、分块和向量化
- 检索增强模块:相似度查询和结果重排
- 上下文记忆模块:维护对话历史
- LLM调用模块:生成最终回答
-
基础服务层:
- 嵌入模型:文本向量化
- 向量数据库:高效相似性检索
- 重排模型:提升检索精度
- 大语言模型:生成自然语言回答
2.2 核心工作流程
系统工作流程分为两个主要阶段:
文档处理阶段(知识入库)
- 解析提取:使用专用库(pypdf、docx2txt等)提取文档文本内容
- 文本分割:按语义将长文本切分为适当大小的块
- 向量化:通过嵌入模型将文本转换为高维向量
- 存储:将文本向量存入向量数据库
用户查询阶段(知识检索与回答)
- 查询向量化:将用户问题转换为向量
- 相似度查询:在向量库中查找最相关的文本块
- 结果重排:对初步结果进行精细排序
- 组合提示词:整合问题、检索结果和上下文
- 生成响应:由LLM生成最终回答
3. 关键技术选型与实现
3.1 技术栈选择
经过综合评估,我们选择了以下技术方案:
- 前端框架:Streamlit(快速构建交互式Web应用)
- 文档处理:LangChain(统一处理不同格式文档)
- 嵌入模型:
- 云端:Qwen的text-embedding-v4
- 本地:BGE-small-zh-v1.5
- 向量数据库:Chroma(轻量级嵌入式方案)
- 大语言模型:
- 默认:Qwen-plus
- 备选:DeepSeek、GPT-4等
- 开发语言:Python 3.8+
3.2 环境配置详解
3.2.1 Conda环境搭建
推荐使用Miniconda管理Python环境:
bash复制# 创建专用环境
conda create -n rag-env python=3.11
conda activate rag-env
# 配置清华镜像源(国内用户)
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free/
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main/
conda config --set show_channel_urls yes
3.2.2 依赖安装
创建requirements.txt文件:
text复制streamlit==1.46.0
langchain==0.3.26
langchain-chroma==0.2.4
python-dotenv==1.1.0
pypdf==5.6.1
dashscope==1.23.5
sentence-transformers==5.1.2
使用阿里云镜像加速安装:
bash复制pip install -r requirements.txt -i https://mirrors.aliyun.com/pypi/simple/
3.2.3 环境变量配置
创建.env文件配置API密钥:
text复制# 千问配置
QWEN_API_KEY=your_api_key
QWEN_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
# OpenAI配置(可选)
OPENAI_API_KEY=your_api_key
OPENAI_BASE_URL=your_base_url
3.3 核心模块实现
3.3.1 自定义嵌入模型适配
为解决官方DashScopeEmbeddings批量处理限制问题,我们重写了相关类:
python复制from langchain_core.embeddings import Embeddings
from pydantic import BaseModel
class DashScopeEmbeddings(BaseModel, Embeddings):
"""自定义千问嵌入模型适配"""
def __init__(self, model="text-embedding-v4", max_retries=5):
self.model = model
self.max_retries = max_retries
self.client = self._init_client()
def _init_client(self):
"""初始化API客户端"""
import dashscope
dashscope.api_key = os.getenv("QWEN_API_KEY")
return dashscope.TextEmbedding
def embed_documents(self, texts: List[str]) -> List[List[float]]:
"""批量生成文档嵌入"""
embeddings = []
batch_size = 6 # 适配API限制
for i in range(0, len(texts), batch_size):
batch = texts[i:i+batch_size]
response = self._embed_with_retry(input=batch)
embeddings.extend([item["embedding"] for item in response])
return embeddings
3.3.2 嵌入模型统一接口
提供多种嵌入模型支持:
python复制def initialize_embedding_model(provider="qwen"):
"""初始化指定提供商的嵌入模型"""
if provider == "qwen":
return DashScopeEmbeddings(model="text-embedding-v4")
elif provider == "local_bge_small":
return HuggingFaceEmbeddings(
model_name="BAAI/bge-small-zh-v1.5",
model_kwargs={'device': 'cpu'},
encode_kwargs={'normalize_embeddings': True}
)
elif provider == "openai":
return OpenAIEmbeddings(model="text-embedding-ada-002")
3.3.3 LLM模型调用封装
支持多种大语言模型:
python复制MODEL_CONFIG = {
"qwen": {
"api_key_env": "QWEN_API_KEY",
"default_model": "qwen-plus"
},
"deepseek": {
"api_key_env": "DEEPSEEK_API_KEY",
"default_model": "deepseek-chat"
}
}
def initialize_llm(model_type="qwen", temperature=0.0):
"""统一LLM初始化接口"""
config = MODEL_CONFIG[model_type]
api_key = os.getenv(config["api_key_env"])
if model_type == "deepseek":
return init_chat_model(
model=config["default_model"],
api_key=api_key,
temperature=temperature
)
else:
return ChatOpenAI(
model=config["default_model"],
api_key=api_key,
temperature=temperature
)
3.3.4 重排模型实现
使用CrossEncoder提升检索精度:
python复制class Reranker:
"""基于CrossEncoder的检索结果重排"""
def __init__(self, model_name="BAAI/bge-reranker-large"):
self.model = CrossEncoder(model_name)
def rerank(self, query: str, documents: List[Document], top_n=3) -> List[Document]:
"""对检索结果进行重排序"""
pairs = [(query, doc.page_content) for doc in documents]
scores = self.model.predict(pairs)
# 关联分数到文档并排序
scored_docs = zip(documents, scores)
sorted_docs = sorted(scored_docs, key=lambda x: x[1], reverse=True)
# 返回top_n结果
return [doc for doc, _ in sorted_docs[:top_n]]
3.3.5 RAG核心服务
完整实现RAG流程:
python复制class RAGService:
"""RAG核心服务类"""
def __init__(self, persist_dir="chroma_db"):
self.vectordb = Chroma(
embedding_function=initialize_embedding_model(),
persist_directory=persist_dir
)
self.llm = initialize_llm()
self.reranker = Reranker()
def ingest_document(self, file_path: str):
"""文档处理入库"""
loader = self._get_loader(file_path)
documents = loader.load()
# 文本分块
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50
)
chunks = text_splitter.split_documents(documents)
# 向量化存储
self.vectordb.add_documents(chunks)
def query(self, question: str) -> str:
"""问答处理流程"""
# 检索相关文档
docs = self.vectordb.similarity_search(question, k=8)
# 结果重排
ranked_docs = self.reranker.rerank(question, docs, top_n=3)
# 构造提示词
context = "\n".join([d.page_content for d in ranked_docs])
prompt = f"基于以下上下文回答问题:\n{context}\n\n问题:{question}"
# 生成回答
return self.llm.invoke(prompt).content
4. 系统部署与使用
4.1 项目结构
完整项目目录结构如下:
code复制simple_rag_assistant/
├── .env # 环境变量配置
├── requirements.txt # 依赖列表
├── main.py # Streamlit主界面
├── models/
│ ├── custom_dashscope_embedding.py # 自定义嵌入模型
│ ├── langchain_embedding.py # 嵌入模型接口
│ ├── langchain_llm.py # LLM调用
│ └── reranker_model.py # 重排模型
└── services/
└── rag_service_stream.py # RAG核心服务
4.2 Streamlit界面实现
创建交互式Web界面:
python复制import streamlit as st
from services.rag_service_stream import RAGService
# 初始化RAG服务
@st.cache_resource
def get_rag_service():
return RAGService()
rag = get_rag_service()
# 页面布局
st.title("📄 本地文档智能问答系统")
st.write("上传文档后即可进行问答")
# 文件上传区
uploaded_file = st.file_uploader("选择文档", type=["pdf", "docx", "txt"])
if uploaded_file:
with st.spinner("处理文档中..."):
# 保存临时文件并处理
with tempfile.NamedTemporaryFile(delete=False) as tmp:
tmp.write(uploaded_file.getvalue())
rag.ingest_document(tmp.name)
st.success("文档处理完成!")
# 聊天交互区
if "messages" not in st.session_state:
st.session_state.messages = []
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.markdown(message["content"])
if prompt := st.chat_input("请输入您的问题"):
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.markdown(prompt)
with st.chat_message("assistant"):
with st.spinner("思考中..."):
response = rag.query(prompt)
st.markdown(response)
st.session_state.messages.append({"role": "assistant", "content": response})
4.3 系统启动
运行以下命令启动系统:
bash复制streamlit run main.py
启动后,浏览器会自动打开系统界面,默认地址为http://localhost:8501。
5. 性能优化与问题排查
5.1 常见问题解决方案
-
文档解析失败
- 现象:上传特定PDF文件时报错
- 原因:加密PDF或特殊格式
- 解决:尝试使用
pdf2text等替代解析工具
-
检索结果不相关
- 现象:回答与问题无关
- 原因:分块大小不合适或嵌入模型不匹配
- 解决:调整
chunk_size或更换嵌入模型
-
响应速度慢
- 现象:问答延迟高
- 原因:LLM API延迟或本地计算资源不足
- 解决:使用更轻量级模型或增加本地GPU资源
5.2 性能优化建议
-
分块策略优化
- 根据文档类型调整分块大小:
- 技术文档:300-500字符
- 叙述性内容:500-800字符
- 设置适当重叠(10-20%)保持上下文
- 根据文档类型调整分块大小:
-
缓存机制
- 对常见问题缓存回答
- 向量数据库持久化避免重复处理
-
异步处理
- 使用异步IO提高并发性能
- 文档处理与问答分离
6. 扩展与进阶
6.1 功能扩展方向
-
多文档管理
- 实现文档分类和标签
- 支持文档删除和更新
-
混合检索策略
- 结合关键词和向量检索
- 加入元数据过滤
-
回答验证
- 自动验证生成内容的准确性
- 提供参考文献和出处
6.2 技术进阶建议
-
微调嵌入模型
- 使用领域数据微调提升相关性
- 尝试更大的嵌入模型
-
提示工程优化
- 设计更有效的提示模板
- 实现动态提示生成
-
评估体系构建
- 建立量化评估指标
- 定期测试系统性能
在实际部署中,我发现以下几个经验特别有价值:
- 对于技术文档,适当减小分块大小(300字符左右)能提高检索精度
- 在重排阶段加入元数据(如标题重要性)能显著改善结果
- 定期清理向量数据库中的陈旧文档可以维持系统性能
- 为常见问题设置快捷回答能大幅提升用户体验
