使用 LangChain + Ollama 构建本地知识库问答系统
很多团队都有这样的困扰：文档散落在各处，关键时刻找不到想要的内容。关键词搜索不智能，看字面对不上意思，只能手动翻文档。
RAG 技术提供了另一种思路：把文档转成向量，用语义检索找到相关内容，再让 LLM 根据这些内容生成答案。这两年很火，很多云服务都支持。
但把文档发给第三方总让人有点顾虑——内部资料嘛，总不希望跑出去。有没有办法完全本地运行？这就是本文要解决的问题。
要解决什么问题
核心需求其实很直接：从本地文档里找到相关的内容，然后根据这些内容回答问题。
具体有几个技术点：
文档格式要支持 txt、md、pdf 这些常见的。文档不能直接丢给 LLM，太长了会超限，需要切成小块。切好的小块要转成向量存起来，方便后续检索。用户提问时，先检索最相关的几个小块，然后把检索结果和问题一起发给 LLM，让它生成答案。
听起来步骤不少，但每个环节都有成熟的工具可以选用。
具体怎么做
准备环境
先装好 Ollama，这是运行本地 LLM 的工具：
# macOS
brew install ollama
Linux
curl -fsSL https://ollama.com/install.sh | sh
再装 Python 依赖：
pip install langchain langchain-community langchain-chroma \
chromadb pypdf python-dotenv \
sentence-transformers
硬件方面，建议 16GB 以上内存，硬盘留出 10GB 空间。向量模型和 LLM 都要占用资源。
放文档
在项目根目录建一个 docs 文件夹，把要索引的文件丢进去。可以是 txt、md 或 pdf。先放几篇测试用，验证流程跑通了再加更多。
启动 Ollama
# 后台运行服务
ollama serve
下载模型（另一终端）
ollama pull llama3
llama3 是个比较均衡的模型，参数和效果都不错。机器配置好的话可以试试 codellama，对代码类任务优化过。
上代码
整个系统其实就三块：文档加载、向量化存库、问答流程。下面是完整实现：
#!/usr/bin/env python3
"""
本地知识库问答系统
基于 LangChain + Ollama
"""
import os
from pathlib import Path
from typing import List
from dotenv import load_dotenv
from langchain_community.document_loaders import (
TextLoader,
PyPDFLoader,
DirectoryLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.llms import Ollama
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
class KnowledgeBaseQA:
"""本地知识库问答系统"""
def __init__(
self,
docs_path: str = "./docs",
embedding_model: str = "sentence-transformers/all-MiniLM-L6-v2",
llm_model: str = "llama3",
chunk_size: int = 500,
chunk_overlap: int = 50
):
self.docs_path = docs_path
self.embedding_model = embedding_model
self.llm_model = llm_model
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.embeddings = None
self.vectorstore = None
self.qa_chain = None
self._init_components()
def _init_components(self):
"""初始化组件"""
print("正在初始化向量嵌入模型...")
self.embeddings = HuggingFaceEmbeddings(
model_name=self.embedding_model,
model_kwargs={'device': 'cpu'}
)
print("正在初始化 LLM...")
self.llm = Ollama(model=self.llm_model)
def load_documents(self) -> List:
"""加载文档"""
print(f"正在加载目录: {self.docs_path}")
documents = []
# 加载文本文件
if os.path.exists(self.docs_path):
for ext in ["*.txt", "*.md"]:
for file_path in Path(self.docs_path).glob(ext):
print(f"加载文件: {file_path}")
try:
loader = TextLoader(str(file_path), encoding='utf-8')
documents.extend(loader.load())
except Exception as e:
print(f"加载 {file_path} 失败: {e}")
# 加载 PDF 文件
for file_path in Path(self.docs_path).glob("*.pdf"):
print(f"加载 PDF: {file_path}")
try:
loader = PyPDFLoader(str(file_path))
documents.extend(loader.load())
except Exception as e:
print(f"加载 {file_path} 失败: {e}")
print(f"共加载 {len(documents)} 个文档")
return documents
def process_documents(self, documents: List) -> List:
"""处理文档：分块"""
print("正在对文档进行分块处理...")
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=self.chunk_size,
chunk_overlap=self.chunk_overlap,
separators=["\n\n", "\n", "。", " ", ""]
)
texts = text_splitter.split_documents(documents)
print(f"文档分块完成，共 {len(texts)} 个文本块")
return texts
def build_vectorstore(self, texts: List):
"""构建向量存储"""
print("正在构建向量数据库...")
persist_directory = "./chroma_db"
self.vectorstore = Chroma.from_documents(
documents=texts,
embedding=self.embeddings,
persist_directory=persist_directory
)
print(f"向量数据库已保存到: {persist_directory}")
def load_vectorstore(self):
"""加载已存在的向量数据库"""
persist_directory = "./chroma_db"
if os.path.exists(persist_directory):
print(f"正在加载已有向量数据库: {persist_directory}")
self.vectorstore = Chroma(
persist_directory=persist_directory,
embedding_function=self.embeddings
)
return True
return False
def setup_qa_chain(self):
"""设置问答 Chain"""
print("正在配置问答 Chain...")
# 自定义提示词
prompt_template = """基于以下上下文回答问题。如果上下文中没有相关信息，请说明无法从提供的文档中找到答案。
上下文：
{context}
问题：{question}
请用中文回答："""
PROMPT = PromptTemplate(
template=prompt_template,
input_variables=["context", "question"]
)
# 创建 RetrievalQA Chain
self.qa_chain = RetrievalQA.from_chain_type(
llm=self.llm,
chain_type="stuff",
retriever=self.vectorstore.as_retriever(
search_kwargs={"k": 3}
),
chain_type_kwargs={"prompt": PROMPT},
return_source_documents=True
)
print("问答系统配置完成！")
def build(self):
"""构建完整的问答系统"""
documents = self.load_documents()
if not documents:
print("警告：未找到任何文档，请检查 docs 目录")
return
texts = self.process_documents(documents)
self.build_vectorstore(texts)
self.setup_qa_chain()
def load(self):
"""加载已构建的系统"""
if self.load_vectorstore():
self.setup_qa_chain()
return True
return False
def ask(self, question: str) -> dict:
"""提问"""
if not self.qa_chain:
print("错误：问答系统未初始化，请先运行 build() 或 load()")
return {"result": None, "sources": []}
print(f"\n正在回答问题: {question}")
result = self.qa_chain({"query": question})
return {
"answer": result["result"],
"sources": [
{
"content": doc.page_content[:200] + "...",
"source": doc.metadata.get("source", "未知")
}
for doc in result.get("source_documents", [])
]
}
def main():
import argparse
parser = argparse.ArgumentParser(description="本地知识库问答系统")
parser.add_argument("--build", action="store_true",
help="构建知识库")
parser.add_argument("--query", type=str,
help="提问")
parser.add_argument("--docs", type=str, default="./docs",
help="文档目录路径")
parser.add_argument("--model", type=str, default="llama3",
help="LLM 模型名称")
parser.add_argument("--embedding", type=str,
default="sentence-transformers/all-MiniLM-L6-v2",
help="嵌入模型名称")
args = parser.parse_args()
qa = KnowledgeBaseQA(
docs_path=args.docs,
llm_model=args.model,
embedding_model=args.embedding
)
if args.build:
qa.build()
print("\n知识库构建完成！")
print("现在可以使用 --query 参数提问")
elif args.query:
if not qa.load():
print("错误：未找到已构建的知识库，请先使用 --build 构建")
return
result = qa.ask(args.query)
print("\n" + "="*50)
print("回答：")
print(result["answer"])
print("\n参考来源：")
for i, source in enumerate(result["sources"], 1):
print(f"\n{i}. {source['source']}")
print(f" {source['content']}")
else:
print("请使用 --build 构建知识库，或使用 --query 提问")
print("示例：")
print(" python knowledge_qa.py --build")
print(" python knowledge_qa.py --query \"什么是 RAG 技术\"")
if __name__ == "__main__":
main()
主要流程是这样的：HuggingFace 的 sentence-transformers 负责把文本转成向量，Chroma 作为向量数据库存这些向量，Ollama 运行的 llama3 负责生成答案。LangChain 把检索和回答串起来，形成完整的 pipeline。
跑起来看看效果
先构建知识库：
python knowledge_qa.py --build
脚本会逐个处理 docs 目录里的文件，切成小块，生成向量，存到 chroma_db 目录。构建完就可以提问了：
python knowledge_qa.py --query "我们的技术架构是什么？"
第一次跑要等模型加载，稍微慢一点。后面的请求会快很多，因为模型已经驻留内存了。
检索结果会同时返回答案和参考来源，方便核实答案的准确性。
几个限制
这个方案有几个实际问题：
对机器有要求。Embeddings 模型和 LLM 同时跑，内存压力不小。首次加载模型需要等几分钟，不是即开即用的。回答速度和本地硬件直接相关，CPU 跑推理比 GPU 慢很多。
不过对于重视数据安全的场景，这些代价值得。完全离线，不用担心文档外传，也不用付 API 费用。
后续可以调整的地方很多：换更大的 LLM 模型、调分块参数、换更快的嵌入模型、或者集成到现有系统里做内部知识库搜索入口。