企业官网建设流程全解析

如果有听前面的课程，看起来会容易很多

一、明确思路

包含四部分代码：

二、代码书写

1、向量存储服务（vector_store.py ）

主要用于得到可以入链的检索器

from langchain_chroma import Chroma import config_data as config class VectorStoreService(object): def __init__(self,embedding): self.embedding = embedding self.vector_store = Chroma( collection_name= config.collection_name, embedding_function=self.embedding, persist_directory= config.persist_directory, ) def get_retriever(self): # as_retriever可以返回一个Runnable接口的子类实例对象，帮助vector_store入链 return self.vector_store.as_retriever(search_kwargs={'k': config.similarity_threshold}) # 验证 if __name__ == "__main__": from langchain_community.embeddings import DashScopeEmbeddings retriever = VectorStoreService(DashScopeEmbeddings(model="text-embedding-v4")).get_retriever() res = retriever.invoke("我的体重110斤，尺码推荐") print(res)

2、长期会话记忆存（file_history_store.py）

import json import os from typing import Sequence from langchain_core.chat_history import BaseChatMessageHistory from langchain_core.messages import BaseMessage, message_to_dict, messages_from_dict def get_history(session_id): return FileChatMessageHistory(session_id, "./chat_history") class FileChatMessageHistory(BaseChatMessageHistory): def __init__(self,session_id,storage_path): self.session_id = session_id # 会话id 一行结尾放逗号，会把这个变量变成元组 tuple！ self.storage_path = storage_path # 不同会话id的存储文件，所在的文件夹路径 # 完整的文件路径 self.file_path = os.path.join(self.storage_path,self.session_id) # 确保文件夹是存在的 os.path.dirname从完整的文件路径中获取到文件夹的名称 os.makedirs(os.path.dirname(self.file_path),exist_ok=True) #如果文件夹不存在，就会创建 def add_messages(self, messages: Sequence[BaseMessage]) -> None: # Sequence序列 类似list、tuple """ list()：是内置函数，用来把别的可迭代对象（消息列表、元组等）转成列表实例 []：字面量语法，直接写一个空列表，或者写带元素的列表 [1,2,3] """ all_messages = list(self.messages) # 已有的消息列表 all_messages.extend(messages) # 新的和已有的融合成一个list # 将数据同步写入到本地文件中 # 类对象写入文件都是一堆二进制 # 为了方便，可以将BaseMessage消息转换为字典（借助json模块以json字符串写入文件） # 官方message_to_dict :单个消息对象（BaseMessage类实例） -> 字典 # new_messages = [] # for message in all_messages: # d = message_to_dict(message) # new_messages.append(d) # 用列表推导式更方便 new_messages = [message_to_dict(message) for message in all_messages] # 将数据写入文件 with open(self.file_path,'w',encoding='utf-8') as f: # 是 file_path json.dump(new_messages,f) """ 读取本地文件，把 json 里保存的对话记录，还原成 LangChain 消息对象列表，供链条 (MessagesPlaceholder) 拿去用。 """ @property # @property装饰器将messages方法变成成员属性用 def messages(self) -> list[BaseMessage]: # 当前文件内： list[字典] try: with open(self.file_path, "r", encoding="utf-8") as f: messages_data = json.load(f) # 返回值就是：list[字典] 把 JSON 字符串 → 变回 Python 对象（list /dict） return messages_from_dict(messages_data) except FileNotFoundError: return [] def clear(self) -> None: with open(self.file_path, "w", encoding="utf-8") as f: json.dump([], f)

3、rag核心服务（rag.py）

from langchain_core.documents import Document from vector_stores import VectorStoreService from langchain_community.embeddings import DashScopeEmbeddings from langchain_core.prompts import ChatPromptTemplate,MessagesPlaceholder from langchain_community.chat_models.tongyi import ChatTongyi import config_data as config from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables.history import RunnableWithMessageHistory , RunnableLambda,RunnablePassthrough from file_histort_store import get_history def print_prompt(prompt): print("="*20) print(prompt.to_string()) print("="*20) return prompt class RagService(object): def __init__(self): self.vector_store = VectorStoreService(DashScopeEmbeddings(model = config.embedding_model_name)) self.prompt_template = ChatPromptTemplate.from_messages( [ ('system',"基于已有资料:{context},用简洁的话，根据用户问题给出回答"), ('system', "并且我提供用户的对话历史记录，如下"), MessagesPlaceholder('{chat_history}'), ('user', "{input}"), ] ) self.chat_model = ChatTongyi(model = config.chat_model_name) self.chain = self.__get_chain() def __get_chain(self): retriever = self.vector_store.get_retriever() def format_document(docs:list[Document]): if docs is None: return "未找到参考资料" formatted_str = "" for doc in docs: formatted_str += f"内容文档片段：{doc.page_content}\n文档元数据：{doc.metadata}\n\n" return formatted_str def format_for_retriever(value:dict) -> str: return value['input'] def format_for_prompt_template(value): # {input,context,history} new_value ={} new_value['input'] = value['input']['input'] new_value['context'] = value['context'] new_value['history'] = value['input']['history'] return new_value chain = ({"input":RunnablePassthrough() ,"context": RunnableLambda(format_for_retriever) | retriever | format_document } |RunnableLambda(format_for_prompt_template) | self.prompt_template | print_prompt | self.chat_model | StrOutputParser()) conversation_chain = RunnableWithMessageHistory( chain, # 被增强的原有chain get_history, # 通过会话id获取InMemoryChatMessageHistory类对象 input_messages_key="input", # 表示用户输入在模板中的占位符 history_messages_key="history" # 表示用户输入在模板中的占位符 ) return conversation_chain

4、项目主程序，启动对话WEB页面（app_qa.py）

from rag import RagService import streamlit as st import config_data as config st.title("智能客服") st.divider() # 在页面下方提供用户输入栏 prompt = st.chat_input() if "message" not in st.session_state: st.session_state['message'] = [{'role':'assistant',"content":'你好，有什么可以帮助你的'}] for message in st.session_state['message']: st.chat_message('role').write(message['content']) if "rag" not in st.session_state: st.session_state['rag'] = RagService() if prompt: # 在页面输出用户提问 st.chat_message('user').write(prompt) st.session_state['message'] .append({'role': 'user', "content": prompt}) ai_res_list=[] with st.spinner('AI思考中...'): res_stream = st.session_state['rag'].chain.stream({"input":prompt},config.session_config) #迭代器 def capture(generator, cache_list): for chunk in generator: cache_list.append(chunk) yield chunk st.chat_message('assistant').write_stream(capture(res_stream,ai_res_list)) # 前面生成气泡，后面展示内容 st.session_state['message'].append({'role': 'assistant', "content": "".join(ai_res_list)})