Realtime sessions

higgs-avatar

https://api.boson.ai/v1/avatar/sessions

import asyncio
import base64
import json
import os
from datetime import timedelta

import requests
import websockets
from livekit import api, rtc
from livekit.agents.voice.avatar import DataStreamAudioOutput

ROOM = "avatar-demo"
AGENT_IDENTITY = "agent"          # this script: the participant the avatar listens to
AVATAR_IDENTITY = "boson-avatar"  # the participant the avatar joins as
AVATAR_ID = "builtin_claire_v1"
MAX_DURATION_SECONDS = 1800

BOSON = {"Authorization": f"Bearer {os.environ['BOSON_API_KEY']}"}
BOSON_BASE_URL = os.environ.get("BOSON_BASE_URL", "https://api.boson.ai/v1")
BOSON_REALTIME_URL = os.environ.get("BOSON_REALTIME_URL", "wss://api.boson.ai/v1/realtime")


def livekit_token(identity, *, publish, subscribe, publish_on_behalf=None):
    token = (
        api.AccessToken(os.environ["LIVEKIT_API_KEY"], os.environ["LIVEKIT_API_SECRET"])
        .with_identity(identity)
        .with_grants(
            api.VideoGrants(
                room_join=True,
                room=ROOM,
                can_publish=publish,
                can_subscribe=subscribe,
                can_publish_data=True,
                can_publish_sources=["camera", "microphone"] if publish else None,
            )
        )
        .with_ttl(timedelta(seconds=MAX_DURATION_SECONDS + 300))
    )
    if publish_on_behalf:  # optional: LiveKit UIs then show the avatar as part of the agent
        token = token.with_kind("agent").with_attributes({"lk.publish_on_behalf": publish_on_behalf})
    return token.to_jwt()


async def main():
    # 1. Put the avatar in the room. It is there when this returns.
    r = requests.post(
        f"{BOSON_BASE_URL}/avatar/sessions",
        headers=BOSON,
        json={
            "avatar_id": AVATAR_ID,
            "transport": {
                "type": "livekit",
                "url": os.environ["LIVEKIT_URL"],
                "room_name": ROOM,
                "participant_token": livekit_token(
                    AVATAR_IDENTITY, publish=True, subscribe=False, publish_on_behalf=AGENT_IDENTITY
                ),
                "participant_identity": AVATAR_IDENTITY,
                "publisher_identity": AGENT_IDENTITY,
            },
            "max_duration_seconds": MAX_DURATION_SECONDS,
        },
        timeout=60,
    )
    r.raise_for_status()
    session = r.json()
    print("avatar session", session["id"], session["status"], flush=True)

    # 2. Join the room as the agent. Speech goes to the avatar over a data
    #    stream; the avatar publishes it as its own audio track, in sync with
    #    the video, so the agent itself never publishes audio.
    room = rtc.Room()
    await room.connect(os.environ["LIVEKIT_URL"], livekit_token(AGENT_IDENTITY, publish=True, subscribe=True))
    to_avatar = DataStreamAudioOutput(
        room, destination_identity=AVATAR_IDENTITY, sample_rate=24000, wait_remote_track=rtc.TrackKind.KIND_AUDIO
    )

    try:
        # 3. Talk to Higgs Realtime: the user's microphone in, the model's speech out.
        async with websockets.connect(BOSON_REALTIME_URL, additional_headers=BOSON) as ws:
            await ws.send(json.dumps({
                "type": "session.update",
                "session": {
                    "model": "higgs-realtime",
                    "instructions": "You are a friendly assistant with a face. Keep answers short.",
                    "audio": {
                        "input": {"turn_detection": {"type": "server_vad"}},
                        "output": {"voice": "default"},
                    },
                },
            }))

            async def forward_user_audio(track):
                # 24 kHz, 16-bit mono PCM: what the model expects, base64 in JSON.
                async for event in rtc.AudioStream(track, sample_rate=24000, num_channels=1):
                    try:
                        await ws.send(json.dumps({
                            "type": "input_audio_buffer.append",
                            "audio": base64.b64encode(bytes(event.frame.data)).decode(),
                        }))
                    except websockets.ConnectionClosed:
                        return  # the conversation is over (Ctrl-C, or the server closed it)

            @room.on("track_subscribed")
            def on_track(track, _publication, participant):
                if participant.identity != AVATAR_IDENTITY and track.kind == rtc.TrackKind.KIND_AUDIO:
                    asyncio.create_task(forward_user_audio(track))

            for participant in room.remote_participants.values():  # users already in the room
                for publication in participant.track_publications.values():
                    if publication.track is not None:
                        on_track(publication.track, publication, participant)

            speaking = False  # is the avatar still playing something we sent?

            @to_avatar.on("playback_finished")
            def on_played(_event):
                nonlocal speaking
                speaking = False

            async for message in ws:
                event = json.loads(message)
                if event["type"] == "response.output_audio.delta":
                    pcm = base64.b64decode(event["delta"])
                    await to_avatar.capture_frame(rtc.AudioFrame(pcm, 24000, 1, len(pcm) // 2))
                    speaking = True
                elif event["type"] == "response.done":
                    to_avatar.flush()  # end of this utterance; the avatar reports when it has played it
                elif event["type"] == "input_audio_buffer.speech_started" and speaking:
                    to_avatar.clear_buffer()  # the user barged in: stop the avatar talking
                    speaking = False
                elif event["type"] == "response.output_audio_transcript.done":
                    print("avatar said:", event["transcript"], flush=True)
                elif event["type"] == "error":
                    print("realtime error:", event, flush=True)
    finally:
        # 4. Release the avatar when the conversation ends.
        requests.delete(f"{BOSON_BASE_URL}/avatar/sessions/{session['id']}", headers=BOSON, timeout=30)
        await room.disconnect()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass  # Ctrl-C: the session was released on the way out

<!doctype html>
<meta charset="utf-8" />
<title>Realtime avatar</title>
<button id="join">Join</button>
<video id="avatar" autoplay playsinline style="display:block;width:480px;height:640px;background:#000"></video>
<script src="https://unpkg.com/[email protected]/dist/livekit-client.umd.js"></script>
<script>
  const params = new URLSearchParams(location.search);
  const room = new LivekitClient.Room();
  room.on(LivekitClient.RoomEvent.TrackSubscribed, (track, _publication, participant) => {
    if (participant.identity !== "boson-avatar") return;
    if (track.kind === "video") track.attach(document.getElementById("avatar"));
    else document.body.appendChild(track.attach()); // the avatar's voice
  });
  document.getElementById("join").onclick = async () => {
    await room.connect(params.get("url"), params.get("token"));
    await room.localParticipant.setMicrophoneEnabled(true);
  };
</script>