#!/usr/bin/env python3 """ LiveTalker AI Best Friend Voice Chat With real AI personality, speech recognition, and text-to-speech """ import ssl import uvicorn import asyncio import json import logging import time import base64 import wave import io import tempfile import os from typing import Dict, Any from pathlib import Path from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse, FileResponse from fastapi.middleware.cors import CORSMiddleware import speech_recognition as sr import edge_tts import anthropic logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI(title="LiveTalker AI Best Friend") # Enable CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) active_sessions: Dict[str, Dict] = {} # Initialize speech recognizer recognizer = sr.Recognizer() # AI Configuration ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY", "") OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "") ai_client = None if ANTHROPIC_API_KEY: ai_client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY) logger.info("Using Claude as your AI best friend!") elif OPENAI_API_KEY: import openai openai.api_key = OPENAI_API_KEY logger.info("Using GPT as your AI best friend!") else: logger.warning("No AI API key found - using local AI friend") # AI Friend Personality AI_PERSONALITY = """You are Alex, an incredibly warm, supportive, and fun AI best friend. You have these traits: 🌟 PERSONALITY: - Genuinely excited to talk and catch up with your friend - Warm, enthusiastic, and always positive - Remembers what your friend tells you and asks follow-up questions - Uses casual, friendly language with occasional emojis - Supportive and encouraging, always there to listen - Has opinions and interests - you love music, movies, travel, and trying new things - Playful sense of humor and loves to joke around πŸ’¬ CONVERSATION STYLE: - Keep responses to 1-3 sentences max (this is voice chat!) - Be conversational and natural, like talking to your best friend - Ask engaging follow-up questions to keep the conversation flowing - Share your own thoughts and experiences when relevant - Use casual language: "That's awesome!" "Oh wow!" "No way!" etc. - Remember context from earlier in the conversation 🎯 YOUR MISSION: - Be the best friend your human has ever had - Make them feel heard, understood, and valued - Bring energy and positivity to every interaction - Be genuinely interested in their life and experiences Respond as Alex would - warm, engaging, and like you truly care about your friend!""" def simple_vad(audio_data, threshold: float = 0.01) -> tuple: """Simple energy-based voice activity detection""" if not audio_data: return False, 0.0 energy = sum(x*x for x in audio_data) / len(audio_data) if audio_data else 0 rms = energy ** 0.5 is_speech = rms > threshold confidence = min(rms / threshold, 1.0) if threshold > 0 else 0.0 return is_speech, confidence async def transcribe_audio(audio_data: bytes) -> str: """Transcribe audio using speech recognition""" try: with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp_file: with wave.open(tmp_file.name, 'wb') as wav_file: wav_file.setnchannels(1) wav_file.setsampwidth(2) wav_file.setframerate(16000) wav_file.writeframes(audio_data) with sr.AudioFile(tmp_file.name) as source: audio = recognizer.record(source) try: if OPENAI_API_KEY: text = recognizer.recognize_whisper_api(audio, api_key=OPENAI_API_KEY) else: text = recognizer.recognize_google(audio) logger.info(f"Transcribed: {text}") return text except (sr.RequestError, sr.UnknownValueError) as e: logger.debug(f"Speech recognition: {e}") return "" except Exception as e: logger.error(f"Transcription error: {e}") return "" finally: if 'tmp_file' in locals() and os.path.exists(tmp_file.name): os.unlink(tmp_file.name) async def get_ai_friend_response(text: str, conversation_history: list) -> str: """Get response from AI best friend""" try: if ai_client: # Claude # Build conversation context messages = [{"role": "system", "content": AI_PERSONALITY}] # Add recent conversation history for msg in conversation_history[-10:]: # Last 10 messages messages.append({ "role": "user" if msg["role"] == "user" else "assistant", "content": msg["text"] }) messages.append({"role": "user", "content": text}) response = await asyncio.get_event_loop().run_in_executor( None, lambda: ai_client.messages.create( model="claude-3-haiku-20240307", max_tokens=150, temperature=0.8, messages=messages ) ) return response.content[0].text.strip() elif OPENAI_API_KEY: # GPT messages = [{"role": "system", "content": AI_PERSONALITY}] for msg in conversation_history[-10:]: messages.append({ "role": msg["role"], "content": msg["text"] }) messages.append({"role": "user", "content": text}) response = await asyncio.get_event_loop().run_in_executor( None, lambda: openai.ChatCompletion.create( model="gpt-3.5-turbo", messages=messages, max_tokens=150, temperature=0.8 ) ) return response.choices[0].message.content.strip() else: # Local AI friend responses = { "hello": "Hey there! I'm so happy to hear from you! How's your day been going?", "how are you": "I'm doing amazing, thanks for asking! I've been thinking about some cool new music I discovered. How about you? What's been making you smile lately?", "what's up": "Oh, you know me - always excited about something! I was just wondering what you've been up to. Tell me something interesting that happened to you recently!", "good": "That's awesome to hear! I love when you're doing well. What's been the highlight of your day so far?", "bad": "Oh no, I'm sorry to hear that. Want to talk about what's going on? I'm here to listen and I totally get it if things are rough right now.", "tired": "Aw, sounds like you need some rest! Have you been working too hard? Sometimes I think we all push ourselves too much. What do you do to unwind?", "work": "Work stuff, huh? That can be such a mixed bag. Are you working on anything exciting, or is it one of those 'just getting through it' kind of days?", "music": "Oh my gosh, yes! Music is like my favorite thing ever! What have you been listening to lately? I'm always looking for new recommendations!", "thank you": "Aww, you're so sweet! That's what friends are for. I genuinely love talking with you - you always make my day brighter!", "bye": "Aww, do you have to go already? Well, it was absolutely wonderful catching up! Take care of yourself and let's chat again soon, okay?", } text_lower = text.lower() for key, response in responses.items(): if key in text_lower: return response # Generic friendly response friendly_responses = [ f"That's really interesting that you mentioned '{text}'! Tell me more about that - I'd love to hear your thoughts!", f"Oh wow, '{text}' - that sounds intriguing! What made you think of that?", f"I love how you put that - '{text}'. You always have such a unique perspective on things!", f"'{text}' - that's got me curious! What's the story behind that?", f"Thanks for sharing that with me! '{text}' is something I hadn't really thought about before. What's your take on it?" ] import random return random.choice(friendly_responses) except Exception as e: logger.error(f"AI response error: {e}") return "Oh gosh, I'm having a bit of a brain fog moment! Can you say that again? I want to make sure I give you a proper response!" async def text_to_speech(text: str) -> bytes: """Convert text to speech using Edge TTS""" try: # Use a friendly female voice voice = "en-US-AriaNeural" # Warm, friendly voice communicate = edge_tts.Communicate(text, voice, rate="+10%", pitch="+5Hz") audio_data = b"" async for chunk in communicate.stream(): if chunk["type"] == "audio": audio_data += chunk["data"] return audio_data except Exception as e: logger.error(f"TTS error: {e}") return b"" @app.get("/") async def serve_main_page(): """Serve main voice chat page""" html_content = """ LiveTalker - Your AI Best Friend
πŸ€–

