Spaces:
Running
Running
File size: 5,534 Bytes
ff470a3 490c58a 13b0a33 ff470a3 13b0a33 ff470a3 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 |
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() |