from qdrant_client import models\nfrom qdrant_client import QdrantClient\nfrom llama_index.embeddings.huggingface import HuggingFaceEmbedding\nfrom llama_index.llms.sambanovasystems import SambaNovaCloud\nfrom llama_index.llms.ollama import Ollama\nimport assemblyai as aai\nfrom typing import List, Dict\n\nfrom llama_index.core.base.llms.types import (\n ChatMessage,\n MessageRole,\n)<\/code><\/pre>\n2.2 Batch Processing and Text Embedding with Hugging Face<\/h4>\n
Here batch_iterate function splits a list of text into smaller chunks, making it easier to process large datasets. The EmbedData class then loads a Hugging Face embedding model, generates embeddings for each batch of text, and collects these embeddings for later use.<\/p>\n
def batch_iterate(lst, batch_size):\n \"\"\"Yield successive n-sized chunks from lst.\"\"\"\n for i in range(0, len(lst), batch_size):\n yield lst[i : i + batch_size]\n\nclass EmbedData:\n\n def __init__(self, embed_model_name=\"BAAI\/bge-large-en-v1.5\", batch_size = 32):\n self.embed_model_name = embed_model_name\n self.embed_model = self._load_embed_model()\n self.batch_size = batch_size\n self.embeddings = []\n \n def _load_embed_model(self):\n embed_model = HuggingFaceEmbedding(model_name=self.embed_model_name, trust_remote_code=True, cache_folder=\".\/hf_cache\")\n return embed_model\n\n def generate_embedding(self, context):\n return self.embed_model.get_text_embedding_batch(context)\n \n def embed(self, contexts):\n \n self.contexts = contexts\n \n for batch_context in batch_iterate(contexts, self.batch_size):\n batch_embeddings = self.generate_embedding(batch_context)\n self.embeddings.extend(batch_embeddings)<\/code><\/pre>\n2.3 Qdrant Vector Database Setup and Ingestion<\/h4>\n\n- QdrantVDB_QB class initializes a Qdrant vector database by setting up key parameters like collection name, vector dimension, and batch size, and it connects to Qdrant while checking for an existing collection (creating one if needed).<\/li>\n
- It efficiently uploads data by batching text contexts with their corresponding embeddings and then updating the collection\u2019s configuration accordingly.<\/li>\n<\/ul>\n
class QdrantVDB_QB:\n\n def __init__(self, collection_name, vector_dim = 768, batch_size=512):\n self.collection_name = collection_name\n self.batch_size = batch_size\n self.vector_dim = vector_dim\n \n def define_client(self):\n \n self.client = QdrantClient(url=\"http:\/\/localhost:6333\", prefer_grpc=True)\n \n def create_collection(self):\n \n if not self.client.collection_exists(collection_name=self.collection_name):\n\n self.client.create_collection(collection_name=f\"{self.collection_name}\",\n \n vectors_config=models.VectorParams(size=self.vector_dim,\n distance=models.Distance.DOT,\n on_disk=True),\n \n optimizers_config=models.OptimizersConfigDiff(default_segment_number=5,\n indexing_threshold=0),\n \n quantization_config=models.BinaryQuantization(\n binary=models.BinaryQuantizationConfig(always_ram=True)),\n )\n \n def ingest_data(self, embeddata):\n \n for batch_context, batch_embeddings in zip(batch_iterate(embeddata.contexts, self.batch_size), \n batch_iterate(embeddata.embeddings, self.batch_size)):\n \n self.client.upload_collection(collection_name=self.collection_name,\n vectors=batch_embeddings,\n payload=[{\"context\": context} for context in batch_context])\n\n self.client.update_collection(collection_name=self.collection_name,\n optimizer_config=models.OptimizersConfigDiff(indexing_threshold=20000)\n )<\/code><\/pre>\n2.4 Query Embedding Retriever<\/h4>\n\n- The Retriever class is designed to bridge the gap between user queries and a vector database by initializing with a vector database client and an embedding model.<\/li>\n
- Its search method transforms a query into an embedding using the model, then performs a vector search on the database with fine-tuned quantization parameters to quickly retrieve relevant results.<\/li>\n<\/ul>\n
class Retriever:\n\n def __init__(self, vector_db, embeddata):\n \n self.vector_db = vector_db\n self.embeddata = embeddata\n\n def search(self, query):\n query_embedding = self.embeddata.embed_model.get_query_embedding(query)\n \n \n result = self.vector_db.client.search(\n collection_name=self.vector_db.collection_name,\n \n query_vector=query_embedding,\n \n search_params=models.SearchParams(\n quantization=models.QuantizationSearchParams(\n ignore=False,\n rescore=True,\n oversampling=2.0,\n )\n ),\n \n timeout=1000,\n )\n\n return result<\/code><\/pre>\n2.5 RAG Smart Query Assistant<\/h4>\n
The RAG class integrates a retriever and an LLM to generate context-aware responses. It retrieves relevant information from a vector database, formats it into a structured prompt, and sends it to the LLM for a response. I am using SambaNovaCloud to access the LLM model through their API for efficient text generation.<\/p>\n
class RAG:\n\n def __init__(self,\n retriever,\n llm_name = \"Meta-Llama-3.1-405B-Instruct\"\n ):\n \n system_msg = ChatMessage(\n role=MessageRole.SYSTEM,\n content=\"You are a helpful assistant that answers questions about the user's document.\",\n )\n self.messages = [system_msg, ]\n self.llm_name = llm_name\n self.llm = self._setup_llm()\n self.retriever = retriever\n self.qa_prompt_tmpl_str = (\"Context information is below.\\n\"\n \"---------------------\\n\"\n \"{context}\\n\"\n \"---------------------\\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\"\n \"Query: {query}\\n\"\n \"Answer: \"\n )\n\n def _setup_llm(self):\n\n return SambaNovaCloud(\n model=self.llm_name,\n temperature=0.7,\n context_window=100000,\n )\n\n # return Ollama(model=self.llm_name,\n # temperature=0.7,\n # context_window=100000,\n # )\n\n def generate_context(self, query):\n\n result = self.retriever.search(query)\n context = [dict(data) for data in result]\n combined_prompt = []\n\n for entry in context[:2]:\n context = entry[\"payload\"][\"context\"]\n\n combined_prompt.append(context)\n\n return \"\\n\\n---\\n\\n\".join(combined_prompt)\n\n def query(self, query):\n context = self.generate_context(query=query)\n \n prompt = self.qa_prompt_tmpl_str.format(context=context, query=query)\n\n user_msg = ChatMessage(role=MessageRole.USER, content=prompt)\n\n # self.messages.append(ChatMessage(role=MessageRole.USER, content=prompt))\n \n streaming_response = self.llm.stream_complete(user_msg.content)\n \n return streaming_response<\/code><\/pre>\n2.6 Audio Transcription<\/h4>\n
Here Transcribe class initializes by setting the AssemblyAI API key and creating a transcriber. It then processes an audio file using a configuration that enables speaker labels, ultimately returning a list of dictionaries where each entry maps a speaker to their transcribed text.<\/p>\n
class Transcribe:\n def __init__(self, api_key: str):\n \"\"\"Initialize the Transcribe class with AssemblyAI API key.\"\"\"\n aai.settings.api_key = api_key\n self.transcriber = aai.Transcriber()\n \n def transcribe_audio(self, audio_path: str) -> List[Dict[str, str]]:\n \"\"\"\n Transcribe an audio file and return speaker-labeled transcripts.\n \n Args:\n audio_path: Path to the audio file\n \n Returns:\n List of dictionaries containing speaker and text information\n \"\"\"\n # Configure transcription with speaker labels\n config = aai.TranscriptionConfig(\n speaker_labels=True,\n speakers_expected=2 # Adjust this based on your needs\n )\n \n # Transcribe the audio\n transcript = self.transcriber.transcribe(audio_path, config=config)\n \n # Extract speaker utterances\n speaker_transcripts = []\n for utterance in transcript.utterances:\n speaker_transcripts.append({\n \"speaker\": f\"Speaker {utterance.speaker}\",\n \"text\": utterance.text\n })\n \n return speaker_transcripts<\/code><\/pre>\n3. Streamlit App<\/h3>\n
Streamlit is a Python library that transforms data scripts into interactive web apps, making it perfect for LLM-based solutions.<\/p>\n
\n- The below code builds a user-friendly app that lets users upload an audio file, view its transcript, and chat accordingly.<\/li>\n
- AssemblyAI transcribes the uploaded audio into speaker-labeled text.<\/li>\n
- The transcript is embedded and stored in a Qdrant vector database for efficient retrieval.<\/li>\n
- A retriever paired with a RAG engine generates context-aware chat responses using these embeddings.<\/li>\n
- Session state manages chat history and file caching to ensure a smooth experience.<\/li>\n<\/ul>\n
import os\nimport gc\nimport uuid\nimport tempfile\nimport base64\nfrom dotenv import load_dotenv\nfrom rag_code import Transcribe, EmbedData, QdrantVDB_QB, Retriever, RAG\nimport streamlit as st\n\nif \"id\" not in st.session_state:\n st.session_state.id = uuid.uuid4()\n st.session_state.file_cache = {}\n\nsession_id = st.session_state.id\ncollection_name = \"chat with audios\"\nbatch_size = 32\n\nload_dotenv()\n\ndef reset_chat():\n st.session_state.messages = []\n st.session_state.context = None\n gc.collect()\n\nwith st.sidebar:\n st.header(\"Add your audio file!\")\n \n uploaded_file = st.file_uploader(\"Choose your audio file\", type=[\"mp3\", \"wav\", \"m4a\"])\n\n if uploaded_file:\n try:\n with tempfile.TemporaryDirectory() as temp_dir:\n file_path = os.path.join(temp_dir, uploaded_file.name)\n \n with open(file_path, \"wb\") as f:\n f.write(uploaded_file.getvalue())\n \n file_key = f\"{session_id}-{uploaded_file.name}\"\n st.write(\"Transcribing with AssemblyAI and storing in vector database...\")\n\n if file_key not in st.session_state.get('file_cache', {}):\n # Initialize transcriber\n transcriber = Transcribe(api_key=os.getenv(\"ASSEMBLYAI_API_KEY\"))\n \n # Get speaker-labeled transcripts\n transcripts = transcriber.transcribe_audio(file_path)\n st.session_state.transcripts = transcripts\n \n # Each speaker segment becomes a separate document for embedding\n documents = [f\"Speaker {t['speaker']}: {t['text']}\" for t in transcripts]\n\n # embed data \n embeddata = EmbedData(embed_model_name=\"BAAI\/bge-large-en-v1.5\", batch_size=batch_size)\n embeddata.embed(documents)\n\n # set up vector database\n qdrant_vdb = QdrantVDB_QB(collection_name=collection_name,\n batch_size=batch_size,\n vector_dim=1024)\n qdrant_vdb.define_client()\n qdrant_vdb.create_collection()\n qdrant_vdb.ingest_data(embeddata=embeddata)\n\n # set up retriever\n retriever = Retriever(vector_db=qdrant_vdb, embeddata=embeddata)\n\n # set up rag\n query_engine = RAG(retriever=retriever, llm_name=\"DeepSeek-R1-Distill-Llama-70B\")\n st.session_state.file_cache[file_key] = query_engine\n else:\n query_engine = st.session_state.file_cache[file_key]\n\n # Inform the user that the file is processed\n st.success(\"Ready to Chat!\")\n \n # Display audio player\n st.audio(uploaded_file)\n \n # Display speaker-labeled transcript\n st.subheader(\"Transcript\")\n with st.expander(\"Show full transcript\", expanded=True):\n for t in st.session_state.transcripts:\n st.text(f\"**{t['speaker']}**: {t['text']}\")\n \n except Exception as e:\n st.error(f\"An error occurred: {e}\")\n st.stop() \n\ncol1, col2 = st.columns([6, 1])\n\nwith col1:\n st.markdown(\"\"\"\n # RAG over Audio powered by ![](\"data:image\/png;base64,{}\")
and \n\"\"\".format(base64.b64encode(open(\"assets\/AssemblyAI.png\", \"rb\").read()).decode(),\n base64.b64encode(open(\"assets\/deep-seek.png\", \"rb\").read()).decode()), unsafe_allow_html=True)\n\nwith col2:\n st.button(\"Clear \u21ba\", on_click=reset_chat)\n\n# Initialize chat history\nif \"messages\" not in st.session_state:\n reset_chat()\n\n# Display chat messages from history on app rerun\nfor message in st.session_state.messages:\n with st.chat_message(message[\"role\"]):\n st.markdown(message[\"content\"])\n\n# Accept user input\nif prompt := st.chat_input(\"Ask about the audio conversation...\"):\n # Add user message to chat history\n st.session_state.messages.append({\"role\": \"user\", \"content\": prompt})\n # Display user message in chat message container\n with st.chat_message(\"user\"):\n st.markdown(prompt)\n\n # Display assistant response in chat message container\n with st.chat_message(\"assistant\"):\n message_placeholder = st.empty()\n full_response = \"\"\n \n # Get streaming response\n streaming_response = query_engine.query(prompt)\n \n for chunk in streaming_response:\n try:\n new_text = chunk.raw[\"choices\"][0][\"delta\"][\"content\"]\n full_response += new_text\n message_placeholder.markdown(full_response + \"\u258c\")\n except:\n pass\n\n message_placeholder.markdown(full_response)\n\n # Add assistant response to chat history\n st.session_state.messages.append({\"role\": \"assistant\", \"content\": full_response})<\/code><\/pre>\nRun the app.py file in the terminal, with the below code, where you can upload an audio file and interact with the chatbot.<\/p>\n
streamlit run app.py<\/code><\/pre>\nYou can see the demo using the app here. And you can download the sample audio file from here.<\/p>\n
\n
![\"Audio](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2025\/03\/Screenshot_2025-03-14_124147-thumbnail_webp-600x300-1.webp\")
<\/figure>\n![\"SambaNova](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2025\/03\/Screenshot_2025-03-14_192534-thumbnail_webp-600x300-1.webp\")
<\/figure>\n![\"Audio](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2025\/03\/Screenshot_2025-03-14_125756-thumbnail_webp-600x300-1.webp\")
<\/figure>\n![\"Audio](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2025\/03\/Screenshot_2025-03-14_122800-thumbnail_webp-600x300-1.webp\")
<\/figure>\n