1.文本上传web服务基于steamlit完成web网页上传服务Streamlit用 Python 写网页应用的工具。基于steamlit完成web网页上传服务 from distutils.command.upload import upload import streamlit as st st.title(知识库更新服务) upload_file st.file_uploader( 请上传文件TXT文件, type[txt], accept_multiple_filesFalse, ) if upload_file is not None: file_nameupload_file.name file_typeupload_file.type file_sizeupload_file.size/1024 st.subhead(f文件名{file_name}) st.write(f格式{file_type} | 大小{file_size}) text upload_file.getvalue().decode(utf-8) st.write(text)上面是生成上传文件的web页面的代码进入该网页的步骤为1.cmd,进入项目所在的盘d:/c:2.cd进入网页代码这个文件佳路径cd 路径3.streamlit run 网页这个py代码2.RAG知识库构建1创建向量数据库目录→创建 Chroma 向量数据库参数一collection_name表明参数二persist_directory数据库存储位置参数三embedding_function文本转换为向量2创建文本分割器使用self.spliter RecursiveCharacterTextSplitter LangChain 的文本分割工具)参数一chunk_size每个文本块最大长度 参数二chunk_overlap文本块之间的重叠参数三separators文本分割优先级 参数四length_function统计文本分割长度3文本上传函数文本上传函数。生成 MD5检查是否存在。4若不存在则判断文本长度太长则切分较短则直接存入。5构建 metadata构建mededata,向量的附加信息。3.连接文件上传和知识库构建1在 Streamlit 会话中创建并保存一个 knowledgeBaseService 实例避免每次页面重新运行时重复初始化。if service not in st.session_state: st.session_state[service]knowledgeBaseService()2对上传的文件调用RAG数据库构建部分的文件对文件内容进行转化、判断、存储。if upload_file is not None: file_nameupload_file.name file_typeupload_file.type file_sizeupload_file.size/1024 st.subheader(f文件名{file_name}) st.write(f格式{file_type} | 大小{file_size}) with st.spinner(载入中): time.sleep(1) text upload_file.getvalue().decode(utf-8) result st.session_state[service].upload_by_str(text,file_name)# 这里 st.success(result)4.加载 Chroma 向量数据库并提供一个基于 embedding 相似度的文本检索接口。当系统需要进行知识检索时它负责加载本地已经保存好的 Chroma 向量数据库并生成一个检索器retriever用于根据用户问题找到最相似的文本内容。class VectorStoreService(object): def __init__(self,embedding): self.embeddingembedding self.vector_storeChroma( #加载向量数据库 collection_nameconfig.collection_name, embedding_functionself.embedding, persist_directoryconfig.persist_directory ) def get_retriever(self): #提供检索器 return self.vector_store.as_retriever(search_kwargs{k:config.similarity_threshold})5.构建链→检索器、提示词、聊天模型链rag开发核心代码部分class RagService(object): def __init__(self): self.vector_service VectorStoreService( #获取检索器 embeddingconfig.embedding ) self.prompt_template ChatPromptTemplate.from_messages( #构建提示词 [ (system,以我提供的已知参考资料为主 简洁和专业的回答用户问题参考资料{context}。), (user,请回答用户提问{input}) ] ) self.chat_model config.chat_model_name #获取聊天模型 self.chain self.__get_chain() #构建链 def __get_chain(self): retriever self.vector_service.get_retriever() #输出是list套ducument def format_document(docs:list[Document]): if not docs: return 无相关资料 formatted_str for doc in docs: formatted_str f文档片段{doc.page_content}\n文档元数据{doc.metadata}\n\n #多个字符串拼接为1个 return formatted_str chain ( { input:RunnablePassthrough(), context:retriever|format_document }|self.prompt_template|self.chat_model|StrOutputParser() ) return chain6.保存长期历史会话记录1根据 session_id会话ID 返回一个聊天历史对象不同用户 / 不同会话→ 对应不同文件def get_history(session_id): return FileChatMessageHistory(session_id,./chat_history)2)自定义聊天历史类class FileChatMessageHistory(BaseChatMessageHistory)只要实现三个方法a加载已有的聊天记录将新的和旧的融合。存储聊天记录时需为字符串形式因为文件存储时必须使用 JSON 可序列化的数据通常是字典/字符串而 LangChain 在进行对话推理时需要的是 BaseMessage 对象。def add_messages(self,messages:Sequence[BaseMessage]) - None: all_messageslist(self.messages) #已有的消息列表 all_messages.extend(messages) #新的和已有的融合成一个list new_messages [] for messages in all_messages: d message_to_dict(messages) #将对话的basemessage格式转换为字典 new_messages.append(d) with open(self.file_path,w,encodingutf-8) as f : json.dump(new_messages,f)b从文件读取历史消息LLM 推理必须使用 Message 对象。这是在第1步读取聊天记录后在通过这个方法转会为message格式模型通过这些进行推理。def messages(self) -list[BaseMessage]: try: with open(self.file_path,r,encodingutf-8)as f: message_data json.load(f) return messages_from_dict(message_data) except FileNotFoundError: return []c通过clear清空聊天记录3把聊天记录通过MessagesPlaceholder(history),插入到会话中让会话具有上下文信息。这里注意插入历史记录后链的输入可能会发生变化。要注意input和每个链的输入比如 #retriever要求输入的是字符串。7.聊天页面开发import time import streamlit as st from streamlit import spinner from rag import RagService import config_data as config #设置一个标题 st.title(智能客服) st.divider() #在 Streamlit 应用中保存用户会话状态让变量在页面重新运行时不会丢失。 if message not in st.session_state: st.session_state[message] [{role:assistant,content:您好请问有什么可以帮您}] if rag not in st.session_state: st.session_state[rag] RagService() for message in st.session_state[message]: st.chat_message(message[role]).write(message[content]) prompt st.chat_input() if prompt: st.chat_message(user).write(prompt) st.session_state[message].append({role:user,content:prompt}) ai_res_list [] with st.spinner(思考中...): #time.sleep(1) def capture(generator,cache_list): #stream的流式输出无法存储在历史中必须转换为字符串 for chunk in generator: cache_list.append(chunk) yield chunk res_stream st.session_state[rag].chain.stream({input:prompt},config.session_config) st.chat_message(assistant).write_stream(capture(res_stream,ai_res_list)) st.session_state[message].append({role: assistant, content:.join(ai_res_list)})