puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-07-29 12:01:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update files

Browse Source
This commit is contained in:
Mohamed Boudra
2026-02-09 13:33:40 +07:00
parent 015c0d7772
commit 5c89ae52ce
8 changed files with 382 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
packages/server/src/server/agent/stt-manager.test.ts
View File

@@ -51,6 +51,49 @@ class FakeStt implements SpeechToTextProvider {
}
}
class SequencedFakeStt implements SpeechToTextProvider {
public readonly id = "fake-sequenced";
constructor(private readonly transcripts: string[]) {}
createSession(_params: {
logger: any;
language?: string;
prompt?: string;
}): StreamingTranscriptionSession {
const emitter = new EventEmitter();
const transcripts = this.transcripts;
let segmentId = "seg-1";
let previousSegmentId: string | null = null;
let idx = 0;
return {
requiredSampleRate: 24000,
async connect() {},
appendPcm16() {},
commit() {
const transcript = transcripts[idx] ?? "";
idx += 1;
(emitter as any).emit("committed", { segmentId, previousSegmentId });
(emitter as any).emit("transcript", {
segmentId,
transcript,
isFinal: true,
language: "en",
isLowConfidence: transcript.length === 0,
});
previousSegmentId = segmentId;
segmentId = `seg-${idx + 1}`;
},
clear() {},
close() {},
on(event: any, handler: any) {
emitter.on(event, handler);
return undefined;
},
};
}
}
describe("STTManager", () => {
it("returns empty text for low-confidence transcriptions", async () => {
const manager = new STTManager(
@@ -79,4 +122,30 @@ describe("STTManager", () => {
expect(result.language).toBe("en");
expect(result.byteLength).toBe(4);
});
it("uses streaming segmentation for batch transcription and concatenates segment finals", async () => {
const original = process.env.PASEO_STT_BATCH_COMMIT_EVERY_SECONDS;
process.env.PASEO_STT_BATCH_COMMIT_EVERY_SECONDS = "1";
try {
const manager = new STTManager(
"s1",
pino({ level: "silent" }),
new SequencedFakeStt(["alpha", "beta", "gamma"])
);
const threeSecondsPcm = Buffer.alloc(24000 * 2 * 3);
const result = await manager.transcribe(threeSecondsPcm, "audio/pcm;rate=24000");
expect(result.text).toBe("alpha beta gamma");
expect(result.language).toBe("en");
expect(result.byteLength).toBe(threeSecondsPcm.length);
} finally {
if (original === undefined) {
delete process.env.PASEO_STT_BATCH_COMMIT_EVERY_SECONDS;
} else {
process.env.PASEO_STT_BATCH_COMMIT_EVERY_SECONDS = original;
}
}
});
});

204
packages/server/src/server/agent/stt-manager.ts
View File

@@ -10,6 +10,22 @@ interface TranscriptionMetadata {
label?: string;
}
const BATCH_APPEND_CHUNK_SECONDS = 1;
const DEFAULT_BATCH_COMMIT_EVERY_SECONDS = 15;
const BATCH_FINAL_TIMEOUT_MS = 120_000;
function resolveBatchCommitEverySeconds(): number {
const fromEnv = process.env.PASEO_STT_BATCH_COMMIT_EVERY_SECONDS;
if (!fromEnv) {
return DEFAULT_BATCH_COMMIT_EVERY_SECONDS;
}
const parsed = Number.parseFloat(fromEnv);
if (!Number.isFinite(parsed) || parsed < 0) {
return DEFAULT_BATCH_COMMIT_EVERY_SECONDS;
}
return parsed;
}
export interface SessionTranscriptionResult extends TranscriptionResult {
debugRecordingPath?: string;
byteLength: number;
@@ -97,38 +113,168 @@ export class STTManager {
try {
const startedAt = Date.now();
const finalEventPromise = new Promise<{
transcript: string;
language?: string;
logprobs?: TranscriptionResult["logprobs"];
avgLogprob?: number;
isLowConfidence?: boolean;
}>((resolve, reject) => {
session.on("error", reject);
session.on("transcript", (payload) => {
if (!payload.isFinal) {
return;
}
resolve({
transcript: payload.transcript,
language: payload.language,
logprobs: payload.logprobs,
avgLogprob: payload.avgLogprob,
isLowConfidence: payload.isLowConfidence,
});
});
await session.connect();
const committedSegmentIds: string[] = [];
const transcriptsBySegmentId = new Map<string, string>();
const finalTranscriptSegmentIds = new Set<string>();
const transcriptMetaBySegmentId = new Map<
string,
{
language?: string;
logprobs?: TranscriptionResult["logprobs"];
avgLogprob?: number;
isLowConfidence?: boolean;
}
>();
let expectedFinals = 0;
let settle: (() => void) | null = null;
let fail: ((error: Error) => void) | null = null;
let settled = false;
const allFinalsReady = new Promise<void>((resolve, reject) => {
settle = resolve;
fail = reject;
});
await session.connect();
session.appendPcm16(pcmForModel);
session.commit();
const finalEvent = await finalEventPromise;
const resolveIfComplete = () => {
if (settled) {
return;
}
if (expectedFinals > 0 && finalTranscriptSegmentIds.size >= expectedFinals) {
settled = true;
settle?.();
return;
}
if (expectedFinals === 0 && finalTranscriptSegmentIds.size > 0) {
settled = true;
settle?.();
}
};
const rejectWith = (error: Error) => {
if (settled) {
return;
}
settled = true;
fail?.(error);
};
session.on("error", (error) => {
rejectWith(error instanceof Error ? error : new Error(String(error)));
});
session.on("committed", ({ segmentId }) => {
committedSegmentIds.push(segmentId);
expectedFinals += 1;
resolveIfComplete();
});
session.on("transcript", (payload) => {
transcriptsBySegmentId.set(payload.segmentId, payload.transcript);
if (!payload.isFinal) {
return;
}
finalTranscriptSegmentIds.add(payload.segmentId);
transcriptMetaBySegmentId.set(payload.segmentId, {
language: payload.language,
logprobs: payload.logprobs,
avgLogprob: payload.avgLogprob,
isLowConfidence: payload.isLowConfidence,
});
resolveIfComplete();
});
const appendChunkBytes = Math.max(
1,
Math.round(session.requiredSampleRate * 2 * BATCH_APPEND_CHUNK_SECONDS)
);
const commitEverySeconds = resolveBatchCommitEverySeconds();
const commitEveryBytes =
commitEverySeconds > 0
? Math.max(1, Math.round(session.requiredSampleRate * 2 * commitEverySeconds))
: 0;
let bytesSinceCommit = 0;
for (let offset = 0; offset < pcmForModel.length; offset += appendChunkBytes) {
const chunk = pcmForModel.subarray(
offset,
Math.min(pcmForModel.length, offset + appendChunkBytes)
);
if (chunk.length === 0) {
continue;
}
session.appendPcm16(chunk);
bytesSinceCommit += chunk.length;
if (commitEveryBytes > 0 && bytesSinceCommit >= commitEveryBytes) {
session.commit();
bytesSinceCommit = 0;
}
}
if (bytesSinceCommit > 0 || expectedFinals === 0) {
session.commit();
}
const finalTimeout = setTimeout(() => {
if (settled) {
return;
}
settled = true;
this.logger.warn(
{
expectedFinals,
receivedFinals: finalTranscriptSegmentIds.size,
label: metadata?.label,
},
"Timed out waiting for final STT segments; returning available transcripts"
);
settle?.();
}, BATCH_FINAL_TIMEOUT_MS);
await allFinalsReady;
clearTimeout(finalTimeout);
const committedSet = new Set(committedSegmentIds);
const orderedSegmentIds: string[] = [...committedSegmentIds];
for (const segmentId of transcriptsBySegmentId.keys()) {
if (!committedSet.has(segmentId)) {
orderedSegmentIds.push(segmentId);
}
}
const transcript = orderedSegmentIds
.map((segmentId) => transcriptsBySegmentId.get(segmentId) ?? "")
.join(" ")
.trim();
const orderedFinalMeta = orderedSegmentIds
.filter((segmentId) => finalTranscriptSegmentIds.has(segmentId))
.map((segmentId) => transcriptMetaBySegmentId.get(segmentId))
.filter(
(
meta
): meta is {
language?: string;
logprobs?: TranscriptionResult["logprobs"];
avgLogprob?: number;
isLowConfidence?: boolean;
} => Boolean(meta)
);
const language = orderedFinalMeta.find((meta) => meta.language)?.language;
const singleSegmentMeta = orderedFinalMeta.length === 1 ? orderedFinalMeta[0] : null;
const allLowConfidence =
orderedFinalMeta.length > 0 &&
orderedFinalMeta.every((meta) => meta.isLowConfidence === true);
const result: TranscriptionResult = {
text: finalEvent.transcript,
language: finalEvent.language,
logprobs: finalEvent.logprobs,
avgLogprob: finalEvent.avgLogprob,
isLowConfidence: finalEvent.isLowConfidence,
text: transcript,
...(language ? { language } : {}),
...(singleSegmentMeta?.logprobs ? { logprobs: singleSegmentMeta.logprobs } : {}),
...(singleSegmentMeta?.avgLogprob !== undefined
? { avgLogprob: singleSegmentMeta.avgLogprob }
: {}),
...(allLowConfidence ? { isLowConfidence: true } : {}),
duration: Date.now() - startedAt,
};

53
packages/server/src/server/dictation/dictation-stream-manager.test.ts
View File

@@ -149,4 +149,57 @@ describe("DictationStreamManager (provider-agnostic provider)", () => {
}
}
});
it("auto-commits while streaming and assembles final transcript in segment order", async () => {
const originalDebug = process.env.PASEO_DICTATION_DEBUG;
process.env.PASEO_DICTATION_DEBUG = "false";
try {
const session = new FakeRealtimeSession();
const emitted: Array<{ type: string; payload: any }> = [];
const manager = new DictationStreamManager({
logger: pino({ level: "silent" }),
emit: (msg) => emitted.push(msg),
sessionId: "s1",
stt: new FakeSttProvider(session),
autoCommitSeconds: 1,
});
await manager.handleStart("d-segmented", "audio/pcm;rate=24000;bits=16");
await manager.handleChunk({
dictationId: "d-segmented",
seq: 0,
audioBase64: buildPcmBase64(2000, 24000),
format: "audio/pcm;rate=24000;bits=16",
});
expect(session.commitCalls).toBe(1);
session.emitCommitted("seg-1");
session.emitTranscript("seg-1", "hello", true);
await manager.handleChunk({
dictationId: "d-segmented",
seq: 1,
audioBase64: buildPcmBase64(2000, 12000),
format: "audio/pcm;rate=24000;bits=16",
});
await manager.handleFinish("d-segmented", 1);
expect(session.commitCalls).toBe(2);
session.emitCommitted("seg-2");
session.emitTranscript("seg-2", "world", true);
await tick();
const final = emitted.find((msg) => msg.type === "dictation_stream_final");
expect(final?.payload.text).toBe("hello world");
} finally {
if (originalDebug === undefined) {
delete process.env.PASEO_DICTATION_DEBUG;
} else {
process.env.PASEO_DICTATION_DEBUG = originalDebug;
}
}
});
});

67
packages/server/src/server/dictation/dictation-stream-manager.ts
View File

@@ -16,11 +16,23 @@ import { parsePcmRateFromFormat, pcm16lePeakAbs } from "../speech/audio.js";
const PCM_CHANNELS = 1;
const PCM_BITS_PER_SAMPLE = 16;
const DEFAULT_DICTATION_FINAL_TIMEOUT_MS = 10000;
const DEFAULT_DICTATION_AUTO_COMMIT_SECONDS = 15;
const DICTATION_SILENCE_PEAK_THRESHOLD = Number.parseInt(
process.env.PASEO_DICTATION_SILENCE_PEAK_THRESHOLD ?? "300",
10
);
function parseNonNegativeNumber(value: string | undefined): number | null {
if (value === undefined) {
return null;
}
const parsed = Number.parseFloat(value);
if (!Number.isFinite(parsed) || parsed < 0) {
return null;
}
return parsed;
}
function convertPCMToWavBuffer(
pcmBuffer: Buffer,
sampleRate: number,
@@ -64,6 +76,7 @@ type DictationStreamState = {
receivedChunks: Map<number, Buffer>;
nextSeqToForward: number;
ackSeq: number;
autoCommitBytes: number;
bytesSinceCommit: number;
peakSinceCommit: number;
committedSegmentIds: string[];
@@ -98,6 +111,7 @@ export class DictationStreamManager {
private readonly sessionId: string;
private readonly stt: SpeechToTextProvider | null;
private readonly finalTimeoutMs: number;
private readonly autoCommitSeconds: number;
private readonly streams = new Map<string, DictationStreamState>();
constructor(params: {
@@ -106,12 +120,17 @@ export class DictationStreamManager {
sessionId: string;
stt: SpeechToTextProvider | null;
finalTimeoutMs?: number;
autoCommitSeconds?: number;
}) {
this.logger = params.logger.child({ component: "dictation-stream-manager" });
this.emit = params.emit;
this.sessionId = params.sessionId;
this.stt = params.stt;
this.finalTimeoutMs = params.finalTimeoutMs ?? DEFAULT_DICTATION_FINAL_TIMEOUT_MS;
this.autoCommitSeconds =
params.autoCommitSeconds ??
parseNonNegativeNumber(process.env.PASEO_DICTATION_AUTO_COMMIT_SECONDS) ??
DEFAULT_DICTATION_AUTO_COMMIT_SECONDS;
}
public cleanupAll(): void {
@@ -213,6 +232,10 @@ export class DictationStreamManager {
);
const outputRate = stt.requiredSampleRate;
const autoCommitBytes =
this.autoCommitSeconds > 0
? Math.max(1, Math.round(this.autoCommitSeconds * outputRate * 2))
: 0;
this.streams.set(dictationId, {
dictationId,
@@ -234,6 +257,7 @@ export class DictationStreamManager {
receivedChunks: new Map(),
nextSeqToForward: 0,
ackSeq: -1,
autoCommitBytes,
bytesSinceCommit: 0,
peakSinceCommit: 0,
committedSegmentIds: [],
@@ -290,6 +314,13 @@ export class DictationStreamManager {
state.debugAudioChunks.push(resampled);
state.bytesSinceCommit += resampled.length;
state.peakSinceCommit = Math.max(state.peakSinceCommit, pcm16lePeakAbs(resampled));
try {
this.maybeAutoCommitDictationSegment(params.dictationId, state);
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
void this.failAndCleanupDictationStream(params.dictationId, message, true);
return;
}
if (state.debugChunkWriter) {
void state.debugChunkWriter.writeChunk(seq, resampled).catch((err) => {
@@ -451,6 +482,42 @@ export class DictationStreamManager {
this.streams.delete(dictationId);
}
private maybeAutoCommitDictationSegment(dictationId: string, state: DictationStreamState): void {
if (state.finishRequested) {
return;
}
if (state.autoCommitBytes <= 0 || state.bytesSinceCommit < state.autoCommitBytes) {
return;
}
if (state.peakSinceCommit < DICTATION_SILENCE_PEAK_THRESHOLD) {
this.logger.debug(
{
dictationId,
autoCommitBytes: state.autoCommitBytes,
bytesSinceCommit: state.bytesSinceCommit,
peakSinceCommit: state.peakSinceCommit,
},
"Dictation auto-segment: clearing silence-only segment"
);
state.stt.clear();
state.bytesSinceCommit = 0;
state.peakSinceCommit = 0;
return;
}
this.logger.debug(
{
dictationId,
autoCommitBytes: state.autoCommitBytes,
bytesSinceCommit: state.bytesSinceCommit,
},
"Dictation auto-segment: committing buffered audio"
);
state.bytesSinceCommit = 0;
state.peakSinceCommit = 0;
state.stt.commit();
}
private maybeSealDictationStreamFinish(dictationId: string): void {
const state = this.streams.get(dictationId);
if (!state) {

2
packages/server/src/server/session.ts
View File

@@ -433,7 +433,7 @@ export class Session {
this.defaultLocalSpeechModelIds =
dictation?.localModels?.defaultModelIds && dictation.localModels.defaultModelIds.length > 0
? [...new Set(dictation.localModels.defaultModelIds)]
: ["parakeet-tdt-0.6b-v3-int8", "pocket-tts-onnx-int8"];
: ["parakeet-tdt-0.6b-v2-int8", "pocket-tts-onnx-int8"];
this.registerVoiceSpeakHandler = voiceBridge?.registerVoiceSpeakHandler;
this.unregisterVoiceSpeakHandler = voiceBridge?.unregisterVoiceSpeakHandler;
this.registerVoiceCallerContext = voiceBridge?.registerVoiceCallerContext;

2
packages/server/src/server/speech/providers/local/runtime.ts
View File

@@ -107,7 +107,7 @@ async function createLocalSttEngine(params: {
}): Promise<LocalSttEngine> {
const { modelId, modelsDir, logger } = params;
if (modelId === "parakeet-tdt-0.6b-v3-int8") {
if (modelId === "parakeet-tdt-0.6b-v3-int8" || modelId === "parakeet-tdt-0.6b-v2-int8") {
const modelDir = getLocalSpeechModelDir(modelsDir, modelId);
return {
kind: "offline",

11
packages/server/src/server/speech/providers/local/sherpa/model-catalog.ts
View File

@@ -39,6 +39,16 @@ export const SHERPA_ONNX_MODEL_CATALOG = {
description: "Streaming Paraformer (often strong accuracy; heavier).",
aliases: ["paraformer"],
},
"parakeet-tdt-0.6b-v2-int8": {
kind: "stt-offline",
archiveUrl:
"https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/sherpa-onnx-nemo-parakeet-tdt-0.6b-v2-int8.tar.bz2",
extractedDir: "sherpa-onnx-nemo-parakeet-tdt-0.6b-v2-int8",
requiredFiles: ["encoder.int8.onnx", "decoder.int8.onnx", "joiner.int8.onnx", "tokens.txt"],
description: "NVIDIA Parakeet TDT v2 (offline NeMo transducer, English).",
aliases: ["parakeet-v2", "parakeet-tdt-v2"],
defaultFor: "stt",
},
"parakeet-tdt-0.6b-v3-int8": {
kind: "stt-offline",
archiveUrl:
@@ -47,7 +57,6 @@ export const SHERPA_ONNX_MODEL_CATALOG = {
requiredFiles: ["encoder.int8.onnx", "decoder.int8.onnx", "joiner.int8.onnx", "tokens.txt"],
description: "NVIDIA Parakeet TDT v3 (offline NeMo transducer, multilingual).",
aliases: ["parakeet", "parakeet-v3", "parakeet-tdt"],
defaultFor: "stt",
},
"kitten-nano-en-v0_1-fp16": {
kind: "tts",

12
packages/server/src/server/speech/speech-config-resolver.test.ts
View File

@@ -34,13 +34,13 @@ describe("resolveSpeechConfig", () => {
modelsDir: path.join(paseoHome, "models", "local-speech"),
autoDownload: true,
models: {
dictationStt: "parakeet-tdt-0.6b-v3-int8",
voiceStt: "parakeet-tdt-0.6b-v3-int8",
dictationStt: "parakeet-tdt-0.6b-v2-int8",
voiceStt: "parakeet-tdt-0.6b-v2-int8",
voiceTts: "pocket-tts-onnx-int8",
},
});
expect(result.speech.local?.models.dictationStt).toBe("parakeet-tdt-0.6b-v3-int8");
expect(result.speech.local?.models.voiceStt).toBe("parakeet-tdt-0.6b-v3-int8");
expect(result.speech.local?.models.dictationStt).toBe("parakeet-tdt-0.6b-v2-int8");
expect(result.speech.local?.models.voiceStt).toBe("parakeet-tdt-0.6b-v2-int8");
expect(result.speech.local?.models.voiceTts).toBe("pocket-tts-onnx-int8");
});
@@ -120,8 +120,8 @@ describe("resolveSpeechConfig", () => {
persisted,
});
expect(result.speech.local?.models.dictationStt).toBe("parakeet-tdt-0.6b-v3-int8");
expect(result.speech.local?.models.voiceStt).toBe("parakeet-tdt-0.6b-v3-int8");
expect(result.speech.local?.models.dictationStt).toBe("parakeet-tdt-0.6b-v2-int8");
expect(result.speech.local?.models.voiceStt).toBe("parakeet-tdt-0.6b-v2-int8");
expect(result.speech.local?.models.voiceTts).toBe("pocket-tts-onnx-int8");
});
});