Keep local voice memory out of the daemon (#1217)
* Move local speech work out of daemon Run local speech models in a lazily spawned worker so native model memory and blocking synthesis/transcription work stay out of the daemon process. The worker exits after idle time so local speech RSS can be reclaimed. * Update supervisor stop test for process title
This commit is contained in:
Mohamed Boudra
@@ -6,7 +6,7 @@
|
||||
*/
|
||||
|
||||
import assert from "node:assert";
|
||||
import { spawn, spawnSync, type ChildProcess } from "node:child_process";
|
||||
import { spawn, type ChildProcess } from "node:child_process";
|
||||
import { mkdtemp, readFile, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
@@ -59,15 +59,6 @@ async function readPidLockState(paseoHome: string): Promise<PidLockState> {
|
||||
}
|
||||
}
|
||||
|
||||
function readProcessCommand(pid: number): string | null {
|
||||
const result = spawnSync("ps", ["-p", String(pid), "-o", "command="], { encoding: "utf8" });
|
||||
if (result.status !== 0 || result.error) {
|
||||
return null;
|
||||
}
|
||||
const command = result.stdout.trim();
|
||||
return command.length > 0 ? command : null;
|
||||
}
|
||||
|
||||
interface DaemonStatus {
|
||||
localDaemon: string | null;
|
||||
pid: number | null;
|
||||
@@ -168,11 +159,10 @@ try {
|
||||
assert(isProcessRunning(daemonPid), "daemon process should be running");
|
||||
const pidLockBeforeStop = await readPidLockState(paseoHome);
|
||||
assert.strictEqual(pidLockBeforeStop.pid, daemonPid, "pid lock should match status pid");
|
||||
const command = readProcessCommand(daemonPid);
|
||||
assert(command !== null, "pid lock pid should resolve to a running process command");
|
||||
assert(
|
||||
command.includes("supervisor-entrypoint.ts") || command.includes("supervisor-entrypoint.js"),
|
||||
`pid lock pid should be supervisor-entrypoint process, got: ${command}`,
|
||||
assert.strictEqual(
|
||||
daemonPid,
|
||||
supervisorProcess.pid,
|
||||
"pid lock pid should be the supervisor-entrypoint process",
|
||||
);
|
||||
console.log(`✓ dev daemon started with daemon pid ${daemonPid}\n`);
|
||||
|
||||
|
||||
@@ -13,6 +13,8 @@ import { runSupervisor } from "./supervisor.js";
|
||||
import { resolveSupervisorLogFile } from "./supervisor-log-config.js";
|
||||
import { applySherpaLoaderEnv } from "../src/server/speech/providers/local/sherpa/sherpa-runtime-env.js";
|
||||
|
||||
process.title = "Paseo Supervisor";
|
||||
|
||||
interface DaemonRunnerConfig {
|
||||
devMode: boolean;
|
||||
workerArgs: string[];
|
||||
|
||||
@@ -4,6 +4,8 @@ import { resolvePaseoHome } from "./paseo-home.js";
|
||||
import { createRootLogger } from "./logger.js";
|
||||
import type { DaemonLifecycleIntent } from "./bootstrap.js";
|
||||
|
||||
process.title = "Paseo Daemon";
|
||||
|
||||
type SupervisorLifecycleMessage =
|
||||
| {
|
||||
type: "paseo:shutdown";
|
||||
|
||||
@@ -5,7 +5,6 @@ import type { SpeechToTextProvider, TextToSpeechProvider } from "../../speech-pr
|
||||
import type { RequestedSpeechProviders } from "../../speech-types.js";
|
||||
import type { TurnDetectionProvider } from "../../turn-detection-provider.js";
|
||||
import {
|
||||
getLocalSpeechModelDir,
|
||||
DEFAULT_LOCAL_STT_MODEL,
|
||||
DEFAULT_LOCAL_TTS_MODEL,
|
||||
LocalSttModelIdSchema,
|
||||
@@ -14,16 +13,12 @@ import {
|
||||
type LocalSttModelId,
|
||||
type LocalTtsModelId,
|
||||
} from "./models.js";
|
||||
import { SherpaOfflineRecognizerEngine } from "./sherpa/sherpa-offline-recognizer.js";
|
||||
import { SherpaOnnxParakeetSTT } from "./sherpa/sherpa-parakeet-stt.js";
|
||||
import { SherpaParakeetRealtimeTranscriptionSession } from "./sherpa/sherpa-parakeet-realtime-session.js";
|
||||
import { SherpaOnnxTTS } from "./sherpa/sherpa-tts.js";
|
||||
import {
|
||||
ensureSileroVadModel,
|
||||
SherpaSileroTurnDetectionProvider,
|
||||
} from "./sherpa/silero-vad-provider.js";
|
||||
|
||||
type LocalSttEngine = SherpaOfflineRecognizerEngine;
|
||||
LocalSpeechWorkerClient,
|
||||
WorkerBackedSpeechToTextProvider,
|
||||
WorkerBackedTextToSpeechProvider,
|
||||
WorkerBackedTurnDetectionProvider,
|
||||
} from "./worker-client.js";
|
||||
|
||||
interface ResolvedLocalModels {
|
||||
dictationLocalSttModel: LocalSttModelId;
|
||||
@@ -50,10 +45,6 @@ export interface InitializedLocalSpeech {
|
||||
cleanup: () => void;
|
||||
}
|
||||
|
||||
function buildModelDownloadHint(modelId: LocalSpeechModelId): string {
|
||||
return `Use 'paseo speech download --model ${modelId}' to download this model.`;
|
||||
}
|
||||
|
||||
function resolveConfiguredLocalModels(speechConfig: PaseoSpeechConfig | null): ResolvedLocalModels {
|
||||
return {
|
||||
dictationLocalSttModel: LocalSttModelIdSchema.parse(
|
||||
@@ -104,131 +95,43 @@ function computeRequiredLocalModelIds(params: {
|
||||
return Array.from(ids);
|
||||
}
|
||||
|
||||
async function createLocalSttEngine(params: {
|
||||
modelId: LocalSttModelId;
|
||||
modelsDir: string;
|
||||
logger: Logger;
|
||||
}): Promise<LocalSttEngine> {
|
||||
const { modelId, modelsDir, logger } = params;
|
||||
|
||||
const modelDir = getLocalSpeechModelDir(modelsDir, modelId);
|
||||
return new SherpaOfflineRecognizerEngine(
|
||||
{
|
||||
model: {
|
||||
kind: "nemo_transducer",
|
||||
encoder: `${modelDir}/encoder.int8.onnx`,
|
||||
decoder: `${modelDir}/decoder.int8.onnx`,
|
||||
joiner: `${modelDir}/joiner.int8.onnx`,
|
||||
tokens: `${modelDir}/tokens.txt`,
|
||||
},
|
||||
numThreads: 2,
|
||||
debug: 0,
|
||||
},
|
||||
logger,
|
||||
);
|
||||
}
|
||||
|
||||
type LocalConfig = NonNullable<PaseoSpeechConfig["local"]>;
|
||||
|
||||
function isLocalProviderEnabled(provider: { enabled?: boolean; provider: string }): boolean {
|
||||
return provider.enabled !== false && provider.provider === "local";
|
||||
}
|
||||
|
||||
async function initializeLocalTurnDetection(
|
||||
localConfig: LocalConfig | null,
|
||||
logger: Logger,
|
||||
): Promise<TurnDetectionProvider> {
|
||||
let vadModelPath: string | undefined;
|
||||
if (localConfig) {
|
||||
try {
|
||||
vadModelPath = await ensureSileroVadModel(localConfig.modelsDir, logger);
|
||||
} catch (err) {
|
||||
logger.warn({ err }, "Failed to provision Silero VAD model, falling back to bundled");
|
||||
}
|
||||
}
|
||||
return new SherpaSileroTurnDetectionProvider({ modelPath: vadModelPath }, logger);
|
||||
function warnLocalConfigMissing(logger: Logger, feature: string): void {
|
||||
logger.warn(
|
||||
{ configured: false },
|
||||
`Local ${feature} selected but local provider config is missing; ${feature} will be unavailable`,
|
||||
);
|
||||
}
|
||||
|
||||
async function initializeLocalVoiceStt(params: {
|
||||
localConfig: LocalConfig | null;
|
||||
modelId: LocalSttModelId;
|
||||
logger: Logger;
|
||||
getLocalSttEngine: (modelId: LocalSttModelId) => Promise<LocalSttEngine | null>;
|
||||
}): Promise<SpeechToTextProvider | null> {
|
||||
const { localConfig, modelId, logger, getLocalSttEngine } = params;
|
||||
if (!localConfig) {
|
||||
logger.warn(
|
||||
{ configured: false },
|
||||
"Local STT selected for voice but local provider config is missing; STT will be unavailable",
|
||||
);
|
||||
return null;
|
||||
}
|
||||
const voiceEngine = await getLocalSttEngine(modelId);
|
||||
return voiceEngine ? new SherpaOnnxParakeetSTT({ engine: voiceEngine }, logger) : null;
|
||||
function initializeLocalTurnDetection(params: {
|
||||
client: LocalSpeechWorkerClient;
|
||||
}): TurnDetectionProvider {
|
||||
const { client } = params;
|
||||
return new WorkerBackedTurnDetectionProvider(client);
|
||||
}
|
||||
|
||||
async function initializeLocalDictationStt(params: {
|
||||
localConfig: LocalConfig | null;
|
||||
modelId: LocalSttModelId;
|
||||
logger: Logger;
|
||||
getLocalSttEngine: (modelId: LocalSttModelId) => Promise<LocalSttEngine | null>;
|
||||
}): Promise<SpeechToTextProvider | null> {
|
||||
const { localConfig, modelId, logger, getLocalSttEngine } = params;
|
||||
if (!localConfig) {
|
||||
logger.warn(
|
||||
{ configured: false },
|
||||
"Local STT selected for dictation but local provider config is missing; dictation STT will be unavailable",
|
||||
);
|
||||
return null;
|
||||
}
|
||||
const dictationEngine = await getLocalSttEngine(modelId);
|
||||
if (dictationEngine) {
|
||||
return {
|
||||
id: "local",
|
||||
createSession: () =>
|
||||
new SherpaParakeetRealtimeTranscriptionSession({ engine: dictationEngine }),
|
||||
};
|
||||
}
|
||||
return null;
|
||||
function initializeLocalVoiceStt(params: {
|
||||
client: LocalSpeechWorkerClient;
|
||||
}): SpeechToTextProvider {
|
||||
const { client } = params;
|
||||
return new WorkerBackedSpeechToTextProvider(client, "voiceStt");
|
||||
}
|
||||
|
||||
async function initializeLocalVoiceTts(params: {
|
||||
localConfig: LocalConfig | null;
|
||||
speechConfig: PaseoSpeechConfig | null;
|
||||
localModels: ResolvedLocalModels;
|
||||
logger: Logger;
|
||||
}): Promise<TextToSpeechProvider | null> {
|
||||
const { localConfig, speechConfig, localModels, logger } = params;
|
||||
if (!localConfig) {
|
||||
logger.warn(
|
||||
{ configured: false },
|
||||
"Local TTS selected for voice but local provider config is missing; TTS will be unavailable",
|
||||
);
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
const modelDir = getLocalSpeechModelDir(localConfig.modelsDir, localModels.voiceLocalTtsModel);
|
||||
return new SherpaOnnxTTS(
|
||||
{
|
||||
preset: localModels.voiceLocalTtsModel,
|
||||
modelDir,
|
||||
speakerId: speechConfig?.local?.models.voiceTtsSpeakerId,
|
||||
speed: speechConfig?.local?.models.voiceTtsSpeed,
|
||||
},
|
||||
logger,
|
||||
);
|
||||
} catch (err) {
|
||||
logger.warn(
|
||||
{
|
||||
err,
|
||||
modelsDir: localConfig.modelsDir,
|
||||
modelId: localModels.voiceLocalTtsModel,
|
||||
hint: buildModelDownloadHint(localModels.voiceLocalTtsModel),
|
||||
},
|
||||
"Local TTS engine unavailable",
|
||||
);
|
||||
return null;
|
||||
}
|
||||
function initializeLocalDictationStt(params: {
|
||||
client: LocalSpeechWorkerClient;
|
||||
}): SpeechToTextProvider {
|
||||
const { client } = params;
|
||||
return new WorkerBackedSpeechToTextProvider(client, "dictationStt");
|
||||
}
|
||||
|
||||
function initializeLocalVoiceTts(params: {
|
||||
client: LocalSpeechWorkerClient;
|
||||
}): TextToSpeechProvider {
|
||||
const { client } = params;
|
||||
return new WorkerBackedTextToSpeechProvider(client);
|
||||
}
|
||||
|
||||
export async function initializeLocalSpeechServices(params: {
|
||||
@@ -251,80 +154,56 @@ export async function initializeLocalSpeechServices(params: {
|
||||
models: localModels,
|
||||
});
|
||||
|
||||
const localSttEngines = new Map<LocalSttModelId, LocalSttEngine>();
|
||||
|
||||
const getLocalSttEngine = async (modelId: LocalSttModelId): Promise<LocalSttEngine | null> => {
|
||||
const existing = localSttEngines.get(modelId);
|
||||
if (existing) {
|
||||
return existing;
|
||||
}
|
||||
if (!localConfig) {
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
const created = await createLocalSttEngine({
|
||||
modelId,
|
||||
modelsDir: localConfig.modelsDir,
|
||||
logger,
|
||||
});
|
||||
localSttEngines.set(modelId, created);
|
||||
return created;
|
||||
} catch (err) {
|
||||
logger.warn(
|
||||
{
|
||||
err,
|
||||
const workerClient = localConfig
|
||||
? new LocalSpeechWorkerClient({
|
||||
config: {
|
||||
modelsDir: localConfig.modelsDir,
|
||||
modelId,
|
||||
hint: buildModelDownloadHint(modelId),
|
||||
voiceSttModel: localModels.voiceLocalSttModel,
|
||||
dictationSttModel: localModels.dictationLocalSttModel,
|
||||
voiceTtsModel: localModels.voiceLocalTtsModel,
|
||||
voiceTtsSpeakerId: speechConfig?.local?.models.voiceTtsSpeakerId,
|
||||
voiceTtsSpeed: speechConfig?.local?.models.voiceTtsSpeed,
|
||||
},
|
||||
"Local STT engine unavailable",
|
||||
);
|
||||
return null;
|
||||
}
|
||||
};
|
||||
})
|
||||
: null;
|
||||
|
||||
if (isLocalProviderEnabled(providers.voiceTurnDetection)) {
|
||||
turnDetectionService = await initializeLocalTurnDetection(localConfig, logger);
|
||||
if (workerClient) {
|
||||
turnDetectionService = initializeLocalTurnDetection({ client: workerClient });
|
||||
} else {
|
||||
warnLocalConfigMissing(logger, "turn detection");
|
||||
}
|
||||
}
|
||||
|
||||
if (isLocalProviderEnabled(providers.voiceStt)) {
|
||||
sttService = await initializeLocalVoiceStt({
|
||||
localConfig,
|
||||
modelId: localModels.voiceLocalSttModel,
|
||||
logger,
|
||||
getLocalSttEngine,
|
||||
});
|
||||
if (workerClient) {
|
||||
sttService = initializeLocalVoiceStt({ client: workerClient });
|
||||
} else {
|
||||
warnLocalConfigMissing(logger, "voice STT");
|
||||
}
|
||||
}
|
||||
|
||||
if (isLocalProviderEnabled(providers.dictationStt)) {
|
||||
dictationSttService = await initializeLocalDictationStt({
|
||||
localConfig,
|
||||
modelId: localModels.dictationLocalSttModel,
|
||||
logger,
|
||||
getLocalSttEngine,
|
||||
});
|
||||
if (workerClient) {
|
||||
dictationSttService = initializeLocalDictationStt({ client: workerClient });
|
||||
} else {
|
||||
warnLocalConfigMissing(logger, "dictation STT");
|
||||
}
|
||||
}
|
||||
|
||||
if (isLocalProviderEnabled(providers.voiceTts)) {
|
||||
localVoiceTtsProvider = await initializeLocalVoiceTts({
|
||||
localConfig,
|
||||
speechConfig,
|
||||
localModels,
|
||||
logger,
|
||||
});
|
||||
if (workerClient) {
|
||||
localVoiceTtsProvider = initializeLocalVoiceTts({ client: workerClient });
|
||||
} else {
|
||||
warnLocalConfigMissing(logger, "voice TTS");
|
||||
}
|
||||
if (localVoiceTtsProvider) {
|
||||
ttsService = localVoiceTtsProvider;
|
||||
}
|
||||
}
|
||||
|
||||
const cleanup = () => {
|
||||
const maybeFreeable = localVoiceTtsProvider as unknown as { free?: () => void } | null;
|
||||
if (typeof maybeFreeable?.free === "function") {
|
||||
maybeFreeable.free();
|
||||
}
|
||||
for (const engine of localSttEngines.values()) {
|
||||
engine.free();
|
||||
}
|
||||
workerClient?.shutdown();
|
||||
};
|
||||
|
||||
return {
|
||||
|
||||
@@ -0,0 +1,212 @@
|
||||
import { EventEmitter } from "node:events";
|
||||
import { once } from "node:events";
|
||||
import pino from "pino";
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
import {
|
||||
LocalSpeechWorkerClient,
|
||||
WorkerBackedSpeechToTextProvider,
|
||||
WorkerBackedTextToSpeechProvider,
|
||||
WorkerBackedTurnDetectionProvider,
|
||||
} from "./worker-client.js";
|
||||
import type {
|
||||
LocalSpeechWorkerRequest,
|
||||
LocalSpeechWorkerToParentMessage,
|
||||
} from "./worker-protocol.js";
|
||||
|
||||
class FakeLocalSpeechWorker extends EventEmitter {
|
||||
public connected = true;
|
||||
public killed = false;
|
||||
public readonly sent: LocalSpeechWorkerRequest[] = [];
|
||||
public disconnects = 0;
|
||||
public kills = 0;
|
||||
|
||||
send(message: LocalSpeechWorkerRequest, callback: (error: Error | null) => void): boolean {
|
||||
this.sent.push(message);
|
||||
queueMicrotask(() => callback(null));
|
||||
return true;
|
||||
}
|
||||
|
||||
disconnect(): void {
|
||||
this.disconnects++;
|
||||
this.connected = false;
|
||||
}
|
||||
|
||||
kill(): boolean {
|
||||
this.kills++;
|
||||
this.killed = true;
|
||||
this.connected = false;
|
||||
return true;
|
||||
}
|
||||
|
||||
respond(request: LocalSpeechWorkerRequest, result?: unknown): void {
|
||||
this.emit("message", {
|
||||
type: "response",
|
||||
requestId: request.requestId,
|
||||
ok: true,
|
||||
result,
|
||||
} satisfies LocalSpeechWorkerToParentMessage);
|
||||
}
|
||||
|
||||
emitWorkerMessage(message: LocalSpeechWorkerToParentMessage): void {
|
||||
this.emit("message", message);
|
||||
}
|
||||
}
|
||||
|
||||
function createClient(options?: { idleTtlMs?: number }) {
|
||||
const workers: FakeLocalSpeechWorker[] = [];
|
||||
const client = new LocalSpeechWorkerClient({
|
||||
config: {
|
||||
modelsDir: "/tmp/models",
|
||||
voiceSttModel: "parakeet-tdt-0.6b-v2-int8",
|
||||
dictationSttModel: "parakeet-tdt-0.6b-v2-int8",
|
||||
voiceTtsModel: "kokoro-en-v0_19",
|
||||
},
|
||||
requestTimeoutMs: 1000,
|
||||
idleTtlMs: options?.idleTtlMs ?? 1000,
|
||||
forkWorker: () => {
|
||||
const worker = new FakeLocalSpeechWorker();
|
||||
workers.push(worker);
|
||||
return worker;
|
||||
},
|
||||
});
|
||||
return { client, workers };
|
||||
}
|
||||
|
||||
async function waitForMicrotasks(): Promise<void> {
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
}
|
||||
|
||||
describe("LocalSpeechWorkerClient", () => {
|
||||
it("does not spawn the worker until first local speech use", () => {
|
||||
const { workers } = createClient();
|
||||
|
||||
expect(workers).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("sends TTS requests through the worker and returns the audio stream", async () => {
|
||||
const { client, workers } = createClient();
|
||||
const provider = new WorkerBackedTextToSpeechProvider(client);
|
||||
|
||||
const pending = provider.synthesizeSpeech("hello");
|
||||
expect(workers).toHaveLength(1);
|
||||
const request = workers[0].sent[0];
|
||||
expect(request).toMatchObject({
|
||||
type: "tts.synthesize",
|
||||
text: "hello",
|
||||
config: {
|
||||
modelsDir: "/tmp/models",
|
||||
voiceTtsModel: "kokoro-en-v0_19",
|
||||
},
|
||||
});
|
||||
|
||||
workers[0].respond(request, {
|
||||
audio: Buffer.from([1, 2, 3, 4]),
|
||||
format: "pcm;rate=24000",
|
||||
});
|
||||
|
||||
const result = await pending;
|
||||
const chunks: Buffer[] = [];
|
||||
for await (const chunk of result.stream) {
|
||||
chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
|
||||
}
|
||||
expect(result.format).toBe("pcm;rate=24000");
|
||||
expect(Buffer.concat(chunks)).toEqual(Buffer.from([1, 2, 3, 4]));
|
||||
});
|
||||
|
||||
it("forwards STT session audio and transcript events through IPC", async () => {
|
||||
const { client, workers } = createClient();
|
||||
const provider = new WorkerBackedSpeechToTextProvider(client, "voiceStt");
|
||||
const session = provider.createSession({ logger: pino({ level: "silent" }) });
|
||||
|
||||
const transcriptPromise = once(session as EventEmitter, "transcript");
|
||||
const committedPromise = once(session as EventEmitter, "committed");
|
||||
|
||||
const connect = session.connect();
|
||||
const createRequest = workers[0].sent[0];
|
||||
expect(createRequest).toMatchObject({ type: "session.create", kind: "voiceStt" });
|
||||
workers[0].respond(createRequest, { requiredSampleRate: 16000 });
|
||||
await connect;
|
||||
|
||||
session.appendPcm16(Buffer.from([9, 8, 7, 6]));
|
||||
await waitForMicrotasks();
|
||||
expect(workers[0].sent[1]).toMatchObject({
|
||||
type: "session.append",
|
||||
sessionId: createRequest.sessionId,
|
||||
audio: Buffer.from([9, 8, 7, 6]),
|
||||
});
|
||||
|
||||
session.commit();
|
||||
await waitForMicrotasks();
|
||||
expect(workers[0].sent[2]).toMatchObject({
|
||||
type: "session.commit",
|
||||
sessionId: createRequest.sessionId,
|
||||
});
|
||||
|
||||
workers[0].emitWorkerMessage({
|
||||
type: "session.committed",
|
||||
sessionId: createRequest.sessionId,
|
||||
payload: { segmentId: "seg-1", previousSegmentId: null },
|
||||
});
|
||||
workers[0].emitWorkerMessage({
|
||||
type: "session.transcript",
|
||||
sessionId: createRequest.sessionId,
|
||||
payload: { segmentId: "seg-1", transcript: "hello", isFinal: true },
|
||||
});
|
||||
|
||||
await expect(committedPromise).resolves.toEqual([
|
||||
{ segmentId: "seg-1", previousSegmentId: null },
|
||||
]);
|
||||
await expect(transcriptPromise).resolves.toEqual([
|
||||
{ segmentId: "seg-1", transcript: "hello", isFinal: true },
|
||||
]);
|
||||
});
|
||||
|
||||
it("forwards VAD session events through the shared worker", async () => {
|
||||
const { client, workers } = createClient();
|
||||
const provider = new WorkerBackedTurnDetectionProvider(client);
|
||||
const session = provider.createSession({ logger: pino({ level: "silent" }) });
|
||||
const startedPromise = once(session as EventEmitter, "speech_started");
|
||||
const stoppedPromise = once(session as EventEmitter, "speech_stopped");
|
||||
|
||||
const connect = session.connect();
|
||||
const createRequest = workers[0].sent[0];
|
||||
expect(createRequest).toMatchObject({ type: "session.create", kind: "vad" });
|
||||
workers[0].respond(createRequest, { requiredSampleRate: 16000 });
|
||||
await connect;
|
||||
|
||||
workers[0].emitWorkerMessage({
|
||||
type: "session.speech_started",
|
||||
sessionId: createRequest.sessionId,
|
||||
});
|
||||
workers[0].emitWorkerMessage({
|
||||
type: "session.speech_stopped",
|
||||
sessionId: createRequest.sessionId,
|
||||
});
|
||||
|
||||
await expect(startedPromise).resolves.toEqual([]);
|
||||
await expect(stoppedPromise).resolves.toEqual([]);
|
||||
});
|
||||
|
||||
it("kills an idle worker and respawns on later use", async () => {
|
||||
const { client, workers } = createClient({ idleTtlMs: 5 });
|
||||
|
||||
const first = client.synthesizeSpeech("first");
|
||||
workers[0].respond(workers[0].sent[0], {
|
||||
audio: Buffer.from([1]),
|
||||
format: "pcm;rate=24000",
|
||||
});
|
||||
await first;
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
expect(workers[0].kills).toBe(1);
|
||||
|
||||
const second = client.synthesizeSpeech("second");
|
||||
expect(workers).toHaveLength(2);
|
||||
workers[1].respond(workers[1].sent[0], {
|
||||
audio: Buffer.from([2]),
|
||||
format: "pcm;rate=24000",
|
||||
});
|
||||
await second;
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,538 @@
|
||||
import { fork } from "node:child_process";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { EventEmitter } from "node:events";
|
||||
import { Readable } from "node:stream";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import type pino from "pino";
|
||||
|
||||
import type {
|
||||
SpeechStreamResult,
|
||||
SpeechToTextProvider,
|
||||
StreamingTranscriptionSession,
|
||||
TextToSpeechProvider,
|
||||
} from "../../speech-provider.js";
|
||||
import type { TurnDetectionProvider, TurnDetectionSession } from "../../turn-detection-provider.js";
|
||||
import { applySherpaLoaderEnv } from "./sherpa/sherpa-runtime-env.js";
|
||||
import type {
|
||||
LocalSpeechCreateSessionResult,
|
||||
LocalSpeechSessionKind,
|
||||
LocalSpeechTranscriptionResult,
|
||||
LocalSpeechTtsResult,
|
||||
LocalSpeechWorkerConfig,
|
||||
LocalSpeechWorkerRequest,
|
||||
LocalSpeechWorkerResponse,
|
||||
LocalSpeechWorkerToParentMessage,
|
||||
} from "./worker-protocol.js";
|
||||
|
||||
const DEFAULT_REQUEST_TIMEOUT_MS = 30000;
|
||||
const DEFAULT_IDLE_TTL_MS = 5 * 60 * 1000;
|
||||
const DEFAULT_LOCAL_SAMPLE_RATE = 16000;
|
||||
|
||||
type LocalSpeechWorkerRequestInput = LocalSpeechWorkerRequest extends infer Request
|
||||
? Request extends LocalSpeechWorkerRequest
|
||||
? Omit<Request, "requestId">
|
||||
: never
|
||||
: never;
|
||||
|
||||
interface LocalSpeechWorkerProcess {
|
||||
connected: boolean;
|
||||
killed: boolean;
|
||||
send(message: LocalSpeechWorkerRequest, callback: (error: Error | null) => void): boolean;
|
||||
disconnect(): void;
|
||||
kill(): boolean;
|
||||
on(event: "message", listener: (message: LocalSpeechWorkerToParentMessage) => void): this;
|
||||
on(event: "exit", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this;
|
||||
}
|
||||
|
||||
interface PendingRequest {
|
||||
resolve: (value: unknown) => void;
|
||||
reject: (error: Error) => void;
|
||||
timeout: ReturnType<typeof setTimeout>;
|
||||
}
|
||||
|
||||
interface LocalSpeechWorkerClientOptions {
|
||||
config: LocalSpeechWorkerConfig;
|
||||
requestTimeoutMs?: number;
|
||||
idleTtlMs?: number;
|
||||
forkWorker?: () => LocalSpeechWorkerProcess;
|
||||
}
|
||||
|
||||
function resolveWorkerUrl(): URL {
|
||||
const currentUrl = import.meta.url;
|
||||
if (currentUrl.endsWith(".ts")) {
|
||||
return new URL("./worker-process.ts", currentUrl);
|
||||
}
|
||||
return new URL("./worker-process.js", currentUrl);
|
||||
}
|
||||
|
||||
function resolveWorkerExecArgv(): string[] {
|
||||
if (!import.meta.url.endsWith(".ts")) {
|
||||
return [];
|
||||
}
|
||||
const loaderUrl = new URL("../../../../terminal/terminal-ts-loader.mjs", import.meta.url).href;
|
||||
const importSource = [
|
||||
'import { register } from "node:module";',
|
||||
'import { pathToFileURL } from "node:url";',
|
||||
`register(${JSON.stringify(loaderUrl)}, pathToFileURL("./"));`,
|
||||
].join(" ");
|
||||
return [
|
||||
"--experimental-strip-types",
|
||||
"--import",
|
||||
`data:text/javascript,${encodeURIComponent(importSource)}`,
|
||||
];
|
||||
}
|
||||
|
||||
function forkLocalSpeechWorker(): LocalSpeechWorkerProcess {
|
||||
const env = { ...process.env };
|
||||
applySherpaLoaderEnv(env);
|
||||
return fork(fileURLToPath(resolveWorkerUrl()), [], {
|
||||
env,
|
||||
execArgv: resolveWorkerExecArgv(),
|
||||
serialization: "advanced",
|
||||
stdio: ["ignore", "ignore", "inherit", "ipc"],
|
||||
}) as LocalSpeechWorkerProcess;
|
||||
}
|
||||
|
||||
function isResponse(
|
||||
message: LocalSpeechWorkerToParentMessage,
|
||||
): message is LocalSpeechWorkerResponse {
|
||||
return message.type === "response";
|
||||
}
|
||||
|
||||
export class LocalSpeechWorkerClient {
|
||||
private readonly config: LocalSpeechWorkerConfig;
|
||||
private readonly requestTimeoutMs: number;
|
||||
private readonly idleTtlMs: number;
|
||||
private readonly forkWorker: () => LocalSpeechWorkerProcess;
|
||||
private readonly pendingRequests = new Map<string, PendingRequest>();
|
||||
private readonly activeSessionIds = new Set<string>();
|
||||
private readonly sessionEmitters = new Map<string, EventEmitter>();
|
||||
private worker: LocalSpeechWorkerProcess | null = null;
|
||||
private inFlightRequests = 0;
|
||||
private idleTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
|
||||
constructor(options: LocalSpeechWorkerClientOptions) {
|
||||
this.config = options.config;
|
||||
this.requestTimeoutMs = options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS;
|
||||
this.idleTtlMs = options.idleTtlMs ?? DEFAULT_IDLE_TTL_MS;
|
||||
this.forkWorker = options.forkWorker ?? forkLocalSpeechWorker;
|
||||
}
|
||||
|
||||
async synthesizeSpeech(text: string): Promise<SpeechStreamResult> {
|
||||
const result = await this.sendRequest<LocalSpeechTtsResult>({
|
||||
type: "tts.synthesize",
|
||||
config: this.config,
|
||||
text,
|
||||
});
|
||||
return {
|
||||
stream: Readable.from([result.audio]),
|
||||
format: result.format,
|
||||
};
|
||||
}
|
||||
|
||||
transcribeVoice(audio: Buffer, format: string): Promise<LocalSpeechTranscriptionResult> {
|
||||
return this.sendRequest<LocalSpeechTranscriptionResult>({
|
||||
type: "stt.transcribe",
|
||||
config: this.config,
|
||||
model: "voice",
|
||||
audio,
|
||||
format,
|
||||
});
|
||||
}
|
||||
|
||||
async createSession(
|
||||
kind: LocalSpeechSessionKind,
|
||||
emitter: EventEmitter,
|
||||
): Promise<{ sessionId: string; requiredSampleRate: number }> {
|
||||
const sessionId = randomUUID();
|
||||
this.activeSessionIds.add(sessionId);
|
||||
this.sessionEmitters.set(sessionId, emitter);
|
||||
try {
|
||||
const result = await this.sendRequest<LocalSpeechCreateSessionResult>({
|
||||
type: "session.create",
|
||||
config: this.config,
|
||||
sessionId,
|
||||
kind,
|
||||
});
|
||||
return { sessionId, requiredSampleRate: result.requiredSampleRate };
|
||||
} catch (err) {
|
||||
this.activeSessionIds.delete(sessionId);
|
||||
this.sessionEmitters.delete(sessionId);
|
||||
this.scheduleIdleShutdownIfReady();
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
appendSessionAudio(sessionId: string, audio: Buffer): void {
|
||||
void this.sendRequest({ type: "session.append", sessionId, audio }).catch((err) => {
|
||||
this.emitSessionError(sessionId, err);
|
||||
});
|
||||
}
|
||||
|
||||
commitSession(sessionId: string): void {
|
||||
void this.sendRequest({ type: "session.commit", sessionId }).catch((err) => {
|
||||
this.emitSessionError(sessionId, err);
|
||||
});
|
||||
}
|
||||
|
||||
clearSession(sessionId: string): void {
|
||||
void this.sendRequest({ type: "session.clear", sessionId }).catch((err) => {
|
||||
this.emitSessionError(sessionId, err);
|
||||
});
|
||||
}
|
||||
|
||||
flushSession(sessionId: string): void {
|
||||
void this.sendRequest({ type: "session.flush", sessionId }).catch((err) => {
|
||||
this.emitSessionError(sessionId, err);
|
||||
});
|
||||
}
|
||||
|
||||
resetSession(sessionId: string): void {
|
||||
void this.sendRequest({ type: "session.reset", sessionId }).catch((err) => {
|
||||
this.emitSessionError(sessionId, err);
|
||||
});
|
||||
}
|
||||
|
||||
closeSession(sessionId: string): void {
|
||||
this.activeSessionIds.delete(sessionId);
|
||||
this.sessionEmitters.delete(sessionId);
|
||||
void this.sendRequest({ type: "session.close", sessionId }).catch(() => {
|
||||
// Closing is best-effort; the parent already dropped the session.
|
||||
});
|
||||
this.scheduleIdleShutdownIfReady();
|
||||
}
|
||||
|
||||
shutdown(): void {
|
||||
this.clearIdleTimer();
|
||||
this.rejectAllPending(new Error("Local speech worker shut down"));
|
||||
this.activeSessionIds.clear();
|
||||
this.sessionEmitters.clear();
|
||||
const worker = this.worker;
|
||||
this.worker = null;
|
||||
if (worker && !worker.killed) {
|
||||
try {
|
||||
worker.disconnect();
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
try {
|
||||
worker.kill();
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private sendRequest<T = void>(input: LocalSpeechWorkerRequestInput): Promise<T> {
|
||||
const worker = this.ensureWorker();
|
||||
const requestId = randomUUID();
|
||||
const message = { ...input, requestId } as LocalSpeechWorkerRequest;
|
||||
this.inFlightRequests++;
|
||||
this.clearIdleTimer();
|
||||
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
const timeout = setTimeout(() => {
|
||||
this.pendingRequests.delete(requestId);
|
||||
this.inFlightRequests = Math.max(0, this.inFlightRequests - 1);
|
||||
this.scheduleIdleShutdownIfReady();
|
||||
reject(new Error(`Local speech worker request timed out: ${input.type}`));
|
||||
}, this.requestTimeoutMs);
|
||||
this.pendingRequests.set(requestId, {
|
||||
resolve: (value) => resolve(value as T),
|
||||
reject,
|
||||
timeout,
|
||||
});
|
||||
|
||||
const sent = worker.send(message, (error) => {
|
||||
if (!error) {
|
||||
return;
|
||||
}
|
||||
const pending = this.pendingRequests.get(requestId);
|
||||
if (!pending) {
|
||||
return;
|
||||
}
|
||||
clearTimeout(pending.timeout);
|
||||
this.pendingRequests.delete(requestId);
|
||||
this.inFlightRequests = Math.max(0, this.inFlightRequests - 1);
|
||||
this.scheduleIdleShutdownIfReady();
|
||||
pending.reject(error);
|
||||
});
|
||||
if (!sent) {
|
||||
const pending = this.pendingRequests.get(requestId);
|
||||
if (pending) {
|
||||
clearTimeout(pending.timeout);
|
||||
this.pendingRequests.delete(requestId);
|
||||
this.inFlightRequests = Math.max(0, this.inFlightRequests - 1);
|
||||
this.scheduleIdleShutdownIfReady();
|
||||
pending.reject(new Error("Local speech worker IPC channel is not writable"));
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private ensureWorker(): LocalSpeechWorkerProcess {
|
||||
if (this.worker && !this.worker.killed && this.worker.connected) {
|
||||
return this.worker;
|
||||
}
|
||||
const worker = this.forkWorker();
|
||||
this.worker = worker;
|
||||
worker.on("message", (message) => this.handleWorkerMessage(message));
|
||||
worker.on("exit", () => this.handleWorkerExit());
|
||||
return worker;
|
||||
}
|
||||
|
||||
private handleWorkerMessage(message: LocalSpeechWorkerToParentMessage): void {
|
||||
if (isResponse(message)) {
|
||||
const pending = this.pendingRequests.get(message.requestId);
|
||||
if (!pending) {
|
||||
return;
|
||||
}
|
||||
clearTimeout(pending.timeout);
|
||||
this.pendingRequests.delete(message.requestId);
|
||||
this.inFlightRequests = Math.max(0, this.inFlightRequests - 1);
|
||||
this.scheduleIdleShutdownIfReady();
|
||||
if (message.ok) {
|
||||
pending.resolve(message.result);
|
||||
} else {
|
||||
pending.reject(new Error(message.error));
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
const emitter = this.sessionEmitters.get(message.sessionId);
|
||||
if (!emitter) {
|
||||
return;
|
||||
}
|
||||
switch (message.type) {
|
||||
case "session.committed":
|
||||
emitter.emit("committed", message.payload);
|
||||
return;
|
||||
case "session.transcript":
|
||||
emitter.emit("transcript", message.payload);
|
||||
return;
|
||||
case "session.speech_started":
|
||||
emitter.emit("speech_started");
|
||||
return;
|
||||
case "session.speech_stopped":
|
||||
emitter.emit("speech_stopped");
|
||||
return;
|
||||
case "session.error":
|
||||
emitter.emit("error", new Error(message.error));
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
private handleWorkerExit(): void {
|
||||
this.worker = null;
|
||||
this.clearIdleTimer();
|
||||
this.rejectAllPending(new Error("Local speech worker exited"));
|
||||
for (const [sessionId, emitter] of this.sessionEmitters) {
|
||||
if (this.activeSessionIds.has(sessionId)) {
|
||||
emitter.emit("error", new Error("Local speech worker exited"));
|
||||
}
|
||||
}
|
||||
this.activeSessionIds.clear();
|
||||
this.sessionEmitters.clear();
|
||||
this.inFlightRequests = 0;
|
||||
}
|
||||
|
||||
private rejectAllPending(error: Error): void {
|
||||
for (const [requestId, pending] of this.pendingRequests) {
|
||||
clearTimeout(pending.timeout);
|
||||
pending.reject(error);
|
||||
this.pendingRequests.delete(requestId);
|
||||
}
|
||||
}
|
||||
|
||||
private emitSessionError(sessionId: string, error: unknown): void {
|
||||
const emitter = this.sessionEmitters.get(sessionId);
|
||||
if (!emitter) {
|
||||
return;
|
||||
}
|
||||
emitter.emit("error", error instanceof Error ? error : new Error(String(error)));
|
||||
}
|
||||
|
||||
private scheduleIdleShutdownIfReady(): void {
|
||||
if (!this.worker || this.inFlightRequests > 0 || this.activeSessionIds.size > 0) {
|
||||
return;
|
||||
}
|
||||
this.clearIdleTimer();
|
||||
this.idleTimer = setTimeout(() => {
|
||||
if (this.inFlightRequests === 0 && this.activeSessionIds.size === 0) {
|
||||
this.shutdown();
|
||||
}
|
||||
}, this.idleTtlMs);
|
||||
}
|
||||
|
||||
private clearIdleTimer(): void {
|
||||
if (this.idleTimer) {
|
||||
clearTimeout(this.idleTimer);
|
||||
this.idleTimer = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export class WorkerBackedTextToSpeechProvider implements TextToSpeechProvider {
|
||||
constructor(private readonly client: LocalSpeechWorkerClient) {}
|
||||
|
||||
synthesizeSpeech(text: string): Promise<SpeechStreamResult> {
|
||||
return this.client.synthesizeSpeech(text);
|
||||
}
|
||||
}
|
||||
|
||||
export class WorkerBackedSpeechToTextProvider implements SpeechToTextProvider {
|
||||
public readonly id = "local" as const;
|
||||
|
||||
constructor(
|
||||
private readonly client: LocalSpeechWorkerClient,
|
||||
private readonly kind: Extract<LocalSpeechSessionKind, "voiceStt" | "dictationStt">,
|
||||
) {}
|
||||
|
||||
createSession(_params: {
|
||||
logger: pino.Logger;
|
||||
language?: string;
|
||||
prompt?: string;
|
||||
}): StreamingTranscriptionSession {
|
||||
return new WorkerBackedTranscriptionSession(this.client, this.kind);
|
||||
}
|
||||
}
|
||||
|
||||
export class WorkerBackedTurnDetectionProvider implements TurnDetectionProvider {
|
||||
public readonly id = "local" as const;
|
||||
|
||||
constructor(private readonly client: LocalSpeechWorkerClient) {}
|
||||
|
||||
createSession(_params: { logger: pino.Logger }): TurnDetectionSession {
|
||||
return new WorkerBackedTurnDetectionSession(this.client);
|
||||
}
|
||||
}
|
||||
|
||||
class WorkerBackedTranscriptionSession
|
||||
extends EventEmitter
|
||||
implements StreamingTranscriptionSession
|
||||
{
|
||||
public requiredSampleRate = DEFAULT_LOCAL_SAMPLE_RATE;
|
||||
private connectedSessionId: string | null = null;
|
||||
private connecting: Promise<void> | null = null;
|
||||
|
||||
constructor(
|
||||
private readonly client: LocalSpeechWorkerClient,
|
||||
private readonly kind: Extract<LocalSpeechSessionKind, "voiceStt" | "dictationStt">,
|
||||
) {
|
||||
super();
|
||||
}
|
||||
|
||||
async connect(): Promise<void> {
|
||||
if (this.connectedSessionId) {
|
||||
return;
|
||||
}
|
||||
if (!this.connecting) {
|
||||
this.connecting = this.connectRemoteSession();
|
||||
}
|
||||
await this.connecting;
|
||||
}
|
||||
|
||||
private async connectRemoteSession(): Promise<void> {
|
||||
try {
|
||||
const result = await this.client.createSession(this.kind, this);
|
||||
this.connectedSessionId = result.sessionId;
|
||||
this.requiredSampleRate = result.requiredSampleRate;
|
||||
} finally {
|
||||
this.connecting = null;
|
||||
}
|
||||
}
|
||||
|
||||
appendPcm16(pcm16le: Buffer): void {
|
||||
const sessionId = this.connectedSessionId;
|
||||
if (!sessionId) {
|
||||
this.emit("error", new Error("Local STT session not connected"));
|
||||
return;
|
||||
}
|
||||
this.client.appendSessionAudio(sessionId, pcm16le);
|
||||
}
|
||||
|
||||
commit(): void {
|
||||
const sessionId = this.connectedSessionId;
|
||||
if (!sessionId) {
|
||||
this.emit("error", new Error("Local STT session not connected"));
|
||||
return;
|
||||
}
|
||||
this.client.commitSession(sessionId);
|
||||
}
|
||||
|
||||
clear(): void {
|
||||
const sessionId = this.connectedSessionId;
|
||||
if (sessionId) {
|
||||
this.client.clearSession(sessionId);
|
||||
}
|
||||
}
|
||||
|
||||
close(): void {
|
||||
const sessionId = this.connectedSessionId;
|
||||
this.connectedSessionId = null;
|
||||
if (sessionId) {
|
||||
this.client.closeSession(sessionId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
class WorkerBackedTurnDetectionSession extends EventEmitter implements TurnDetectionSession {
|
||||
public requiredSampleRate = DEFAULT_LOCAL_SAMPLE_RATE;
|
||||
private connectedSessionId: string | null = null;
|
||||
private connecting: Promise<void> | null = null;
|
||||
|
||||
constructor(private readonly client: LocalSpeechWorkerClient) {
|
||||
super();
|
||||
}
|
||||
|
||||
async connect(): Promise<void> {
|
||||
if (this.connectedSessionId) {
|
||||
return;
|
||||
}
|
||||
if (!this.connecting) {
|
||||
this.connecting = this.connectRemoteSession();
|
||||
}
|
||||
await this.connecting;
|
||||
}
|
||||
|
||||
private async connectRemoteSession(): Promise<void> {
|
||||
try {
|
||||
const result = await this.client.createSession("vad", this);
|
||||
this.connectedSessionId = result.sessionId;
|
||||
this.requiredSampleRate = result.requiredSampleRate;
|
||||
} finally {
|
||||
this.connecting = null;
|
||||
}
|
||||
}
|
||||
|
||||
appendPcm16(pcm16le: Buffer): void {
|
||||
const sessionId = this.connectedSessionId;
|
||||
if (!sessionId) {
|
||||
this.emit("error", new Error("Local turn-detection session not connected"));
|
||||
return;
|
||||
}
|
||||
this.client.appendSessionAudio(sessionId, pcm16le);
|
||||
}
|
||||
|
||||
flush(): void {
|
||||
const sessionId = this.connectedSessionId;
|
||||
if (sessionId) {
|
||||
this.client.flushSession(sessionId);
|
||||
}
|
||||
}
|
||||
|
||||
reset(): void {
|
||||
const sessionId = this.connectedSessionId;
|
||||
if (sessionId) {
|
||||
this.client.resetSession(sessionId);
|
||||
}
|
||||
}
|
||||
|
||||
close(): void {
|
||||
const sessionId = this.connectedSessionId;
|
||||
this.connectedSessionId = null;
|
||||
if (sessionId) {
|
||||
this.client.closeSession(sessionId);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,335 @@
|
||||
import pino from "pino";
|
||||
|
||||
import type { StreamingTranscriptionSession } from "../../speech-provider.js";
|
||||
import type { TurnDetectionSession } from "../../turn-detection-provider.js";
|
||||
import { getLocalSpeechModelDir, type LocalSttModelId, type LocalTtsModelId } from "./models.js";
|
||||
import { SherpaOfflineRecognizerEngine } from "./sherpa/sherpa-offline-recognizer.js";
|
||||
import { SherpaOnnxParakeetSTT } from "./sherpa/sherpa-parakeet-stt.js";
|
||||
import { SherpaParakeetRealtimeTranscriptionSession } from "./sherpa/sherpa-parakeet-realtime-session.js";
|
||||
import { SherpaOnnxTTS } from "./sherpa/sherpa-tts.js";
|
||||
import {
|
||||
ensureSileroVadModel,
|
||||
SherpaSileroTurnDetectionProvider,
|
||||
} from "./sherpa/silero-vad-provider.js";
|
||||
import type {
|
||||
LocalSpeechWorkerConfig,
|
||||
LocalSpeechWorkerRequest,
|
||||
LocalSpeechWorkerToParentMessage,
|
||||
} from "./worker-protocol.js";
|
||||
|
||||
process.title = "Paseo Voice";
|
||||
|
||||
type LocalSttEngine = SherpaOfflineRecognizerEngine;
|
||||
|
||||
const logger = pino({
|
||||
level: process.env.PASEO_LOG_LEVEL ?? "info",
|
||||
}).child({ module: "speech", component: "local-worker" });
|
||||
|
||||
const sttEngines = new Map<string, LocalSttEngine>();
|
||||
const sttProviders = new Map<string, SherpaOnnxParakeetSTT>();
|
||||
const ttsProviders = new Map<string, SherpaOnnxTTS>();
|
||||
const sessions = new Map<string, StreamingTranscriptionSession | TurnDetectionSession>();
|
||||
const unsubscribeBySessionId = new Map<string, Array<() => void>>();
|
||||
let ipcClosing = false;
|
||||
|
||||
function sendToParent(message: LocalSpeechWorkerToParentMessage): void {
|
||||
if (ipcClosing || !process.connected || !process.send) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
process.send(message, (error) => {
|
||||
if (error) {
|
||||
ipcClosing = true;
|
||||
}
|
||||
});
|
||||
} catch {
|
||||
ipcClosing = true;
|
||||
}
|
||||
}
|
||||
|
||||
function sttModelId(
|
||||
config: LocalSpeechWorkerConfig,
|
||||
model: "voice" | "dictation",
|
||||
): LocalSttModelId {
|
||||
return (model === "voice" ? config.voiceSttModel : config.dictationSttModel) as LocalSttModelId;
|
||||
}
|
||||
|
||||
function ttsModelId(config: LocalSpeechWorkerConfig): LocalTtsModelId {
|
||||
return config.voiceTtsModel as LocalTtsModelId;
|
||||
}
|
||||
|
||||
function sttEngineKey(config: LocalSpeechWorkerConfig, modelId: LocalSttModelId): string {
|
||||
return `${config.modelsDir}:${modelId}`;
|
||||
}
|
||||
|
||||
function ttsKey(config: LocalSpeechWorkerConfig): string {
|
||||
return [
|
||||
config.modelsDir,
|
||||
config.voiceTtsModel,
|
||||
config.voiceTtsSpeakerId ?? 0,
|
||||
config.voiceTtsSpeed ?? 1,
|
||||
].join(":");
|
||||
}
|
||||
|
||||
function getSttEngine(
|
||||
config: LocalSpeechWorkerConfig,
|
||||
model: "voice" | "dictation",
|
||||
): LocalSttEngine {
|
||||
const modelId = sttModelId(config, model);
|
||||
const key = sttEngineKey(config, modelId);
|
||||
const existing = sttEngines.get(key);
|
||||
if (existing) {
|
||||
return existing;
|
||||
}
|
||||
const modelDir = getLocalSpeechModelDir(config.modelsDir, modelId);
|
||||
const created = new SherpaOfflineRecognizerEngine(
|
||||
{
|
||||
model: {
|
||||
kind: "nemo_transducer",
|
||||
encoder: `${modelDir}/encoder.int8.onnx`,
|
||||
decoder: `${modelDir}/decoder.int8.onnx`,
|
||||
joiner: `${modelDir}/joiner.int8.onnx`,
|
||||
tokens: `${modelDir}/tokens.txt`,
|
||||
},
|
||||
numThreads: 2,
|
||||
debug: 0,
|
||||
},
|
||||
logger,
|
||||
);
|
||||
sttEngines.set(key, created);
|
||||
return created;
|
||||
}
|
||||
|
||||
function getSttProvider(
|
||||
config: LocalSpeechWorkerConfig,
|
||||
model: "voice" | "dictation",
|
||||
): SherpaOnnxParakeetSTT {
|
||||
const modelId = sttModelId(config, model);
|
||||
const key = sttEngineKey(config, modelId);
|
||||
const existing = sttProviders.get(key);
|
||||
if (existing) {
|
||||
return existing;
|
||||
}
|
||||
const created = new SherpaOnnxParakeetSTT({ engine: getSttEngine(config, model) }, logger);
|
||||
sttProviders.set(key, created);
|
||||
return created;
|
||||
}
|
||||
|
||||
function getTtsProvider(config: LocalSpeechWorkerConfig): SherpaOnnxTTS {
|
||||
const key = ttsKey(config);
|
||||
const existing = ttsProviders.get(key);
|
||||
if (existing) {
|
||||
return existing;
|
||||
}
|
||||
const modelDir = getLocalSpeechModelDir(config.modelsDir, ttsModelId(config));
|
||||
const created = new SherpaOnnxTTS(
|
||||
{
|
||||
preset: ttsModelId(config),
|
||||
modelDir,
|
||||
speakerId: config.voiceTtsSpeakerId,
|
||||
speed: config.voiceTtsSpeed,
|
||||
},
|
||||
logger,
|
||||
);
|
||||
ttsProviders.set(key, created);
|
||||
return created;
|
||||
}
|
||||
|
||||
function cleanupSession(sessionId: string): void {
|
||||
const unsubscribe = unsubscribeBySessionId.get(sessionId);
|
||||
if (unsubscribe) {
|
||||
for (const fn of unsubscribe) {
|
||||
try {
|
||||
fn();
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
}
|
||||
unsubscribeBySessionId.delete(sessionId);
|
||||
const session = sessions.get(sessionId);
|
||||
sessions.delete(sessionId);
|
||||
try {
|
||||
session?.close();
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
|
||||
function trackTranscriptionSession(
|
||||
sessionId: string,
|
||||
session: StreamingTranscriptionSession,
|
||||
): void {
|
||||
session.on("committed", (payload) => {
|
||||
sendToParent({ type: "session.committed", sessionId, payload });
|
||||
});
|
||||
session.on("transcript", (payload) => {
|
||||
sendToParent({ type: "session.transcript", sessionId, payload });
|
||||
});
|
||||
session.on("error", (err) => {
|
||||
sendToParent({
|
||||
type: "session.error",
|
||||
sessionId,
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
});
|
||||
});
|
||||
unsubscribeBySessionId.set(sessionId, []);
|
||||
}
|
||||
|
||||
function trackTurnDetectionSession(sessionId: string, session: TurnDetectionSession): void {
|
||||
session.on("speech_started", () => {
|
||||
sendToParent({ type: "session.speech_started", sessionId });
|
||||
});
|
||||
session.on("speech_stopped", () => {
|
||||
sendToParent({ type: "session.speech_stopped", sessionId });
|
||||
});
|
||||
session.on("error", (err) => {
|
||||
sendToParent({
|
||||
type: "session.error",
|
||||
sessionId,
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
});
|
||||
});
|
||||
unsubscribeBySessionId.set(sessionId, []);
|
||||
}
|
||||
|
||||
async function createSession(
|
||||
message: Extract<LocalSpeechWorkerRequest, { type: "session.create" }>,
|
||||
) {
|
||||
cleanupSession(message.sessionId);
|
||||
if (message.kind === "vad") {
|
||||
let vadModelPath: string | undefined;
|
||||
try {
|
||||
vadModelPath = await ensureSileroVadModel(message.config.modelsDir, logger);
|
||||
} catch (err) {
|
||||
logger.warn({ err }, "Failed to provision Silero VAD model, falling back to bundled");
|
||||
}
|
||||
const provider = new SherpaSileroTurnDetectionProvider({ modelPath: vadModelPath }, logger);
|
||||
const session = provider.createSession({ logger });
|
||||
trackTurnDetectionSession(message.sessionId, session);
|
||||
await session.connect();
|
||||
sessions.set(message.sessionId, session);
|
||||
return { requiredSampleRate: session.requiredSampleRate };
|
||||
}
|
||||
|
||||
const model = message.kind === "voiceStt" ? "voice" : "dictation";
|
||||
const engine = getSttEngine(message.config, model);
|
||||
const session =
|
||||
message.kind === "voiceStt"
|
||||
? getSttProvider(message.config, "voice").createSession({ logger })
|
||||
: new SherpaParakeetRealtimeTranscriptionSession({ engine });
|
||||
trackTranscriptionSession(message.sessionId, session);
|
||||
await session.connect();
|
||||
sessions.set(message.sessionId, session);
|
||||
return { requiredSampleRate: session.requiredSampleRate };
|
||||
}
|
||||
|
||||
function sendOk(requestId: string, result?: unknown): void {
|
||||
sendToParent({ type: "response", requestId, ok: true, result });
|
||||
}
|
||||
|
||||
type LocalSpeechWorkerSessionRequest = Extract<
|
||||
LocalSpeechWorkerRequest,
|
||||
{
|
||||
type:
|
||||
| "session.append"
|
||||
| "session.commit"
|
||||
| "session.clear"
|
||||
| "session.flush"
|
||||
| "session.reset"
|
||||
| "session.close";
|
||||
}
|
||||
>;
|
||||
|
||||
function handleSessionRequest(message: LocalSpeechWorkerSessionRequest): void {
|
||||
if (message.type === "session.close") {
|
||||
cleanupSession(message.sessionId);
|
||||
sendOk(message.requestId);
|
||||
return;
|
||||
}
|
||||
|
||||
const session = sessions.get(message.sessionId);
|
||||
switch (message.type) {
|
||||
case "session.append":
|
||||
session?.appendPcm16(message.audio);
|
||||
break;
|
||||
case "session.commit":
|
||||
if (session && "commit" in session) {
|
||||
session.commit();
|
||||
}
|
||||
break;
|
||||
case "session.clear":
|
||||
if (session && "clear" in session) {
|
||||
session.clear();
|
||||
}
|
||||
break;
|
||||
case "session.flush":
|
||||
if (session && "flush" in session) {
|
||||
session.flush();
|
||||
}
|
||||
break;
|
||||
case "session.reset":
|
||||
if (session && "reset" in session) {
|
||||
session.reset();
|
||||
}
|
||||
break;
|
||||
}
|
||||
sendOk(message.requestId);
|
||||
}
|
||||
|
||||
async function handleRequest(message: LocalSpeechWorkerRequest): Promise<void> {
|
||||
if (message.type === "tts.synthesize") {
|
||||
const result = await getTtsProvider(message.config).synthesizeSpeech(message.text);
|
||||
const chunks: Buffer[] = [];
|
||||
for await (const chunk of result.stream) {
|
||||
chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
|
||||
}
|
||||
sendOk(message.requestId, { audio: Buffer.concat(chunks), format: result.format });
|
||||
return;
|
||||
}
|
||||
|
||||
if (message.type === "stt.transcribe") {
|
||||
const result = await getSttProvider(message.config, message.model).transcribeAudio(
|
||||
message.audio,
|
||||
message.format,
|
||||
);
|
||||
sendOk(message.requestId, result);
|
||||
return;
|
||||
}
|
||||
|
||||
if (message.type === "session.create") {
|
||||
const result = await createSession(message);
|
||||
sendOk(message.requestId, result);
|
||||
return;
|
||||
}
|
||||
|
||||
handleSessionRequest(message);
|
||||
}
|
||||
|
||||
process.on("message", (message: LocalSpeechWorkerRequest) => {
|
||||
void handleRequest(message).catch((error: unknown) => {
|
||||
sendToParent({
|
||||
type: "response",
|
||||
requestId: message.requestId,
|
||||
ok: false,
|
||||
error: error instanceof Error ? error.message : "Local speech worker request failed",
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
process.once("disconnect", () => {
|
||||
ipcClosing = true;
|
||||
for (const sessionId of Array.from(sessions.keys())) {
|
||||
cleanupSession(sessionId);
|
||||
}
|
||||
for (const tts of ttsProviders.values()) {
|
||||
try {
|
||||
tts.free();
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
for (const engine of sttEngines.values()) {
|
||||
engine.free();
|
||||
}
|
||||
});
|
||||
@@ -0,0 +1,122 @@
|
||||
import type {
|
||||
StreamingTranscriptionCommittedEvent,
|
||||
StreamingTranscriptionEvent,
|
||||
TranscriptionResult,
|
||||
} from "../../speech-provider.js";
|
||||
|
||||
export interface LocalSpeechWorkerConfig {
|
||||
modelsDir: string;
|
||||
voiceSttModel: string;
|
||||
dictationSttModel: string;
|
||||
voiceTtsModel: string;
|
||||
voiceTtsSpeakerId?: number;
|
||||
voiceTtsSpeed?: number;
|
||||
}
|
||||
|
||||
export type LocalSpeechSessionKind = "voiceStt" | "dictationStt" | "vad";
|
||||
|
||||
export type LocalSpeechWorkerRequest =
|
||||
| {
|
||||
type: "tts.synthesize";
|
||||
requestId: string;
|
||||
config: LocalSpeechWorkerConfig;
|
||||
text: string;
|
||||
}
|
||||
| {
|
||||
type: "stt.transcribe";
|
||||
requestId: string;
|
||||
config: LocalSpeechWorkerConfig;
|
||||
model: "voice" | "dictation";
|
||||
audio: Buffer;
|
||||
format: string;
|
||||
}
|
||||
| {
|
||||
type: "session.create";
|
||||
requestId: string;
|
||||
config: LocalSpeechWorkerConfig;
|
||||
sessionId: string;
|
||||
kind: LocalSpeechSessionKind;
|
||||
}
|
||||
| {
|
||||
type: "session.append";
|
||||
requestId: string;
|
||||
sessionId: string;
|
||||
audio: Buffer;
|
||||
}
|
||||
| {
|
||||
type: "session.commit";
|
||||
requestId: string;
|
||||
sessionId: string;
|
||||
}
|
||||
| {
|
||||
type: "session.clear";
|
||||
requestId: string;
|
||||
sessionId: string;
|
||||
}
|
||||
| {
|
||||
type: "session.flush";
|
||||
requestId: string;
|
||||
sessionId: string;
|
||||
}
|
||||
| {
|
||||
type: "session.reset";
|
||||
requestId: string;
|
||||
sessionId: string;
|
||||
}
|
||||
| {
|
||||
type: "session.close";
|
||||
requestId: string;
|
||||
sessionId: string;
|
||||
};
|
||||
|
||||
export type LocalSpeechWorkerResponse =
|
||||
| {
|
||||
type: "response";
|
||||
requestId: string;
|
||||
ok: true;
|
||||
result?: unknown;
|
||||
}
|
||||
| {
|
||||
type: "response";
|
||||
requestId: string;
|
||||
ok: false;
|
||||
error: string;
|
||||
};
|
||||
|
||||
export type LocalSpeechWorkerEvent =
|
||||
| {
|
||||
type: "session.committed";
|
||||
sessionId: string;
|
||||
payload: StreamingTranscriptionCommittedEvent;
|
||||
}
|
||||
| {
|
||||
type: "session.transcript";
|
||||
sessionId: string;
|
||||
payload: StreamingTranscriptionEvent;
|
||||
}
|
||||
| {
|
||||
type: "session.speech_started";
|
||||
sessionId: string;
|
||||
}
|
||||
| {
|
||||
type: "session.speech_stopped";
|
||||
sessionId: string;
|
||||
}
|
||||
| {
|
||||
type: "session.error";
|
||||
sessionId: string;
|
||||
error: string;
|
||||
};
|
||||
|
||||
export type LocalSpeechWorkerToParentMessage = LocalSpeechWorkerResponse | LocalSpeechWorkerEvent;
|
||||
|
||||
export interface LocalSpeechCreateSessionResult {
|
||||
requiredSampleRate: number;
|
||||
}
|
||||
|
||||
export interface LocalSpeechTtsResult {
|
||||
audio: Buffer;
|
||||
format: string;
|
||||
}
|
||||
|
||||
export type LocalSpeechTranscriptionResult = TranscriptionResult;
|
||||
Reference in New Issue
Block a user