当前位置：首页 > news >正文

Day6：RAG项目实战（1）

news 2026/6/2 19:30:22

前言

hi，这里是惬鹤频道！
又过去几天了，感觉应该来更新一下我的最新进度了。
最近正式学完了RAG相关的基础知识，进入了项目实战。本来想着应该有很多可以和大家聊聊的东西，但是。。。

真的好难！！！

我不是一个能在脑袋里快速记下大概结构的人，对于RAG项目这种需要在前期进行结构思考，想清楚代码内所需的基本方法，并想好怎么把它们都拼接起来的工作，实在有点艰难。
到现在已经两天了，才写好了两个代码文件，所有py文件加起来大概有六七个，接下来还需要熬一熬才能完成。
所以这一期和大家介绍一下我写的两个代码文件吧，没有啥需要分享的问题，大家感兴趣的也可以学学。
好了不说太多了，最近的文章就短一点吧，有点累了。

app_file_uploader.py

""" 文件介绍 文件名：app_file_uploader.py 结构：利用streamlit，对接knowledge_base.py文件，创建一个可以上传，读取，解析文本文件的网页。 """# 导入依赖importstreamlitasstimporttimefromknowledge_baseimportKnowledgeBase# 创建页面st.title("请上传文本")# 设置上传文件入口the_file=st.file_uploader(label="请上传文件",type=['txt'],accept_multiple_files=False,)# 创建数据存储服务if"service"notinst.session_state:st.session_state["service"]=KnowledgeBase()ifthe_fileisnotNone:# 解析文件的名称，类型和大小（大小默认为B，转成KB比较好）file_name=the_file.name file_type=the_file.typefile_size_kb=the_file.size/1024text=the_file.getvalue().decode("utf-8")st.subheader("以下是文件的简要数据：")st.write(f"文件名：{file_name}，文件类型：{file_type}，文件大小：{file_size_kb:.2f}KB")withst.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进制且固定长度。 """# 导入依赖importosimporthashlibimportconfig_md5fromlangchain_chromaimportChromafromlangchain_community.embeddingsimportDashScopeEmbeddingsfromlangchain_text_splittersimportRecursiveCharacterTextSplitterfromdatetimeimportdatetime# 定义三个md5相关处理方法# 检测是否有这个文件""" 解释：md5.txt用来存放所有曾经处理过的文本，为避免出现相同的文本，这里需要进行去重。 """defcheck_md5(input_str:str):ifnotos.path.exists(config_md5.md5_path):# 如果这个文件不存在，那么就打开后再关闭，相当于创建它。open(config_md5.md5_path,'w',encoding="utf-8").close()returnFalseelse:# 如果这个文件存在，那么接下来判断是否处理过它。（需要先去除前后换行符和空格）texts=open(config_md5.md5_path,'r',encoding="utf-8").readlines()fortextintexts:text=text.strip()iftext==input_str:returnTruereturnFalse# save_md5负责往md5.txt里面输入处理过的文本（经过16进制转化的）defsave_md5(input_md5:str):withopen(config_md5.md5_path,'a',encoding="utf-8")asf:f.write(input_md5+'\n')# get_string_md5负责将普通的文本转化为16进制格式，好处就是不论文本多长多大，转化后的长度相同。defget_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进制数据returnhash_str.hexdigest()# 创建知识库基础服务类classKnowledgeBase(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,)defupload_by_str(self,data:str,filename:str):md5_hex=get_string_md5(data)ifcheck_md5(md5_hex):return"[错误]已被记录，不允许重复的记录"# 判断：当前的文本长度是否需要切割？iflen(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=[metadatafor_inknowledge_chunk],)# 记录到md5文件中save_md5(md5_hex)return"[成功]已创建记录"# 测试if__name__=='__main__':service=KnowledgeBase()r=service.upload_by_str("zzy是天才","test_file")print(r)