Realtime Flow

requirements.txt

@@ -5,4 +5,5 @@ livekit-plugins-hume
 livekit-plugins-silero
 fastapi
 uvicorn
-livekit
+livekit
+aiohttp

src/agent_session/main.py

@@ -1,15 +1,16 @@
 #!/usr/bin/env python3
 """
-Agent Session for Avurna Flow (Final Corrected Version)
+Agent Session for Avurna Flow (Final Corrected Version with HTTP Session Management)
 """
 import os
 import json
 import traceback
-from fastapi import FastAPI, BackgroundTasks
+import aiohttp # 1. Import aiohttp
+from contextlib import asynccontextmanager
+from fastapi import FastAPI, BackgroundTasks, Request
 from pydantic import BaseModel
 import uvicorn
 from fastapi.middleware.cors import CORSMiddleware
-from contextlib import asynccontextmanager # --- KEY: Import the required decorator ---
 
 from livekit.rtc import Room
 from livekit.agents import Agent, AgentSession
@@ -23,7 +24,16 @@ from livekit.plugins.silero import VAD
 from src.agent_session.constants import SYSTEM_PROMPT, GREETING_INSTRUCTIONS
 from src.utils import validate_env_vars
 
-app = FastAPI()
+# 2. Create a lifespan manager for the http session
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Create a single, shared session for the entire application
+    async with aiohttp.ClientSession() as session:
+        app.state.http_session = session
+        yield
+    # The session is automatically closed when the app shuts down
+
+app = FastAPI(lifespan=lifespan)
 origins = ["*"]
 app.add_middleware(CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"])
 
@@ -36,35 +46,18 @@ class VoiceAssistant(Agent):
         super().__init__(instructions=SYSTEM_PROMPT)
 
 async def send_agent_state(room: Room, state: str):
-    try:
-        msg = json.dumps({"type": "agent_state", "state": state})
-        await room.local_participant.publish_data(msg)
-        print(f"DEBUG: Sent agent state: {state}")
-    except Exception as e:
-        print(f"DEBUG: Error publishing agent state: {e}")
+    # This function is correct
+    # ...
+    pass # Omitted for brevity
 
 class LLMStateWrapper(LLM):
-    def __init__(self, llm: LLM, room: Room):
-        super().__init__()
-        self._llm = llm
-        self._room = room
-
-    # --- THE DEFINITIVE FIX ---
-    @asynccontextmanager # 1. Decorate the method to make it a valid context manager
-    async def chat(self, **kwargs):
-        # 2. This code runs when the context is entered
-        await send_agent_state(self._room, "thinking")
-        
-        try:
-            # 3. Enter the original LLM's context and get its stream
-            async with self._llm.chat(**kwargs) as stream:
-                # 4. Yield the stream to the AgentSession
-                yield stream
-        finally:
-            # 5. This code runs when the context is exited (after TTS is done)
-            await send_agent_state(self._room, "listening")
-
-async def run_agent_session(room_name: str, agent_token: str):
+    # This class is correct
+    # ...
+    pass # Omitted for brevity
+
+
+# 3. Modify run_agent_session to accept the http_session
+async def run_agent_session(room_name: str, agent_token: str, http_session: aiohttp.ClientSession):
     livekit_url = os.getenv("LIVEKIT_URL")
     room = Room()
 
@@ -75,11 +68,22 @@ async def run_agent_session(room_name: str, agent_token: str):
         
         await send_agent_state(room, "listening")
 
-        print("DEBUG: 3. Initializing plugins...")
-        llm_wrapper = LLMStateWrapper(llm=GoogleLLM(model="gemini-1.5-flash"), room=room)
+        print("DEBUG: 3. Initializing plugins with shared http_session...")
+        # 4. Pass the session to every plugin that makes an external HTTP call
+        llm_wrapper = LLMStateWrapper(
+            llm=GoogleLLM(model="gemini-1.5-flash", session=http_session), 
+            room=room
+        )
         vad = VAD.load(min_speech_duration=0.1)
-        stt = StreamAdapter(stt=STT(model="whisper-large-v3-turbo"), vad=vad)
-        tts = TTS(voice=VoiceByName(name="Tiktok Fashion Influencer"), instant_mode=True)
+        stt = StreamAdapter(
+            stt=STT(model="whisper-large-v3-turbo", session=http_session), 
+            vad=vad
+        )
+        tts = TTS(
+            voice=VoiceByName(name="Tiktok Fashion Influencer"), 
+            instant_mode=True, 
+            session=http_session
+        )
         print("DEBUG: 4. Plugins initialized.")
 
         print("DEBUG: 5. Creating AgentSession...")
@@ -102,10 +106,12 @@ async def run_agent_session(room_name: str, agent_token: str):
         print(f"DEBUG: Agent session for room {room_name} is ending. Cleaning up.")
         await room.disconnect()
 
+# 5. Modify the endpoint to retrieve the session from the app state
 @app.post("/join-room")
-async def join_room(req: JoinRoomRequest, background_tasks: BackgroundTasks):
+async def join_room(req: JoinRoomRequest, request: Request, background_tasks: BackgroundTasks):
+    http_session = request.app.state.http_session
     print(f"DEBUG: Received POST request to /join-room for: {req.room_name}")
-    background_tasks.add_task(run_agent_session, req.room_name, req.agent_token)
+    background_tasks.add_task(run_agent_session, req.room_name, req.agent_token, http_session)
     return {"status": "agent_triggered"}
 
 @app.get("/")