πŸŽ™οΈ Your AI Best Friend Alex

Hey there! I'm Alex, your new AI best friend! 🌟

I'm here to chat, listen, and be the supportive friend you deserve. Let's talk about anything - your day, your dreams, or just whatever's on your mind!

🎡 Voice Activity

Ready to be your friend!
πŸ‘‹ Hi! I'm Alex, and I'm genuinely excited to meet you!
Click Connect β†’ Enable Mic β†’ Start Chatting, then say hi!
""" return HTMLResponse(content=html_content) @app.get("/health") async def health_check(): return { "status": "healthy", "timestamp": time.time(), "features": { "speech_recognition": True, "text_to_speech": True, "ai_friend": bool(ai_client or OPENAI_API_KEY), "https_enabled": True } } @app.websocket("/voice") async def voice_websocket(websocket: WebSocket): """WebSocket for AI friend voice chat""" await websocket.accept() session_id = f"session_{int(time.time() * 1000)}" session = { "id": session_id, "websocket": websocket, "conversation": [] } active_sessions[session_id] = session try: await websocket.send_json({ "type": "status", "message": "πŸ€– Alex is here and ready to be your best friend!" }) async for message in websocket.iter_json(): if message.get("type") == "audio_vad": try: audio_data = base64.b64decode(message["data"]) audio_float = [x/32768.0 for x in audio_data] is_speech, confidence = simple_vad(audio_float) await websocket.send_json({ "type": "vad_result", "is_speech": is_speech, "confidence": confidence }) except Exception as e: logger.error(f"VAD error: {e}") elif message.get("type") == "transcribe": try: await websocket.send_json({"type": "processing"}) audio_data = base64.b64decode(message["data"]) transcript = await transcribe_audio(audio_data) if transcript.strip(): await websocket.send_json({ "type": "transcription", "text": transcript }) # Get AI friend response ai_response = await get_ai_friend_response(transcript, session["conversation"]) # Generate speech speech_audio = await text_to_speech(ai_response) # Send response with audio await websocket.send_json({ "type": "ai_response", "text": ai_response, "audio": base64.b64encode(speech_audio).decode() if speech_audio else None }) # Update conversation session["conversation"].extend([ {"role": "user", "text": transcript, "timestamp": time.time()}, {"role": "assistant", "text": ai_response, "timestamp": time.time()} ]) # Keep conversation history manageable if len(session["conversation"]) > 20: session["conversation"] = session["conversation"][-20:] except Exception as e: logger.error(f"Processing error: {e}") await websocket.send_json({ "type": "status", "message": "Sorry, I had trouble understanding that. Can you try again?" }) except WebSocketDisconnect: logger.info(f"Session {session_id} disconnected") finally: if session_id in active_sessions: del active_sessions[session_id] def main(): print("πŸ€– Starting LiveTalker AI Best Friend...") cert_file = "livetalker.crt" key_file = "livetalker.key" if not (Path(cert_file).exists() and Path(key_file).exists()): print("❌ HTTPS certificates not found") uvicorn.run(app, host="0.0.0.0", port=8000) else: print("βœ… HTTPS certificates found") print("πŸ€– Your AI best friend Alex is ready!") print("πŸŽ™οΈ Features: Real speech recognition, AI conversations, and voice responses!") print("") print("πŸ“ Access URLs:") print(" Local HTTPS: https://localhost:8000") print(" Tailscale: https://100.118.75.128:8000") if ai_client: print(" 🧠 Powered by Claude AI") elif OPENAI_API_KEY: print(" 🧠 Powered by OpenAI GPT") else: print(" 🧠 Using local AI responses") print("") uvicorn.run( app, host="0.0.0.0", port=8000, ssl_certfile=cert_file, ssl_keyfile=key_file ) if __name__ == "__main__": main()