本地deepseek构建私有知识库

本文通过python传统方式调用大模型，实现本地deepseek私有知识库的搭建。

逻辑图

1. 前提条件

在本地环境中，我们已经完成了以下准备工作：

已安装 Ollama，用于下载、装载、管理大模型
已下载 deepseek-r1:1.5b 作为基础问答大模型
已在 Ollama 中下载 nomic-embed-text 作为向量化模型
本地已准备好一个文本文件 finance.txt，其中包含希望用于增强模型回答的数据
本地已经下载并按照好vscode代码编辑软件，编译环境中已经安装python库、chromadb库

2. 目标

我们希望实现基于 RAG（检索增强生成） 技术的本地化微调，使 DeepSeek-R1:1.5B 能够依托 finance.txt 提供精准回答，而不是凭空生成（即减少幻觉）。

RAG 技术流程如下：

文本向量化：将 finance.txt 处理成向量，存入向量数据库
查询增强：当用户提问时，从向量数据库中检索相关内容，并将其与问题一起传递给模型
模型回答：DeepSeek-R1:1.5B 结合检索到的信息，生成更加准确的回答

3. 本地finance.txt

说明：finance.txt里面存放了一些贷款相关的知识，这是一个示例文件，我们希望在询问deepseek贷款相关问题，他不要随意发挥，要依托我们的finance.txt里面的背景来组织语言，内容如下。

【贷款利率政策】
当前贷款利率由央行基准利率决定，并受到市场因素影响。对于个人住房贷款，银行一般采用 LPR（贷款市场报价利率）+ 加点的方式确定最终利率。企业贷款利率则根据信用评级、行业风险等进行调整。

【贷款审批条件】
银行对贷款申请人的信用评分、收入水平、资产状况进行综合评估。一般要求：
1. 信用评分在 600 分以上，无重大逾期记录；
2. 个人住房贷款首付款比例至少 20%-30%；
3. 企业贷款需提供财务报表，部分情况需抵押物。

【还款方式】
贷款还款方式主要有两种：等额本息与等额本金。等额本息前期压力较小，但总利息较高；等额本金前期还款较多，但利息总支出较少。借款人可根据自身经济情况选择适合的方式。

【提前还款政策】
大部分银行允许提前还款，但可能收取违约金。提前还款是否划算，需考虑贷款利率、剩余期限等因素。一般来说，如果贷款剩余年限较长，提前还款可减少利息支出。

【企业贷款优惠政策】
部分政府支持行业（如科技创新、绿色能源）可获得低息贷款支持。中小企业可申请信用贷款或政府贴息贷款，以降低融资成本。

4. python代码实现

import os
import requests
import chromadb
from chromadb.utils import embedding_functions

# 🔹 定义 ChromaDB 存储路径
DB_DIR = “./chroma_db”

# 🔹 定义 Ollama API 地址
OLLAMA_EMBEDDING_URL = “http://127.0.0.1:11434/api/embeddings”

# 🔹 定义本地模型
OLLAMA_MODEL = “deepseek-r1:1.5b”

# 1️⃣ **加载 finance.txt 并切分文本**
def load_and_split_text(file_path, chunk_size=100, overlap=20):
“”” 读取文本文件并进行分块 “””
try:
with open(file_path, “r”, encoding=”utf-8″) as f:
text = f.read()

chunks = []
for i in range(0, len(text), chunk_size – overlap):
chunks.append(text[i : i + chunk_size])

print(f”✅ 成功加载 {file_path}，共分割 {len(chunks)} 个块”)
return chunks
except Exception as e:
print(f”❌ 读取 {file_path} 失败: {e}”)
return []

# 2️⃣ **调用 Ollama API 获取文本向量**
def get_embedding(text):
“”” 通过 Ollama API 获取文本的向量表示 “””
try:
data = {“model”: “nomic-embed-text”, “prompt”: text}
response = requests.post(OLLAMA_EMBEDDING_URL, json=data, timeout=10)

if response.status_code == 200:
return response.json().get(“embedding”, [])
else:
print(f”❌ 向量化失败，状态码: {response.status_code}, 响应: {response.text}”)
return None
except requests.exceptions.RequestException as e:
print(f”❌ 请求 Ollama 失败: {e}”)
return None

# 3️⃣ **存入 ChromaDB**
def store_in_chroma(text_chunks):
“”” 将文本向量存入 ChromaDB 持久化数据库 “””
try:
client = chromadb.PersistentClient(path=DB_DIR)
collection = client.get_or_create_collection(name=”finance”)

for i, chunk in enumerate(text_chunks):
embedding = get_embedding(chunk)
if embedding:
collection.add(
ids=[f”doc_{i}”], # 设定唯一 ID
embeddings=[embedding],
documents=[chunk]
)

print(“✅ 数据已成功存入 ChromaDB！”)
except Exception as e:
print(f”❌ 存入 ChromaDB 失败: {e}”)

# 4️⃣ **查询 ChromaDB**
def query_chroma(query_text, top_k=3):
“”” 从 ChromaDB 查询相关文本 “””
try:
client = chromadb.PersistentClient(path=DB_DIR)
collection = client.get_collection(name=”finance”)

query_vector = get_embedding(query_text)
if query_vector:
results = collection.query(
query_embeddings=[query_vector],
n_results=top_k
)
return results.get(“documents”, [[]])[0]
else:
return []
except Exception as e:
print(f”❌ 查询 ChromaDB 失败: {e}”)
return []

# 5️⃣ **让 deepseek-r1:1.5b 回答问题**
def ask_local_model(context, question):
“”” 让本地 deepseek-r1:1.5b 模型回答问题 “””
try:
ollama_url = “http://127.0.0.1:11434/api/generate”
data = {
“model”: OLLAMA_MODEL,
“prompt”: f”根据以下内容回答问题：\n{context}\n\n问题：{question}”,
“stream”: False
}

response = requests.post(ollama_url, json=data, timeout=40)
if response.status_code == 200:
return response.json().get(“response”, “⚠️ 无法生成回答”)
else:
print(f”❌ deepseek-r1:1.5b生成回答失败，状态码: {response.status_code}, 响应: {response.text}”)
return “⚠️ deepseek-r1:1.5b生成回答失败”
except requests.exceptions.RequestException as e:
print(f”❌ 请求 deepseek-r1:1.5b 失败: {e}”)
return “⚠️ deepseek-r1:1.5b 生成回答失败”

# **运行流程**
if __name__ == “__main__”:
try:
# 1️⃣ 读取 finance.txt 并存入 ChromaDB
finance_file = “finance.txt”
if os.path.exists(finance_file):
text_chunks = load_and_split_text(finance_file)
if text_chunks:
store_in_chroma(text_chunks)
else:
print(f”❌ 文件 {finance_file} 不存在！”)

# 2️⃣ 处理查询
question = “贷款审批条件有哪些？”
retrieved_docs = query_chroma(question)
print(“\n🔹 **retrieved_docs结果** 🔹\n”, retrieved_docs)

if retrieved_docs:
context = “\n”.join(retrieved_docs)
answer = ask_local_model(context, question)
print(“\n🔹 **最终回答** 🔹\n”, answer)
else:
print(“⚠️ 没有找到相关信息！”)

except Exception as e:
print(f”❌ 运行主流程时发生错误: {e}”)