大语言模型虽然能说会道，但它懂的东西全是训练时塞进去的。你要是问它你们公司的报销流程、某个产品的技术参数，它大概率会瞎编——这就是所谓的「幻觉」。所以很多公司做私有知识库的时候，都不用那些云端的 API，就是怕数据泄露。
RAG（检索增强生成）就是来解决这个问题的。简单说就是把用户的提问和从知识库拉来的相关资料一起喂给大模型，让它基于真实内容回答，而不是凭记忆编。本地部署 RAG 的话，数据全在自己机器上，不用传给第三方，安全感拉满。
一、环境准备
先装 Python 依赖：
pip install langchain langchain-community langchain-ollama \
chroma pypdf python-docx sentence-transformers
然后装 Ollama。Linux 系统直接运行：
curl -fsSL https://ollama.com/install.sh | sh
macOS 或者 Windows 直接去官网下载安装包。
模型拉取命令：
ollama pull llama3:8b
ollama pull nomic-embed-text
消费级显卡跑得动 llama3:8b 或者 qwen:7b，显存不够的话可以选更小的版本。
二、文档加载与处理
PDF、Markdown、TXT 都能处理：
from langchain_community.document_loaders import (
PyPDFLoader,
TextLoader,
UnstructuredMarkdownLoader
)
def load_documents(directory_path):
"""加载指定目录下的所有文档"""
from pathlib import Path
documents = []
path = Path(directory_path)
for file_path in path.rglob("*"):
if not file_path.is_file():
continue
try:
if file_path.suffix == ".pdf":
loader = PyPDFLoader(str(file_path))
elif file_path.suffix == ".md":
loader = UnstructuredMarkdownLoader(str(file_path))
elif file_path.suffix == ".txt":
loader = TextLoader(str(file_path), encoding="utf-8")
else:
continue
docs = loader.load()
for doc in docs:
doc.metadata["source"] = str(file_path)
documents.extend(docs)
except Exception as e:
print(f"加载 {file_path} 失败: {e}")
return documents三、文档分块
分块是个技术活。块太小了上下文丢失，块太大了掺进杂质。技术文档推荐这样设置：
from langchain.text_splitter import RecursiveCharacterTextSplitter
def chunk_documents(documents, chunk_size=500, chunk_overlap=50):
"""对文档进行分块处理"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len,
separators=[
"\n\n",
"\n",
"。",
". ",
" ",
""
],
keep_separator=True,
add_start_index=True
)
chunks = text_splitter.split_documents(documents)
# 过滤太小的块
chunks = [chunk for chunk in chunks
if len(chunk.page_content) > 50]
return chunks这个配置会优先按段落分割，然后按句子，最后按字符。块之间有 50 个字符的重叠，保证重要内容不会卡在边界上。
四、向量存储与检索
用 Chroma 当向量数据库，搭配支持中文的 embedding 模型：
from langchain_community.vectorstores import Chroma
from langchain.embeddings import HuggingFaceEmbeddings
def create_vector_store(chunks, persist_directory="./chroma_db"):
"""创建向量存储"""
embeddings = HuggingFaceEmbeddings(
model_name="BAAI/bge-small-zh-v1.5",
model_kwargs={"device": "cpu"}
)
if os.path.exists(persist_directory):
vectorstore = Chroma(
persist_directory=persist_directory,
embedding_function=embeddings
)
else:
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory=persist_directory
)
return vectorstore
def retrieve_documents(vectorstore, query, k=5):
"""检索最相关的文档块"""
results = vectorstore.similarity_search(
query=query,
k=k
)
return resultsbge-small-zh-v1.5 这个模型向量维度只有 512，推理速度快，效果也不错，适合本地跑。
五、构建问答链
把检索结果和 Ollama 模型接上：
from langchain_ollama import ChatOllama
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough
def build_qa_chain(vectorstore):
"""构建问答链"""
llm = ChatOllama(
model="llama3:8b",
base_url="http://localhost:11434",
temperature=0.3,
num_ctx=4096
)
template = """请基于以下参考资料回答用户问题。如果资料中没有相关信息，请明确说明。
参考资料：
{context}
用户问题：{question}
请给出准确、详细的回答。如果涉及代码，请提供完整的示例。"""
prompt = PromptTemplate.from_template(template)
def format_docs(docs):
return "\n\n".join(
f"来源 {i+1}: {doc.page_content}"
for i, doc in enumerate(docs)
)
qa_chain = (
{
"context": vectorstore.as_retriever(k=5) | format_docs,
"question": RunnablePassthrough()
}
| prompt
| llm
)
return qa_chain这里 temperature 设成 0.3 是为了让输出稳定一点，别太天马行空。num_ctx 是上下文窗口大小，根据模型能力自己调。
六、完整示例
把以上代码拼起来就是一个能跑的应用：
import os
from pathlib import Path
from utils import (
load_documents,
chunk_documents,
create_vector_store,
build_qa_chain
)
def main():
# 1. 加载文档
docs_dir = "./knowledge_base"
documents = load_documents(docs_dir)
print(f"共加载 {len(documents)} 个文档")
# 2. 分块处理
chunks = chunk_documents(documents)
print(f"共生成 {len(chunks)} 个块")
# 3. 创建向量存储
vectorstore = create_vector_store(chunks)
print("向量存储创建完成")
# 4. 构建问答链
qa_chain = build_qa_chain(vectorstore)
print("问答链构建完成，开始交互...")
# 5. 交互问答
while True:
query = input("\n请输入问题 (输入 q 退出): ")
if query.lower() == "q":
break
try:
response = qa_chain.invoke(query)
print(f"\n回答: {response.content}")
except Exception as e:
print(f"处理出错: {e}")
if __name__ == "__main__":
main()启动前确保知识库目录里有文档文件，运行后会提示输入问题，直接问就行。
七、运行效果
实际跑起来大概是这个样子的：
请输入问题 (输入 q 退出): 如何在 Python 中连接 PostgreSQL 数据库
回答: 可以使用 psycopg2 库。下面是基本用法：
1. 安装：
pip install psycopg2-binary2. 连接：
import psycopg2conn = psycopg2.connect(
host=”localhost”,
port=5432,
database=”mydb”,
user=”postgres”,
password=”your_password”
)3. 查询：
cursor = conn.cursor()
cursor.execute(“SELECT * FROM users WHERE id = %s”, (1,))
result = cursor.fetchone()cursor.close()
conn.close()需要连接池的话，用 DBUtils 或者 SQLAlchemy。
知识库里要是有 PostgreSQL 相关的技术文档，系统就能把这些内容检索出来，结合模型的理解能力给出完整的答案。
八、总结
Ollama + LangChain 这套组合跑本地 RAG 效果不错。数据全在本地，不用担心外传。模型可以选 llama3、qwen、mistral 这些开源的，根据自己显卡显存看着办。
后续可以捣鼓的方向：加个缓存层提速、弄个多模型切换、在检索前面套个 rerank 模型进一步提升准确度。RAG 这东西熟练了之后应用场景挺广的，企业内部知识库、技术文档问答、客服辅助都能用上。