Voice Agent (Electron + Pulse + Lightning)

import asyncio
import json
import os
import websockets
from openai import OpenAI
SMALLEST = os.environ["SMALLEST_API_KEY"]
client = OpenAI(
    base_url="https://api.smallest.ai/waves/v1",
    api_key=SMALLEST,
)
SYSTEM_PROMPT = """You are a friendly customer support agent for Acme Corp.
Keep responses concise — under three sentences.
Use the get_account_info tool when the user asks about their account.
"""
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_account_info",
            "description": "Look up account information for the current caller.",
            "parameters": {
                "type": "object",
                "properties": {
                    "account_id": {"type": "string"},
                },
                "required": ["account_id"],
            },
        },
    }
]
conversation = [{"role": "system", "content": SYSTEM_PROMPT}]
async def transcribe_loop(audio_in, transcripts_out):
    """Stream audio to Pulse, push final transcripts onto the queue."""
    url = "wss://api.smallest.ai/waves/v1/pulse/get_text?language=en"
    async with websockets.connect(
        url, additional_headers={"Authorization": f"Bearer {SMALLEST}"}
    ) as ws:
        async def send():
            async for chunk in audio_in:
                await ws.send(chunk)
        async def recv():
            async for msg in ws:
                event = json.loads(msg)
                # Pulse WS frames: {"type": "transcription", "is_final": true|false, "transcript": "..."}
                # See /models/documentation/speech-to-text-pulse/realtime-web-socket/response-format
                if event.get("type") == "transcription" and event.get("is_final"):
                    await transcripts_out.put(event["transcript"])
        await asyncio.gather(send(), recv())
async def respond(user_text):
    """Send to Electron, stream tokens. Returns the assistant message + any tool calls."""
    conversation.append({"role": "user", "content": user_text})
    stream = client.chat.completions.create(
        model="electron",
        messages=conversation,
        tools=TOOLS,
        stream=True,
        stream_options={"include_usage": True},
    )
    content = ""
    tool_calls = []   # accumulated
    for chunk in stream:
        if not chunk.choices:
            continue
        delta = chunk.choices[0].delta
        if delta.content:
            content += delta.content
            # 🎙️ feed delta.content to Lightning for TTS immediately
            await tts_speak_delta(delta.content)
        if delta.tool_calls:
            # accumulate tool_calls (function name + arguments stream as deltas)
            accumulate_tool_calls(tool_calls, delta.tool_calls)
    # Whatever Electron emitted gets recorded into history.
    assistant_msg = {"role": "assistant", "content": content or None}
    if tool_calls:
        assistant_msg["tool_calls"] = tool_calls
    conversation.append(assistant_msg)
    return assistant_msg
async def handle_tools(assistant_msg):
    """Run any tool calls, append results, re-prompt Electron."""
    if not assistant_msg.get("tool_calls"):
        return
    for call in assistant_msg["tool_calls"]:
        args = json.loads(call["function"]["arguments"])
        result = run_tool(call["function"]["name"], args)   # your implementation
        conversation.append({
            "role": "tool",
            "tool_call_id": call["id"],
            "content": json.dumps(result),
        })
    # Re-prompt with tool results — this gives the final spoken response.
    await respond_continuation()
async def respond_continuation():
    stream = client.chat.completions.create(
        model="electron",
        messages=conversation,
        tools=TOOLS,
        stream=True,
        stream_options={"include_usage": True},
    )
    content = ""
    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            delta = chunk.choices[0].delta.content
            content += delta
            await tts_speak_delta(delta)
    conversation.append({"role": "assistant", "content": content})
async def turn_loop(audio_in):
    """One iteration per user utterance."""
    transcripts = asyncio.Queue()
    asyncio.create_task(transcribe_loop(audio_in, transcripts))
    while True:
        user_text = await transcripts.get()
        assistant_msg = await respond(user_text)
        await handle_tools(assistant_msg)