前言
在2026年的企业数字化转型浪潮中，AI智能客服已经成为各行业降本增效的核心工具。与传统关键词匹配客服不同，基于RAG（检索增强生成）技术的智能客服能够理解上下文、从企业知识库中精准检索信息，并生成自然流畅的回答。本文将带你从零构建一个企业级知识库问答系统，涵盖文档处理、向量检索、大模型调用、提示词优化等核心环节。完成这个项目后，你将掌握构建真实AI应用的全套技能。
项目概述与技术架构
项目目标
我们将构建一个企业知识库智能问答系统，具备以下核心能力：支持上传企业文档（PDF、Word、TXT等格式）、自动向量化存储、基于语义检索匹配相关文档片段、大模型生成准确答案。这个系统可以部署到企业内网，为员工提供产品文档查询、内部流程咨询、技术问题解答等服务。
技术栈选型
本项目采用以下技术组合，每个组件都经过2026年企业级应用验证：
LangChain框架负责整体流程编排，它提供了文档加载、文本分割、向量存储、检索链等完整工具链，版本选择langchain-0.3.x系列。向量数据库选用Chroma，这是一个轻量级但功能完整的向量库，支持本地部署，0.5.x版本对检索性能做了显著优化，Windows、macOS、Linux三大平台均可直接安装。Embedding模型使用sentence-transformers的all-MiniLM-L6-v2，这个模型体积小（约90MB）、速度快、效果稳定，中文支持良好。大模型层可以根据实际需求选择OpenAI GPT-4o、国内的通义千问V5、DeepSeek-V3，或者通过Ollama本地部署的开源模型如Llama 3、Qwen2。
整体架构分为五个层级：文档层负责处理各类企业文档；处理层完成文本分割和向量化；存储层管理Chroma向量数据库；检索层实现语义相似度匹配；生成层调用大模型输出最终答案。
环境准备与依赖安装
系统要求
确保你的开发环境满足以下条件：Python版本3.10或更高，Windows系统需要Windows 10/11，macOS需要10.15及以上，Linux推荐Ubuntu 20.04或CentOS 8。磁盘空间建议预留5GB以上用于存储向量数据库。
安装步骤
创建项目目录并进入：
Windows系统打开PowerShell执行：
powershell
mkdir ai-knowledge-qa && cd ai-knowledge-qa
python -m venv venv
.\venv\Scripts\activate
pip install langchain langchain-community langchain-openai chromadb sentence-transformers pypdf python-docx python-dotenv tiktoken
macOS和Linux系统打开终端执行：
bash
mkdir -p ai-knowledge-qa && cd ai-knowledge-qa
python3 -m venv venv
source venv/bin/activate
pip install langchain langchain-community langchain-openai chromadb sentence-transformers pypdf python-docx python-dotenv tiktoken
如果使用本地模型（通过Ollama部署），需要额外安装：
bash
pip install langchain-ollama
并确保Ollama已在本地运行：
bash
# macOS/Linux安装Ollama
curl -fsSL https://ollama.com/install.sh | sh
# Windows从 https://ollama.com/download 下载安装
# 拉取模型
ollama pull qwen2.5:7b # 中文支持好的开源模型
项目目录结构
创建完成后，项目结构如下：
plaintext
ai-knowledge-qa/
├── config/
│ └── settings.py # 配置文件
├── data/
│ └── knowledge_base/ # 知识库文档目录
├── src/
│ ├── document_loader.py # 文档加载模块
│ ├── text_splitter.py # 文本分割模块
│ ├── vector_store.py # 向量存储模块
│ ├── retriever.py # 检索模块
│ ├── chain.py # 问答链模块
│ └── main.py # 主程序入口
├── .env # 环境变量（API密钥等）
├── requirements.txt # 依赖清单
└── README.md # 项目说明
核心模块实现
配置管理
在config/settings.py中集中管理所有配置，便于后续调整：
python
# config/settings.py
import os
from dotenv import load_dotenv
load_dotenv()
class Settings:
# 大模型配置
LLM_PROVIDER = os.getenv("LLM_PROVIDER", "openai") # openai, ollama, zhipu
# OpenAI配置（如使用OpenAI）
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o")
OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
# Ollama配置（如使用本地模型）
OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434")
OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "qwen2.5:7b")
# 向量数据库配置
CHROMA_PERSIST_DIRECTORY = os.getenv("CHROMA_PERSIST_DIRECTORY", "./data/chroma_db")
EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "sentence-transformers/all-MiniLM-L6-v2")
# 文本分割配置
CHUNK_SIZE = int(os.getenv("CHUNK_SIZE", 500))
CHUNK_OVERLAP = int(os.getenv("CHUNK_OVERLAP", 50))
# 检索配置
TOP_K = int(os.getenv("TOP_K", 4)) # 检索返回的相关文档数
settings = Settings()
.env文件示例：
env
# 选择大模型提供商：openai, ollama, zhipu
LLM_PROVIDER=ollama
# Ollama配置（使用本地模型）
OLLAMA_BASE_URL=http://localhost:11434
OLLAMA_MODEL=qwen2.5:7b
# 向量数据库路径
CHROMA_PERSIST_DIRECTORY=./data/chroma_db
# 文本分割参数
CHUNK_SIZE=500
CHUNK_OVERLAP=50
# 检索数量
TOP_K=4
文档加载模块
支持多种文档格式的加载是企业知识库的基础需求：
python
# src/document_loader.py
from langchain_community.document_loaders import (
PyPDFLoader, # PDF文档
Docx2txtLoader, # Word文档(.docx)
TextLoader, # 纯文本文件
UnstructuredMarkdownLoader # Markdown文件
)
from langchain.schema import Document
from pathlib import Path
import logging
logger = logging.getLogger(__name__)
class DocumentLoader:
"""企业文档加载器，支持多种格式"""
def __init__(self):
self.loader_map = {
'.pdf': PyPDFLoader,
'.docx': Docx2txtLoader,
'.txt': TextLoader,
'.md': UnstructuredMarkdownLoader
}
def load_file(self, file_path: str) -> list[Document]:
"""加载单个文件"""
path = Path(file_path)
suffix = path.suffix.lower()
if suffix not in self.loader_map:
logger.warning(f"不支持的文件格式: {suffix}")
return []
try:
loader_class = self.loader_map[suffix]
loader = loader_class(str(path))
documents = loader.load()
# 添加元数据：来源文件名
for doc in documents:
doc.metadata["source"] = path.name
logger.info(f"成功加载文件: {path.name}, 包含 {len(documents)} 个文档块")
return documents
except Exception as e:
logger.error(f"加载文件失败 {path.name}: {str(e)}")
return []
def load_directory(self, directory: str, extensions: list = None) -> list[Document]:
"""批量加载目录下的所有文档"""
if extensions is None:
extensions = ['.pdf', '.docx', '.txt', '.md']
all_docs = []
dir_path = Path(directory)
for ext in extensions:
for file_path in dir_path.rglob(f"*{ext}"):
docs = self.load_file(str(file_path))
all_docs.extend(docs)
logger.info(f"目录 {directory} 共加载 {len(all_docs)} 个文档块")
return all_docs
# 使用示例
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
loader = DocumentLoader()
# 加载单个文件
docs = loader.load_file("./data/knowledge_base/product_manual.pdf")
print(f"加载了 {len(docs)} 个文档块")
文本分割模块
将长文档合理分割是检索效果的关键。分割太短会丢失上下文，分割太长则引入噪声：
python
# src/text_splitter.py
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document
from config.settings import settings
class TextSplitter:
"""智能文本分割器，保持语义完整性"""
def __init__(
self,
chunk_size: int = None,
chunk_overlap: int = None,
length_function = len
):
self.chunk_size = chunk_size or settings.CHUNK_SIZE
self.chunk_overlap = chunk_overlap or settings.CHUNK_OVERLAP
self.length_function = length_function
# 使用递归字符分割器，按优先级尝试不同的分隔符
self.splitter = RecursiveCharacterTextSplitter(
separators=[
"\n\n", # 段落分隔（优先，保持语义完整）
"\n", # 换行符
"。", # 中文句号
"！", # 中文感叹号
"？", # 中文问号
"，", # 中文逗号
" ", # 英文空格
"" # 单字符（最后兜底）
],
chunk_size=self.chunk_size,
chunk_overlap=self.chunk_overlap,
length_function=length_function,
add_start_index=True # 添加起始位置索引
)
def split_documents(self, documents: list[Document]) -> list[Document]:
"""分割文档列表"""
if not documents:
return []
# 去重：检查是否有重复内容的文档块
seen_content = set()
unique_docs = []
for doc in documents:
content_hash = hash(doc.page_content)
if content_hash not in seen_content:
seen_content.add(content_hash)
unique_docs.append(doc)
# 执行分割
splits = self.splitter.split_documents(unique_docs)
print(f"原始文档块: {len(unique_docs)}, 分割后: {len(splits)}")
return splits
def split_text(self, text: str) -> list[str]:
"""分割单段文本"""
return self.splitter.split_text(text)
# 使用示例
if __name__ == "__main__":
from src.document_loader import DocumentLoader
loader = DocumentLoader()
docs = loader.load_file("./data/knowledge_base/sample.txt")
splitter = TextSplitter(chunk_size=500, chunk_overlap=50)
splits = splitter.split_documents(docs)
print(f"分割后的文档块数: {len(splits)}")
for i, split in enumerate(splits[:3]):
print(f"\n--- 文档块 {i+1} ---")
print(f"内容长度: {len(split.page_content)} 字符")
print(f"内容预览: {split.page_content[:100]}...")
向量存储模块
使用Chroma存储文档向量，支持持久化和增量更新：
python
# src/vector_store.py
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain.schema import Document
from config.settings import settings
from pathlib import Path
import logging
logger = logging.getLogger(__name__)
class VectorStore:
"""向量数据库管理器"""
def __init__(self):
self.embeddings = HuggingFaceEmbeddings(
model_name=settings.EMBEDDING_MODEL,
model_kwargs={'device': 'cpu'}, # 可改为 'cuda' 使用GPU加速
encode_kwargs={'normalize_embeddings': True} # 归一化向量
)
self.vectorstore = None
def create_vectorstore(self, documents: list[Document], persist_directory: str = None) -> Chroma:
"""从文档创建向量存储"""
if not documents:
raise ValueError("文档列表不能为空")
persist_dir = persist_directory or settings.CHROMA_PERSIST_DIRECTORY
Path(persist_dir).parent.mkdir(parents=True, exist_ok=True)
logger.info(f"开始创建向量数据库，文档数: {len(documents)}")
self.vectorstore = Chroma.from_documents(
documents=documents,
embedding=self.embeddings,
persist_directory=persist_dir
)
logger.info(f"向量数据库创建完成，存储路径: {persist_dir}")
return self.vectorstore
def load_vectorstore(self, persist_directory: str = None) -> Chroma:
"""加载已存在的向量数据库"""
persist_dir = persist_directory or settings.CHROMA_PERSIST_DIRECTORY
if not Path(persist_dir).exists():
raise FileNotFoundError(f"向量数据库不存在: {persist_dir}")
self.vectorstore = Chroma(
persist_directory=persist_dir,
embedding_function=self.embeddings
)
logger.info(f"向量数据库加载完成，文档数: {self.vectorstore._collection.count()}")
return self.vectorstore
def add_documents(self, documents: list[Document]):
"""向现有向量库添加文档"""
if self.vectorstore is None:
raise RuntimeError("向量数据库未初始化，请先创建或加载")
self.vectorstore.add_documents(documents)
logger.info(f"添加了 {len(documents)} 个文档块")
def get_retriever(self, search_kwargs: dict = None):
"""获取检索器"""
if self.vectorstore is None:
raise RuntimeError("向量数据库未初始化")
default_search_kwargs = {"k": settings.TOP_K}
if search_kwargs:
default_search_kwargs.update(search_kwargs)
return self.vectorstore.as_retriever(
search_type="similarity", # 基于相似度的检索
search_kwargs=default_search_kwargs
)
def similarity_search(self, query: str, k: int = None) -> list[Document]:
"""直接执行相似度搜索"""
if self.vectorstore is None:
raise RuntimeError("向量数据库未初始化")
k = k or settings.TOP_K
return self.vectorstore.similarity_search(query, k=k)
def similarity_search_with_score(self, query: str, k: int = None) -> list[tuple]:
"""带相似度分数的搜索"""
if self.vectorstore is None:
raise RuntimeError("向量数据库未初始化")
k = k or settings.TOP_K
return self.vectorstore.similarity_search_with_score(query, k=k)
def delete_collection(self):
"""删除向量库集合"""
if self.vectorstore:
self.vectorstore.delete_collection()
self.vectorstore = None
logger.info("向量库已清空")
# 使用示例
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
from src.document_loader import DocumentLoader
from src.text_splitter import TextSplitter
# 加载文档
loader = DocumentLoader()
docs = loader.load_directory("./data/knowledge_base")
# 分割文档
splitter = TextSplitter()
splits = splitter.split_documents(docs)
# 创建向量库
vector_store = VectorStore()
vector_store.create_vectorstore(splits)
# 测试检索
results = vector_store.similarity_search("产品的退换货政策是什么？")
print(f"\n检索到 {len(results)} 个相关文档块")
for i, doc in enumerate(results):
print(f"\n--- 结果 {i+1} ---")
print(doc.page_content[:200])
检索模块
检索质量直接影响最终回答的准确性，这里实现几种高级检索策略：
python
# src/retriever.py
from langchain.schema import Document
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from typing import Optional
class Retriever:
"""高级检索器，支持多种检索策略"""
def __init__(self, vectorstore, documents: list[Document] = None):
self.vectorstore = vectorstore
self.documents = documents
def get_basic_retriever(self, search_kwargs: dict = None):
"""获取基础向量检索器"""
return self.vectorstore.as_retriever(
search_type="similarity",
search_kwargs=search_kwargs or {"k": 4}
)
def get_ensemble_retriever(self, weight_vector: float = 0.5, weight_bm25: float = 0.5):
"""获取融合检索器（向量检索 + BM25关键词检索）"""
if not self.documents:
raise ValueError("融合检索需要提供原始文档列表")
# 向量检索器
vector_retriever = self.get_basic_retriever()
# BM25关键词检索器
bm25_retriever = BM25Retriever.from_documents(self.documents)
bm25_retriever.k = 4
# 融合两种检索结果
ensemble_retriever = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[weight_vector, weight_bm25] # 权重可调整
)
return ensemble_retriever
def get_contextual_compression_retriever(self, base_retriever, compressor):
"""获取上下文压缩检索器，减少检索结果中的冗余内容"""
from langchain.retrievers import ContextualCompressionRetriever
return ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=base_retriever
)
class SearchResultFormatter:
"""检索结果格式化工具"""
@staticmethod
def format_results(results: list[Document], max_content_length: int = 500) -> str:
"""格式化检索结果，便于查看和调试"""
if not results:
return "未找到相关结果"
formatted = []
for i, doc in enumerate(results, 1):
content = doc.page_content
# 截断过长内容
if len(content) > max_content_length:
content = content[:max_content_length] + "..."
source = doc.metadata.get("source", "未知来源")
formatted.append(
f"【结果 {i}】来源: {source}\n"
f"内容: {content}\n"
)
return "\n".join(formatted)
@staticmethod
def extract_sources(results: list[Document]) -> list[str]:
"""提取检索结果的所有来源"""
sources = []
for doc in results:
source = doc.metadata.get("source", "未知")
if source not in sources:
sources.append(source)
return sources
# 使用示例
if __name__ == "__main__":
from src.vector_store import VectorStore
vector_store = VectorStore()
vector_store.load_vectorstore()
retriever_manager = Retriever(vector_store.vectorstore)
# 获取基础检索器
retriever = retriever_manager.get_basic_retriever()
# 执行检索
query = "如何申请产品退换货？"
docs = retriever.get_relevant_documents(query)
# 格式化输出
formatter = SearchResultFormatter()
print(formatter.format_results(docs))
print(f"\n涉及来源: {formatter.extract_sources(docs)}")
问答链模块
整合检索和生成，构建完整的RAG问答链：
python
# src/chain.py
from langchain_openai import ChatOpenAI
from langchain_ollama import ChatOllama
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
from langchain.schema import BaseRetriever
from config.settings import settings
import logging
logger = logging.getLogger(__name__)
class QAChain:
"""问答链管理器"""
def __init__(self, retriever: BaseRetriever):
self.retriever = retriever
self.llm = self._create_llm()
self.qa_chain = self._create_qa_chain()
def _create_llm(self):
"""创建大模型实例"""
provider = settings.LLM_PROVIDER.lower()
if provider == "openai":
return ChatOpenAI(
model=settings.OPENAI_MODEL,
api_key=settings.OPENAI_API_KEY,
base_url=settings.OPENAI_BASE_URL,
temperature=0.3, # 较低的随机性，保持回答一致性
max_tokens=1000 # 限制回答长度
)
elif provider == "ollama":
return ChatOllama(
base_url=settings.OLLAMA_BASE_URL,
model=settings.OLLAMA_MODEL,
temperature=0.3,
options={"num_ctx": 4096} # 上下文窗口大小
)
elif provider == "zhipu":
# 智谱AI配置
return ChatOpenAI(
model="glm-4",
api_key=settings.OPENAI_API_KEY,
base_url="https://open.bigmodel.cn/api/paas/v4",
temperature=0.3
)
else:
raise ValueError(f"不支持的大模型提供商: {provider}")
def _create_qa_chain(self) -> RetrievalQA:
"""创建问答链"""
# 自定义提示词模板，引导模型基于检索结果回答
prompt_template = """你是一个企业知识库问答助手。请根据以下检索到的文档内容，回答用户的问题。
如果检索结果中没有相关信息，请直接说明"抱歉，知识库中暂无相关内容"，不要编造答案。
如果检索结果中有相关信息，请基于内容给出准确回答，并在回答末尾注明参考来源。
---
检索到的文档内容:
{context}
---
用户问题: {question}
请给出回答:"""
PROMPT = PromptTemplate(
template=prompt_template,
input_variables=["context", "question"]
)
# 创建检索问答链
qa_chain = RetrievalQA.from_chain_type(
llm=self.llm,
chain_type="stuff", # 将所有检索结果拼接后一次输入
retriever=self.retriever,
return_source_documents=True, # 返回参考文档
chain_type_kwargs={
"prompt": PROMPT,
"document_variable_name": "context"
},
verbose=True
)
return qa_chain
def ask(self, question: str) -> dict:
"""向知识库提问"""
logger.info(f"收到问题: {question}")
result = self.qa_chain({"query": question})
return {
"answer": result["result"],
"source_documents": result["source_documents"]
}
def ask_with_sources(self, question: str) -> str:
"""提问并格式化返回带来源的回答"""
result = self.ask(question)
answer = result["answer"]
sources = result["source_documents"]
# 添加参考来源
if sources:
source_names = list(set([doc.metadata.get("source", "未知") for doc in sources]))
answer += f"\n\n📚 参考来源: {', '.join(source_names)}"
return answer
# 使用示例
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
from src.vector_store import VectorStore
from src.retriever import Retriever
# 加载向量库
vector_store = VectorStore()
vector_store.load_vectorstore()
# 创建检索器
retriever_manager = Retriever(vector_store.vectorstore)
retriever = retriever_manager.get_basic_retriever()
# 创建问答链
qa_chain = QAChain(retriever)
# 提问测试
question = "我们公司的产品质保期是多久？"
answer = qa_chain.ask_with_sources(question)
print(f"\n问题: {question}")
print(f"\n回答:\n{answer}")
主程序入口
整合所有模块，提供命令行交互界面：
python
# src/main.py
import argparse
import logging
from pathlib import Path
from config.settings import settings
from src.document_loader import DocumentLoader
from src.text_splitter import TextSplitter
from src.vector_store import VectorStore
from src.retriever import Retriever, SearchResultFormatter
from src.chain import QAChain
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class KnowledgeBaseQA:
"""企业知识库问答系统主类"""
def __init__(self, knowledge_base_path: str = None):
self.knowledge_base_path = knowledge_base_path or "./data/knowledge_base"
self.vector_store = None
self.qa_chain = None
self.loader = DocumentLoader()
self.splitter = TextSplitter()
self.retriever_manager = None
def initialize(self, rebuild: bool = False):
"""初始化系统"""
print("=" * 50)
print("🏢 企业知识库智能问答系统")
print("=" * 50)
# 检查或创建向量库
vector_db_path = Path(settings.CHROMA_PERSIST_DIRECTORY)
if rebuild or not vector_db_path.exists():
logger.info("正在构建向量数据库...")
self._build_vectorstore()
else:
logger.info("正在加载已有向量数据库...")
self.vector_store = VectorStore()
self.vector_store.load_vectorstore()
# 创建检索和问答链
self.retriever_manager = Retriever(self.vector_store.vectorstore)
retriever = self.retriever_manager.get_basic_retriever()
self.qa_chain = QAChain(retriever)
print("✅ 系统初始化完成！")
print(f"📁 知识库路径: {self.knowledge_base_path}")
print(f"💬 大模型: {settings.LLM_PROVIDER} - {getattr(settings, f'{settings.LLM_PROVIDER.upper()}_MODEL', 'N/A')}")
print("-" * 50)
def _build_vectorstore(self):
"""构建向量数据库"""
# 加载文档
print("📂 正在加载文档...")
documents = self.loader.load_directory(self.knowledge_base_path)
if not documents:
print("⚠️ 未找到任何文档，请先将文档放入知识库目录！")
print(f" 路径: {self.knowledge_base_path}")
return
print(f" 已加载 {len(documents)} 个文档块")
# 分割文档
print("✂️ 正在分割文档...")
splits = self.splitter.split_documents(documents)
print(f" 分割为 {len(splits)} 个文本块")
# 创建向量库
print("🔢 正在生成向量并存储...")
self.vector_store = VectorStore()
self.vector_store.create_vectorstore(splits)
def chat(self, question: str) -> str:
"""回答问题"""
if not self.qa_chain:
return "系统未初始化，请先运行 initialize()"
try:
return self.qa_chain.ask_with_sources(question)
except Exception as e:
logger.error(f"问答失败: {str(e)}")
return f"抱歉，处理您的问题时出现错误: {str(e)}"
def interactive(self):
"""交互式问答"""
print("\n💡 输入您的问题，按回车发送。输入 'quit' 或 'exit' 退出。")
print("💡 输入 'reload' 重新加载知识库。\n")
while True:
try:
question = input("👤 您: ").strip()
if question.lower() in ['quit', 'exit', 'q']:
print("\n👋 再见！")
break
if question.lower() == 'reload':
self.initialize(rebuild=True)
continue
if not question:
continue
print("\n🤖 AI: ", end="")
answer = self.chat(question)
print(answer)
print()
except KeyboardInterrupt:
print("\n\n👋 再见！")
break
except Exception as e:
print(f"\n❌ 错误: {str(e)}")
def main():
parser = argparse.ArgumentParser(description="企业知识库智能问答系统")
parser.add_argument("--path", "-p", type=str, default="./data/knowledge_base",
help="知识库文档目录路径")
parser.add_argument("--rebuild", "-r", action="store_true",
help="强制重建向量数据库")
args = parser.parse_args()
# 创建系统实例
qa_system = KnowledgeBaseQA(knowledge_base_path=args.path)
qa_system.initialize(rebuild=args.rebuild)
qa_system.interactive()
if __name__ == "__main__":
main()
运行与测试
准备测试数据
在data/knowledge_base/目录下放置一些测试文档，例如创建product_faq.txt：
plaintext
产品退换货政策
1. 退换货时限
自收到商品之日起7天内可以申请退换货，15天内可以申请换货。
2. 退换货条件
- 商品未被人为损坏
- 包装配件完整
- 附带购买凭证
3. 退换货流程
登录官网 -> 进入"我的订单" -> 选择订单 -> 点击"申请退换货" -> 填写原因 -> 提交审核
4. 运费说明
- 因质量问题退换货：由我们承担运费
- 因个人原因退换货：需自行承担运费
启动系统
Windows系统：
powershell
cd ai-knowledge-qa
.\venv\Scripts\activate
python src/main.py --path ./data/knowledge_base
macOS/Linux系统：
bash
cd ai-knowledge-qa
source venv/bin/activate
python src/main.py --path ./data/knowledge_base
测试示例
系统启动后，输入以下问题进行测试：
plaintext
👤 您: 产品的退换货政策是什么？
🤖 AI: 根据知识库中的信息，我们公司的产品退换货政策如下：
1. **退换货时限**：自收到商品之日起7天内可申请退换货，15天内可申请换货。
2. **退换货条件**：
- 商品未被人为损坏
- 包装配件完整
- 附带购买凭证
3. **退换货流程**：登录官网 → 进入"我的订单" → 选择订单 → 点击"申请退换货" → 填写原因 → 提交审核
4. **运费说明**：
- 因质量问题退换货由我们承担运费
- 因个人原因退换货需自行承担运费
📚 参考来源: product_faq.txt
性能优化建议
检索优化
向量检索的关键参数是top_k值和similarity_threshold。如果检索结果噪声过多，可以提高检索数量后通过重排序（rerank）筛选最相关的几条。中文场景建议使用专门的中文Embedding模型，如moka-ai/m3e-base，能显著提升中文语义理解效果。
回答质量优化
通过调整prompt模板可以改善回答质量。增加”请分点回答”等约束让回答更有条理，添加”如果不确定请明确说明”可以减少幻觉问题。对于专业术语较多的行业，可以在prompt中加入领域背景信息。
大模型选择
在线模型（如GPT-4o、通义千问）效果较好但有API成本，本地模型（如Llama 3、Qwen2）可以离线使用但需要较高硬件配置。建议开发测试阶段使用在线模型，生产部署根据数据敏感性选择本地模型或私有化部署。
扩展功能
添加流式输出
修改QAChain以支持流式输出，提升用户体验：
python
def ask_stream(self, question: str):
"""流式问答"""
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
# 创建流式回调处理器
streaming_handler = StreamingStdOutCallbackHandler()
# 创建支持流式输出的链
streaming_chain = RetrievalQA.from_chain_type(
llm=self.llm,
chain_type="stuff",
retriever=self.retriever,
return_source_documents=True,
callbacks=[streaming_handler]
)
return streaming_chain({"query": question})
支持多轮对话
集成对话记忆功能，实现多轮上下文理解：
python
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationalRetrievalChain
def create_conversational_chain(retriever):
"""创建支持多轮对话的检索链"""
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True,
output_key="answer"
)
condense_question_prompt = """给定对话历史和用户问题，重新组织成一个独立、完整的问题。
对话历史:
{chat_history}
用户问题: {question}
独立完整的问题:"""
CONDENSE_QUESTION_PROMPT = PromptTemplate(
template=condense_question_prompt,
input_variables=["chat_history", "question"]
)
qa_chain = ConversationalRetrievalChain.from_llm(
llm=self.llm,
retriever=retriever,
memory=memory,
condense_question_prompt=CONDENSE_QUESTION_PROMPT,
return_source_documents=True
)
return qa_chain
总结
通过本文的实战项目，你已经掌握了构建企业级RAG知识库问答系统的完整技能。从文档处理、文本分割、向量化存储到检索问答，每个环节都有深入讲解。这个系统可以灵活扩展，支持多种文档格式、多种大模型，以及流式输出、多轮对话等高级功能。
关键要点回顾：使用LangChain作为核心框架简化开发流程；Chroma作为轻量级向量库便于快速原型开发；合理的文本分割策略直接影响检索效果；通过prompt工程可以优化回答质量和格式；根据实际场景选择在线API或本地部署的大模型。
希望这个项目能帮助你在实际工作中快速落地AI应用，如果你在实践过程中遇到问题，欢迎在评论区交流讨论。代码已同步到GitHub，可以直接clone下来运行测试。
你学会了吗？动手试试构建自己的知识库问答系统吧！
发表回复