app.py

import gradio as gr
import pixeltable as pxt
from pixeltable.iterators import FrameIterator, StringSplitter
from pixeltable.functions.video import extract_audio
from pixeltable.functions.audio import get_metadata
from pixeltable.functions import openai
import os
import getpass
import numpy as np
from pixeltable.functions.huggingface import sentence_transformer

# Store OpenAI API Key
if 'OPENAI_API_KEY' not in os.environ:
    os.environ['OPENAI_API_KEY'] = getpass.getpass('Enter your OpenAI API key:')

MAX_VIDEO_SIZE_MB = 35

def process_video(video_file, progress=gr.Progress()):

    progress(0, desc="Initializing...")
    
    try:
        # Create a Table, a View, and Computed Columns
        pxt.drop_dir('gong_demo', force=True)
        pxt.create_dir('gong_demo')
        
        calls_table = pxt.create_table(
            'gong_demo.calls', {
            "video": pxt.VideoType(nullable=True),
            }
        )
        
        # Create computed columns to store transformations and persist outputs
        calls_table['audio'] = extract_audio(calls_table.video, format='mp3')
        calls_table['metadata'] = get_metadata(calls_table.audio)
        calls_table['transcription'] = openai.transcriptions(audio=calls_table.audio, model='whisper-1')
        calls_table['transcription_text'] = calls_table.transcription.text.astype(pxt.StringType())

        sentences_view = pxt.create_view(
            'gong_demo.sentences',
            calls_table,
            iterator=StringSplitter.create(
                text=calls_table.transcription_text,
                separators='sentence'
            )
        )

        @pxt.expr_udf
        def e5_embed(text: str) -> np.ndarray:
            return sentence_transformer(text, model_id='intfloat/e5-large-v2')

        sentences_view.add_embedding_index('text', string_embed=e5_embed)

        progress(0.2, desc="Creating UDFs...")

        # Custom User-Defined Function (UDF) for Generating Insights
        @pxt.udf
        def generate_insights(transcription: str) -> list[dict]:
            system_msg = 'You are an AI assistant that analyzes call transcriptions. Analyze the following call transcription and provide insights on: 1. Main topics discussed 2. Action items 3. Sentiment analysis 4. Key questions asked'
            user_msg = f'Transcription: "{transcription}"'
            return [
                {'role': 'system', 'content': system_msg},
                {'role': 'user', 'content': user_msg}
            ]
        
        # Apply the UDF to create a new column
        calls_table['insights_prompt'] = generate_insights(calls_table.transcription_text)
        
        progress(0.4, desc="Generating insights...")
        
        # Generate insights using OpenAI's chat completion API
        calls_table['insights_response'] = openai.chat_completions(messages=calls_table.insights_prompt, model='gpt-3.5-turbo', max_tokens=500)
        
        # Extract the content of the response
        calls_table['insights'] = calls_table.insights_response.choices[0].message.content
        
        if not video_file:
            return "Please upload a video file.", ""

        # Check video file size
        video_size = os.path.getsize(video_file) / (1024 * 1024)  # Convert to MB
        if video_size > MAX_VIDEO_SIZE_MB:
            return f"The video file is larger than {MAX_VIDEO_SIZE_MB} MB. Please upload a smaller file.", ""

        progress(0.6, desc="Processing video...")
        
        # Insert a video into the table
        calls_table.insert([{"video": video_file}])

        progress(0.8, desc="Retrieving results...")
        
        # Retrieve transcription and insights
        result = calls_table.select(calls_table.transcription_text, calls_table.insights, calls_table.audio).tail(1)
        transcription = result['transcription_text'][0]
        insights = result['insights'][0]
        
        audio = calls_table.select(calls_table.audio).tail(1)['audio'][0]

        progress(1.0, desc="Processing complete")

        return transcription, insights, audio, "Processing complete"

    except Exception as e:
        return f"An error occurred during video processing: {str(e)}", ""

# Perform similarity search
def similarity_search(query, num_results, progress=gr.Progress()):
    sentences_view = pxt.get_table('gong_demo.sentences')
    
    progress(0.5, desc="Performing search...")

    sim = sentences_view.text.similarity(query)
    results = sentences_view.order_by(sim, asc=False).limit(num_results).select(sentences_view.text).collect().to_pandas()
    
    progress(1.0, desc="Search complete")
    return results

def chatbot_response(message, chat_history):
    @pxt.udf
    def create_chatbot_prompt(context: str, question: str) -> list[dict]:
        system_message = "You are an AI assistant that answers questions about a call based on the provided context. If the answer cannot be found in the context, say that you don't know."
        user_message = f"Context:\n{context}\n\nQuestion: {question}"
        return [
            {"role": "system", "content": system_message},
            {"role": "user", "content": user_message}
        ]

    try:
        sentences_view = pxt.get_table('gong_demo.sentences')

        # Perform similarity search to get relevant context
        sim = sentences_view.text.similarity(message)
        context = sentences_view.order_by(sim, asc=False).limit(5).select(sentences_view.text, sim=sim).collect()
        
        # Prepare the context for the prompt
        context_text = "\n".join([row['text'] for row in context])

        # Create a temporary table for the chatbot interaction
        temp_table = pxt.create_table('gong_demo.temp_chatbot', {'question': pxt.StringType()})
        temp_table.insert([{'question': message}])

        # Create computed columns for the prompt and response
        temp_table['chatbot_prompt'] = create_chatbot_prompt(context_text, temp_table.question)
        temp_table['chatbot_response'] = openai.chat_completions(
            messages=temp_table.chatbot_prompt,
            model='gpt-4o-mini-2024-07-18',
            max_tokens=300
        ) 
        temp_table['answer'] = temp_table.chatbot_response.choices[0].message.content
        
        answer = temp_table.select(temp_table.answer).collect()['answer'][0]
        
        # Clean up the temporary table
        pxt.drop_table('gong_demo.temp_chatbot', force=True)
        
        chat_history.append((message, answer))
        return "", chat_history  # Return both expected outputs
    except Exception as e:
        error_message = f"An error occurred: {str(e)}"
        chat_history.append((message, error_message))
        return "", chat_history  # Return both expec

