想象一下，你正在忙着手头的活儿，突然想起一个复杂的问题需要答案。如果这时候你只需要随口问一句，设备就能用语音给你一个精准的回复——这种体验是不是很酷？

这就是语音驱动的RAG（Audio RAG）系统的魅力。它把语音命令的便捷性和智能问答的能力结合在了一起。用户可以通过语音直接和知识库对话，整个过程利用了语音识别和自然语言处理的最新进展，把语音问题转成文本，再从庞大的知识库里找到最相关的信息。

Audio RAG 的工作原理

整个流程可以拆解成几个清晰的步骤：

1. 录制音频
   ：系统先用麦克风捕捉你的语音输入，你只需要说出问题就行。

2. 语音转文本
   ：录制好的音频会被送入语音识别引擎。这个引擎会分析语音的声学特征，将其转换成文本，同时保持问题的含义和上下文。

3. 执行问答
   ：拿到转录后的文本，系统利用先进的语言模型来理解问题的意图，然后搜索知识库，找到最相关的信息，给出简洁准确的答案。

4. 文本转语音
   ：最后，大语言模型生成的答案会被转成音频文件，方便你直接收听。

使用的技术栈

搭建这样一套系统，用到的技术栈相当扎实：

1. GROQ
   ：一个由LPU™ AI推理技术驱动的快速AI推理平台，主打速度快、成本低、效率高。

2. gTTS
   ：Google Text-to-Speech的Python库，能对接Google翻译的文本转语音API，把文本转成MP3文件。

3. pydub
   ：一个Python库，提供了简单易用的高级接口，方便对音频文件进行各种操作。

4. LangChain
   ：大语言模型应用开发的框架，是构建复杂链路的基石。

5. HuggingFace嵌入模型
   ：Hugging Face平台上的预训练模型，用于生成文本的密集向量表示，可用于语义搜索、文本分类等任务。

6. Streamlit
   ：开源的Python框架，用于快速构建和分享交互式数据应用。

7. ChromaDB
   ：开源向量数据库，专门用于存储和检索向量嵌入，非常适合大语言模型和语义搜索场景。

代码实现

接下来，我们一步步来看如何用代码实现这个系统。

安装所需依赖

!pip install langchain langchain_community langchain_groq chromadb sentence_transformers
!pip install -U langchain-huggingface
!pip install -U langchain-chroma
!pip install python-dotenv

创建 .env 文件并设置Groq API密钥：

GROQ_API_KEY=<你的API密钥>

设置Groq API密钥：

import os
from dotenv import load_dotenv

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

设置LLM：

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

设置RetrievalQA辅助函数：

# RetrievalQA
def 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']

使用Groq distil-whisper-large-v3-en转录音频的辅助函数。这个模型是OpenAI Whisper的压缩版本，专为英语语音识别设计，速度更快、成本更低，同时保持了相当的准确性。

import os
from groq import Groq

# 初始化 Groq 客户端
client = Groq()

# 指定音频文件的路径
filename = "/content/recorded_audio.wa v"

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.sa ve(mp3_file)
    
    return mp3_file

为Streamlit应用程序准备的完整代码实现。编写一个名为 audio_rag.py 的Python脚本：

import streamlit as st
from time import sleep
#from st_audiorec import st_audiorec
from streamlit_mic_recorder import mic_recorder
from streamlit_chat import message
import os
from groq import Groq
from langchain_groq import ChatGroq
from langchain_community.embeddings import HuggingFaceBgeEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import PyPDFLoader
from langchain_chroma import Chroma
from langchain.chains import RetrievalQA
from chromadb.config import Settings
import chromadb
from gtts import gTTS
from pydub import AudioSegment
import os
from 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_file = "temp_audio.mp3"
    tts.sa ve(mp3_file)
    return mp3_file

# 保存上传的文件
def sa ve_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.wa v"

# 转录音频的辅助函数
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="?",
    layout="wide"
)

# 创建两列布局
col1, col2 = st.columns([1, 2])

# 在左列显示内容
with col1:
    st.markdown(
        """
        

        ? 启用音频的 ? 知识应用程序
        
        

        您的音频问答系统一站式解决方案！
        
        """,
        unsafe_allow_html=True
    )
    st.write("欢迎使用启用音频的 RAG 应用程序！")
    st.image("audio.jpeg", caption="音频驱动的 RAG", output_format="auto")
    
    if st.button("停止进程"):
        st.session_state.stop = 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:
        sa ve_uploaded_file(uploaded_file, upload_dir)
        file_name = uploaded_file.name
        loader = 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 = 100
            for 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_list
        selected_option = st.selectbox("选择一个选项：", options)
        file_name = newest("uploaded_files") if selected_option == "none" else selected_option
        st.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/wa v')
            with open("recorded_audio.wa v", "wb") as f:
                f.write(audio['bytes'])
            st.success("音频录制完成！")
            
            with st.spinner("正在进行音频转录..."):
                text = transcribe_audio(filename)
                transcription = text
                st.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/wa v', start_time=0)

运行Streamlit应用程序：

streamlit run audio_rag.py

应用程序操作步骤

1. 浏览并上传PDF文件。
2. 向量存储加载完成后，点击“启动进程”按钮。
3. 开始录制你的问题。问题录制完成后，RAG流程会自动触发。屏幕上会显示答案以及对应的音频。
4. 如果想在已上传的文档上进行问答，可以在知识来源选择框中选择相应选项。
5. 上传另一个文档。如果你只想在当前文档中问答，选择“无”选项。
6. 按下停止按钮以停止进程。
7. 刷新屏幕以重新开始。

结论

这篇文章详细探讨了如何使用Python构建一个音频驱动的问答系统（Audio RAG）。从录制音频、语音转文本，到利用RAG模型进行问答，这里罗列了关键步骤。通过整合这些技术，我们可以打造一个强大的系统，让用户直接用语音命令与知识库互动，彻底解放双手。