AI应用中基于语音驱动的RAG如何做？

GROQ_API_KEY=<你的API密钥>

import osfrom dotenv import load_dotenv
# 从 .env 文件加载环境变量load_dotenv()

llm = ChatGroq(model_name="llama3-70b-8192",temperature=0.1,max_tokens=1000,)

from langchain_huggingface import HuggingFaceEmbeddings
model_name = "BAAI/bge-small-en-v1.5"model_kwargs = {"device": "cpu"}encode_kwargs = {"normalize_embeddings": False}embeddings = HuggingFaceEmbeddings(model_name=model_name,model_kwargs=model_kwargs,encode_kwargs=encode_kwargs)

# 文本分割器def text_splitter():text_splitter = RecursiveCharacterTextSplitter(chunk_size=512,chunk_overlap=20,length_function=len,)return text_splitter

# RetrievalQAdef answer_question(question):retriever = vectorstore.as_retriever(search_kwargs={"k": 5})qa = RetrievalQA.from_chain_type(llm=llm,chain_type="stuff",retriever=retriever,return_source_documents=True)result = qa.invoke({"query": question})return result['result']

import osfrom groq import Groq
# 初始化 Groq 客户端client = Groq()
# 指定音频文件的路径filename = "/content/recorded_audio.wav"
def transcribe_audio(filename):# 打开音频文件with open(filename, "rb") as file:# 创建音频文件的转录transcription = client.audio.transcriptions.create(file=(filename, file.read()),# 必需的音频文件model="distil-whisper-large-v3-en",# 必需的用于转录的模型prompt="Specify context or spelling",# 可选的上下文或拼写提示response_format="json",# 可选的响应格式language="en",# 可选的语言temperature=0.0# 可选的温度参数)# 打印转录文本print(transcription.text)return transcription.text

from gtts import gTTS
def text_to_audio(text):# 将文本转换为语音tts = gTTS(text=text, lang='en', slow=False)# 将音频保存为MP3文件mp3_file = "temp_audio.mp3"tts.save(mp3_file)return mp3_file

