import json import re # from crewai import Agent, Task, Process,Crew # from langchain_groq import ChatGroq import tempfile import os import streamlit as st from deepgram import DeepgramClient, PrerecordedOptions, FileSource import os from dotenv import load_dotenv # Load the environment variables from the .env file load_dotenv() # Access the API key DG_KEY = os.getenv("DG_KEY") # # Initialize the Deepgram client # DG_KEY = "88b968f3e3cfc8eaf5a596e15c579ffca9a59aed" deepgram = DeepgramClient(DG_KEY) # #creating llm # llm = ChatGroq( # model_name="llama3-8b-8192", # api_key= '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' # ) # Function to transcribe an audio file def transcribe_audio_file(audio_file_path): # Read the audio file from the local path with open(audio_file_path, "rb") as audio_file: buffer_data = audio_file.read() # Define the transcription options options = { "model": "nova-2", "smart_format": True, "language": "hi", #alternatively 'en' "diarize": True, "profanity_filter": False } payload = { "buffer": buffer_data, } # Call the transcribe_file method with the audio buffer and options response = deepgram.listen.prerecorded.v("1").transcribe_file(payload, options) return response def process_diarized_transcript(res): transcript = res['results']['channels'][0]['alternatives'][0] words = res['results']['channels'][0]['alternatives'][0]['words'] current_speaker = None current_sentence = [] output = [] for word in words: # This checks if the speaker has changed from the previous word. if current_speaker != word['speaker']: if current_sentence: output.append((current_speaker, ' '.join(current_sentence))) current_sentence = [] current_speaker = word['speaker'] # This updates the current speaker. current_sentence.append(word['punctuated_word']) # adds current word to the sentence being built. # This checks if the current word ends a sentence (by punctuation). if word['punctuated_word'].endswith(('.', '?', '!')): output.append((current_speaker, ' '.join(current_sentence))) current_sentence = [] # adds any remaining words as a final sentence. if current_sentence: output.append((current_speaker, ' '.join(current_sentence))) return output def format_speaker(speaker_num): return f"speaker {speaker_num}" def transcribe_and_process_audio(audio_file_path): # Transcribe the audio file res = transcribe_audio_file(audio_file_path) # Process the diarized transcript diarized_result = process_diarized_transcript(res) # Check if the result is available if not diarized_result: return "No transcription available. The audio might still be too low quality or silent." # Initialize an empty string variable to store the transcription transcription = "" # Open a text file to write the result with tempfile.NamedTemporaryFile(delete=False, suffix=".txt") as temp_file: file_path = temp_file.name # Iterate over the diarized result for speaker, sentence in diarized_result: # Format the speaker and sentence line = f"{format_speaker(speaker)}: {sentence}\n" # Append the line to the transcription variable transcription += line # Write the line to the text file temp_file.write(line.encode('utf-8')) return transcription # #creating class for agent and task # class meeting_assistant(): # def meeting_assistant(self): # return Agent( # role='expert meeting assistant', # goal='Your goal is to understand the complete meeting conversation and extract important information such as summary, key points discussed, key action item and owener of action item with agreed schedule, meeting sentiment analysis ', # backstory=('''With the critical eye on the each discussed topic in the meeting. # You provide an exact parts of coversation where the key points are discussed. # you identify them with strong experience in understanding the conversation. # You provide a overall sentiment analysis of the meeting. # You recognise an idividual who owns the responsibility of a perticular action and also agrees to a schedule on which he/she takes/completes the action. # ''' # ), # verbose=False, # max_iter=20, # allow_delegation=False, # llm = llm # ) # class assisatnt_tasks(): # def meeting_assistance_task(self, agent): # return Task( # description=( # ''' # "Analyze the provided conversation text {text} of a meeting and extract the following key information:" # "1. Summary: Provide a concise summary of the entire meeting, capturing the main topics discussed and the overall purpose of the meeting." # "2. Key Points Discussed: Identify and list the key points discussed during the meeting. These should be the main ideas or topics that were addressed, without including minor details." # "3. Key Action Items: Extract the key action items that were agreed upon during the meeting. For each action item, include the following details:" # " a. Description of the action item." # " b. Owner of the action item (the person responsible for completing the task)." # " c. Agreed schedule (the deadline or time frame within which the action item should be completed)." # "4. Meeting Sentiment Analysis: Perform sentiment analysis on the meeting conversation. Determine the overall sentiment (positive, negative, neutral) and provide specific examples or quotes that illustrate the sentiment." # "Provide the output strictly in the following structured format:" # "{meeting_structure}" # "Ensure the information is clear, concise, and accurately reflects the content of the meeting conversation in a given {meeting_structure}." # ''' # ), # expected_output=( # 'A structured output as a {meeting_structure} highlighting the summary, key points discussed, key action items with owners and agreed schedules, and meeting sentiment analysis.' # ), # allow_delegation=False, # agent=agent # ) # # structure of outputs # meeting_structure=''' # { # "Summary": "A detailed summary of the meeting.", # "KeyPointsDiscussed": [ # "First key point discussed in the meeting.", # "Second key point discussed in the meeting.", # "Third key point discussed in the meeting." # ], # "ActionItems": [ # { # "Description": "Description of the action to be taken.", # "Owner": "Name of the owner responsible.", # "DueDate": "Date or time mentioned for completion." # }, # { # "Description": "Description of the action to be taken.", # "Owner": "Name of the owner responsible.", # "DueDate": "Date or time mentioned for completion." # } # ], # "SentimentAnalysis": { # "OverallSentiment": "Overall sentiment of the meeting (e.g., positive, negative, neutral).", # "Comments": "Specific comments related to the sentiment." # } # }''' # #calling classes # agent = meeting_assistant() # task = assisatnt_tasks() # # Call the agent methods to get BaseAgent instances # meeting_agent = agent.meeting_assistant() # meeting_task = task.meeting_assistance_task(meeting_agent) # def text_analysis(text): # crew = Crew(agents=[meeting_agent], # tasks=[meeting_task], # process= Process.sequential, # verbose=False # ) # output = crew.kickoff(inputs={'text':text, 'meeting_structure':meeting_structure}) # # Extract the raw output as a string # output_string = output.raw # return output_string # #formatting the jsonstring # def format_json_output(output_str): # # Apply regex to extract JSON part # json_str = re.search(r'{.*}', output_str, re.DOTALL).group(0) # # Convert the cleaned string to a JSON object # try: # json_obj = json.loads(json_str) # except json.JSONDecodeError: # return "Invalid JSON format" # # Define the format # formatted_text = [] # # Add summary # formatted_text.append(f"Summary:\n{json_obj.get('Summary', 'No summary available')}\n") # # Add Key Points Discussed # formatted_text.append("Key Points Discussed:") # key_points = json_obj.get('KeyPointsDiscussed', []) # if key_points: # for point in key_points: # formatted_text.append(f" - {point}") # else: # formatted_text.append(" No key points discussed") # formatted_text.append("") # # Add Action Items # formatted_text.append("Action Items:") # action_items = json_obj.get('ActionItems', []) # if action_items: # for item in action_items: # description = item.get('Description', 'No description') # owner = item.get('Owner', 'No owner') # due_date = item.get('DueDate', 'No due date') # formatted_text.append(f" - {description}\n Owner: {owner}\n Due Date: {due_date}") # else: # formatted_text.append(" No action items") # formatted_text.append("") # # Add Sentiment Analysis # sentiment_analysis = json_obj.get('SentimentAnalysis', {}) # overall_sentiment = sentiment_analysis.get('OverallSentiment', 'No sentiment analysis') # comments = sentiment_analysis.get('Comments', 'No comments') # formatted_text.append(f"Sentiment Analysis:\n Overall Sentiment: {overall_sentiment}\n Comments: {comments}") # # Join and return the formatted text # return "\n".join(formatted_text) # Streamlit interface st.title("Audio Transcription and Diarization") uploaded_file = st.file_uploader("Choose an audio file", type=["mp3", "wav", "m4a"]) if uploaded_file is not None: with tempfile.NamedTemporaryFile(delete=False) as temp_audio_file: temp_audio_file.write(uploaded_file.read()) temp_audio_file_path = temp_audio_file.name st.write("Transcribing audio...") transcription = transcribe_and_process_audio(temp_audio_file_path) st.write("Transcription:") st.text(transcription) # st.write('text_analysis') # output_string = text_analysis(transcription) # formatted_string = format_json_output(output_string) # st.write(formatted_string)