Skip to main content

F-08 Voice live (duplex)

Brick F of the Enterprise AI Platform. Technology: LiveKit (SFU, agents framework) + WebRTC duplex, WebSocket fallback /v1/realtime/ws/{session_id}. French version: ../../fr/bricks/F-08-voice-live.md.

Role and responsibilities​

  • Full duplex voice conversations between a user and an agent: barge-in, turn detection, low latency.
  • Pipeline: VAD (F-09) -> STT streaming (F-05) -> agent/LLM -> TTS streaming (F-07), or native speech to speech models through realtime adapters.

Position and interactions​

  • Execution path: realtime duplex session.
  • POST /v1/realtime/sessions issues a short lived LiveKit token scoped to one room and takes a quota reservation for the max duration.
  • usage.update events stream consumption during the call; POST /v1/realtime/sessions/{id}/end commits the reservation.

Open source and developed parts​

  • Open source reused: LiveKit server, LiveKit agents, WebRTC.
  • Developed in the platform: Session service, token issuance, event protocol, budget enforcement mid-call, transcript persistence policy.

Data model​

Control Plane tables (schema eai): eai.realtime_sessions, eai.usage_events, eai.quota_reservations, eai.telemetry_spans.

Every tenant table has tenant_id uuid NOT NULL and a row level security policy on current_setting('app.tenant_id', true)::uuid. ClickHouse usage_events is the analytics replica of eai.usage_events (ReplacingMergeTree keyed by event_id).

Contract: contracts/v1/multimodal-contract.schema.json (schema_version 1.0). Usage is always reported with contracts/v1/usage-event.schema.json, audit with contracts/v1/audit-event.schema.json.

API endpoints​

MethodPath
POST/v1/realtime/sessions
WebSocket/v1/realtime/ws/{session_id}
POST/v1/realtime/sessions/{id}/end

POST /v1/realtime/sessions​

Request:

{
  "room": "support-voice-42",
  "pipeline": "stt",
  "ttl_seconds": 900,
  "config": {
    "language": "fr",
    "sample_rate": 16000,
    "dlp_mode": "mask"
  }
}

Response:

{
  "session_id": "rts_3K",
  "livekit_url": "wss://<livekit-host>",
  "token": "eyJhbGciOi...",
  "ws_url": "wss://<control-plane-host>/v1/realtime/ws/rts_3K",
  "expires_at": "2026-10-29T10:15:00Z"
}

WebSocket /v1/realtime/ws/{session_id}​

Request:

{
  "type": "input_audio.append",
  "audio": "<base64 pcm16 20ms>"
}

Response:

{
  "type": "transcript.delta",
  "text": "Bonjour",
  "final": false
}

POST /v1/realtime/sessions/{id}/end​

No request body.

Response:

{
  "session_id": "rts_3K",
  "status": "ended",
  "usage": {
    "audio_seconds": 312.4,
    "input_tokens": 2210,
    "output_tokens": 1650,
    "audio_tokens": 7800,
    "sessions": 1
  }
}

SDK examples​

Python​

from enterprise_ai import EnterpriseAI

client = EnterpriseAI(base_url="<control-plane-url>", api_key="eai_live_xxx", tenant="acme", locale="en")
session = client.realtime.create_session("stt", config={"language": "fr"})
client.realtime.end_session(session["session_id"])

TypeScript​

import { EnterpriseAI } from "@enterprise-ai/sdk";

const client = new EnterpriseAI({ baseUrl: "<control-plane-url>", apiKey: process.env.EAI_API_KEY, tenant: "acme", locale: "en" });
const session = await client.realtime.createSession({ pipeline: "stt", config: { language: "fr" } });
await client.realtime.endSession(session.session_id as string);

Java​

EnterpriseAIClient client = EnterpriseAIClient.builder()
    .baseUrl("<control-plane-url>")
    .apiKey(System.getenv("EAI_API_KEY"))
    .tenant("acme")
    .locale(Locale.ENGLISH)
    .build();
JsonNode session = client.realtime().createSession(Map.of("pipeline", "stt", "config", Map.of("language", "fr")));
client.realtime().endSession(session.get("session_id").asText());

Go​

client, err := eai.NewClient(eai.WithBaseURL("<control-plane-url>"), eai.WithAPIKey(os.Getenv("EAI_API_KEY")), eai.WithTenant("acme"), eai.WithLocale("en"))
if err != nil { log.Fatal(err) }
ctx := context.Background()
session, err := client.Realtime.CreateSession(ctx, map[string]any{"pipeline": "stt", "config": map[string]any{"language": "fr"}})
var s struct{ SessionID string `json:"session_id"` }
_ = json.Unmarshal(session, &s)
_, err = client.Realtime.EndSession(ctx, s.SessionID)

TypeScript realtime client (@enterprise-ai/realtime)​

