app.py

import gradio as gr
from gradio_webrtc import WebRTC, ReplyOnPause, AdditionalOutputs
import anthropic
from pyht import Client as PyHtClient, TTSOptions
import dataclasses
import os
import numpy as np
from huggingface_hub import InferenceClient
import io
from pydub import AudioSegment
from dotenv import load_dotenv

load_dotenv()

account_sid = os.environ.get("TWILIO_ACCOUNT_SID")
auth_token = os.environ.get("TWILIO_AUTH_TOKEN")

if account_sid and auth_token:
    from twilio.rest import Client
    client = Client(account_sid, auth_token)

    token = client.tokens.create()

    rtc_configuration = {
        "iceServers": token.ice_servers,
        "iceTransportPolicy": "relay",
    }
else:
    rtc_configuration = None


@dataclasses.dataclass
class Clients:
    claude: anthropic.Anthropic
    play_ht: PyHtClient
    hf: InferenceClient


tts_options = TTSOptions(voice="s3://voice-cloning-zero-shot/775ae416-49bb-4fb6-bd45-740f205d20a1/jennifersaad/manifest.json",
                         sample_rate=24000)


def aggregate_chunks(chunks_iterator):
    leftover = b''  # Store incomplete bytes between chunks
    
    for chunk in chunks_iterator:
        # Combine with any leftover bytes from previous chunk
        current_bytes = leftover + chunk
        
        # Calculate complete samples
        n_complete_samples = len(current_bytes) // 2  # int16 = 2 bytes
        bytes_to_process = n_complete_samples * 2
        
        # Split into complete samples and leftover
        to_process = current_bytes[:bytes_to_process]
        leftover = current_bytes[bytes_to_process:]
        
        if to_process:  # Only yield if we have complete samples
            audio_array = np.frombuffer(to_process, dtype=np.int16).reshape(1, -1)
            yield audio_array


def audio_to_bytes(audio: tuple[int, np.ndarray]) -> bytes:
    audio_buffer = io.BytesIO()
    segment = AudioSegment(
        audio[1].tobytes(),
        frame_rate=audio[0],
        sample_width=audio[1].dtype.itemsize,
        channels=1,
    )
    segment.export(audio_buffer, format="mp3")
    return audio_buffer.getvalue()


def set_api_key(claude_key, play_ht_username, play_ht_key):
    try:
        claude_client = anthropic.Anthropic(api_key=claude_key)
        play_ht_client = PyHtClient(user_id=play_ht_username, api_key=play_ht_key)
    except:
        raise gr.Error("Invalid API keys. Please try again.")
    gr.Info("Successfully set API keys.", duration=3)
    return Clients(claude=claude_client, play_ht=play_ht_client,
                   hf=InferenceClient()), gr.skip()



def response(audio: tuple[int, np.ndarray], conversation_llm_format: list[dict],
             chatbot: list[dict], client_state: Clients):
    if not client_state:
        raise gr.Error("Please set your API keys first.")
    prompt = client_state.hf.automatic_speech_recognition(audio_to_bytes(audio)).text
    conversation_llm_format.append({"role": "user", "content": prompt})
    response = client_state.claude.messages.create(
        model="claude-3-5-haiku-20241022",
        max_tokens=512,
        messages=conversation_llm_format,
    )
    response_text = " ".join(block.text for block in response.content if getattr(block, "type", None) == "text")
    conversation_llm_format.append({"role": "assistant", "content": response_text})
    chatbot.append({"role": "user", "content": prompt})
    chatbot.append({"role": "assistant", "content": response_text})
    yield AdditionalOutputs(conversation_llm_format, chatbot)
    iterator = client_state.play_ht.tts(response_text, options=tts_options, voice_engine="Play3.0")
    for chunk in aggregate_chunks(iterator):
        audio_array = np.frombuffer(chunk, dtype=np.int16).reshape(1, -1)
        yield (24000, audio_array, "mono")


with gr.Blocks() as demo:
    with gr.Group():
        with gr.Row():
            chatbot = gr.Chatbot(label="Conversation", type="messages")
        with gr.Row(equal_height=True):
            with gr.Column(scale=1):
                with gr.Row():
                    claude_key = gr.Textbox(type="password", value=os.getenv("ANTHROPIC_API_KEY"),
                                            label="Enter your Anthropic API Key")
                    play_ht_username = gr.Textbox(type="password",
                                                value=os.getenv("PLAY_HT_USER_ID"),
                                                label="Enter your PlayHt Username")
                    play_ht_key = gr.Textbox(type="password",
                                            value=os.getenv("PLAY_HT_API_KEY"),
                                            label="Enter your PlayHt API Key")
                with gr.Row():
                    set_key_button = gr.Button("Set Keys", variant="primary")
            with gr.Column(scale=5):
                audio = WebRTC(modality="audio", mode="send-receive",
                                label="Audio Stream",
                                rtc_configuration=rtc_configuration)

    client_state = gr.State(None)
    conversation_llm_format = gr.State([])

    set_key_button.click(set_api_key, inputs=[claude_key, play_ht_username, play_ht_key],
                         outputs=[client_state, set_key_button])
    audio.stream(
        ReplyOnPause(response),
        inputs=[audio, conversation_llm_format, chatbot, client_state],
        outputs=[audio]
    )
    audio.on_additional_outputs(lambda l, g: (l, g), outputs=[conversation_llm_format, chatbot])


if __name__ == "__main__":
    demo.launch()