Voice Agent

import uvicorn
from starlette.applications import Starlette
from starlette.routing import Route, WebSocketRoute
from starlette.websockets import WebSocket
from dotenv import load_dotenv

from langchain_openai_voice import OpenAIVoiceReactAgent
from server.utils import websocket_stream
from server.tools import TOOLS
from chatbot import loadCharacters, process_character_config
from server.prompt import BASE_INSTRUCTIONS

load_dotenv(override=True)

# Track active connections
active_connections = 0
MAX_CONNECTIONS_PER_INSTANCE = 10

async def websocket_endpoint(websocket: WebSocket, character_file: str):
    global active_connections
    
    # Connection limiting
    if active_connections >= MAX_CONNECTIONS_PER_INSTANCE:
        await websocket.accept()
        await websocket.send_text(
            json.dumps({
                "type": "error",
                "message": "Server at capacity. Please try again later."
            })
        )
        await websocket.close()
        return
    
    active_connections += 1
    try:
        await websocket.accept()
        browser_receive_stream = websocket_stream(websocket)
        
        # Load character configuration
        character = loadCharacters(character_file)[0]
        personality = process_character_config(character)
        
        # Combine base instructions with character config
        full_instructions = BASE_INSTRUCTIONS.format(
            character_instructions=personality,
            character_name=character["name"],
            adjectives=", ".join(character.get("adjectives", [])),
            topics=", ".join(character.get("topics", [])),
        )
        
        # Create voice agent
        agent = OpenAIVoiceReactAgent(
            model="gpt-4o-realtime-preview",
            tools=TOOLS,
            instructions=full_instructions,
            voice="verse",  # Options: alloy, ash, ballad, coral, echo, sage, shimmer, verse
        )
        
        # Connect agent to WebSocket streams
        await agent.aconnect(browser_receive_stream, websocket.send_text)
    finally:
        active_connections -= 1

async def rolypoly_websocket_endpoint(websocket: WebSocket):
    await websocket_endpoint(websocket, "characters/rolypoly.json")

async def chainyoda_websocket_endpoint(websocket: WebSocket):
    await websocket_endpoint(websocket, "characters/chainyoda.json")

routes = [
    Route("/", homepage),
    Route("/health", health_check),
    WebSocketRoute("/ws", websocket_endpoint),
    WebSocketRoute("/ws/rolypoly", rolypoly_websocket_endpoint),
    WebSocketRoute("/ws/chainyoda", chainyoda_websocket_endpoint),
]

app = Starlette(debug=True, routes=routes)
app.mount("/", StaticFiles(directory="server/src/server/static"), name="static")

BASE_INSTRUCTIONS = """
You are {character_name}

# Character Identity Configuration
{character_instructions}

# Operational Guidelines
You are a helpful assistant with access to tools. Maintain these core principles:

1. Personality Enforcement:
- Speak in English
- Be concise but insightful
- Prioritize character bio/lore/knowledge over general knowledge

2. Tool Usage Priorities:
- When asked about GPUs: Only state model names from get_available_gpus
- For podcast questions ("The Podcast", "The Rollup"): 
  * Use podcast_query_tool first
  * Cite speakers and exact quotes
  * Cross-reference with character knowledge

3. Interaction Style:
- Maintain this tone: {adjectives}
- Adhere strictly to style guidelines
- Focus on these topics: {topics}
"""

import psutil

async def health_check(request):
    # Get system metrics
    memory = psutil.virtual_memory()
    
    # Check if we're in a healthy state
    is_healthy = (
        memory.percent < 90 and 
        active_connections < MAX_CONNECTIONS_PER_INSTANCE
    )
    
    # Return detailed health information
    return JSONResponse(
        {
            "status": "healthy" if is_healthy else "unhealthy",
            "memory": {
                "used_percent": memory.percent,
                "available_mb": memory.available / (1024 * 1024),
            },
            "connections": {
                "active": active_connections,
                "max": MAX_CONNECTIONS_PER_INSTANCE,
            },
        },
        status_code=200 if is_healthy else 503,
    )

# OpenAI (Required for voice agent)
OPENAI_API_KEY=your_openai_api_key

# Character Configuration
CHARACTER_FILE=characters/default.json

# Server Configuration
PORT=3000
HOST=0.0.0.0

# Tool Configuration (same as chat agent)
USE_COINBASE_TOOLS=true
USE_HYPERBOLIC_TOOLS=true
USE_PODCAST_KNOWLEDGE_BASE=true

# Install dependencies
pip install uvicorn starlette langchain-openai-voice psutil

# Start the server
python server/src/server/app.py

const ws = new WebSocket('ws://localhost:3000/ws/chainyoda');

ws.onopen = () => {
  console.log('Connected to voice agent');
  
  // Start audio capture from microphone
  navigator.mediaDevices.getUserMedia({ audio: true })
    .then(stream => {
      const mediaRecorder = new MediaRecorder(stream);
      
      mediaRecorder.ondataavailable = (event) => {
        // Send audio data to server
        ws.send(event.data);
      };
      
      mediaRecorder.start(100); // Send chunks every 100ms
    });
};

ws.onmessage = (event) => {
  const data = JSON.parse(event.data);
  
  if (data.type === 'audio') {
    // Play received audio from agent
    playAudioChunk(data.audio);
  } else if (data.type === 'transcript') {
    console.log('Agent said:', data.text);
  }
};

ws.onerror = (error) => {
  console.error('WebSocket error:', error);
};

from langchain.tools import Tool
from langchain_community.tools import DuckDuckGoSearchRun

TOOLS = [
    DuckDuckGoSearchRun(name="web_search"),
    Tool(
        name="query_podcast_knowledge_base",
        func=lambda q: podcast_kb.query_knowledge_base(q),
        description="Query the podcast knowledge base for insights"
    ),
    # Add more tools as needed
]

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY server/ ./server/
COPY characters/ ./characters/
COPY chatbot.py .

EXPOSE 3000

CMD ["python", "server/src/server/app.py"]

docker build -t voice-agent .
docker run -p 3000:3000 --env-file .env voice-agent