音頻也能“對話”？用 AssemblyAI、Qdrant 和 DeepSeek-R1 構建音頻 RAG 聊天機器人 原創 精華
在信息爆炸的時代，音頻內容也如潮水般涌來。無論是會議記錄、播客還是采訪，我們常常需要從海量音頻中提取關鍵信息。但手動篩選音頻不僅耗時費力，還容易錯過重要細節。今天，就讓我們一起探索如何用 AssemblyAI、Qdrant 和 DeepSeek-R1 構建一個強大的 AI 驅動的聊天機器人，將音頻轉化為可交互的對話內容，讓音頻檢索變得輕松又高效！
一、初識利器：AssemblyAI、SambaNova Cloud、Qdrant 和 DeepSeek-R1
（一）AssemblyAI：精準轉錄的“專家”
AssemblyAI 是音頻轉錄領域的佼佼者。它就像一位精通多國語言的速記員，無論是帶有口音的演講，還是嘈雜背景下的對話，都能準確無誤地將其轉化為文字。無論是轉錄播客、分析客戶電話，還是為視頻添加字幕，AssemblyAI 都能輕松應對，為我們的音頻處理工作打下堅實基礎。
（二）SambaNova Cloud：讓大模型運行“飛”起來
想象一下，如果能以 10 倍的速度運行像 DeepSeek-R1 這樣龐大的開源模型，而且無需擔心復雜的基礎設施問題，那該有多輕松？SambaNova Cloud 正是為此而生。它不依賴傳統的 GPU，而是采用 RDUs（可重構數據流單元），帶來驚人的性能提升。它擁有海量的內存存儲，無需頻繁重新加載模型；數據流設計高效，專為高吞吐量任務優化；還能在微秒級瞬間切換模型。在 SambaNova Cloud 上，你可以輕松訓練、微調模型，一切都在同一個平臺上完成。
（三）Qdrant：快速檢索的“搜索引擎”
Qdrant 是一款超快速的向量數據庫，堪稱 AI 應用的加速器。它就像是在海量數據中尋找針尖的高手，無論是構建推薦系統、圖像搜索工具，還是聊天機器人，Qdrant 都能快速找到與復雜數據（如文本嵌入或視覺特征）最相似的匹配項。有了它，我們的音頻轉錄內容可以被高效存儲和檢索，為后續的智能對話提供有力支持。
（四）DeepSeek-R1：自然語言理解的“大師”
DeepSeek-R1 是一款極具創新性的語言模型，它將人類般的適應性與前沿 AI 技術完美融合，在自然語言處理領域獨樹一幟。無論是撰寫內容、翻譯語言、調試代碼，還是總結復雜報告，DeepSeek-R1 都能精準理解上下文、語氣和意圖，給出自然流暢而非機械生硬的回答。它不僅僅是一個工具，更是讓我們窺見未來 AI 與人類自然交流的窗口。
二、搭建 RAG 模型：讓音頻“活”起來
（一）搭建前的準備
在開始搭建 RAG 模型之前，我們需要做好一些準備工作。首先，從 GitHub 上克隆項目倉庫（??https://github.com/karthikponna/chat_with_audios.git??），然后進入項目目錄。接下來，根據操作系統創建并激活虛擬環境，安裝所需的依賴包，并設置好 AssemblyAI 和 SambaNova 的 API 密鑰。這些步驟就像是為我們的項目搭建起穩固的“腳手架”，確保后續開發工作順利進行。
git clone https://github.com/karthikponna/chat_with_audios.git
cd chat_with_audios
# 創建虛擬環境
python3 -m venv venv
source venv/bin/activate # macOS 和 Linux
# 對于 Windows：
# python -m venv venv
# .\venv\Scripts\activate
# 安裝依賴
pip install -r requirements.txt
# 設置環境變量
touch .env # macOS 和 Linux
# 對于 Windows：
# type nul > .env
echo"ASSEMBLYAI_API_KEY=your_assemblyai_api_key_string" >> .env
echo"SAMBANOVA_API_KEY=your_sambanova_api_key_string" >> .env（二）Retrieval Augmented Generation（RAG）：融合檢索與生成
RAG 是一種將大型語言模型與外部數據相結合的技術，它能夠在查詢時獲取相關信息，從而生成更準確、更具上下文的答復。這種技術確保了回答不僅依賴于模型的訓練數據，而是基于真實可靠的數據，讓聊天機器人變得更加智能和實用。
（三）代碼實現：一步步構建 RAG 系統
1. 批量處理與文本嵌入
我們首先定義了一個 ??batch_iterate??? 函數，它可以將文本列表分割成更小的塊，方便后續處理大規模數據集。接著，創建了一個 ??EmbedData?? 類，它加載 Hugging Face 嵌入模型，為每一塊文本生成嵌入向量，并將這些嵌入向量收集起來，為后續的存儲和檢索做好準備。
from typing import List
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
def batch_iterate(lst: List[str], batch_size: int) -> List[List[str]]:
"""將文本列表分割成小塊"""
for i in range(0, len(lst), batch_size):
yield lst[i : i + batch_size]
class EmbedData:
def __init__(self, embed_model_name: str = "BAAI/bge-large-en-v1.5", batch_size: int = 32):
self.embed_model_name = embed_model_name
self.embed_model = self._load_embed_model()
self.batch_size = batch_size
self.embeddings = []
def _load_embed_model(self) -> HuggingFaceEmbedding:
"""加載 Hugging Face 嵌入模型"""
embed_model = HuggingFaceEmbedding(model_name=self.embed_model_name, trust_remote_code=True, cache_folder='./hf_cache')
return embed_model
def generate_embedding(self, context: List[str]) -> List[List[float]]:
"""生成文本嵌入向量"""
return self.embed_model.get_text_embedding_batch(context)
def embed(self, contexts: List[str]) -> None:
"""為所有文本生成嵌入向量"""
self.contexts = contexts
for batch_context in batch_iterate(contexts, self.batch_size):
batch_embeddings = self.generate_embedding(batch_context)
self.embeddings.extend(batch_embeddings)2. Qdrant 向量數據庫設置與數據導入
??QdrantVDB_QB?? 類負責初始化 Qdrant 向量數據庫。它設置了關鍵參數，如集合名稱、向量維度和批量大小，并連接到 Qdrant 數據庫。如果指定的集合不存在，它會自動創建一個新的集合。然后，它通過批量處理的方式，將文本上下文及其對應的嵌入向量高效地上傳到數據庫中，并根據需要更新集合的配置。
from qdrant_client import QdrantClient, models
class QdrantVDB_QB:
def __init__(self, collection_name: str, vector_dim: int = 768, batch_size: int = 512):
self.collection_name = collection_name
self.batch_size = batch_size
self.vector_dim = vector_dim
def define_client(self) -> None:
"""定義 Qdrant 客戶端"""
self.client = QdrantClient(url="http://localhost:6333", prefer_grpc=True)
def create_collection(self) -> None:
"""創建 Qdrant 集合"""
ifnot self.client.collection_exists(collection_name=self.collection_name):
self.client.create_collection(
collection_name=self.collection_name,
vectors_cnotallow=models.VectorParams(
size=self.vector_dim,
distance=models.Distance.DOT,
on_disk=True
),
optimizers_cnotallow=models.OptimizersConfigDiff(
default_segment_number=5,
indexing_threshold=0
),
quantization_cnotallow=models.BinaryQuantization(
binary=models.BinaryQuantizationConfig(always_ram=True)
)
)
def ingest_data(self, embeddata: EmbedData) -> None:
"""將嵌入數據導入 Qdrant"""
for batch_context, batch_embeddings in zip(
batch_iterate(embeddata.contexts, self.batch_size),
batch_iterate(embeddata.embeddings, self.batch_size)
):
self.client.upload_collection(
collection_name=self.collection_name,
vectors=batch_embeddings,
payload=[{"context": context} for context in batch_context]
)
self.client.update_collection(
collection_name=self.collection_name,
optimizer_cnotallow=models.OptimizersConfigDiff(indexing_threshold=20000)
)3. 查詢嵌入檢索器
??Retriever??? 類是連接用戶查詢和向量數據庫的橋梁。它初始化時接收一個向量數據庫客戶端和一個嵌入模型。其 ??search?? 方法將用戶查詢轉化為嵌入向量，然后在數據庫中進行向量搜索，通過精細調整量化參數，快速檢索出與查詢最相關的結果。
class Retriever:
def __init__(self, vector_db: QdrantVDB_QB, embeddata: EmbedData):
self.vector_db = vector_db
self.embeddata = embeddata
def search(self, query: str) -> List[dict]:
"""根據用戶查詢檢索相關上下文"""
query_embedding = self.embeddata.embed_model.get_query_embedding(query)
result = self.vector_db.client.search(
collection_name=self.vector_db.collection_name,
query_vector=query_embedding,
search_params=models.SearchParams(
quantizatinotallow=models.QuantizationSearchParams(
ignore=False,
rescore=True,
oversampling=2.0
)
),
timeout=1000
)
return result4. RAG 智能查詢助手
??RAG?? 類將檢索器和大型語言模型（LLM）整合在一起，用于生成具有上下文意識的回應。它從向量數據庫中檢索相關信息，將其格式化為結構化的提示文本，然后發送給 LLM 以獲取回應。在這里，我們通過 SambaNova Cloud 的 API 訪問 LLM 模型，實現高效的文本生成。
from llama_index.llms.sambanovasystems import SambaNovaCloud
from llama_index.core.base.llms.types import ChatMessage, MessageRole
class RAG:
def __init__(self, retriever: Retriever, llm_name: str = "Meta-Llama-3.1-405B-Instruct"):
system_msg = ChatMessage(role=MessageRole.SYSTEM, cnotallow="You are a helpful assistant that answers questions about the user's document.")
self.messages = [system_msg]
self.llm_name = llm_name
self.llm = self._setup_llm()
self.retriever = retriever
self.qa_prompt_tmpl_str = (
"Context information is below.\n"
"---------------------\n"
"{context}\n"
"---------------------\n"
"Given the context information above I want you to think step by step to answer the query in a crisp manner, incase case you don't know the answer say 'I don't know!'.\n"
"Query: {query}\n"
"Answer: "
)
def _setup_llm(self) -> SambaNovaCloud:
"""設置 LLM"""
return SambaNovaCloud(
model=self.llm_name,
temperature=0.7,
context_window=100000
)
def generate_context(self, query: str) -> str:
"""生成上下文"""
result = self.retriever.search(query)
context = [dict(data) for data in result]
combined_prompt = []
for entry in context[:2]:
combined_prompt.append(entry["payload"]["context"])
return"\n\n---\n\n".join(combined_prompt)
def query(self, query: str) -> str:
"""處理用戶查詢"""
context = self.generate_context(query)
prompt = self.qa_prompt_tmpl_str.format(cnotallow=context, query=query)
user_msg = ChatMessage(role=MessageRole.USER, cnotallow=prompt)
streaming_response = self.llm.stream_complete(user_msg.content)
return streaming_response5. 音頻轉錄
??Transcribe?? 類負責初始化 AssemblyAI API 密鑰并創建轉錄器。它使用配置了說話者標簽的轉錄器處理音頻文件，最終返回一個字典列表，其中每個條目都將說話者與其轉錄文本相對應。這讓我們能夠清楚地知道每個說話者在音頻中說了什么內容。
import assemblyai as aai
from typing import List, Dict
class Transcribe:
def __init__(self, api_key: str):
"""初始化轉錄器"""
aai.settings.api_key = api_key
self.transcriber = aai.Transcriber()
def transcribe_audio(self, audio_path: str) -> List[Dict[str, str]]:
"""轉錄音頻文件"""
config = aai.TranscriptionConfig(speaker_labels=True, speakers_expected=2)
transcript = self.transcriber.transcribe(audio_path, cnotallow=config)
speaker_transcripts = []
for utterance in transcript.utterances:
speaker_transcripts.append({
"speaker": f"Speaker {utterance.speaker}",
"text": utterance.text
})
return speaker_transcripts三、Streamlit 應用：讓交互變得簡單有趣
Streamlit 是一個強大的 Python 庫，它可以將數據腳本轉換為交互式的 Web 應用程序，非常適合基于 LLM 的解決方案。我們利用 Streamlit 構建了一個用戶友好的應用程序，用戶可以通過它上傳音頻文件，查看轉錄內容，并與聊天機器人進行實時互動。
當用戶上傳音頻文件后，應用程序會使用 AssemblyAI 將音頻轉錄為帶有說話者標簽的文本。然后，這些文本會被嵌入并存儲在 Qdrant 向量數據庫中，以便快速檢索。檢索器與 RAG 引擎配合，利用這些嵌入向量生成具有上下文意識的聊天回應。同時，會話狀態管理聊天歷史和文件緩存，確保用戶體驗流暢。
import os
import gc
import uuid
import tempfile
import base64
from dotenv import load_dotenv
import streamlit as st
# 加載環境變量
load_dotenv()
# 初始化會話狀態
if"id"notin st.session_state:
st.session_state.id = uuid.uuid4()
st.session_state.file_cache = {}
# 側邊欄：上傳音頻文件
with st.sidebar:
st.header("Add your audio file!")
uploaded_file = st.file_uploader("Choose your audio file", type=["mp3", "wav", "m4a"])
if uploaded_file:
try:
with tempfile.TemporaryDirectory() as temp_dir:
file_path = os.path.join(temp_dir, uploaded_file.name)
with open(file_path, "wb") as f:
f.write(uploaded_file.getvalue())
file_key = f"{st.session_state.id}-{uploaded_file.name}"
st.write("Transcribing with AssemblyAI and storing in vector database...")
if file_key notin st.session_state.get('file_cache', {}):
# 初始化轉錄器
transcriber = Transcribe(api_key=os.getenv("ASSEMBLYAI_API_KEY"))
# 獲取帶有說話者標簽的轉錄內容
transcripts = transcriber.transcribe_audio(file_path)
st.session_state.transcripts = transcripts
# 將每個說話者片段作為單獨的文檔進行嵌入
documents = [f"Speaker {t['speaker']}: {t['text']}"for t in transcripts]
embeddata = EmbedData(embed_model_name="BAAI/bge-large-en-v1.5", batch_size=32)
embeddata.embed(documents)
# 設置向量數據庫
qdrant_vdb = QdrantVDB_QB(collection_name="chat_with_audios", batch_size=32, vector_dim=1024)
qdrant_vdb.define_client()
qdrant_vdb.create_collection()
qdrant_vdb.ingest_data(embeddata=embeddata)
# 設置檢索器
retriever = Retriever(vector_db=qdrant_vdb, embeddata=embeddata)
# 設置 RAG 引擎
query_engine = RAG(retriever=retriever, llm_name="DeepSeek-R1-Distill-Llama-70B")
st.session_state.file_cache[file_key] = query_engine
else:
query_engine = st.session_state.file_cache[file_key]
# 提示用戶文件已準備好
st.success("Ready to Chat!")
st.audio(uploaded_file)
# 顯示帶有說話者標簽的轉錄內容
st.subheader("Transcript")
with st.expander("Show full transcript", expanded=True):
for t in st.session_state.transcripts:
st.text(f"**{t['speaker']}**: {t['text']}")
except Exception as e:
st.error(f"An error occurred: {e}")
st.stop()
# 初始化聊天歷史
if"messages"notin st.session_state:
st.session_state.messages = []
# 顯示聊天歷史
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.markdown(message["content"])
# 接受用戶輸入
if prompt := st.chat_input("Ask about the audio conversation..."):
# 將用戶消息添加到聊天歷史中
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.markdown(prompt)
# 顯示助手回應
with st.chat_message("assistant"):
message_placeholder = st.empty()
full_response = ""
streaming_response = query_engine.query(prompt)
for chunk in streaming_response:
try:
new_text = chunk.raw["choices"][0]["delta"]["content"]
full_response += new_text
message_placeholder.markdown(full_response + "▌")
except:
pass
message_placeholder.markdown(full_response)
# 將助手回應添加到聊天歷史中
st.session_state.messages.append({"role": "assistant", "content": full_response})四、總結與展望
通過將 AssemblyAI、SambaNova Cloud、Qdrant 和 DeepSeek-R1 結合起來，我們成功構建了一個基于音頻的聊天機器人，它利用檢索增強生成（RAG）技術，為用戶提供了一個高效、智能的音頻檢索和對話體驗。??rag_code.py??? 文件管理著整個 RAG 工作流程，而 ??app.py?? 文件則提供了一個簡潔的 Streamlit 界面，讓整個系統易于使用和擴展。
這個項目的成功離不開各個組件的協同合作：AssemblyAI 提供了精準的音頻轉錄，為后續的對話體驗奠定了基礎；Qdrant 確保了快速的向量檢索，讓聊天機器人能夠迅速找到相關上下文；RAG 方法將檢索和生成相結合，確保回答基于真實數據；SambaNova Cloud 為 LLM 提供了強大的語言理解能力，讓對話更加自然流暢；Streamlit 則為我們提供了一個簡單易用的用戶界面，簡化了音頻聊天機器人的部署過程。
現在，你可以通過運行 ??streamlit run app.py??? 來啟動這個應用程序，上傳音頻文件并與聊天機器人互動。你還可以嘗試使用不同的音頻文件，調整代碼，添加新功能，探索音頻聊天解決方案的無限可能。GitHub 倉庫（??https://github.com/karthikponna/chat_with_audios/tree/main??）中包含了完整的代碼和相關資源，等待你去挖掘和創新。
音頻不再是靜態的信息載體，而是可以與我們互動、為我們提供幫助的智能伙伴。讓我們一起期待，在未來，音頻技術將如何繼續改變我們的生活和工作方式！
五、常見問題解答
（一）Qdrant 是如何實現快速檢索的？
Qdrant 通過高效的向量索引和優化的搜索算法，能夠快速在海量數據中找到與查詢向量最相似的匹配項。它支持多種距離度量方式，并且可以根據數據特點和查詢需求進行靈活配置，從而實現快速準確的檢索。
（二）DeepSeek-R1 與其他語言模型相比有什么優勢？
DeepSeek-R1 的優勢在于其出色的人類適應性和自然語言理解能力。它能夠精準把握上下文、語氣和意圖，生成自然流暢的回答，而不是機械生硬的文本。這使得它在處理復雜的語言任務時表現出色，能夠更好地滿足用戶的需求。
（三）Streamlit 應用程序如何管理聊天歷史和文件緩存？
Streamlit 應用程序通過會話狀態（session state）來管理聊天歷史和文件緩存。會話狀態是一個全局的存儲空間，可以在應用程序的不同部分之間共享數據。當用戶上傳文件或發送消息時，這些數據會被存儲在會話狀態中，以便后續的處理和顯示。同時，應用程序會根據需要對文件進行緩存，避免重復處理相同的文件，提高運行效率。
（四）如果我想擴展這個項目，可以添加哪些新功能？
你可以嘗試添加語音識別功能，讓用戶可以通過語音與聊天機器人互動；或者增加多語言支持，讓聊天機器人能夠處理不同語言的音頻內容；還可以優化用戶界面，增加更多的交互元素，如圖表、音頻標注等，提升用戶體驗。此外，你還可以探索將這個系統與其他應用程序或服務集成，實現更廣泛的應用場景。
（五）如何確保音頻數據的安全性和隱私性？
在處理音頻數據時，確保數據的安全性和隱私性至關重要。你可以采取以下措施：在傳輸和存儲音頻數據時使用加密技術；限制對音頻數據的訪問權限，只有授權用戶才能上傳和查看數據；在應用程序中添加數據刪除功能，允許用戶隨時刪除自己的音頻數據；遵守相關的法律法規和隱私政策，確保用戶數據的合法使用和保護。
六、結語
今天，我們深入探討了如何利用 AssemblyAI、Qdrant、SambaNova Cloud 和 DeepSeek-R1 構建一個基于音頻的 RAG 聊天機器人。這個項目不僅展示了各個技術組件的強大功能，還體現了它們協同合作的巨大潛力。通過這個項目，我們看到了音頻技術在智能交互領域的廣闊前景，也感受到了技術創新為我們的生活和工作帶來的便利。
在未來，隨著技術的不斷發展和創新，音頻聊天機器人將變得更加智能、高效和人性化。它將不僅僅是一個工具，更將成為我們生活中的得力助手，幫助我們更好地處理音頻信息，提高工作效率，豐富我們的生活體驗。讓我們一起期待這個充滿無限可能的未來吧！
本文轉載自公眾號Halo咯咯 作者：基咯咯
