#!/usr/bin/env python3
"""
LiveTalker Real-time Voice Chat with AI Agent
==============================================
Complete voice conversation system with STT, LLM, and TTS
"""
import asyncio
import json
import logging
import time
import base64
import ssl
import numpy as np
from typing import Dict, List, Any, Optional
from datetime import datetime, timedelta
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
import ipaddress
# FastAPI and WebSocket imports
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
import uvicorn
# Speech processing imports
import speech_recognition as sr
from gtts import gTTS
import io
import threading
import queue
import wave
import tempfile
import os
# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Global state
app = FastAPI()
active_connections: Dict[str, Dict] = {}
recognizer = sr.Recognizer()
class VoiceChatAgent:
"""Simple conversational AI agent"""
def __init__(self):
self.conversation_history = []
self.personality = {
"name": "LiveTalker Assistant",
"style": "friendly, helpful, and conversational",
"traits": ["responsive", "empathetic", "knowledgeable"]
}
def generate_response(self, user_input: str) -> str:
"""Generate AI response (placeholder for actual LLM integration)"""
# Store user input
self.conversation_history.append({"role": "user", "content": user_input})
# Simple rule-based responses for testing
user_lower = user_input.lower()
if "hello" in user_lower or "hi" in user_lower:
response = "Hello! I'm your LiveTalker voice assistant. How can I help you today?"
elif "how are you" in user_lower:
response = "I'm doing great, thank you for asking! I'm here to have a conversation with you."
elif "what can you do" in user_lower:
response = "I can have real-time voice conversations with you! I listen to your speech, understand what you're saying, and respond back with voice. We can chat about anything you'd like."
elif "weather" in user_lower:
response = "I don't have access to current weather data, but I'd be happy to chat about the weather or anything else!"
elif "time" in user_lower:
current_time = datetime.now().strftime("%I:%M %p")
response = f"The current time is {current_time}. Is there anything else I can help you with?"
elif "name" in user_lower:
response = "I'm LiveTalker, your real-time voice assistant! What's your name?"
elif "goodbye" in user_lower or "bye" in user_lower:
response = "Goodbye! It was great talking with you. Feel free to start another conversation anytime!"
elif len(user_input.split()) < 3:
response = "I heard you, but could you tell me a bit more? I'd love to have a longer conversation!"
else:
response = f"That's interesting! You said: '{user_input}'. I'm here to chat and help however I can. What would you like to talk about?"
# Store assistant response
self.conversation_history.append({"role": "assistant", "content": response})
return response
def simple_vad(audio: np.ndarray, threshold: float = 0.02) -> tuple:
"""Simple VAD based on energy"""
if len(audio) == 0:
return False, 0.0
# Energy-based VAD
energy = np.sqrt(np.mean(audio**2))
is_speech = energy > threshold
confidence = min(energy / threshold, 1.0) if is_speech else energy / threshold
return bool(is_speech), float(confidence)
def audio_to_text(audio_data: bytes) -> Optional[str]:
"""Convert audio to text using speech recognition"""
try:
# Create temporary WAV file
with tempfile.NamedTemporaryFile(delete=False, suffix='.wav') as tmp_file:
# Convert raw audio to WAV format
with wave.open(tmp_file.name, 'wb') as wav_file:
wav_file.setnchannels(1) # Mono
wav_file.setsampwidth(2) # 16-bit
wav_file.setframerate(16000) # 16kHz
wav_file.writeframes(audio_data)
# Recognize speech
with sr.AudioFile(tmp_file.name) as source:
audio = recognizer.record(source)
text = recognizer.recognize_google(audio)
os.unlink(tmp_file.name) # Clean up
return text
except sr.UnknownValueError:
return None # Could not understand audio
except sr.RequestError as e:
logger.error(f"Speech recognition error: {e}")
return None
except Exception as e:
logger.error(f"Audio to text error: {e}")
return None
def text_to_speech(text: str) -> Optional[bytes]:
"""Convert text to speech audio"""
try:
tts = gTTS(text=text, lang='en', slow=False)
audio_buffer = io.BytesIO()
tts.write_to_fp(audio_buffer)
audio_buffer.seek(0)
return audio_buffer.read()
except Exception as e:
logger.error(f"Text to speech error: {e}")
return None
def create_ssl_certificate():
"""Create self-signed SSL certificate for HTTPS"""
try:
# Generate private key
private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
)
# Create certificate
subject = issuer = x509.Name([
x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, "CA"),
x509.NameAttribute(NameOID.LOCALITY_NAME, "Local"),
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "LiveTalker"),
x509.NameAttribute(NameOID.COMMON_NAME, "localhost"),
])
cert = x509.CertificateBuilder().subject_name(
subject
).issuer_name(
issuer
).public_key(
private_key.public_key()
).serial_number(
x509.random_serial_number()
).not_valid_before(
datetime.utcnow()
).not_valid_after(
datetime.utcnow() + timedelta(days=365)
).add_extension(
x509.SubjectAlternativeName([
x509.DNSName("localhost"),
x509.DNSName("*.localhost"),
x509.IPAddress(ipaddress.IPv4Address("127.0.0.1")),
x509.IPAddress(ipaddress.IPv4Address("0.0.0.0")),
]),
critical=False,
).sign(private_key, hashes.SHA256())
# Write certificate and key files
cert_path = "livetalker.crt"
key_path = "livetalker.key"
with open(cert_path, "wb") as f:
f.write(cert.public_bytes(serialization.Encoding.PEM))
with open(key_path, "wb") as f:
f.write(private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption()
))
print(f"ā SSL certificate created: {cert_path}")
print(f"ā SSL private key created: {key_path}")
return cert_path, key_path
except Exception as e:
print(f"ā Failed to create SSL certificate: {e}")
return None, None
@app.get("/")
async def root():
"""Real-time voice chat interface"""
html_content = """
LiveTalker Real-time Voice Chat
šļø LiveTalker Voice Chat
Welcome to LiveTalker! Click "Start Conversation" to begin voice chat.
Voice Controls
Ready to chat
Live Stats
Connection:Connecting...
VAD Active:No
Speech Confidence:0%
Messages:0
Session Time:00:00
"""
return HTMLResponse(content=html_content)
@app.websocket("/voice-chat")
async def voice_chat_endpoint(websocket: WebSocket):
"""WebSocket endpoint for real-time voice chat"""
await websocket.accept()
session_id = f"chat_{int(time.time() * 1000)}"
# Initialize voice chat agent
agent = VoiceChatAgent()
session = {
"id": session_id,
"websocket": websocket,
"agent": agent,
"audio_buffer": [],
"is_listening": False,
"last_speech": 0,
"processing": False
}
active_connections[session_id] = session
logger.info(f"Voice chat session started: {session_id}")
try:
await websocket.send_json({
"type": "config",
"session_id": session_id,
"message": "Voice chat ready!",
"agent_name": agent.personality["name"]
})
async for message in websocket.iter_json():
await handle_voice_message(session, message)
except WebSocketDisconnect:
logger.info(f"Voice chat session ended: {session_id}")
except Exception as e:
logger.error(f"Voice chat error: {e}")
try:
await websocket.send_json({
"type": "error",
"error": str(e)
})
except:
pass
finally:
if session_id in active_connections:
del active_connections[session_id]
async def handle_voice_message(session: Dict, message: Dict[str, Any]):
"""Handle voice chat messages"""
msg_type = message.get("type")
if msg_type == "start_conversation":
session["is_listening"] = True
await session["websocket"].send_json({
"type": "conversation_started",
"message": "š¤ Voice chat started! Speak into your microphone."
})
elif msg_type == "audio" and session["is_listening"] and not session["processing"]:
try:
# Decode audio
audio_data = base64.b64decode(message["data"])
audio_np = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0
if len(audio_np) > 0:
# VAD check
is_speech, confidence = simple_vad(audio_np, threshold=0.025)
await session["websocket"].send_json({
"type": "vad_result",
"is_speech": bool(is_speech),
"confidence": float(confidence)
})
# Accumulate speech
if is_speech:
session["audio_buffer"].extend(audio_data)
session["last_speech"] = time.time()
# Process if we have enough audio or silence detected
if len(session["audio_buffer"]) > 32000: # ~2 seconds of 16kHz audio
await process_voice_conversation(session)
elif (len(session["audio_buffer"]) > 16000 and # At least 1 second
time.time() - session["last_speech"] > 1.0): # 1 second silence
await process_voice_conversation(session)
except Exception as e:
await session["websocket"].send_json({
"type": "error",
"error": f"Audio processing error: {str(e)}"
})
elif msg_type == "stop_listening":
session["is_listening"] = False
if session["audio_buffer"]:
await process_voice_conversation(session)
async def process_voice_conversation(session: Dict):
"""Process complete voice conversation turn"""
if not session["audio_buffer"] or session["processing"]:
return
session["processing"] = True
try:
# Convert audio to text
audio_bytes = bytes(session["audio_buffer"])
session["audio_buffer"] = [] # Clear buffer
logger.info(f"Processing {len(audio_bytes)} bytes of audio for transcription")
# Run STT in thread to avoid blocking
def transcribe_audio():
return audio_to_text(audio_bytes)
# Simple async wrapper for thread
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(transcribe_audio)
transcript = future.result(timeout=10) # 10 second timeout
if transcript:
logger.info(f"Transcript: {transcript}")
# Send transcript to user
await session["websocket"].send_json({
"type": "transcript",
"text": transcript
})
# Generate AI response
response_text = session["agent"].generate_response(transcript)
logger.info(f"Agent response: {response_text}")
# Convert response to speech
def generate_speech():
return text_to_speech(response_text)
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(generate_speech)
audio_response = future.result(timeout=15) # 15 second timeout
# Send response to client
response_data = {
"type": "agent_response",
"text": response_text
}
if audio_response:
response_data["audio"] = base64.b64encode(audio_response).decode('utf-8')
await session["websocket"].send_json(response_data)
else:
logger.info("No transcript generated from audio")
except Exception as e:
logger.error(f"Voice conversation processing error: {e}")
await session["websocket"].send_json({
"type": "error",
"error": f"Conversation processing error: {str(e)}"
})
finally:
session["processing"] = False
await session["websocket"].send_json({
"type": "processing_complete"
})
if __name__ == "__main__":
print("šļø LiveTalker Real-time Voice Chat")
print("=" * 50)
# Create SSL certificates
cert_path, key_path = create_ssl_certificate()
if not cert_path or not key_path:
print("ā Failed to create SSL certificates")
exit(1)
print("ā SSL certificate ready")
print("ā Speech recognition ready")
print("ā Text-to-speech ready")
print("ā AI agent ready")
# Get local IP
import socket
hostname = socket.gethostname()
local_ip = socket.gethostbyname(hostname)
print("\nš HTTPS URLs:")
print(f" Local: https://localhost:8000")
print(f" Network: https://{local_ip}:8000")
print("\nš§ Features:")
print(" ā Real-time voice conversation")
print(" ā Speech-to-text transcription")
print(" ā AI agent responses")
print(" ā Text-to-speech synthesis")
print(" ā SSL/TLS encryption")
print("\nā ļø You may see browser security warnings for the self-signed certificate.")
print(" Click 'Advanced' ā 'Continue to site' to proceed.")
# Configure SSL
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ssl_context.load_cert_chain(cert_path, key_path)
# Start server
uvicorn.run(app, host="0.0.0.0", port=8000, ssl=ssl_context)