F-10 Audio streaming
Brick F of the Enterprise AI Platform. Technology: LiveKit tracks and low latency audio services (Opus, PCM16 frames of 20 ms) between clients, STT, agents and TTS. French version: ../../fr/bricks/F-10-audio-streaming.md.
Role and responsibilities
- Transport audio frames with jitter buffers, resampling and codec negotiation.
- Expose a WebSocket protocol for clients without WebRTC (servers, telephony bridges).
Position and interactions
- Execution path: realtime transport.
- Events: session.update, input_audio.append, transcript.delta, response.audio.delta, usage.update, error.
Open source and developed parts
- Open source reused: LiveKit, Opus, SIP bridge from LiveKit for telephony.
- Developed in the platform: Event protocol versioning, backpressure, resampling service.
Data model
Control Plane tables (schema eai): eai.realtime_sessions, 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
| Method | Path |
|---|---|
| WebSocket | /v1/realtime/ws/{session_id} |
| POST | /v1/realtime/sessions |
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
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"
}
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"})
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" } });
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")));
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()
_, err = client.Realtime.CreateSession(ctx, map[string]any{"pipeline": "stt", "config": map[string]any{"language": "fr"}})
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, EAI_LIVEKIT_API_SECRET.
Helm values keys (deploy/helm/enterprise-ai/values.yaml): global.networkPolicy.
Security
- WebSocket requires the session token; origin checks for browsers.
- 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
- Frame drops, buffer underruns, end to end latency per hop.
- W3C traceparent propagated, X-Request-Id returned, spans carry eai.tenant_id, eai.project, eai.agent_id, eai.run_id.
FinOps units
- Counted as part of voice_live (duration_ms, bytes_transferred).
Failure modes and resilience
- Client too slow -> server drops oldest frames and sends an error event with code backpressure.
Standalone adoption
Telephony integrators use only the WebSocket protocol with their SIP stack.
Acceptance tests
- Round trip of 1000 frames without loss on a clean network.
- Protocol version mismatch returns an error event, not a silent close.
Kit L4 voice evidence
Response audio streamed in 100 ms response.audio.delta chunks paced at real time, cancelled cleanly by input_audio.speech_started or response.cancel (MULT-010). See the voice runtime guide.