import { EnterpriseAI } from "@enterprise-ai/sdk";
import { RealtimeClient } from "@enterprise-ai/realtime";

const client = new EnterpriseAI({ baseUrl: "<control-plane-url>", apiKey: process.env.EAI_API_KEY, tenant: "acme", locale: "fr" });
const rtClient = new RealtimeClient({ baseUrl: "<control-plane-url>", apiKey: process.env.EAI_API_KEY, tenant: "acme", locale: "fr" });
const s = await rtClient.createSession({ pipeline: "stt" });
const rt = await rtClient.connect(s, { language: "fr", turn_detection: { type: "server_vad" } });
rt.on("transcript.delta", (e) => render(e.delta));
rt.on("usage.update", (e) => showBudget(e.usage));
rt.on("error", (e) => console.error(e.error.code, e.error.message));
rt.appendAudio(pcm20ms, 16000);
rt.close();
await rtClient.endSession(s.session_id);

Go realtime package​

raw, err := client.Realtime.CreateSession(ctx, map[string]any{"pipeline": "stt"})
if err != nil { return err }
var s struct{ SessionID string `json:"session_id"` }
if err := json.Unmarshal(raw, &s); err != nil { return err }
conn, err := realtime.Dial(ctx, client, s.SessionID)
if err != nil { return err }
defer conn.Close()
_ = conn.UpdateSession(map[string]any{"language": "fr"})
_ = conn.AppendAudio(pcm20ms)
for {
    ev, err := conn.Recv()
    if err != nil { return err }
    if ev.Type == realtime.TranscriptDelta {
        var d struct{ Delta string `json:"delta"` }
        _ = ev.Decode(&d)
        log.Println(d.Delta)
    }
}

Configuration​

Environment variables: EAI_LIVEKIT_URL, EAI_LIVEKIT_API_KEY.

Helm values keys (deploy/helm/enterprise-ai/values.yaml): controlPlane.config.EAI_LIVEKIT_URL, controlPlane.secrets, global.networkPolicy.

Security​

  • Tokens expire quickly and grant only publish/subscribe in the session room; SRTP/DTLS for media.
  • Recording is off by default; consent prompt is required by policy when enabled.
  • Tenant resolved from the credential, never from the body; PostgreSQL row level security on every tenant table.
  • Every mutation and policy decision writes a hash chained audit event (G-11).
  • Errors use the standard envelope with message_i18n en and fr.

Observability​

  • Mouth to ear latency, jitter, packet loss and interruption counts per session.
  • W3C traceparent propagated, X-Request-Id returned, spans carry eai.tenant_id, eai.project, eai.agent_id, eai.run_id.

FinOps units​

  • usage_type voice_live: audio_seconds, audio_tokens, input/output text tokens, sessions=1, duration_ms.

Failure modes and resilience​

  • Quota reached during the call -> warning event then graceful end with a spoken message.
  • WebRTC blocked by the network -> TURN over TLS 443, then WebSocket fallback.

Standalone adoption​

A client of a client embeds a voice agent with @enterprise-ai/realtime in its web app using only realtime and FinOps bricks.

Acceptance tests​

  1. p95 time from end of user speech to first agent audio below 800 ms on the reference network.
  2. Barge-in stops TTS playback within 200 ms.
  3. Ending a session commits exactly one Usage Event with sessions=1.

WebSocket security​

  • ws_token is short lived (EAI_REALTIME_WS_TOKEN_TTL_SECONDS, 300 s by default), single use and bound to the tenant, the session and the identity. Its signature and expiry are checked before any database read, then the session is read with the principal, under RLS. Invalid, expired, replayed or other tenant token: close 4401.
  • POST /v1/realtime/sessions/{id}/token issues a new token for a reconnection (409 session_ended when the session has ended). The TypeScript Realtime SDK calls it before every automatic reconnection.
  • Limits: message size EAI_REALTIME_WS_MAX_MESSAGE_BYTES (1 MiB, also uvicorn ws_max_size, close 1009); rate EAI_REALTIME_WS_MAX_EVENTS_PER_SECOND (50) with burst EAI_REALTIME_WS_BURST (100), close 4429; duration EAI_REALTIME_WS_MAX_SESSION_SECONDS (3600) and idle EAI_REALTIME_WS_IDLE_TIMEOUT_SECONDS (120), close 4408.
  • The connection reserves the cost of the maximum duration on the quotas (4429 quota_exceeded otherwise); session.end commits the measured amount, a disconnection releases the reservation.

Implementation roadmap

Kit L4 voice evidence​

Consent per participant, identity (participant, user, agent, trace) in every turn, barge-in, real LiveKit 1.13.11 server: token validated, room joined over signalling, ListParticipants (MULT-001, MULT-002 server side, MULT-008, MULT-012, RELE-011). WebRTC media not tested. See the voice runtime guide.