import streamlit as stfrom time import sleep#from st_audiorec import st_audiorecfrom streamlit_mic_recorder import mic_recorderfrom streamlit_chat import messageimport osfrom groq import Groqfrom langchain_groq import ChatGroqfrom langchain_community.embeddings import HuggingFaceBgeEmbeddingsfrom langchain.text_splitter import RecursiveCharacterTextSplitterfrom langchain.document_loaders import PyPDFLoaderfrom langchain_chroma import Chromafrom langchain.chains import RetrievalQAfrom chromadb.config import Settingsimport chromadbfrom gtts import gTTSfrom pydub import AudioSegmentimport osfrom dotenv import load_dotenv
# 从 .env 文件加载环境变量load_dotenv()设置 Chroma 配置
chroma_setting = Settings(anonymized_telemetry=False)获取最新的文件路径
def newest(path):files = os.listdir(path)paths = [os.path.join(path, basename) for basename in files]newest_file_path = max(paths, key=os.path.getctime)return os.path.basename(newest_file_path)文本转换为音频的辅助函数
def text_to_audio(text):# 将文本转换为语音tts = gTTS(text=text, lang='en', slow=False)# 将音频保存为MP3文件mp3_file = "temp_audio.mp3"tts.save(mp3_file)return mp3_file保存上传的文件
def save_uploaded_file(uploaded_file, directory):try:with open(os.path.join(directory, uploaded_file.name), "wb") as f:f.write(uploaded_file.getbuffer())return st.success(f"已保存文件：{uploaded_file.name} 到 {directory}")except Exception as e:return st.error(f"保存文件出错：{e}")创建用于保存上传文件的目录
upload_dir = "uploaded_files"os.makedirs(upload_dir, exist_ok=True)设置 LLM
llm = ChatGroq(model_name="llama3-70b-8192",temperature=0.1,max_tokens=1000,)设置嵌入模型
model_name ="BAAI/bge-small-en-v1.5"model_kwargs ={"device":"cpu"}encode_kwargs ={"normalize_embeddings":False}embeddings = HuggingFaceBgeEmbeddings(model_name=model_name, model_kwargs=model_kwargs, encode_kwargs=encode_kwargs)设置文本分割器
def text_splitter():text_splitter = RecursiveCharacterTextSplitter(chunk_size=512,chunk_overlap=20,length_function=len,)return text_splitter设置 RetrievalQA
def answer_question(question,vectorstore):retriever = vectorstore.as_retriever(search_kwargs={"k": 5})qa = RetrievalQA.from_chain_type(llm=llm, chain_type="stuff", retriever=retriever, return_source_documents=True)result = qa.invoke({"query": question})return result['result']初始化 Groq 客户端
groq_client = Groq()指定音频文件的路径
filename = "recorded_audio.wav"转录音频的辅助函数
def transcribe_audio(filename):# 打开音频文件with open(filename, "rb") as file:# 创建音频文件的转录transcription = groq_client.audio.transcriptions.create(file=(filename, file.read()), # 必需的音频文件model="distil-whisper-large-v3-en", # 必需的用于转录的模型prompt="指定上下文或拼写",# 可选response_format="json",# 可选language="en",# 可选temperature=0.0# 可选)# 打印转录文本print(transcription.text)return transcription.text初始化 session state 变量
if 'stop' not in st.session_state:st.session_state.stop = False设置页面配置
st.set_page_config(page_title="音频和书籍应用程序",page_icon="?",# 您可以使用表情符号或图片URLlayout="wide")创建两列布局
col1, col2 = st.columns([1, 2])# 调整比例控制列的宽度在左列显示内容
with col1:st.markdown("""<h1 style='text-align: center;'>? 启用音频的 ? 知识应用程序</h1><h5 style='text-align: center;'>您的音频问答系统一站式解决方案！</h5>""",unsafe_allow_html=True)st.write("欢迎使用启用音频的 RAG 应用程序！")st.image("audio.jpeg", caption="音频驱动的 RAG", output_format="auto")if st.button("停止进程"):st.session_state.stop = True# 设置停止标志为 True
if st.session_state.stop:st.write("进程已停止。您可以刷新页面重新开始。")在右列显示 PDF 上传和阅读器
with col2:st.title("PDF 上传和阅读器")uploaded_file = st.file_uploader("选择一个 PDF 文件", type="pdf")persist_directory_path = "chromanew"if uploaded_file is not None:save_uploaded_file(uploaded_file, upload_dir)file_name = uploaded_file.nameloader = PyPDFLoader(f"uploaded_files/{file_name}")pages = loader.load_and_split(text_splitter())persist_directory = persist_directory_path + "_" + file_name.split(".")[0]if os.path.exists(persist_directory):client = chromadb.PersistentClient(path=persist_directory, settings=chroma_setting)vectorstore = Chroma(embedding_function=embeddings, client=client, persist_directory=persist_directory, collection_name=file_name.split(".")[0], client_settings=chroma_setting,)print(f"向量存储中加载的文档数量：{len(vectorstore.get()['documents'])}")else:client = chromadb.PersistentClient(path=persist_directory, settings=chroma_setting)vectorstore = Chroma(embedding_function=embeddings, client=client, persist_directory=persist_directory, collection_name=file_name.split(".")[0], client_settings=chroma_setting)MAX_BATCH_SIZE = 100for i in range(0, len(pages), MAX_BATCH_SIZE):i_end = min(len(pages), i+MAX_BATCH_SIZE)batch = pages[i:i_end]vectorstore.add_documents(batch)print(f"向量存储中加载的文档数量：{len(vectorstore.get()['documents'])}")创建一个按钮来启动进程
if 'start_process' not in st.session_state:st.session_state.start_process = False
if st.button("启动进程"):st.session_state.start_process = True
if st.session_state.start_process:options = os.listdir("uploaded_files")none_list = ["none"]options += none_listselected_option = st.selectbox("选择一个选项：", options)file_name = newest("uploaded_files") if selected_option == "none" else selected_optionst.write(f"您选择了：{selected_option}")st.title("音频录制 - 根据所选选项提问")
with st.spinner("正在进行音频录制..."):audio = mic_recorder(start_prompt="开始录制",stop_prompt="停止录制",just_once=False,key='recorder')if audio:st.audio(audio['bytes'], format='audio/wav')with open("recorded_audio.wav", "wb") as f:f.write(audio['bytes'])st.success("音频录制完成！")with st.spinner("正在进行音频转录..."):text = transcribe_audio(filename)transcription = textst.markdown(text)if "chat_history" not in st.session_state:st.session_state.chat_history = []for i, chat in enumerate(st.session_state.chat_history):message(chat["question"], is_user=True, key=f"question_{i}")message(chat["response"], is_user=False, key=f"response_{i}")if transcription:with st.spinner("正在生成回应..."):persist_directory = persist_directory_path + "_" + file_name.split(".")[0]client = chromadb.PersistentClient(path=persist_directory, settings=chroma_setting)vectorstore = Chroma(embedding_function=embeddings,client=client,persist_directory=persist_directory,collection_name=file_name.split(".")[0],client_settings=chroma_setting)response = answer_question(transcription, vectorstore)st.success("回应已生成")aud_file = text_to_audio(response)st.session_state.chat_history.append({"question": transcription, "response": response})message(transcription, is_user=True)message(response, is_user=False)st.title("音频播放")st.audio(aud_file, format='audio/wav', start_time=0)

streamlit run audio_rag.py