import requests
import time
import os
import json
import threading
from deepgram import DeepgramClient
from deepgram.core.events import EventType
from deepgram.agent.v1.types import (
AgentV1Settings,
AgentV1SettingsAgent,
AgentV1SettingsAudio,
AgentV1SettingsAudioInput,
AgentV1SettingsAudioOutput,
AgentV1SettingsAgentListen,
AgentV1SettingsAgentListenProvider_V1,
)
from deepgram.types.think_settings_v1 import ThinkSettingsV1
from deepgram.types.think_settings_v1provider import ThinkSettingsV1Provider_OpenAi
from deepgram.types.speak_settings_v1 import SpeakSettingsV1
from deepgram.types.speak_settings_v1provider import SpeakSettingsV1Provider_Deepgram
def main():
try:
api_key = os.getenv("DEEPGRAM_API_KEY")
if not api_key:
raise ValueError("DEEPGRAM_API_KEY environment variable is not set")
client = DeepgramClient(api_key=api_key)
with client.agent.v1.connect() as connection:
print("Created WebSocket connection...")
settings = AgentV1Settings(
audio=AgentV1SettingsAudio(
input=AgentV1SettingsAudioInput(
encoding="linear16",
sample_rate=24000,
),
output=AgentV1SettingsAudioOutput(
encoding="linear16",
sample_rate=24000,
container="wav",
),
),
agent=AgentV1SettingsAgent(
language="en",
listen=AgentV1SettingsAgentListen(
provider=AgentV1SettingsAgentListenProvider_V1(
type="deepgram",
model="nova-3",
)
),
think=ThinkSettingsV1(
provider=ThinkSettingsV1Provider_OpenAi(
type="open_ai",
model="gpt-4o-mini",
),
prompt="You are a friendly AI assistant.",
),
speak=SpeakSettingsV1(
provider=SpeakSettingsV1Provider_Deepgram(
type="deepgram",
model="aura-2-thalia-en",
)
),
greeting="Hello! How can I help you today?",
),
)
audio_buffer = bytearray()
file_counter = 0
processing_complete = False
def on_message(message):
nonlocal audio_buffer, file_counter, processing_complete
if isinstance(message, bytes):
audio_buffer.extend(message)
return
msg_type = getattr(message, "type", "Unknown")
if msg_type == "ConversationText":
print(f"Conversation: {message}")
with open("chatlog.txt", 'a') as chatlog:
chatlog.write(f"{json.dumps(message.__dict__)}\n")
elif msg_type == "AgentAudioDone":
print("Agent audio done")
if len(audio_buffer) > 0:
with open(f"output-{file_counter}.wav", 'wb') as f:
f.write(create_wav_header())
f.write(audio_buffer)
print(f"Created output-{file_counter}.wav")
audio_buffer = bytearray()
file_counter += 1
processing_complete = True
connection.on(EventType.MESSAGE, on_message)
connection.send_settings(settings)
listener_thread = threading.Thread(target=connection.start_listening, daemon=True)
listener_thread.start()
time.sleep(1)
print("Streaming audio...")
response = requests.get("https://dpgr.am/spacewalk.wav", stream=True)
header = response.raw.read(44)
for chunk in response.iter_content(chunk_size=8192):
if chunk:
connection.send_media(chunk)
time.sleep(0.1)
print("Waiting for agent response...")
start_time = time.time()
while not processing_complete and (time.time() - start_time) < 30:
time.sleep(1)
print("Finished")
except Exception as e:
print(f"Error: {str(e)}")
def create_wav_header(sample_rate=24000, bits_per_sample=16, channels=1):
byte_rate = sample_rate * channels * (bits_per_sample // 8)
block_align = channels * (bits_per_sample // 8)
header = bytearray(44)
header[0:4] = b'RIFF'
header[4:8] = b'\x00\x00\x00\x00'
header[8:12] = b'WAVE'
header[12:16] = b'fmt '
header[16:20] = b'\x10\x00\x00\x00'
header[20:22] = b'\x01\x00'
header[22:24] = channels.to_bytes(2, 'little')
header[24:28] = sample_rate.to_bytes(4, 'little')
header[28:32] = byte_rate.to_bytes(4, 'little')
header[32:34] = block_align.to_bytes(2, 'little')
header[34:36] = bits_per_sample.to_bytes(2, 'little')
header[36:40] = b'data'
header[40:44] = b'\x00\x00\x00\x00'
return header
if __name__ == "__main__":
main()
