Personal AI Assistant

import asyncio import base64 import json from pathlib import Path import os import numpy as np import openai from dotenv import load_dotenv from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse, StreamingResponse from fastrtc import ( AdditionalOutputs, AsyncStreamHandler, Stream, get_twilio_turn_credentials, wait_for_item, ) from gradio.utils import get_space from openai.types.beta.realtime import ResponseAudioTranscriptDoneEvent import httpx from typing import Optional, List, Dict import gradio as gr import io from scipy import signal import wave import aiosqlite from langdetect import detect, LangDetectException from datetime import datetime import uuid load_dotenv() SAMPLE_RATE = 24000 # Use Persistent Storage path for Hugging Face Space # In HF Spaces, persistent storage is at /data if os.path.exists("/data"): PERSISTENT_DIR = "/data" else: PERSISTENT_DIR = "./data" os.makedirs(PERSISTENT_DIR, exist_ok=True) DB_PATH = os.path.join(PERSISTENT_DIR, "personal_assistant.db") print(f"Using persistent directory: {PERSISTENT_DIR}") print(f"Database path: {DB_PATH}") # HTML content embedded as a string HTML_CONTENT = """ Personal AI Assistant

설정

웹 검색

사용자 이름:

기억된 정보

대화 기록

대화

""" class BraveSearchClient: """Brave Search API client""" def __init__(self, api_key: str): self.api_key = api_key self.base_url = "https://api.search.brave.com/res/v1/web/search" async def search(self, query: str, count: int = 10) -> List[Dict]: """Perform a web search using Brave Search API""" if not self.api_key: return [] headers = { "Accept": "application/json", "X-Subscription-Token": self.api_key } params = { "q": query, "count": count, "lang": "ko" } async with httpx.AsyncClient() as client: try: response = await client.get(self.base_url, headers=headers, params=params) response.raise_for_status() data = response.json() results = [] if "web" in data and "results" in data["web"]: for result in data["web"]["results"][:count]: results.append({ "title": result.get("title", ""), "url": result.get("url", ""), "description": result.get("description", "") }) return results except Exception as e: print(f"Brave Search error: {e}") return [] # Database helper class class PersonalAssistantDB: """Database manager for personal assistant""" @staticmethod async def init(): """Initialize database tables""" async with aiosqlite.connect(DB_PATH) as db: # Conversations table await db.execute(""" CREATE TABLE IF NOT EXISTS conversations ( id TEXT PRIMARY KEY, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, summary TEXT ) """) # Messages table await db.execute(""" CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, detected_language TEXT, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (session_id) REFERENCES conversations(id) ) """) # User memories table - stores personal information await db.execute(""" CREATE TABLE IF NOT EXISTS user_memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, category TEXT NOT NULL, content TEXT NOT NULL, confidence REAL DEFAULT 1.0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, source_session_id TEXT, FOREIGN KEY (source_session_id) REFERENCES conversations(id) ) """) # Create indexes for better performance await db.execute("CREATE INDEX IF NOT EXISTS idx_memories_category ON user_memories(category)") await db.execute("CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id)") await db.commit() @staticmethod async def create_session(session_id: str): """Create a new conversation session""" async with aiosqlite.connect(DB_PATH) as db: await db.execute( "INSERT INTO conversations (id) VALUES (?)", (session_id,) ) await db.commit() @staticmethod async def save_message(session_id: str, role: str, content: str): """Save a message to the database""" # Check for None or empty content if not content: print(f"[SAVE_MESSAGE] Empty content for {role} message, skipping") return # Detect language detected_language = None try: if content and len(content) > 10: detected_language = detect(content) except (LangDetectException, Exception) as e: print(f"Language detection error: {e}") async with aiosqlite.connect(DB_PATH) as db: await db.execute( """INSERT INTO messages (session_id, role, content, detected_language) VALUES (?, ?, ?, ?)""", (session_id, role, content, detected_language) ) # Update conversation's updated_at timestamp await db.execute( "UPDATE conversations SET updated_at = CURRENT_TIMESTAMP WHERE id = ?", (session_id,) ) # Update conversation summary (use first user message as summary) if role == "user": cursor = await db.execute( "SELECT summary FROM conversations WHERE id = ?", (session_id,) ) row = await cursor.fetchone() if row and not row[0]: summary = content[:100] + "..." if len(content) > 100 else content await db.execute( "UPDATE conversations SET summary = ? WHERE id = ?", (summary, session_id) ) await db.commit() @staticmethod async def get_recent_conversations(limit: int = 10): """Get recent conversations""" async with aiosqlite.connect(DB_PATH) as db: cursor = await db.execute( """SELECT id, created_at, summary FROM conversations ORDER BY updated_at DESC LIMIT ?""", (limit,) ) rows = await cursor.fetchall() return [ { "id": row[0], "created_at": row[1], "summary": row[2] or "새 대화" } for row in rows ] @staticmethod async def get_conversation_messages(session_id: str): """Get all messages for a conversation""" async with aiosqlite.connect(DB_PATH) as db: cursor = await db.execute( """SELECT role, content, detected_language, timestamp FROM messages WHERE session_id = ? ORDER BY timestamp ASC""", (session_id,) ) rows = await cursor.fetchall() return [ { "role": row[0], "content": row[1], "detected_language": row[2], "timestamp": row[3] } for row in rows ] @staticmethod async def save_memory(category: str, content: str, session_id: str = None, confidence: float = 1.0): """Save or update a user memory""" async with aiosqlite.connect(DB_PATH) as db: # Check if similar memory exists cursor = await db.execute( """SELECT id, content FROM user_memories WHERE category = ? AND content LIKE ? LIMIT 1""", (category, f"%{content[:20]}%") ) existing = await cursor.fetchone() if existing: # Update existing memory await db.execute( """UPDATE user_memories SET content = ?, confidence = ?, updated_at = CURRENT_TIMESTAMP, source_session_id = ? WHERE id = ?""", (content, confidence, session_id, existing[0]) ) else: # Insert new memory await db.execute( """INSERT INTO user_memories (category, content, confidence, source_session_id) VALUES (?, ?, ?, ?)""", (category, content, confidence, session_id) ) await db.commit() @staticmethod async def get_all_memories(): """Get all user memories""" async with aiosqlite.connect(DB_PATH) as db: cursor = await db.execute( """SELECT category, content, confidence, updated_at FROM user_memories ORDER BY category, updated_at DESC""" ) rows = await cursor.fetchall() return [ { "category": row[0], "content": row[1], "confidence": row[2], "updated_at": row[3] } for row in rows ] @staticmethod async def extract_and_save_memories(session_id: str): """Extract memories from conversation and save them""" # Get all messages from the session messages = await PersonalAssistantDB.get_conversation_messages(session_id) if not messages: return # Prepare conversation text for analysis conversation_text = "\n".join([ f"{msg['role']}: {msg['content']}" for msg in messages if msg.get('content') ]) # Use GPT to extract memories client = openai.AsyncOpenAI() try: response = await client.chat.completions.create( model="gpt-4.1-mini", messages=[ { "role": "system", "content": """You are a memory extraction system. Extract personal information from conversations. Categories to extract: - personal_info: 이름, 나이, 성별, 직업, 거주지 - preferences: 좋아하는 것, 싫어하는 것, 취향 - important_dates: 생일, 기념일, 중요한 날짜 - relationships: 가족, 친구, 동료 관계 - hobbies: 취미, 관심사 - health: 건강 상태, 알레르기, 의료 정보 - goals: 목표, 계획, 꿈 - routines: 일상, 습관, 루틴 - work: 직장, 업무, 프로젝트 - education: 학력, 전공, 학습 Return as JSON array with format: [ { "category": "category_name", "content": "extracted information in Korean", "confidence": 0.0-1.0 } ] Only extract clear, factual information. Do not make assumptions.""" }, { "role": "user", "content": f"Extract memories from this conversation:\n\n{conversation_text}" } ], temperature=0.3, max_tokens=2000 ) # Parse and save memories memories_text = response.choices[0].message.content # Extract JSON from response import re json_match = re.search(r'\[.*\]', memories_text, re.DOTALL) if json_match: memories = json.loads(json_match.group()) for memory in memories: if memory.get('content') and len(memory['content']) > 5: await PersonalAssistantDB.save_memory( category=memory.get('category', 'general'), content=memory['content'], session_id=session_id, confidence=memory.get('confidence', 0.8) ) print(f"Extracted and saved {len(memories)} memories from session {session_id}") except Exception as e: print(f"Error extracting memories: {e}") # Initialize search client globally brave_api_key = os.getenv("BSEARCH_API") search_client = BraveSearchClient(brave_api_key) if brave_api_key else None print(f"Search client initialized: {search_client is not None}, API key present: {bool(brave_api_key)}") # Store connection settings connection_settings = {} # Initialize OpenAI client for text chat client = openai.AsyncOpenAI() def update_chatbot(chatbot: list[dict], response: ResponseAudioTranscriptDoneEvent): chatbot.append({"role": "assistant", "content": response.transcript}) return chatbot def format_memories_for_prompt(memories: Dict[str, List[str]]) -> str: """Format memories for inclusion in system prompt""" if not memories: return "" memory_text = "\n\n=== 기억된 정보 ===\n" for category, items in memories.items(): if items and isinstance(items, list): memory_text += f"\n[{category}]\n" for item in items: if item: # Check if item is not None or empty memory_text += f"- {item}\n" return memory_text async def process_text_chat(message: str, web_search_enabled: bool, session_id: str, user_name: str = "", memories: Dict = None) -> Dict[str, str]: """Process text chat using GPT-4o-mini model""" try: # Check for empty or None message if not message: return {"error": "메시지가 비어있습니다."} # Check for stop words stop_words = ["중단", "그만", "스톱", "stop", "닥쳐", "멈춰", "중지"] if any(word in message.lower() for word in stop_words): return { "response": "대화를 중단합니다.", "detected_language": "ko" } # Build system prompt with memories base_prompt = f"""You are a personal AI assistant for {user_name if user_name else 'the user'}. You remember all previous conversations and personal information about the user. Be friendly, helpful, and personalized in your responses. Always use the information you remember to make conversations more personal and relevant. IMPORTANT: Give only ONE response. Do not repeat or give multiple answers.""" # Add memories to prompt if memories: memory_text = format_memories_for_prompt(memories) base_prompt += memory_text messages = [{"role": "system", "content": base_prompt}] # Handle web search if enabled if web_search_enabled and search_client and message: search_keywords = ["날씨", "기온", "비", "눈", "뉴스", "소식", "현재", "최근", "오늘", "지금", "가격", "환율", "주가", "weather", "news", "current", "today", "price", "2024", "2025"] should_search = any(keyword in message.lower() for keyword in search_keywords) if should_search: search_results = await search_client.search(message) if search_results: search_context = "웹 검색 결과:\n\n" for i, result in enumerate(search_results[:5], 1): search_context += f"{i}. {result['title']}\n{result['description']}\n\n" messages.append({ "role": "system", "content": "다음 웹 검색 결과를 참고하여 답변하세요:\n\n" + search_context }) messages.append({"role": "user", "content": message}) # Call GPT-4o-mini response = await client.chat.completions.create( model="gpt-4.1-mini", messages=messages, temperature=0.7, max_tokens=2000 ) response_text = response.choices[0].message.content # Detect language detected_language = None try: if response_text and len(response_text) > 10: detected_language = detect(response_text) except: pass # Save messages to database if session_id: await PersonalAssistantDB.save_message(session_id, "user", message) await PersonalAssistantDB.save_message(session_id, "assistant", response_text) return { "response": response_text, "detected_language": detected_language } except Exception as e: print(f"Error in text chat: {e}") return {"error": str(e)} class OpenAIHandler(AsyncStreamHandler): def __init__(self, web_search_enabled: bool = False, webrtc_id: str = None, session_id: str = None, user_name: str = "", memories: Dict = None) -> None: super().__init__( expected_layout="mono", output_sample_rate=SAMPLE_RATE, output_frame_size=480, input_sample_rate=SAMPLE_RATE, ) self.connection = None self.output_queue = asyncio.Queue() self.search_client = search_client self.function_call_in_progress = False self.current_function_args = "" self.current_call_id = None self.webrtc_id = webrtc_id self.web_search_enabled = web_search_enabled self.session_id = session_id self.user_name = user_name self.memories = memories or {} self.is_responding = False # Track if already responding self.should_stop = False # Track if conversation should stop print(f"[INIT] Handler created with web_search={web_search_enabled}, session_id={session_id}, user={user_name}") def copy(self): if connection_settings: recent_ids = sorted(connection_settings.keys(), key=lambda k: connection_settings[k].get('timestamp', 0), reverse=True) if recent_ids: recent_id = recent_ids[0] settings = connection_settings[recent_id] print(f"[COPY] Copying settings from {recent_id}:") return OpenAIHandler( web_search_enabled=settings.get('web_search_enabled', False), webrtc_id=recent_id, session_id=settings.get('session_id'), user_name=settings.get('user_name', ''), memories=settings.get('memories', {}) ) print(f"[COPY] No settings found, creating default handler") return OpenAIHandler(web_search_enabled=False) async def search_web(self, query: str) -> str: """Perform web search and return formatted results""" if not self.search_client or not self.web_search_enabled: return "웹 검색이 비활성화되어 있습니다." print(f"Searching web for: {query}") results = await self.search_client.search(query) if not results: return f"'{query}'에 대한 검색 결과를 찾을 수 없습니다." formatted_results = [] for i, result in enumerate(results, 1): formatted_results.append( f"{i}. {result['title']}\n" f" URL: {result['url']}\n" f" {result['description']}\n" ) return f"웹 검색 결과 '{query}':\n\n" + "\n".join(formatted_results) async def process_text_message(self, message: str): """Process text message from user""" if self.connection: await self.connection.conversation.item.create( item={ "type": "message", "role": "user", "content": [{"type": "input_text", "text": message}] } ) await self.connection.response.create() async def start_up(self): """Connect to realtime API""" if connection_settings and self.webrtc_id: if self.webrtc_id in connection_settings: settings = connection_settings[self.webrtc_id] self.web_search_enabled = settings.get('web_search_enabled', False) self.session_id = settings.get('session_id') self.user_name = settings.get('user_name', '') self.memories = settings.get('memories', {}) print(f"[START_UP] Updated settings from storage for {self.webrtc_id}") self.client = openai.AsyncOpenAI() print(f"[REALTIME API] Connecting...") # Build system prompt with memories base_instructions = f"""You are a personal AI assistant for {self.user_name if self.user_name else 'the user'}. You remember all previous conversations and personal information about the user. Be friendly, helpful, and personalized in your responses. Always use the information you remember to make conversations more personal and relevant. IMPORTANT: Give only ONE response per user input. Do not repeat yourself or give multiple answers.""" # Add memories to prompt if self.memories: memory_text = format_memories_for_prompt(self.memories) base_instructions += memory_text # Define the web search function tools = [] if self.web_search_enabled and self.search_client: tools = [{ "type": "function", "function": { "name": "web_search", "description": "Search the web for current information. Use this for weather, news, prices, current events, or any time-sensitive topics.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "The search query" } }, "required": ["query"] } } }] search_instructions = ( "\n\nYou have web search capabilities. " "Use web_search for current information like weather, news, prices, etc." ) instructions = base_instructions + search_instructions else: instructions = base_instructions async with self.client.beta.realtime.connect( model="gpt-4o-mini-realtime-preview-2024-12-17" ) as conn: session_update = { "turn_detection": { "type": "server_vad", "threshold": 0.5, "prefix_padding_ms": 300, "silence_duration_ms": 200 }, "instructions": instructions, "tools": tools, "tool_choice": "auto" if tools else "none", "temperature": 0.7, "max_response_output_tokens": 4096, "modalities": ["text", "audio"], "voice": "alloy" } try: await conn.session.update(session=session_update) self.connection = conn print(f"Connected with tools: {len(tools)} functions") print(f"Session update successful") except Exception as e: print(f"Error updating session: {e}") raise async for event in self.connection: # Debug log for all events if hasattr(event, 'type'): if event.type not in ["response.audio.delta", "response.audio.done"]: print(f"[EVENT] Type: {event.type}") # Handle user input audio transcription if event.type == "conversation.item.input_audio_transcription.completed": if hasattr(event, 'transcript') and event.transcript: user_text = event.transcript.lower() stop_words = ["중단", "그만", "스톱", "stop", "닥쳐", "멈춰", "중지"] if any(word in user_text for word in stop_words): print(f"[STOP DETECTED] User said: {event.transcript}") self.should_stop = True if self.connection: try: await self.connection.response.cancel() except: pass continue # Save user message to database if self.session_id: await PersonalAssistantDB.save_message(self.session_id, "user", event.transcript) # Handle user transcription for stop detection (alternative event) elif event.type == "conversation.item.created": if hasattr(event, 'item') and hasattr(event.item, 'role') and event.item.role == "user": if hasattr(event.item, 'content') and event.item.content: for content_item in event.item.content: if hasattr(content_item, 'transcript') and content_item.transcript: user_text = content_item.transcript.lower() stop_words = ["중단", "그만", "스톱", "stop", "닥쳐", "멈춰", "중지"] if any(word in user_text for word in stop_words): print(f"[STOP DETECTED] User said: {content_item.transcript}") self.should_stop = True if self.connection: try: await self.connection.response.cancel() except: pass continue # Save user message to database if self.session_id: await PersonalAssistantDB.save_message(self.session_id, "user", content_item.transcript) elif event.type == "response.audio_transcript.done": # Prevent multiple responses if self.is_responding: print("[DUPLICATE RESPONSE] Skipping duplicate response") continue self.is_responding = True print(f"[RESPONSE] Transcript: {event.transcript[:100] if event.transcript else 'None'}...") # Detect language detected_language = None try: if event.transcript and len(event.transcript) > 10: detected_language = detect(event.transcript) except Exception as e: print(f"Language detection error: {e}") # Save to database if self.session_id and event.transcript: await PersonalAssistantDB.save_message(self.session_id, "assistant", event.transcript) output_data = { "event": event, "detected_language": detected_language } await self.output_queue.put(AdditionalOutputs(output_data)) elif event.type == "response.done": # Reset responding flag when response is complete self.is_responding = False self.should_stop = False print("[RESPONSE DONE] Response completed") elif event.type == "response.audio.delta": # Check if we should stop if self.should_stop: continue if hasattr(event, 'delta'): await self.output_queue.put( ( self.output_sample_rate, np.frombuffer( base64.b64decode(event.delta), dtype=np.int16 ).reshape(1, -1), ), ) # Handle errors elif event.type == "error": print(f"[ERROR] {event}") self.is_responding = False # Handle function calls elif event.type == "response.function_call_arguments.start": print(f"Function call started") self.function_call_in_progress = True self.current_function_args = "" self.current_call_id = getattr(event, 'call_id', None) elif event.type == "response.function_call_arguments.delta": if self.function_call_in_progress: self.current_function_args += event.delta elif event.type == "response.function_call_arguments.done": if self.function_call_in_progress: print(f"Function call done, args: {self.current_function_args}") try: args = json.loads(self.current_function_args) query = args.get("query", "") # Emit search event to client await self.output_queue.put(AdditionalOutputs({ "type": "search", "query": query })) # Perform the search search_results = await self.search_web(query) print(f"Search results length: {len(search_results)}") # Send function result back to the model if self.connection and self.current_call_id: await self.connection.conversation.item.create( item={ "type": "function_call_output", "call_id": self.current_call_id, "output": search_results } ) await self.connection.response.create() except Exception as e: print(f"Function call error: {e}") finally: self.function_call_in_progress = False self.current_function_args = "" self.current_call_id = None async def receive(self, frame: tuple[int, np.ndarray]) -> None: if not self.connection: print(f"[RECEIVE] No connection, skipping") return try: if frame is None or len(frame) < 2: print(f"[RECEIVE] Invalid frame") return _, array = frame if array is None: print(f"[RECEIVE] Null array") return array = array.squeeze() audio_message = base64.b64encode(array.tobytes()).decode("utf-8") await self.connection.input_audio_buffer.append(audio=audio_message) except Exception as e: print(f"Error in receive: {e}") async def emit(self) -> tuple[int, np.ndarray] | AdditionalOutputs | None: item = await wait_for_item(self.output_queue) if isinstance(item, dict) and item.get('type') == 'text_message': await self.process_text_message(item['content']) return None return item async def shutdown(self) -> None: print(f"[SHUTDOWN] Called") if self.connection: await self.connection.close() self.connection = None print("[REALTIME API] Connection closed") # Create initial handler instance handler = OpenAIHandler(web_search_enabled=False) # Create components chatbot = gr.Chatbot(type="messages") # Create stream with handler instance stream = Stream( handler, mode="send-receive", modality="audio", additional_inputs=[chatbot], additional_outputs=[chatbot], additional_outputs_handler=update_chatbot, rtc_configuration=get_twilio_turn_credentials() if get_space() else None, concurrency_limit=5 if get_space() else None, time_limit=300 if get_space() else None, ) app = FastAPI() # Mount stream stream.mount(app) # Initialize database on startup @app.on_event("startup") async def startup_event(): try: await PersonalAssistantDB.init() print(f"Database initialized at: {DB_PATH}") print(f"Persistent directory: {PERSISTENT_DIR}") print(f"DB file exists: {os.path.exists(DB_PATH)}") # Check if we're in Hugging Face Space if os.path.exists("/data"): print("Running in Hugging Face Space with persistent storage") # List files in persistent directory try: files = os.listdir(PERSISTENT_DIR) print(f"Files in persistent directory: {files}") except Exception as e: print(f"Error listing files: {e}") except Exception as e: print(f"Error during startup: {e}") # Try to create directory if it doesn't exist os.makedirs(PERSISTENT_DIR, exist_ok=True) await PersonalAssistantDB.init() # Intercept offer to capture settings @app.post("/webrtc/offer", include_in_schema=False) async def custom_offer(request: Request): """Intercept offer to capture settings""" body = await request.json() webrtc_id = body.get("webrtc_id") web_search_enabled = body.get("web_search_enabled", False) session_id = body.get("session_id") user_name = body.get("user_name", "") memories = body.get("memories", {}) print(f"[OFFER] Received offer with webrtc_id: {webrtc_id}") print(f"[OFFER] web_search_enabled: {web_search_enabled}") print(f"[OFFER] session_id: {session_id}") print(f"[OFFER] user_name: {user_name}") # Store settings with timestamp if webrtc_id: connection_settings[webrtc_id] = { 'web_search_enabled': web_search_enabled, 'session_id': session_id, 'user_name': user_name, 'memories': memories, 'timestamp': asyncio.get_event_loop().time() } print(f"[OFFER] Stored settings for {webrtc_id}") # Remove our custom route temporarily custom_route = None for i, route in enumerate(app.routes): if hasattr(route, 'path') and route.path == "/webrtc/offer" and route.endpoint == custom_offer: custom_route = app.routes.pop(i) break # Forward to stream's offer handler print(f"[OFFER] Forwarding to stream.offer()") response = await stream.offer(body) # Re-add our custom route if custom_route: app.routes.insert(0, custom_route) print(f"[OFFER] Response status: {response.get('status', 'unknown') if isinstance(response, dict) else 'OK'}") return response @app.post("/session/new") async def create_new_session(): """Create a new chat session""" session_id = str(uuid.uuid4()) await PersonalAssistantDB.create_session(session_id) return {"session_id": session_id} @app.post("/session/end") async def end_session(request: Request): """End session and extract memories""" body = await request.json() session_id = body.get("session_id") if not session_id: return {"error": "session_id required"} # Extract and save memories from the conversation await PersonalAssistantDB.extract_and_save_memories(session_id) return {"status": "ok"} @app.post("/message/save") async def save_message(request: Request): """Save a message to the database""" body = await request.json() session_id = body.get("session_id") role = body.get("role") content = body.get("content") if not all([session_id, role, content]): return {"error": "Missing required fields"} await PersonalAssistantDB.save_message(session_id, role, content) return {"status": "ok"} @app.get("/history/recent") async def get_recent_history(): """Get recent conversation history""" conversations = await PersonalAssistantDB.get_recent_conversations() return conversations @app.get("/history/{session_id}") async def get_conversation(session_id: str): """Get messages for a specific conversation""" messages = await PersonalAssistantDB.get_conversation_messages(session_id) return messages @app.get("/memory/all") async def get_all_memories(): """Get all user memories""" memories = await PersonalAssistantDB.get_all_memories() return memories @app.post("/chat/text") async def chat_text(request: Request): """Handle text chat messages using GPT-4o-mini""" try: body = await request.json() message = body.get("message", "") web_search_enabled = body.get("web_search_enabled", False) session_id = body.get("session_id") user_name = body.get("user_name", "") memories = body.get("memories", {}) if not message: return {"error": "메시지가 비어있습니다."} # Process text chat result = await process_text_chat(message, web_search_enabled, session_id, user_name, memories) return result except Exception as e: print(f"Error in chat_text endpoint: {e}") return {"error": "채팅 처리 중 오류가 발생했습니다."} @app.post("/text_message/{webrtc_id}") async def receive_text_message(webrtc_id: str, request: Request): """Receive text message from client""" body = await request.json() message = body.get("content", "") # Find the handler for this connection if webrtc_id in stream.handlers: handler = stream.handlers[webrtc_id] # Queue the text message for processing await handler.output_queue.put({ 'type': 'text_message', 'content': message }) return {"status": "ok"} @app.get("/outputs") async def outputs(webrtc_id: str): """Stream outputs including search events""" async def output_stream(): async for output in stream.output_stream(webrtc_id): if hasattr(output, 'args') and output.args: # Check if it's a search event if isinstance(output.args[0], dict) and output.args[0].get('type') == 'search': yield f"event: search\ndata: {json.dumps(output.args[0])}\n\n" # Regular transcript event with language info elif isinstance(output.args[0], dict) and 'event' in output.args[0]: event_data = output.args[0] if 'event' in event_data and hasattr(event_data['event'], 'transcript'): data = { "role": "assistant", "content": event_data['event'].transcript, "detected_language": event_data.get('detected_language') } yield f"event: output\ndata: {json.dumps(data)}\n\n" return StreamingResponse(output_stream(), media_type="text/event-stream") @app.get("/") async def index(): """Serve the HTML page""" rtc_config = get_twilio_turn_credentials() if get_space() else None html_content = HTML_CONTENT.replace("__RTC_CONFIGURATION__", json.dumps(rtc_config)) return HTMLResponse(content=html_content) if __name__ == "__main__": import uvicorn mode = os.getenv("MODE") if mode == "UI": stream.ui.launch(server_port=7860) elif mode == "PHONE": stream.fastphone(host="0.0.0.0", port=7860) else: uvicorn.run(app, host="0.0.0.0", port=7860)