jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Add streaming ASR partial transcripts and reset hosts to env default

Browse Source 
- Stream partial transcripts to the client during dictation for live feedback
- Use OpenAI semantic VAD instead of manual commit intervals for better word boundaries
- Append silence tail on finish to flush final utterance through VAD
- Skip committing silence-only audio tails
- Reset hosts to env-defined daemon when env var is present, overriding stored state
This commit is contained in:
Mohamed Boudra
2026-01-24 19:01:14 +07:00
parent 9b82c02fd2
commit b498c9ea81
8 changed files with 290 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/openai-realtime-transcription.ts
+53 -4
View File
@@ -2,6 +2,21 @@ import type pino from "pino";
import WebSocket from "ws";
import { EventEmitter } from "node:events";
type OpenAITurnDetection =
| null
| {
type: "server_vad";
create_response?: boolean;
threshold?: number;
prefix_padding_ms?: number;
silence_duration_ms?: number;
}
| {
type: "semantic_vad";
create_response?: boolean;
eagerness?: "low" | "medium" | "high";
};
type OpenAIClientEvent =
| {
type: "session.update";
@@ -15,7 +30,7 @@ type OpenAIClientEvent =
language?: string;
prompt?: string;
};
turn_detection: null;
turn_detection: OpenAITurnDetection;
};
};
};
@@ -31,6 +46,13 @@ type OpenAIServerEvent =
item_id: string;
previous_item_id: string | null;
}
| { type: "input_audio_buffer.speech_started" }
| { type: "input_audio_buffer.speech_stopped" }
| {
type: "conversation.item.input_audio_transcription.delta";
item_id: string;
delta: string;
}
| {
type: "conversation.item.input_audio_transcription.completed";
item_id: string;
@@ -43,22 +65,29 @@ export class OpenAIRealtimeTranscriptionSession extends EventEmitter {
private readonly logger: pino.Logger;
private readonly transcriptionModel: string;
private readonly language?: string;
private readonly prompt?: string;
private readonly turnDetection: OpenAITurnDetection;
private ws: WebSocket | null = null;
private ready: Promise<void> | null = null;
private closing = false;
private partialByItemId = new Map<string, string>();
constructor(params: {
apiKey: string;
logger: pino.Logger;
transcriptionModel: string;
language?: string;
prompt?: string;
turnDetection?: OpenAITurnDetection;
}) {
super();
this.apiKey = params.apiKey;
this.logger = params.logger.child({ provider: "openai", component: "realtime-transcription" });
this.transcriptionModel = params.transcriptionModel;
this.language = params.language;
this.prompt = params.prompt;
this.turnDetection = params.turnDetection ?? null;
}
public async connect(): Promise<void> {
@@ -99,9 +128,9 @@ export class OpenAIRealtimeTranscriptionSession extends EventEmitter {
transcription: {
model: this.transcriptionModel,
...(this.language ? { language: this.language } : {}),
...(this.prompt ? { prompt: this.prompt } : {}),
},
// We commit periodically ourselves; no server-side VAD for dictation.
turn_detection: null,
turn_detection: this.turnDetection,
},
},
},
@@ -138,8 +167,28 @@ export class OpenAIRealtimeTranscriptionSession extends EventEmitter {
return;
}
if (event.type === "input_audio_buffer.speech_started") {
this.emit("speech_started");
return;
}
if (event.type === "input_audio_buffer.speech_stopped") {
this.emit("speech_stopped");
return;
}
if (event.type === "conversation.item.input_audio_transcription.delta") {
const replaceDelta = this.transcriptionModel === "whisper-1";
const prev = this.partialByItemId.get(event.item_id) ?? "";
const next = replaceDelta ? event.delta : prev + event.delta;
this.partialByItemId.set(event.item_id, next);
this.emit("transcript", { itemId: event.item_id, transcript: next, isFinal: false });
return;
}
if (event.type === "conversation.item.input_audio_transcription.completed") {
this.emit("transcript", { itemId: event.item_id, transcript: event.transcript });
this.partialByItemId.set(event.item_id, event.transcript);
this.emit("transcript", { itemId: event.item_id, transcript: event.transcript, isFinal: true });
return;
}
packages/server/src/server/session.ts
+141 -54
View File
@@ -141,20 +141,63 @@ const MIN_STREAMING_SEGMENT_BYTES = Math.round(
const DICTATION_PCM_INPUT_RATE = 16000;
const DICTATION_PCM_OUTPUT_RATE = 24000;
const DICTATION_COMMIT_INTERVAL_MS = Number.parseInt(
process.env.OPENAI_REALTIME_DICTATION_COMMIT_MS ?? "5000",
const DICTATION_FINAL_TIMEOUT_MS = 120000;
const DICTATION_SILENCE_PEAK_THRESHOLD = Number.parseInt(
process.env.OPENAI_REALTIME_DICTATION_SILENCE_PEAK_THRESHOLD ?? "300",
10
);
const DICTATION_COMMIT_TARGET_BYTES = Math.round(
(DICTATION_PCM_OUTPUT_RATE * 2 * DICTATION_COMMIT_INTERVAL_MS) / 1000
const DICTATION_TURN_DETECTION = (process.env.OPENAI_REALTIME_DICTATION_TURN_DETECTION ?? "semantic_vad").trim();
const DICTATION_SEMANTIC_VAD_EAGERNESS = (process.env.OPENAI_REALTIME_DICTATION_SEMANTIC_VAD_EAGERNESS ?? "medium").trim();
const DICTATION_FLUSH_SILENCE_MS = Number.parseInt(
process.env.OPENAI_REALTIME_DICTATION_FLUSH_SILENCE_MS ?? "800",
10
);
const DICTATION_MIN_COMMIT_MS = 100;
const DICTATION_MIN_COMMIT_BYTES = Math.round(
(DICTATION_PCM_OUTPUT_RATE * 2 * DICTATION_MIN_COMMIT_MS) / 1000
);
const DICTATION_FINAL_TIMEOUT_MS = 120000;
const SAFE_GIT_REF_PATTERN = /^[A-Za-z0-9._\/-]+$/;
function pcm16lePeakAbs(pcm16le: Buffer): number {
if (pcm16le.length === 0) {
return 0;
}
if (pcm16le.length % 2 !== 0) {
throw new Error(`PCM16 chunk byteLength must be even, got ${pcm16le.length}`);
}
const samples = new Int16Array(
pcm16le.buffer,
pcm16le.byteOffset,
pcm16le.byteLength / 2
);
let peak = 0;
for (let i = 0; i < samples.length; i += 1) {
const v = samples[i]!;
const abs = v < 0 ? -v : v;
if (abs > peak) {
peak = abs;
if (peak >= 32767) {
break;
}
}
}
return peak;
}
function parseDictationTurnDetection():
| null
| { type: "server_vad"; create_response: false; threshold?: number; prefix_padding_ms?: number; silence_duration_ms?: number }
| { type: "semantic_vad"; create_response: false; eagerness?: "low" | "medium" | "high" } {
if (!DICTATION_TURN_DETECTION || DICTATION_TURN_DETECTION === "none" || DICTATION_TURN_DETECTION === "null") {
return null;
}
if (DICTATION_TURN_DETECTION === "server_vad") {
return { type: "server_vad", create_response: false };
}
const eagerness =
DICTATION_SEMANTIC_VAD_EAGERNESS === "low" ||
DICTATION_SEMANTIC_VAD_EAGERNESS === "high"
? (DICTATION_SEMANTIC_VAD_EAGERNESS as "low" | "high")
: ("medium" as const);
return { type: "semantic_vad", create_response: false, eagerness };
}
/**
* Type for present_artifact tool arguments
*/
@@ -278,10 +321,11 @@ export class Session {
nextSeqToForward: number;
ackSeq: number;
bytesSinceCommit: number;
expectedCommits: number;
committedCount: number;
peakSinceCommit: number;
committedItemIds: string[];
transcriptsByItemId: Map<string, string>;
finalTranscriptItemIds: Set<string>;
awaitingFinalCommit: boolean;
finishRequested: boolean;
finalSeq: number | null;
finalTimeout: ReturnType<typeof setTimeout> | null;
@@ -1258,6 +1302,13 @@ export class Session {
this.dictationStreams.delete(dictationId);
}
private emitDictationPartial(dictationId: string, text: string): void {
this.emit({
type: "dictation_stream_partial",
payload: { dictationId, text },
});
}
private maybeFinalizeDictationStream(dictationId: string): void {
const state = this.dictationStreams.get(dictationId);
if (!state) {
@@ -1270,12 +1321,16 @@ export class Session {
if (state.ackSeq < state.finalSeq) {
return;
}
if (state.committedCount < state.expectedCommits) {
if (state.awaitingFinalCommit) {
return;
}
if (state.committedItemIds.length === 0) {
return;
}
const allTranscriptsReady = state.committedItemIds.every((itemId) =>
state.transcriptsByItemId.has(itemId)
state.finalTranscriptItemIds.has(itemId)
);
if (!allTranscriptsReady) {
return;
@@ -1308,12 +1363,17 @@ export class Session {
const transcriptionModel =
process.env.OPENAI_REALTIME_TRANSCRIPTION_MODEL ?? "gpt-4o-transcribe";
const transcriptionPrompt =
process.env.OPENAI_REALTIME_DICTATION_TRANSCRIPTION_PROMPT ??
"Transcribe only what the speaker says. Do not add words. Preserve punctuation and casing. If the audio is silence or non-speech noise, return an empty transcript.";
const openai = new OpenAIRealtimeTranscriptionSession({
apiKey,
logger: this.sessionLogger.child({ dictationId }),
transcriptionModel,
language: "en",
prompt: transcriptionPrompt,
turnDetection: parseDictationTurnDetection(),
});
openai.on("committed", ({ itemId }: { itemId: string }) => {
@@ -1321,23 +1381,46 @@ export class Session {
if (!state) {
return;
}
state.committedCount += 1;
state.committedItemIds.push(itemId);
state.bytesSinceCommit = 0;
state.peakSinceCommit = 0;
// When finishing, we require at least one commit after finish if we flushed pending audio.
if (state.finishRequested && state.awaitingFinalCommit) {
state.awaitingFinalCommit = false;
}
this.maybeFinalizeDictationStream(dictationId);
});
openai.on("transcript", ({ itemId, transcript }: { itemId: string; transcript: string }) => {
const state = this.dictationStreams.get(dictationId);
if (!state) {
return;
openai.on(
"transcript",
({ itemId, transcript, isFinal }: { itemId: string; transcript: string; isFinal: boolean }) => {
const state = this.dictationStreams.get(dictationId);
if (!state) {
return;
}
state.transcriptsByItemId.set(itemId, transcript);
if (isFinal) {
state.finalTranscriptItemIds.add(itemId);
}
const orderedIds = state.committedItemIds.includes(itemId)
? state.committedItemIds
: [...state.committedItemIds, itemId];
const partialText = orderedIds
.map((id) => state.transcriptsByItemId.get(id) ?? "")
.join(" ")
.trim();
this.emitDictationPartial(dictationId, partialText);
this.maybeFinalizeDictationStream(dictationId);
}
state.transcriptsByItemId.set(itemId, transcript);
this.maybeFinalizeDictationStream(dictationId);
});
);
openai.on("error", (err) => {
const error = err instanceof Error ? err.message : String(err);
this.failDictationStream(dictationId, error, true);
const message = err instanceof Error ? err.message : String(err);
this.failDictationStream(dictationId, message, true);
this.cleanupDictationStream(dictationId);
});
@@ -1355,10 +1438,11 @@ export class Session {
nextSeqToForward: 0,
ackSeq: -1,
bytesSinceCommit: 0,
expectedCommits: 0,
committedCount: 0,
peakSinceCommit: 0,
committedItemIds: [],
transcriptsByItemId: new Map(),
finalTranscriptItemIds: new Set(),
awaitingFinalCommit: false,
finishRequested: false,
finalSeq: null,
finalTimeout: null,
@@ -1403,20 +1487,14 @@ export class Session {
if (resampled.length > 0) {
state.openai.appendPcm16Base64(resampled.toString("base64"));
state.bytesSinceCommit += resampled.length;
state.peakSinceCommit = Math.max(state.peakSinceCommit, pcm16lePeakAbs(resampled));
}
state.nextSeqToForward += 1;
state.ackSeq = state.nextSeqToForward - 1;
if (state.bytesSinceCommit >= DICTATION_COMMIT_TARGET_BYTES) {
state.expectedCommits += 1;
state.openai.commit();
state.bytesSinceCommit = 0;
}
}
this.emitDictationAck(msg.dictationId, state.ackSeq);
this.maybeFinalizeDictationStream(msg.dictationId);
}
private async handleDictationStreamFinish(
@@ -1441,8 +1519,7 @@ export class Session {
ackSeq: state.ackSeq,
nextSeqToForward: state.nextSeqToForward,
receivedChunks: state.receivedChunks.size,
expectedCommits: state.expectedCommits,
committedCount: state.committedCount,
bytesSinceCommit: state.bytesSinceCommit,
},
"Dictation finish: no chunks received (failing fast)"
);
@@ -1455,31 +1532,41 @@ export class Session {
return;
}
// Commit any remaining uncommitted audio.
// IMPORTANT: OpenAI requires at least ~100ms of audio in the buffer for a commit.
// Also, committing an empty buffer throws "buffer too small ... 0.00ms".
// Force-flush any remaining audio so we get a final transcription even if VAD hasn't committed yet.
if (state.bytesSinceCommit > 0) {
this.sessionLogger.debug(
{ dictationId: msg.dictationId, bytesSinceCommit: state.bytesSinceCommit },
"Dictation finish: committing pending audio"
);
if (state.bytesSinceCommit < DICTATION_MIN_COMMIT_BYTES) {
const padBytes = DICTATION_MIN_COMMIT_BYTES - state.bytesSinceCommit;
if (state.peakSinceCommit < DICTATION_SILENCE_PEAK_THRESHOLD) {
this.sessionLogger.debug(
{ dictationId: msg.dictationId, padBytes },
"Dictation finish: padding to minimum commit size"
{
dictationId: msg.dictationId,
bytesSinceCommit: state.bytesSinceCommit,
peakSinceCommit: state.peakSinceCommit,
},
"Dictation finish: clearing silence-only tail (skip final commit)"
);
state.openai.appendPcm16Base64(Buffer.alloc(padBytes).toString("base64"));
state.bytesSinceCommit += padBytes;
state.openai.clear();
state.bytesSinceCommit = 0;
state.peakSinceCommit = 0;
state.awaitingFinalCommit = false;
} else {
// We do not manually commit for dictation (to avoid chopping words). Instead we rely on
// OpenAI semantic VAD to commit utterances. When the user presses OK, we append a short
// silence tail so VAD has enough silence to close out the final utterance promptly.
const silenceBytes = Math.max(
0,
Math.round((DICTATION_PCM_OUTPUT_RATE * 2 * DICTATION_FLUSH_SILENCE_MS) / 1000)
);
if (silenceBytes > 0) {
this.sessionLogger.debug(
{ dictationId: msg.dictationId, silenceMs: DICTATION_FLUSH_SILENCE_MS, silenceBytes },
"Dictation finish: appending silence tail for semantic VAD flush"
);
state.openai.appendPcm16Base64(Buffer.alloc(silenceBytes).toString("base64"));
state.bytesSinceCommit += silenceBytes;
}
state.awaitingFinalCommit = true;
}
state.expectedCommits += 1;
state.openai.commit();
state.bytesSinceCommit = 0;
} else {
this.sessionLogger.debug(
{ dictationId: msg.dictationId },
"Dictation finish: no pending audio to commit"
);
state.awaitingFinalCommit = false;
}
if (state.finalTimeout) {
packages/server/src/shared/messages.ts
+9
View File
@@ -813,6 +813,14 @@ export const DictationStreamAckMessageSchema = z.object({
}),
});
export const DictationStreamPartialMessageSchema = z.object({
type: z.literal("dictation_stream_partial"),
payload: z.object({
dictationId: z.string(),
text: z.string(),
}),
});
export const DictationStreamFinalMessageSchema = z.object({
type: z.literal("dictation_stream_final"),
payload: z.object({
@@ -1324,6 +1332,7 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [
AudioOutputMessageSchema,
TranscriptionResultMessageSchema,
DictationStreamAckMessageSchema,
DictationStreamPartialMessageSchema,
DictationStreamFinalMessageSchema,
DictationStreamErrorMessageSchema,
StatusMessageSchema,