# Gradio interface
with gr.Blocks(theme=gr.themes.Base()) as demo:
    gr.Markdown(
        """
        <div style="text-align: left; margin-bottom: 20px;">
            <img src="https://raw.githubusercontent.com/pixeltable/pixeltable/main/docs/source/data/pixeltable-logo-large.png" alt="Pixeltable" style="max-width: 150px;" />
            <h1 style="margin-top: 10px;">Call Analysis AI Tool</h1>
        </div>
        """
    )
    gr.HTML(
        """
        <p>
            <a href="https://github.com/pixeltable/pixeltable" target="_blank" style="color: #F25022; text-decoration: none; font-weight: bold;">Pixeltable</a> is a declarative interface for working with text, images, embeddings, and even video, enabling you to store, transform, index, and iterate on data.
        </p>
        """
    )
    
    with gr.Row():
        with gr.Column():
            with gr.Accordion("🎯 What does it do?", open=False):
                gr.Markdown("""
                - 🎙️ Transcribes call audio to text
                - 💡 Generates insights and key points
                - 🔍 Enables content-based similarity search
                - 🤖 Provides an AI chatbot for in-depth analysis
                - 📊 Offers summaries of call data
                """)
        with gr.Column():
            with gr.Accordion("🛠️ How does it work?", open=False):
                gr.Markdown("""
                1. 📤 Upload your call recording (video)
                2. ⚙️ AI processes and analyzes the content
                3. 📝 Review the transcript and generated insights
                4. 🔎 Use similarity search to explore specific topics
                5. 💬 Interact with the AI chatbot for deeper understanding
                """)
    
    with gr.Row():
        with gr.Column(scale=1):
            video_file = gr.Video(
                label=f"Upload Call Recording (max {MAX_VIDEO_SIZE_MB} MB)",
                include_audio=True,
                autoplay=False
            )
            process_btn = gr.Button("Analyze Call", variant="primary")
            status_output = gr.Textbox(label="Status", interactive=False)

        with gr.Column(scale=2):
            with gr.Tabs() as tabs:
                with gr.TabItem("📝 Transcript"):
                    output_transcription = gr.Textbox(label="Call Transcription", lines=10)
                
                with gr.TabItem("💡 Insights"):
                    output_insights = gr.Textbox(label="Key Takeaways", lines=20)

                with gr.TabItem("🎵 Audio"):
                    audio = gr.Audio(label="Extracted audio", type="filepath", show_download_button=True)
            
                with gr.TabItem("🔍 Search"):
                    with gr.Row():
                        similarity_query = gr.Textbox(label="Search Query", placeholder="Enter a topic or phrase to search for")
                        num_results = gr.Slider(minimum=1, maximum=20, value=5, step=1, label="Number of Results")
                    similarity_search_btn = gr.Button("Search", variant="secondary")
                    similarity_results = gr.DataFrame(
                        headers=["Relevant Text"],
                        label="Search Results",
                        wrap=True
                    )
            
                with gr.TabItem("🤖 AI Assistant"):
                    chatbot = gr.Chatbot(height=400, label="Chat with AI about the call")
                    with gr.Row():
                        msg = gr.Textbox(label="Ask a question about the call", placeholder="e.g., What were the main points discussed?", scale=4)
                        send_btn = gr.Button("Send", variant="secondary", scale=1)
                    clear = gr.Button("Clear Chat")

                    gr.Examples(
                        examples=[
                            "What were the main topics discussed in this call?",
                            "Can you summarize the action items mentioned?",
                            "What was the overall sentiment of the conversation?",
                            "Were there any objections raised by the client?",
                            "What features or products were highlighted during the call?",
                        ],
                        inputs=msg,
                    )

    process_btn.click(
        process_video,
        inputs=[video_file],
        outputs=[output_transcription, output_insights, audio, status_output],
        show_progress="full"
    )

    similarity_search_btn.click(
        similarity_search,
        inputs=[similarity_query, num_results],
        outputs=[similarity_results]
    )

    msg.submit(chatbot_response, [msg, chatbot], [msg, chatbot])
    send_btn.click(chatbot_response, [msg, chatbot], [msg, chatbot])
    clear.click(lambda: None, None, chatbot, queue=False)

if __name__ == "__main__":
    demo.launch(show_api=False, allowed_paths=[os.path.expanduser("~/.pixeltable/media")])