前言
hi，这里是惬鹤频道！
又过去几天了，感觉应该来更新一下我的最新进度了。
最近正式学完了RAG相关的基础知识，进入了项目实战。本来想着应该有很多可以和大家聊聊的东西，但是。。。
真的好难！！！
我不是一个能在脑袋里快速记下大概结构的人，对于RAG项目这种需要在前期进行结构思考，想清楚代码内所需的基本方法，并想好怎么把它们都拼接起来的工作，实在有点艰难。
到现在已经两天了，才写好了两个代码文件，所有py文件加起来大概有六七个，接下来还需要熬一熬才能完成。
所以这一期和大家介绍一下我写的两个代码文件吧，没有啥需要分享的问题，大家感兴趣的也可以学学。
好了不说太多了，最近的文章就短一点吧，有点累了。
app_file_uploader.py
"""
文件介绍
文件名：app_file_uploader.py
结构：利用streamlit，对接knowledge_base.py文件，创建一个可以上传，读取，解析文本文件的网页。
"""
# 导入依赖
import streamlit as st
import time
from knowledge_base import KnowledgeBase
# 创建页面
st.title("请上传文本")
# 设置上传文件入口
the_file = st.file_uploader(
label="请上传文件",
type = ['txt'],
accept_multiple_files=False,
)
# 创建数据存储服务
if "service" not in st.session_state:
st.session_state["service"] = KnowledgeBase()
if the_file is not None:
# 解析文件的名称，类型和大小（大小默认为B，转成KB比较好）
file_name = the_file.name
file_type = the_file.type
file_size_kb = the_file.size / 1024
text = the_file.getvalue().decode("utf-8")
st.subheader("以下是文件的简要数据：")
st.write(f"文件名：{file_name}，文件类型：{file_type}，文件大小：{file_size_kb:.2f}KB")
with st.spinner("请稍等，正在处理中。。。"):
time.sleep(1)
st.write(st.session_state["service"].upload_by_str(text, file_name))
嗯，这是一个基于streamlit的代码，可以在命令提示符中执行，自动生成一个网页。
knowledge_base.py
"""
文件介绍
文件名：knowledge_base.py
结构：在类KnowledgeBase中可以进行文本文档的chroma上传，并基于MD5格式，进行文本文档的md5转换，记录，去重。
类内方法：upload_by_str
传入字符串和文件名，调用类外方法，将文本记录至md5文档和数据库
类外方法：check_md5，save_md5，get_string_md5
check_md5：检查传入的字符串是否在MD5文档中已经存在，如果有则拒绝再次入库。
save_md5：保存字符串到MD5文档中
get_string_md5：将普通的文本转化为16进制且固定长度。
"""
# 导入依赖
import os
import hashlib
import config_md5
from langchain_chroma import Chroma
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from datetime import datetime
# 定义三个md5相关处理方法
# 检测是否有这个文件
"""
解释：md5.txt用来存放所有曾经处理过的文本，为避免出现相同的文本，这里需要进行去重。
"""
def check_md5(input_str:str):
if not os.path.exists(config_md5.md5_path):
# 如果这个文件不存在，那么就打开后再关闭，相当于创建它。
open(config_md5.md5_path,'w',encoding="utf-8").close()
return False
else:
# 如果这个文件存在，那么接下来判断是否处理过它。（需要先去除前后换行符和空格）
texts = open(config_md5.md5_path,'r',encoding="utf-8").readlines()
for text in texts:
text = text.strip()
if text == input_str:
return True
return False
# save_md5负责往md5.txt里面输入处理过的文本（经过16进制转化的）
def save_md5(input_md5:str):
with open(config_md5.md5_path,'a',encoding="utf-8") as f:
f.write(input_md5 + '\\n')
# get_string_md5负责将普通的文本转化为16进制格式，好处就是不论文本多长多大，转化后的长度相同。
def get_string_md5(input_str:str):
# 先转化为bytes类型
byte_str = input_str.encode(encoding='utf-8')
# 创建一个md5哈希对象
hash_str = hashlib.md5()
# 将要转换的bytes类型文件传入其中
hash_str.update(byte_str)
# 转换完成，返回16进制数据
return hash_str.hexdigest()
# 创建知识库基础服务类
class KnowledgeBase(object):
def __init__(self):
# 先确保数据库存储的文件夹存在
os.makedirs(config_md5.persist_directory,exist_ok=True)
# 创建数据库
self.chroma = Chroma(
collection_name=config_md5.collection_name,
embedding_function=DashScopeEmbeddings(model="text-embedding-v4"),
persist_directory=config_md5.persist_directory,
)
# 定义文本分割器
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=config_md5.chunk_size,
chunk_overlap=config_md5.chunk_overlap,
separators=config_md5.separator,
length_function=len,
)
def upload_by_str(self,data: str,filename: str):
md5_hex = get_string_md5(data)
if check_md5(md5_hex):
return "[错误]已被记录，不允许重复的记录"
# 判断：当前的文本长度是否需要切割？
if len(data) > config_md5.max_split_char_number:
knowledge_chunk: list[str] = self.text_splitter.split_text(data)
else:
knowledge_chunk = [data]
# 源数据定义
metadata = {
"source":filename,
"create_time":datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"operator":"zzy",
}
# 将处理过的文本放到数据库中，附上源数据
self.chroma.add_texts(
knowledge_chunk,
metadatas = [metadata for _ in knowledge_chunk],
)
# 记录到md5文件中
save_md5(md5_hex)
return "[成功]已创建记录"
# 测试
if __name__ == '__main__':
service = KnowledgeBase()
r = service.upload_by_str("zzy是天才","test_file")
print(r)
这是用于将数据上传至chroma数据库，同时检测，记录是否是已经载入的数据的代码文件。
这两个文件合起来后，可以在生成的网页段中上传，解析，记录文本。
结尾
大概就是这些，具体的说明可以看看代码的开头，和代码行的注释，这两个代码文件其实原理还不算难，但是抓细节的话需要记的东西还很多。
总之，明天继续努力！我先去睡了，byebye！
