""" FastAPI server that: 1. Accepts raw audio from ESP32 over WebSocket (same port 8000) 2. Streams audio to Deepgram via raw WebSocket (no SDK) 3. Pushes transcripts to a browser via WebSocket 4. When user presses Q/Stop, sends full transcript to Gemini via LangChain and streams the conversational response + follow-up question back to browser """ import asyncio import json import os from contextlib import asynccontextmanager import websockets from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse from langchain_google_genai import ChatGoogleGenerativeAI # ─── Configuration ─────────────────────────────────────────────────────────── DEEPGRAM_API_KEY = "Enter DeepGram API Key" GOOGLE_API_KEY = "Enter Google API Key" GOOGLE_API_KEY_CHAT = "Enter Google API Key" SAMPLE_RATE = 16000 FASTAPI_HOST = "0.0.0.0" FASTAPI_PORT = int(os.environ.get("PORT", 8000)) DEEPGRAM_URL = ( f"wss://api.deepgram.com/v1/listen" f"?model=nova-3&language=en&encoding=linear16" f"&sample_rate={SAMPLE_RATE}&channels=1" f"&interim_results=true&punctuate=true" f"&smart_format=true&endpointing=500" f"&utterance_end_ms=1500" ) # ─── Shared state ──────────────────────────────────────────────────────────── browser_clients: list[WebSocket] = [] # Accumulates FINAL transcript lines during an active ESP32 session transcript_buffer: list[str] = [] # Conversation history for multi-turn LLM summarization conversation_history: list[tuple] = [] # Separate chat history for user ↔ Sync AI conversations chat_history: list[tuple] = [] # Stores ALL generated summaries so chat can reference them summary_history: list[str] = [] # ─── LLM setup ─────────────────────────────────────────────────────────────── # LLM for transcript summarization (Sync AI button) llm = ChatGoogleGenerativeAI( model="gemini-2.5-flash", google_api_key=GOOGLE_API_KEY, temperature=1.0, max_tokens=None, timeout=None, max_retries=2, ) # Separate LLM for chat conversations (summary-aware Q&A) chat_llm = ChatGoogleGenerativeAI( model="gemini-2.5-flash", google_api_key=GOOGLE_API_KEY_CHAT, temperature=0.7, max_tokens=None, timeout=None, max_retries=2, ) LLM_SYSTEM_PROMPT = ( "You are Sync AI, a smart and concise assistant. " "When the user provides a transcript or message, summarise the key points " "from the conversation so far in 2-3 sentences. " "Then ask exactly 2 relevant follow-up questions to the user " "to help deepen the discussion or clarify important details." ) CHAT_SYSTEM_PROMPT = ( "You are Sync AI, a knowledgeable and friendly conversational assistant. " "You have access to ALL the AI-generated summaries from the user's audio " "transcription session. Use these summaries as your primary knowledge base " "to answer the user's questions accurately. " "If the user asks about something covered in the summaries, reference the " "relevant summary details in your answer. " "Keep answers clear and concise (2-4 sentences) unless the user asks for detail. " "Be natural and conversational." ) async def run_llm_on_transcript(full_transcript: str): """ Send the buffered transcript to Gemini via LangChain. Uses tuple-based messages and run_in_executor (same as new_with_llmv1.py). Streams the response back to browser clients. Each call is independent — previous summaries are NOT carried forward. """ if not full_transcript.strip(): await broadcast("__LLM_ERROR__:No transcript to process.") return await broadcast("__LLM_START__") print(f"[LLM] Processing transcript ({len(full_transcript)} chars)...") # Clear conversation history so each sync is independent (no appending) conversation_history.clear() # Build message list using tuples: (role, content) messages = [("system", LLM_SYSTEM_PROMPT)] # Add current transcript as the only user turn (no previous turns) messages.append(("human", full_transcript)) try: # Call Gemini in a thread (same pattern as new_with_llmv1.py) loop = asyncio.get_event_loop() ai_msg = await loop.run_in_executor(None, llm.invoke, messages) full_response = ai_msg.content # Send the full response to browser await broadcast(f"__LLM_TOKEN__:{full_response}") # Save this summary so chat can reference it later summary_history.append(full_response) # Clear the transcript buffer so next sync only gets NEW lines transcript_buffer.clear() await broadcast("__LLM_DONE__") print(f"[LLM] Response: {full_response[:100]}...") except Exception as e: err = str(e) print(f"[LLM] Error: {err}") await broadcast(f"__LLM_ERROR__:{err}") async def run_chat_response(user_message: str): """ Handle a chat message from the user. Uses chat_llm (separate Gemini instance) with ALL generated summaries as context so the user can ask questions about any summary. Runs in parallel with STT — does not block audio processing. """ if not user_message.strip(): await broadcast("__CHAT_ERROR__:Empty message.") return await broadcast("__CHAT_START__") print(f"[Chat] User: {user_message[:80]}...") # Build message list with chat system prompt messages = [("system", CHAT_SYSTEM_PROMPT)] # Inject ALL generated summaries as context if summary_history: summaries_context = "\n\n".join( f"--- Summary {i+1} ---\n{s}" for i, s in enumerate(summary_history) ) messages.append(( "system", f"Here are all the AI-generated summaries from this session " f"({len(summary_history)} total). Use these to answer the user's " f"questions:\n\n{summaries_context}" )) # Include current (unsummarized) transcript if available if transcript_buffer: transcript_context = " ".join(transcript_buffer) messages.append(( "system", f"Current live transcript (not yet summarized): {transcript_context}" )) # Replay previous chat turns for role, content in chat_history: messages.append((role, content)) # Add current user message messages.append(("human", user_message)) chat_history.append(("human", user_message)) try: loop = asyncio.get_event_loop() ai_msg = await loop.run_in_executor(None, chat_llm.invoke, messages) full_response = ai_msg.content await broadcast(f"__CHAT_TOKEN__:{full_response}") chat_history.append(("ai", full_response)) await broadcast("__CHAT_DONE__") print(f"[Chat] Sync AI: {full_response[:100]}...") except Exception as e: err = str(e) print(f"[Chat] Error: {err}") await broadcast(f"__CHAT_ERROR__:{err}") # ─── Helpers ───────────────────────────────────────────────────────────────── async def broadcast(message: str): """Send a message to every connected browser client.""" disconnected = [] for ws in browser_clients: try: await ws.send_text(message) except Exception: disconnected.append(ws) for ws in disconnected: browser_clients.remove(ws) # ─── FastAPI lifecycle ─────────────────────────────────────────────────────── @asynccontextmanager async def lifespan(app: FastAPI): print("[FastAPI] Server ready") yield app = FastAPI(lifespan=lifespan) # ─── WebSocket: browser clients ───────────────────────────────────────────── @app.websocket("/ws") async def websocket_browser(ws: WebSocket): await ws.accept() browser_clients.append(ws) try: while True: msg = await ws.receive_text() # Browser sends "PROCESS" when user clicks Stop/Q if msg == "PROCESS": full_text = " ".join(transcript_buffer) asyncio.create_task(run_llm_on_transcript(full_text)) # Browser sends chat message as "CHAT:message" elif msg.startswith("CHAT:"): user_msg = msg[5:] asyncio.create_task(run_chat_response(user_msg)) # Browser sends "CLEAR_HISTORY" to reset conversation elif msg == "CLEAR_HISTORY": conversation_history.clear() chat_history.clear() transcript_buffer.clear() summary_history.clear() await broadcast("__STATUS__:history_cleared") except WebSocketDisconnect: if ws in browser_clients: browser_clients.remove(ws) # ─── WebSocket: ESP32 audio → Deepgram (fully async, no threads) ──────────── @app.websocket("/ws/audio") async def websocket_audio(esp_ws: WebSocket): await esp_ws.accept() await broadcast("__STATUS__:esp32_connected") print("[ESP32] Connected via WebSocket") # Clear buffer for new session transcript_buffer.clear() dg_ws = None recv_task = None chunks = 0 try: headers = {"Authorization": f"Token {DEEPGRAM_API_KEY}"} print(f"[Deepgram] Connecting to: {DEEPGRAM_URL[:80]}...") print(f"[Deepgram] API key: {DEEPGRAM_API_KEY[:8]}...{DEEPGRAM_API_KEY[-4:]}") dg_ws = await websockets.connect( DEEPGRAM_URL, additional_headers=headers, ping_interval=20, ping_timeout=20, close_timeout=5, ) await broadcast("__STATUS__:deepgram_connected") print(f"[Deepgram] Connected (state: open={dg_ws.protocol.state if hasattr(dg_ws, 'protocol') else 'n/a'})") async def receive_transcripts(): """Receive and process Deepgram transcript messages.""" nonlocal dg_ws print("[Deepgram] Receive task started — waiting for messages...") msg_count = 0 try: while True: try: msg = await dg_ws.recv() except websockets.ConnectionClosed as e: print(f"[Deepgram] Connection closed during recv: code={e.code} reason={e.reason}") await broadcast(f"__DEBUG__:[DG] Closed: {e.code} {e.reason}") break msg_count += 1 # Log first few raw messages for debugging if msg_count <= 3: raw_preview = str(msg)[:300] if isinstance(msg, str) else f"" print(f"[Deepgram] Raw msg #{msg_count}: {raw_preview}") try: data = json.loads(msg) except (json.JSONDecodeError, TypeError) as je: print(f"[Deepgram] JSON parse error: {je} — raw: {str(msg)[:200]}") continue msg_type = data.get("type", "") if msg_type != "Results": print(f"[Deepgram] Event: {msg_type} → {str(data)[:200]}") await broadcast(f"__DEBUG__:[DG] {msg_type}: {str(data)[:150]}") continue channel = data.get("channel") if not isinstance(channel, dict): continue alternatives = channel.get("alternatives", []) if not alternatives: continue transcript = alternatives[0].get("transcript", "") if transcript: is_final = data.get("is_final", False) prefix = "FINAL" if is_final else "INTERIM" await broadcast(f"__{prefix}__:{transcript}") print(f"[{prefix}] {transcript}") # Buffer only final lines for LLM if is_final: transcript_buffer.append(transcript) except asyncio.CancelledError: print(f"[Deepgram] Receive task cancelled after {msg_count} messages") except Exception as e: import traceback print(f"[Deepgram] Receive error: {type(e).__name__}: {e}") traceback.print_exc() await broadcast(f"__DEBUG__:[DG] Recv error: {e}") def task_exception_callback(task: asyncio.Task): """Catch any unhandled exception from the receive task.""" if task.cancelled(): return exc = task.exception() if exc: print(f"[Deepgram] TASK EXCEPTION: {type(exc).__name__}: {exc}") recv_task = asyncio.create_task(receive_transcripts()) recv_task.add_done_callback(task_exception_callback) while True: audio = await esp_ws.receive_bytes() try: await dg_ws.send(audio) except Exception as send_err: print(f"[Deepgram] Send failed: {type(send_err).__name__}: {send_err}") print("[Deepgram] Connection lost, reconnecting...") try: dg_ws = await websockets.connect( DEEPGRAM_URL, additional_headers=headers, ping_interval=20, ping_timeout=20, close_timeout=5, ) recv_task.cancel() recv_task = asyncio.create_task(receive_transcripts()) recv_task.add_done_callback(task_exception_callback) await dg_ws.send(audio) print("[Deepgram] Reconnected!") except Exception as re_err: print(f"[Deepgram] Reconnect failed: {re_err}") break chunks += 1 if chunks == 1: print(f"[Audio] First chunk: {len(audio)} bytes") if chunks % 500 == 0: print(f"[Audio] Forwarded {chunks} chunks") # Check if receive task is still alive if recv_task.done(): print("[Deepgram] WARNING: Receive task died! Restarting...") recv_task = asyncio.create_task(receive_transcripts()) recv_task.add_done_callback(task_exception_callback) except WebSocketDisconnect: print(f"[ESP32] Disconnected after {chunks} chunks") except Exception as e: import traceback print(f"[Error] {type(e).__name__}: {e}") traceback.print_exc() finally: if recv_task and not recv_task.done(): recv_task.cancel() try: if dg_ws: await dg_ws.close() except Exception: pass await broadcast("__STATUS__:esp32_disconnected") await broadcast("__STATUS__:deepgram_disconnected") print("[Cleanup] Session ended") # ─── Serve the single-page UI ─────────────────────────────────────────────── PAGE_HTML = """\ SyncScribe

SyncScribe

WebSocket
Deepgram
ESP32
Sync AI
Waiting for audio stream from ESP32 ...
ESP32 connects via WebSocket on /ws/audio
0 lines
⬡ Debug Log
""" @app.get("/", response_class=HTMLResponse) async def index(): return PAGE_HTML # ─── Entrypoint ────────────────────────────────────────────────────────────── if __name__ == "__main__": import uvicorn uvicorn.run(app, host=FASTAPI_HOST, port=FASTAPI_PORT)