python + ollama 手敲实现私有大模型知识库
在不依赖 LangChain、LlamaIndex 等框架，以及各种知识问答软件的情况下，尽量减少第三方库的使用，仅通过 Ollama 和 NumPy 两个外部库来实现 RAG（Retrieval-Augmented Generation）应用。
一、安装python
下载：https://python.org/downloads/
安装：一路下一步即可，安装完成后运行以下代码，即可查看对应版本号
python --version
二、安装PyCharm
（推荐，也可以用vscode或windows记事本等编辑软件）
PyCharm的优点：
1、自动提示
2、创建项目时自动创建python虚拟环境
3、整合本地终端（且基于python对应本项目的虚拟环境）
三、安装ollama
下载：https://ollama.com/download
安装：一路下一步。安装完成后用以下命令检验
ollama --version
windows安装的ollama默认不适用GPU，也可以使用，如果你手痒且有独立显卡，可先执行此命令，查看下GPU是否可用
（启用GPU的这一步也可省略哦）
nvidia-smi
执行后（如下图）能找到如①标识，说明有GPU可用。
然后设置下系统变量：CUDA_VISIBLE_DEVICES = 0（此值为上图①）
设置了系统变量后，重启下ollama，就可以如上图②，已经有一个叫ollama_llama_server.exe的程序在使用gpu了。
四、在ollama中安装开源chat大模型：qwen2.5 或deepseek-r1
ollama官网models（https://ollama.com/search）页面可以找到很多开源大模型，我们下载用的比较多的千问中文大模型（显卡差的建议安装ollama2.5:0.5b）
ollama run qwen2.5
五、在ollama中安装开源embedding大模型：milkey/m3e nomic-embed-text
embedding作用是将问题和知识库文本转换成向量，便于查询。
原本打算使用nomic-embed-text模型，但是使用时效果不好，发现m3e的效果不错，所以改用m3e
ollama run milkey/m3e
六、在项目python虚拟环境中安装ollama
pip install ollama
七、安装numpy
pip install numpy
八、编写代码
项目文件目录结构如下图
1. kb.py
import numpy as np
from ollama import embeddings
class Kb:
def __init__(self, filepath):
# 读取文件内容
content = self.read_file(filepath)
# print(content)
# 读取拆分好的数组
self.chunks = self.split_content(content)
# print(chunks)
# for chunk in chunks:
# print(chunk)
# print('=' * 10)
# 转换成向量
self.embeds = self.get_embeddings(self.chunks)
# 读取文件
def read_file(self, filepath):
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
return content
# 拆分知识库
@staticmethod
def split_content(content):
chunks = content.split('# ')
# 过滤掉空块
chunks = [chunk.strip() for chunk in chunks if chunk.strip()]
return chunks
# 字符串转向量(embeddings)
def get_embedding(self, chunk):
# milkey/m3e 0.642084887746903
# bge-m3 0.6073383067378445
# nomic-embed-text 完全找不到
res = embeddings(model='milkey/m3e', prompt=chunk)
# print(chunk)
# print(res)
# print(res['embedding'])
return res['embedding']
def get_embeddings(self, chunks):
embeds = []
for chunk in chunks:
embed = self.get_embedding(chunk)
embeds.append(embed)
return np.array(embeds)
# 查询相似性向量
def search(self, text):
print(text)
max_similarity = 0
max_similarity_index = 0
ask_embed = self.get_embedding(text)
for kb_embed_index, kb_embed in enumerate(self.embeds):
similarity = self.similarity(kb_embed, ask_embed)
# print(similarity)
# print(self.chunks[kb_embed_index])
if similarity > max_similarity:
max_similarity = similarity
max_similarity_index = kb_embed_index
print(max_similarity)
print(self.chunks[max_similarity_index])
# print(self.embeds[max_similarity_index])
# 返回查到的相关文本
return self.chunks[max_similarity_index]
# 相似度
@staticmethod
def similarity(A, B):
# 计算点积
dot_product = np.dot(A, B)
# 计算范数
norm_A = np.linalg.norm(A)
norm_B = np.linalg.norm(B)
# 计算余弦相似度
cosine_sim = dot_product / (norm_A * norm_B)
return cosine_sim
2. rag.py
from kb import Kb
from ollama import chat, Message
class Rag:
def __init__(self, model, kb_filepath):
self.kb_filepath = kb_filepath
self.kb = Kb(kb_filepath)
self.model = model
self.prompt_template = """
基于：%s
回答：%s
"""
def chat(self, message):
# 用户消息检索相关上下文
context = self.kb.search(message)
# print(context)
# prompt = self.prompt_template % (context, message)
prompt = '请基于以下内容回答问题：\n' + context
response = chat(self.model, [Message(role='system', content=prompt), Message(role='user', content=message)])
return response['message']
3. index.py
from rag import Rag
rag = Rag('deepseek-r1:14b', '私人知识库.txt')
msg = rag.chat('请介绍下刘芳')
print(msg)
4. 私人知识库.txt
MIS部门人员名单
# 1. 张小刚
姓名：张小刚
性别：男
爱好：打篮球、踢足球
电话：13223344422
籍贯：山东菏泽
# 2. 李光亮
姓名：李光亮
性别：男
爱好：踢足球、打排球
电话：1595959559
籍贯：河南平顶山
# 3. 王丽丽
姓名：王丽丽
性别：女
爱好：游泳、阅读
电话：13812345678
籍贯：江苏南京
# 4. 陈大明
姓名：陈大明
性别：男
爱好：跑步、爬山
电话：13987654321
籍贯：浙江杭州
# 5. 刘芳
姓名：刘芳
性别：女
爱好：瑜伽、绘画
电话：13711223344
籍贯：广东深圳
# 6. 赵强
姓名：赵强
性别：男
爱好：打羽毛球、钓鱼
电话：13566554433
籍贯：四川成都
# 7. 孙婷婷
姓名：孙婷婷
性别：女
爱好：跳舞、唱歌
电话：13677889900
籍贯：福建厦门
# 8. 周伟
姓名：周伟
性别：男
爱好：打乒乓球、下棋
电话：13455667788
籍贯：湖南长沙
# 9. 吴晓梅
姓名：吴晓梅
性别：女
爱好：摄影、旅行
电话：13344556677
籍贯：湖北武汉
# 10. 郑小龙
姓名：郑小龙
性别：男
爱好：打篮球、游泳
电话：13233445566
籍贯：陕西西安
# 11. 高静
姓名：高静
性别：女
爱好：阅读、写作
电话：13122334455
籍贯：辽宁沈阳
# 12. 林浩
姓名：林浩
性别：男
爱好：踢足球、跑步
电话：13011223344
籍贯：广西南宁
# 13. 黄雅婷
姓名：黄雅婷
性别：女
爱好：跳舞、瑜伽
电话：13900112233
籍贯：云南昆明
# 14. 徐志强
姓名：徐志强
性别：男
爱好：打排球、爬山
电话：13899001122
籍贯：贵州贵阳
# 15. 何丽
姓名：何丽
性别：女
爱好：绘画、摄影
电话：13788990011
籍贯：江西南昌
# 16. 马超
姓名：马超
性别：男
爱好：打篮球、钓鱼
电话：13677889900
籍贯：山西太原
# 17. 郭晓燕
姓名：郭晓燕
性别：女
爱好：唱歌、旅行
电话：13566778899
籍贯：河北石家庄
# 18. 罗志勇
姓名：罗志勇
性别：男
爱好：踢足球、下棋
电话：13455667788
籍贯：吉林长春
# 19. 邓丽丽
姓名：邓丽丽
性别：女
爱好：瑜伽、阅读
电话：13344556677
籍贯：黑龙江哈尔滨
# 20. 许文强
姓名：许文强
性别：男
爱好：打羽毛球、跑步
电话：13233445566
籍贯：安徽合肥
# 21. 韩雪
姓名：韩雪
性别：女
爱好：跳舞、摄影
电话：13122334455
籍贯：甘肃兰州
# 22. 曹阳
姓名：曹阳
性别：男
爱好：打篮球、爬山
电话：13011223344
籍贯：青海西宁
# 23. 谢婷婷
姓名：谢婷婷
性别：女
爱好：唱歌、绘画
电话：13900112233
籍贯：宁夏银川
# 24. 董志刚
姓名：董志刚
性别：男
爱好：踢足球、钓鱼
电话：13899001122
籍贯：新疆乌鲁木齐
# 25. 苏静
姓名：苏静
性别：女
爱好：阅读、旅行
电话：13788990011
籍贯：内蒙古呼和浩特
# 26. 潘伟
姓名：潘伟
性别：男
爱好：打乒乓球、下棋
电话：13677889900
籍贯：海南海口
# 27. 钟丽
姓名：钟丽
性别：女
爱好：瑜伽、摄影
电话：13566778899
籍贯：重庆
# 28. 田小龙
姓名：田小龙
性别：男
爱好：打篮球、游泳
电话：13455667788
籍贯：天津
# 29. 白晓梅
姓名：白晓梅
性别：女
爱好：跳舞、唱歌
电话：13344556677
籍贯：北京
# 30. 石浩
姓名：石浩
性别：男
爱好：踢足球、跑步
电话：13233445566
籍贯：上海
通过以上代码，可基本实现用RAG技术，搭建本地知识库问答AI助理，非流式回复(steam=False)。需要等待大模型输出所有文字之后，才能全部返回，太慢，肯定要实现一般聊天大模型的流式回复（steam=True）
首先，在rag.py中增加以下方法
def stream_chat(self, message):
context = self.kb.search(message)
prompt = '请基于以下内容回答问题：\n' + context
response = chat(self.model, [Message(role='system', content=prompt), Message(role='user', content=message)], stream=True)
# 遍历流式响应
for chunk in response:
print(chunk['message']['content'], end='', flush=True) # 实时打印输出
print() # 输出完成后换行
然后，调整index.py代码为
from rag import Rag
rag = Rag('deepseek-r1:14b', '私人知识库.txt')
rag.stream_chat('请介绍下刘芳')
经过以上的修改，就可以看到AI的流式答复了。
注：以上RAG只是获取到了本地知识库中的最为匹配的一条记录，如果需要实现一次性获取多个关联内容，并根据多个关联内容进行答复，需要再调整代码。