feat(media): transcribe audio and video into subtitles as a media job
With the media runtime in place, transcription becomes an operation like the others: bots call the transcribe tool, the job runs in the background and the card shows the subtitle files. - createTranscriptionOperations registers transcribe: available when a transcription server is configured, refuses artifacts the user cannot read before the job exists, and writes <name>.srt and <name>.vtt next to the source (a .txt when the server returns no timings). With ffmpeg the audio is converted to 16 kHz mono WAV and long media is transcribed in upload-sized ranges on one timeline; without it audio is sent as stored and video is refused with a clear message. A server that is down or busy (unreachable, 429, 502, 503) defers the job instead of failing it, since nothing has been stored yet. The bot gets the transcript, cut at a word. - The transcribe tool definition uses the shared media schema helper and continue_after; dispatch stays generic. - The main bot's run creates Studio once a media server is configured, right before the specialist directory is built. - Dictation: when the user's voice provider cannot transcribe (or there is none), /api/voice/transcribe uses the self-hosted transcription server, and the voice status reports server dictation so the web app uses it instead of the browser's speech recognition. A voice provider that transcribes keeps being used as before. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MRtGG9ARfqqhma3t85Ro2u
This commit is contained in:
Claude
parent
1722ac1a0f
commit
c380e7820a
@ -221,6 +221,7 @@ import {
|
||||
listVoiceCatalog,
|
||||
loadDefaultVoiceCredential,
|
||||
loadVoiceCredential,
|
||||
mediaTranscriptionSettings,
|
||||
persistVoiceCredential,
|
||||
prepareVoice,
|
||||
toVoiceCredential,
|
||||
@ -4782,8 +4783,11 @@ export function createRouter(deps: RouterDeps) {
|
||||
voice: {
|
||||
catalog: authed.voice.catalog.handler(async () => listVoiceCatalog()),
|
||||
status: authed.voice.status.handler(async ({ context }) => {
|
||||
const cred = await findDefaultVoiceCredential(deps.prisma, context.actor);
|
||||
return toVoiceStatus(cred);
|
||||
const [cred, server] = await Promise.all([
|
||||
findDefaultVoiceCredential(deps.prisma, context.actor),
|
||||
mediaTranscriptionSettings(deps),
|
||||
]);
|
||||
return toVoiceStatus(cred, { mediaTranscription: Boolean(server) });
|
||||
}),
|
||||
credentials: authed.voice.credentials.handler(async ({ context }) => {
|
||||
const rows = await deps.prisma.userVoiceCredential.findMany({
|
||||
@ -4845,7 +4849,9 @@ export function createRouter(deps: RouterDeps) {
|
||||
{ isolationLevel: Prisma.TransactionIsolationLevel.Serializable },
|
||||
),
|
||||
);
|
||||
return toVoiceStatus(cred);
|
||||
return toVoiceStatus(cred, {
|
||||
mediaTranscription: Boolean(await mediaTranscriptionSettings(deps)),
|
||||
});
|
||||
}),
|
||||
voices: authed.voice.voices.handler(async ({ context, input }) => {
|
||||
const loaded = await loadDefaultVoiceCredential(deps, context.actor);
|
||||
|
||||
89
apps/api/src/voice-transcribe.test.ts
Normal file
89
apps/api/src/voice-transcribe.test.ts
Normal file
@ -0,0 +1,89 @@
|
||||
import { createServer, type Server } from "node:http";
|
||||
import type { AddressInfo } from "node:net";
|
||||
import { NoVoiceConfigured } from "@rakazo/adapters";
|
||||
import type { Actor } from "@rakazo/contracts";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { toVoiceStatus, transcribeVoice, type VoiceDeps } from "./voice.js";
|
||||
|
||||
const actor: Actor = {
|
||||
userId: "user-1",
|
||||
spaceId: "space-1",
|
||||
email: "user@rakazo.test",
|
||||
isDeploymentOwner: false,
|
||||
};
|
||||
|
||||
/** A user without a voice credential, on a deployment whose media settings are `settings`. */
|
||||
function deps(settings: object | null, ciphertext: string): VoiceDeps {
|
||||
return {
|
||||
prisma: {
|
||||
spaceVoicePreference: { findFirst: async () => null },
|
||||
integrationProviderConfig: {
|
||||
findUnique: async () => (settings ? { id: "media", ciphertext } : null),
|
||||
},
|
||||
},
|
||||
secrets: { load: () => JSON.stringify(settings) },
|
||||
} as unknown as VoiceDeps;
|
||||
}
|
||||
|
||||
describe("dictation through the self-hosted transcription server", () => {
|
||||
let server: Server;
|
||||
let baseUrl = "";
|
||||
const uploads: string[] = [];
|
||||
|
||||
beforeAll(async () => {
|
||||
server = createServer((request, response) => {
|
||||
const chunks: Buffer[] = [];
|
||||
request.on("data", (chunk: Buffer) => chunks.push(chunk));
|
||||
request.on("end", () => {
|
||||
uploads.push(`${request.url} ${Buffer.concat(chunks).toString("latin1")}`);
|
||||
response.setHeader("content-type", "application/json");
|
||||
response.end(JSON.stringify({ text: " Remind me to call Ada. " }));
|
||||
});
|
||||
});
|
||||
await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
|
||||
baseUrl = `http://127.0.0.1:${(server.address() as AddressInfo).port}/v1`;
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await new Promise((resolve) => server.close(resolve));
|
||||
});
|
||||
|
||||
it("transcribes when no voice provider can", async () => {
|
||||
const result = await transcribeVoice(
|
||||
deps({ comfyui: null, transcription: { baseUrl } }, "media-a"),
|
||||
actor,
|
||||
{ audio: new Uint8Array([1, 2, 3]), mimeType: "audio/webm;codecs=opus" },
|
||||
);
|
||||
expect(result).toEqual({ text: "Remind me to call Ada." });
|
||||
expect(uploads[0]).toContain("/v1/audio/transcriptions");
|
||||
expect(uploads[0]).toContain('filename="audio.webm"');
|
||||
});
|
||||
|
||||
it("keeps asking for a voice key when no server is set up either", async () => {
|
||||
await expect(
|
||||
transcribeVoice(deps(null, "none"), actor, {
|
||||
audio: new Uint8Array([1]),
|
||||
mimeType: "audio/webm",
|
||||
}),
|
||||
).rejects.toBeInstanceOf(NoVoiceConfigured);
|
||||
});
|
||||
|
||||
it("refuses an empty recording before uploading it", async () => {
|
||||
const before = uploads.length;
|
||||
await expect(
|
||||
transcribeVoice(deps({ comfyui: null, transcription: { baseUrl } }, "media-b"), actor, {
|
||||
audio: new Uint8Array(),
|
||||
mimeType: "audio/webm",
|
||||
}),
|
||||
).rejects.toThrow("That recording is empty or too large.");
|
||||
expect(uploads).toHaveLength(before);
|
||||
});
|
||||
|
||||
it("reports server dictation in the voice status", () => {
|
||||
expect(toVoiceStatus(null).transcribe).toBe(false);
|
||||
expect(toVoiceStatus(null, { mediaTranscription: true })).toMatchObject({
|
||||
configured: false,
|
||||
transcribe: true,
|
||||
});
|
||||
});
|
||||
});
|
||||
@ -1,10 +1,12 @@
|
||||
import { ORPCError } from "@orpc/server";
|
||||
import type { AdapterContext } from "@rakazo/adapter-kit";
|
||||
import {
|
||||
createOpenAiCompatibleTranscriptionProvider,
|
||||
createVoiceProvider,
|
||||
type EncryptedSecretStore,
|
||||
isVoiceProviderId,
|
||||
listVoiceCatalog,
|
||||
loadMediaSettings,
|
||||
MAX_SPEAK_CHARS,
|
||||
MAX_TRANSCRIBE_BYTES,
|
||||
NoVoiceConfigured,
|
||||
@ -51,12 +53,20 @@ export function catalogEntry(provider: string) {
|
||||
return voiceCatalogEntry(provider);
|
||||
}
|
||||
|
||||
export function toVoiceStatus(cred: { provider: string; voiceId: string } | null): VoiceStatus {
|
||||
/** The deployment's self-hosted transcription server, when the owner set one up. */
|
||||
export async function mediaTranscriptionSettings(deps: VoiceDeps) {
|
||||
return (await loadMediaSettings(deps).catch(() => null))?.transcription ?? null;
|
||||
}
|
||||
|
||||
export function toVoiceStatus(
|
||||
cred: { provider: string; voiceId: string } | null,
|
||||
options: { mediaTranscription?: boolean } = {},
|
||||
): VoiceStatus {
|
||||
const entry = cred ? catalogEntry(cred.provider) : undefined;
|
||||
return {
|
||||
configured: Boolean(cred),
|
||||
ready: Boolean(cred?.voiceId),
|
||||
transcribe: Boolean(entry?.transcribe && cred),
|
||||
transcribe: Boolean(entry?.transcribe && cred) || Boolean(options.mediaTranscription),
|
||||
provider: cred?.provider ?? null,
|
||||
voiceId: cred?.voiceId ?? "",
|
||||
};
|
||||
@ -277,9 +287,22 @@ export async function transcribeVoice(
|
||||
input: { audio: Uint8Array; mimeType: string; signal?: AbortSignal },
|
||||
) {
|
||||
const loaded = await loadDefaultVoiceCredential(deps, actor);
|
||||
if (!loaded) throw new NoVoiceConfigured("key");
|
||||
const provider = createVoiceProvider(loaded.cred.provider);
|
||||
if (!provider.transcribe) {
|
||||
const provider = loaded ? createVoiceProvider(loaded.cred.provider) : null;
|
||||
if (!loaded || !provider?.transcribe) {
|
||||
// Without a voice provider that transcribes, the self-hosted media server can.
|
||||
const server = await mediaTranscriptionSettings(deps);
|
||||
if (server) {
|
||||
if (input.audio.byteLength === 0 || input.audio.byteLength > MAX_TRANSCRIBE_BYTES) {
|
||||
throw new ORPCError("BAD_REQUEST", { message: "That recording is empty or too large." });
|
||||
}
|
||||
const transcript = await createOpenAiCompatibleTranscriptionProvider(server).transcribe({
|
||||
audio: input.audio,
|
||||
mimeType: input.mimeType || "audio/webm",
|
||||
signal: input.signal,
|
||||
});
|
||||
return { text: transcript.text };
|
||||
}
|
||||
if (!loaded) throw new NoVoiceConfigured("key");
|
||||
throw new ORPCError("BAD_REQUEST", {
|
||||
message: "This voice provider does not transcribe audio. Use on-device dictation instead.",
|
||||
});
|
||||
|
||||
@ -5,9 +5,13 @@ import {
|
||||
MODEL_TIERS,
|
||||
SecretAskPurpose,
|
||||
SecretHttpRequest,
|
||||
TranscribeInputSchema,
|
||||
USER_FACT_CATEGORIES,
|
||||
} from "@rakazo/contracts";
|
||||
import { z } from "zod";
|
||||
import { mediaToolInputSchema } from "./media-tools.js";
|
||||
|
||||
type MediaToolSchema = Parameters<typeof mediaToolInputSchema>[0];
|
||||
|
||||
export const SUBAGENT_MODEL_TIER_DESCRIPTION =
|
||||
"Optional model tier: fast for quick simple work, code for programming, standard otherwise. Uses the model this space maps to the tier, or your own model when it maps none. Ignored when model_provider and model_id are set.";
|
||||
@ -326,6 +330,12 @@ export const builtinAgentTools: ConnectorTool[] = [
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "transcribe",
|
||||
description:
|
||||
"Transcribe speech in an audio or video artifact into SRT and VTT subtitle files, shown on a media card. Set continue_after to get the transcript back, e.g. to burn the .srt in with edit_video.",
|
||||
inputSchema: mediaToolInputSchema(z.toJSONSchema(TranscribeInputSchema) as MediaToolSchema),
|
||||
},
|
||||
{
|
||||
name: "add_mcp_server",
|
||||
description:
|
||||
|
||||
@ -357,6 +357,7 @@ import {
|
||||
skillReadFromTool,
|
||||
skillUpdateFromTool,
|
||||
} from "./skill-tools.js";
|
||||
import { ensureStudioWhenMediaConfigured } from "./studio-specialist.js";
|
||||
import {
|
||||
continueRunClaimFence,
|
||||
DESKTOP_HELD_FOR_TAKEOVER_MESSAGE,
|
||||
@ -3620,6 +3621,8 @@ export function createRunExecutor(deps: ExecutorDeps) {
|
||||
});
|
||||
}
|
||||
const runtimeHistory = [...historicalContext, ...history];
|
||||
// Studio joins the directory once a media server is set up.
|
||||
if (exposedToolNames.has("delegate_task")) await ensureStudioWhenMediaConfigured(deps, run);
|
||||
// The main agent sees every specialist it can delegate to, across the user's spaces.
|
||||
// Other runs see the teammates they can address with message_bot; runs without it
|
||||
// (every group run) list their reachable members in <group_members> instead.
|
||||
|
||||
@ -1,4 +1,5 @@
|
||||
import type { MediaSettings } from "@rakazo/contracts";
|
||||
import { createTranscriptionOperations } from "./media/transcription.js";
|
||||
import type { MediaOperationRegistration } from "./media-runtime.js";
|
||||
|
||||
export interface MediaOperationDeps {
|
||||
@ -10,6 +11,6 @@ export interface MediaOperationDeps {
|
||||
* Every media operation this deployment can run. The API and the worker build their runtime
|
||||
* from this one list, so both processes run the same operations.
|
||||
*/
|
||||
export function createMediaOperations(_deps: MediaOperationDeps): MediaOperationRegistration[] {
|
||||
return [];
|
||||
export function createMediaOperations(deps: MediaOperationDeps): MediaOperationRegistration[] {
|
||||
return [...createTranscriptionOperations({ loadSettings: deps.loadSettings })];
|
||||
}
|
||||
|
||||
@ -1,10 +1,24 @@
|
||||
import { mkdtemp, rm, writeFile } from "node:fs/promises";
|
||||
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import { EmulatorTranscriptionProvider } from "../transcription-emulator.js";
|
||||
import {
|
||||
type MediaJobContext,
|
||||
MediaJobDeferredError,
|
||||
type MediaJobInputFile,
|
||||
type MediaJobOutputFile,
|
||||
} from "@rakazo/adapter-kit";
|
||||
import type { MediaOutput, MediaSettings } from "@rakazo/contracts";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import { createOpenAiCompatibleTranscriptionProvider } from "../openai-compatible-transcription.js";
|
||||
import {
|
||||
EmulatorTranscriptionProvider,
|
||||
OpenAiCompatibleTranscriptionServerEmulator,
|
||||
} from "../transcription-emulator.js";
|
||||
import {
|
||||
createTranscriptionOperations,
|
||||
planAudioChunks,
|
||||
TRANSCRIPT_SUMMARY_MAX_CHARS,
|
||||
type TranscriptionAudioTools,
|
||||
transcribeChunks,
|
||||
transcriptionChunkSeconds,
|
||||
transcriptSummary,
|
||||
@ -100,3 +114,215 @@ describe("transcriptSummary", () => {
|
||||
expect(body.endsWith("word …")).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
const TRANSCRIPTION: MediaSettings = {
|
||||
comfyui: null,
|
||||
transcription: { baseUrl: "http://127.0.0.1:8080/v1", model: undefined, apiKey: undefined },
|
||||
};
|
||||
|
||||
async function jobContext(source: Omit<MediaJobInputFile, "path">, input: unknown) {
|
||||
const tempDir = await mkdtemp(join(tmpdir(), "transcribe-job-"));
|
||||
dirs.push(tempDir);
|
||||
const sourcePath = join(tempDir, `input${source.name.slice(source.name.lastIndexOf("."))}`);
|
||||
await writeFile(sourcePath, new Uint8Array([82, 73, 70, 70, 1, 2, 3]));
|
||||
const outputs: Array<MediaJobOutputFile & { body: string }> = [];
|
||||
const progress: number[] = [];
|
||||
const context: MediaJobContext = {
|
||||
jobId: "job-1",
|
||||
spaceId: "space-1",
|
||||
userId: "user-1",
|
||||
botId: "bot-1",
|
||||
input,
|
||||
tempDir,
|
||||
signal: new AbortController().signal,
|
||||
progress: async (fraction) => {
|
||||
progress.push(fraction);
|
||||
},
|
||||
fetchArtifact: async (artifactId) => {
|
||||
if (artifactId !== source.artifactId) throw new Error("unknown artifact");
|
||||
return { ...source, path: sourcePath };
|
||||
},
|
||||
addOutput: async (file) => {
|
||||
outputs.push({ ...file, body: await readFile(file.path, "utf8") });
|
||||
return { artifactId: `out-${outputs.length}` } as MediaOutput;
|
||||
},
|
||||
};
|
||||
return { context, outputs, progress };
|
||||
}
|
||||
|
||||
const audioSource = {
|
||||
artifactId: "audio-1",
|
||||
name: "Interview take 2.m4a",
|
||||
mimeType: "audio/mp4",
|
||||
kind: "audio" as const,
|
||||
size: 7,
|
||||
};
|
||||
|
||||
function operation(
|
||||
settings: MediaSettings = TRANSCRIPTION,
|
||||
options: { flavor?: "openai" | "text"; audio?: TranscriptionAudioTools; maxBytes?: number } = {},
|
||||
) {
|
||||
const server = new OpenAiCompatibleTranscriptionServerEmulator({ flavor: options.flavor });
|
||||
const [registration] = createTranscriptionOperations({
|
||||
loadSettings: async () => settings,
|
||||
audio: options.audio,
|
||||
createProvider: (transcription) => {
|
||||
const provider = createOpenAiCompatibleTranscriptionProvider({
|
||||
...transcription,
|
||||
fetch: server.fetch,
|
||||
});
|
||||
if (!options.maxBytes) return provider;
|
||||
const maxBytes = options.maxBytes;
|
||||
return {
|
||||
transcribe: provider.transcribe,
|
||||
describe: () => {
|
||||
const descriptor = provider.describe();
|
||||
return { ...descriptor, capabilities: { ...descriptor.capabilities, maxBytes } };
|
||||
},
|
||||
};
|
||||
},
|
||||
});
|
||||
return { registration: registration!, server };
|
||||
}
|
||||
|
||||
describe("transcribe operation", () => {
|
||||
it("is available only with a transcription server and reads the source artifact", async () => {
|
||||
expect(await operation().registration.available({ spaceId: "space-1" })).toBe(true);
|
||||
expect(
|
||||
await operation({ comfyui: null, transcription: null }).registration.available({
|
||||
spaceId: "space-1",
|
||||
}),
|
||||
).toBe(false);
|
||||
const { registration } = operation();
|
||||
const input = registration.inputSchema.parse({ artifactId: "audio-1", language: "de" });
|
||||
expect(registration.artifactIds?.(input)).toEqual(["audio-1"]);
|
||||
expect(registration.title(input)).toBe("Transcript");
|
||||
expect(
|
||||
registration.inputSchema.safeParse({ artifactId: "audio-1", language: "German!" }).success,
|
||||
).toBe(false);
|
||||
});
|
||||
|
||||
it("turns an audio artifact into SRT and VTT files next to it", async () => {
|
||||
const { registration, server } = operation();
|
||||
const job = await jobContext(audioSource, { artifactId: "audio-1", language: "de" });
|
||||
const result = await registration.handler.run(job.context);
|
||||
expect(server.requests).toEqual([
|
||||
expect.objectContaining({
|
||||
language: "de",
|
||||
fileName: "Interview_take_2.m4a",
|
||||
fileType: "audio/mp4",
|
||||
}),
|
||||
]);
|
||||
expect(
|
||||
job.outputs.map(({ name, mimeType, parentArtifactId }) => [name, mimeType, parentArtifactId]),
|
||||
).toEqual([
|
||||
["Interview take 2.srt", "application/x-subrip", "audio-1"],
|
||||
["Interview take 2.vtt", "text/vtt", "audio-1"],
|
||||
]);
|
||||
expect(job.outputs[0]?.body).toContain("00:00:00,000 --> 00:00:02,400\nWelcome to the studio.");
|
||||
expect(job.outputs[1]?.body.startsWith("WEBVTT\n")).toBe(true);
|
||||
expect(job.progress.at(-1)).toBeCloseTo(0.95);
|
||||
expect(result.summary).toBe(
|
||||
"Transcript of Interview take 2.m4a: Welcome to the studio. Today we cut a vertical clip.\nSubtitles are attached as SRT and VTT.",
|
||||
);
|
||||
});
|
||||
|
||||
it("writes a text file when the server returns no timings", async () => {
|
||||
const { registration } = operation(TRANSCRIPTION, { flavor: "text" });
|
||||
const job = await jobContext(audioSource, { artifactId: "audio-1" });
|
||||
const result = await registration.handler.run(job.context);
|
||||
expect(job.outputs.map((output) => [output.name, output.mimeType])).toEqual([
|
||||
["Interview take 2.txt", "text/plain"],
|
||||
]);
|
||||
expect(job.outputs[0]?.body).toBe("Welcome to the studio. Today we cut a vertical clip.\n");
|
||||
expect(result.summary).toContain("attached as text");
|
||||
});
|
||||
|
||||
it("refuses what it cannot transcribe without ffmpeg", async () => {
|
||||
const video = {
|
||||
...audioSource,
|
||||
artifactId: "video-1",
|
||||
name: "clip.mp4",
|
||||
kind: "video" as const,
|
||||
};
|
||||
const { registration } = operation();
|
||||
await expect(
|
||||
registration.handler.run((await jobContext(video, { artifactId: "video-1" })).context),
|
||||
).rejects.toThrow("Transcribing video needs ffmpeg on the server.");
|
||||
const image = {
|
||||
...audioSource,
|
||||
artifactId: "image-1",
|
||||
name: "still.png",
|
||||
kind: "image" as const,
|
||||
};
|
||||
await expect(
|
||||
registration.handler.run((await jobContext(image, { artifactId: "image-1" })).context),
|
||||
).rejects.toThrow("Only audio and video can be transcribed.");
|
||||
const long = operation(TRANSCRIPTION, { maxBytes: 4 });
|
||||
await expect(
|
||||
long.registration.handler.run(
|
||||
(await jobContext(audioSource, { artifactId: "audio-1" })).context,
|
||||
),
|
||||
).rejects.toThrow("too long to transcribe without ffmpeg");
|
||||
});
|
||||
|
||||
it("converts with ffmpeg and transcribes long media in chunks on one timeline", async () => {
|
||||
const extracted: Array<{ startMs: number; endMs: number; mimeType: string }> = [];
|
||||
const audio: TranscriptionAudioTools = {
|
||||
available: async () => true,
|
||||
durationMs: async () => 150_000,
|
||||
extract: async (_input, output, options) => {
|
||||
extracted.push({
|
||||
startMs: options.startMs,
|
||||
endMs: options.endMs,
|
||||
mimeType: options.mimeType,
|
||||
});
|
||||
await writeFile(output, new Uint8Array([1, 2, 3]));
|
||||
},
|
||||
};
|
||||
// 2 MiB per upload: 62 seconds of 16 kHz mono WAV.
|
||||
const { registration, server } = operation(TRANSCRIPTION, { audio, maxBytes: 2 * 1024 * 1024 });
|
||||
const video = {
|
||||
...audioSource,
|
||||
artifactId: "video-1",
|
||||
name: "talk.mp4",
|
||||
mimeType: "video/mp4",
|
||||
kind: "video" as const,
|
||||
};
|
||||
const job = await jobContext(video, { artifactId: "video-1" });
|
||||
await registration.handler.run(job.context);
|
||||
expect(extracted).toEqual([
|
||||
{ startMs: 0, endMs: 62_000, mimeType: "video/mp4" },
|
||||
{ startMs: 62_000, endMs: 124_000, mimeType: "video/mp4" },
|
||||
{ startMs: 124_000, endMs: 150_000, mimeType: "video/mp4" },
|
||||
]);
|
||||
expect(server.requests.map((request) => [request.fileName, request.fileType])).toEqual([
|
||||
["talk.wav", "audio/wav"],
|
||||
["talk.wav", "audio/wav"],
|
||||
["talk.wav", "audio/wav"],
|
||||
]);
|
||||
expect(job.outputs[0]?.body).toContain("00:02:04,000 --> 00:02:06,400");
|
||||
});
|
||||
|
||||
it("waits for a transcription server that is down instead of failing", async () => {
|
||||
const { registration, server } = operation();
|
||||
server.failNextStatus = 503;
|
||||
const job = await jobContext(audioSource, { artifactId: "audio-1" });
|
||||
await expect(registration.handler.run(job.context)).rejects.toBeInstanceOf(
|
||||
MediaJobDeferredError,
|
||||
);
|
||||
expect(job.outputs).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("planAudioChunks", () => {
|
||||
it("covers the whole duration in upload-sized ranges", () => {
|
||||
expect(planAudioChunks(0, 25 * 1024 * 1024)).toEqual([]);
|
||||
expect(planAudioChunks(30_000, 25 * 1024 * 1024)).toEqual([{ startMs: 0, endMs: 30_000 }]);
|
||||
expect(planAudioChunks(1_500_000, 25 * 1024 * 1024)).toEqual([
|
||||
{ startMs: 0, endMs: 720_000 },
|
||||
{ startMs: 720_000, endMs: 1_440_000 },
|
||||
{ startMs: 1_440_000, endMs: 1_500_000 },
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
@ -1,5 +1,23 @@
|
||||
import { readFile } from "node:fs/promises";
|
||||
import type { Transcript, TranscriptionProvider, TranscriptSegment } from "@rakazo/adapter-kit";
|
||||
import { readFile, writeFile } from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import {
|
||||
MediaJobDeferredError,
|
||||
type Transcript,
|
||||
type TranscriptionProvider,
|
||||
type TranscriptSegment,
|
||||
} from "@rakazo/adapter-kit";
|
||||
import {
|
||||
type MediaSettings,
|
||||
type TranscribeInput,
|
||||
TranscribeInputSchema,
|
||||
type TranscriptionSettings,
|
||||
} from "@rakazo/contracts";
|
||||
import { toSrt, toVtt } from "@rakazo/core";
|
||||
import { MediaJobRejectedError, type MediaOperationRegistration } from "../media-runtime.js";
|
||||
import {
|
||||
createOpenAiCompatibleTranscriptionProvider,
|
||||
TranscriptionServerUnavailableError,
|
||||
} from "../openai-compatible-transcription.js";
|
||||
|
||||
/** Bytes per second of the 16 kHz mono 16-bit WAV that audio is converted to before upload. */
|
||||
export const TRANSCRIPTION_WAV_BYTES_PER_SECOND = 16_000 * 2;
|
||||
@ -77,3 +95,163 @@ export function transcriptSummary(name: string, transcript: Transcript): string
|
||||
: "The server returned no timings, so the transcript is attached as text.";
|
||||
return `Transcript of ${name}: ${cut}\n${timing}`;
|
||||
}
|
||||
|
||||
/** Retry delay while the transcription server is down or busy. */
|
||||
const UNAVAILABLE_RETRY_MS = 15_000;
|
||||
|
||||
/** ffmpeg steps the operation uses when the server has ffmpeg. */
|
||||
export interface TranscriptionAudioTools {
|
||||
available(): Promise<boolean>;
|
||||
durationMs(input: string, options: { signal: AbortSignal; mimeType: string }): Promise<number>;
|
||||
/** Writes `[startMs, endMs)` of the first audio track as 16 kHz mono WAV. */
|
||||
extract(
|
||||
input: string,
|
||||
output: string,
|
||||
options: { signal: AbortSignal; mimeType: string; startMs: number; endMs: number },
|
||||
): Promise<void>;
|
||||
}
|
||||
|
||||
/** Time ranges of converted audio that each fit one upload. */
|
||||
export function planAudioChunks(
|
||||
durationMs: number,
|
||||
maxBytes: number,
|
||||
): Array<{ startMs: number; endMs: number }> {
|
||||
const chunkMs = transcriptionChunkSeconds(maxBytes) * 1000;
|
||||
const ranges: Array<{ startMs: number; endMs: number }> = [];
|
||||
for (let startMs = 0; startMs < durationMs; startMs += chunkMs) {
|
||||
ranges.push({ startMs, endMs: Math.min(durationMs, startMs + chunkMs) });
|
||||
}
|
||||
return ranges;
|
||||
}
|
||||
|
||||
export interface TranscriptionOperationDeps {
|
||||
loadSettings(): Promise<MediaSettings>;
|
||||
/** Without ffmpeg, audio is sent as stored and video cannot be transcribed. */
|
||||
audio?: TranscriptionAudioTools;
|
||||
/** The client for the configured server; tests pass one backed by an emulator. */
|
||||
createProvider?(settings: TranscriptionSettings): TranscriptionProvider;
|
||||
}
|
||||
|
||||
/**
|
||||
* `transcribe`: speech in an audio or video artifact becomes SRT and VTT subtitle files (or a
|
||||
* text file when the server returns no timings), and the bot gets the transcript back.
|
||||
*/
|
||||
export function createTranscriptionOperations(
|
||||
deps: TranscriptionOperationDeps,
|
||||
): MediaOperationRegistration<TranscribeInput>[] {
|
||||
const createProvider = deps.createProvider ?? createOpenAiCompatibleTranscriptionProvider;
|
||||
return [
|
||||
{
|
||||
operation: "transcribe",
|
||||
inputSchema: TranscribeInputSchema,
|
||||
title: () => "Transcript",
|
||||
available: async () => Boolean((await deps.loadSettings()).transcription),
|
||||
artifactIds: (input) => [input.artifactId],
|
||||
handler: {
|
||||
operation: "transcribe",
|
||||
async run(context) {
|
||||
const input = TranscribeInputSchema.parse(context.input);
|
||||
const settings = (await deps.loadSettings()).transcription;
|
||||
if (!settings) throw new MediaJobRejectedError("Transcription is not set up.");
|
||||
const provider = createProvider(settings);
|
||||
const source = await context.fetchArtifact(input.artifactId);
|
||||
if (source.kind !== "audio" && source.kind !== "video") {
|
||||
throw new MediaJobRejectedError("Only audio and video can be transcribed.");
|
||||
}
|
||||
const audio = await prepareAudio(deps.audio, provider, source, context);
|
||||
await context.progress(0.05);
|
||||
let transcript: Transcript;
|
||||
try {
|
||||
transcript = await transcribeChunks(provider, audio.chunks, {
|
||||
mimeType: audio.mimeType,
|
||||
fileName: audio.fileName,
|
||||
...(input.language ? { language: input.language } : {}),
|
||||
signal: context.signal,
|
||||
onProgress: (fraction) => context.progress(0.05 + fraction * 0.9),
|
||||
});
|
||||
} catch (error) {
|
||||
// Nothing is stored yet, so the job can wait for the server to come back.
|
||||
if (error instanceof TranscriptionServerUnavailableError) {
|
||||
throw new MediaJobDeferredError(error.message, UNAVAILABLE_RETRY_MS);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
if (!transcript.text.trim() && transcript.segments.length === 0) {
|
||||
throw new MediaJobRejectedError(`No speech found in ${source.name}.`);
|
||||
}
|
||||
const base = path.parse(source.name).name || "transcript";
|
||||
const files =
|
||||
transcript.segments.length > 0
|
||||
? [
|
||||
{
|
||||
name: `${base}.srt`,
|
||||
mimeType: "application/x-subrip",
|
||||
body: toSrt(transcript.segments),
|
||||
},
|
||||
{ name: `${base}.vtt`, mimeType: "text/vtt", body: toVtt(transcript.segments) },
|
||||
]
|
||||
: [
|
||||
{
|
||||
name: `${base}.txt`,
|
||||
mimeType: "text/plain",
|
||||
body: `${transcript.text.trim()}\n`,
|
||||
},
|
||||
];
|
||||
for (const [index, file] of files.entries()) {
|
||||
const filePath = path.join(
|
||||
context.tempDir,
|
||||
`transcript-${index}${path.extname(file.name)}`,
|
||||
);
|
||||
await writeFile(filePath, file.body, { mode: 0o600 });
|
||||
await context.addOutput({
|
||||
path: filePath,
|
||||
name: file.name,
|
||||
mimeType: file.mimeType,
|
||||
parentArtifactId: source.artifactId,
|
||||
});
|
||||
}
|
||||
return { summary: transcriptSummary(source.name, transcript) };
|
||||
},
|
||||
},
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
async function prepareAudio(
|
||||
tools: TranscriptionAudioTools | undefined,
|
||||
provider: TranscriptionProvider,
|
||||
source: { path: string; name: string; mimeType: string; kind: string; size: number },
|
||||
context: { tempDir: string; signal: AbortSignal },
|
||||
): Promise<{ chunks: AudioChunk[]; mimeType: string; fileName: string }> {
|
||||
const maxBytes = provider.describe().capabilities.maxBytes;
|
||||
if (!tools || !(await tools.available())) {
|
||||
if (source.kind === "video") {
|
||||
throw new MediaJobRejectedError("Transcribing video needs ffmpeg on the server.");
|
||||
}
|
||||
if (source.size > maxBytes) {
|
||||
throw new MediaJobRejectedError(
|
||||
"This recording is too long to transcribe without ffmpeg on the server.",
|
||||
);
|
||||
}
|
||||
return {
|
||||
chunks: [{ path: source.path, offsetMs: 0 }],
|
||||
mimeType: source.mimeType,
|
||||
fileName: source.name,
|
||||
};
|
||||
}
|
||||
// Whisper servers expect 16 kHz mono; converting also keeps long files within upload limits.
|
||||
const options = { signal: context.signal, mimeType: source.mimeType };
|
||||
const durationMs = await tools.durationMs(source.path, options);
|
||||
const chunks: AudioChunk[] = [];
|
||||
for (const [index, range] of planAudioChunks(durationMs, maxBytes).entries()) {
|
||||
const chunkPath = path.join(context.tempDir, `speech-${index}.wav`);
|
||||
await tools.extract(source.path, chunkPath, { ...options, ...range });
|
||||
chunks.push({ path: chunkPath, offsetMs: range.startMs });
|
||||
}
|
||||
if (chunks.length === 0) throw new MediaJobRejectedError(`No speech found in ${source.name}.`);
|
||||
return {
|
||||
chunks,
|
||||
mimeType: "audio/wav",
|
||||
fileName: `${path.parse(source.name).name || "speech"}.wav`,
|
||||
};
|
||||
}
|
||||
|
||||
@ -23,6 +23,14 @@ export const OPENAI_COMPATIBLE_TRANSCRIPTION_DEFAULT_MODEL = "whisper-1";
|
||||
const TRANSCRIPTION_TIMEOUT_MS = 15 * 60_000;
|
||||
const MAX_TRANSCRIPTION_RESPONSE_BYTES = 8 * 1024 * 1024;
|
||||
|
||||
/** The server is down, restarting or busy; trying again later may succeed. */
|
||||
export class TranscriptionServerUnavailableError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
this.name = "TranscriptionServerUnavailableError";
|
||||
}
|
||||
}
|
||||
|
||||
export interface OpenAiCompatibleTranscriptionConfig {
|
||||
baseUrl: string;
|
||||
model?: string;
|
||||
@ -94,7 +102,11 @@ export function createOpenAiCompatibleTranscriptionProvider(
|
||||
) {
|
||||
return { rejectedFormat: true };
|
||||
}
|
||||
throw new Error(httpErrorMessage(response.status, message));
|
||||
const failure = httpErrorMessage(response.status, message);
|
||||
if (response.status === 429 || response.status === 502 || response.status === 503) {
|
||||
throw new TranscriptionServerUnavailableError(failure);
|
||||
}
|
||||
throw new Error(failure);
|
||||
}
|
||||
return parseTranscriptionResponse(text, response.headers.get("content-type"));
|
||||
} catch (error) {
|
||||
@ -102,7 +114,7 @@ export function createOpenAiCompatibleTranscriptionProvider(
|
||||
if (timeout.aborted) throw new Error("Transcription server timed out");
|
||||
// undici reports refused or reset connections as a bare "fetch failed".
|
||||
if (error instanceof TypeError && error.message === "fetch failed") {
|
||||
throw new Error("Transcription server is unreachable");
|
||||
throw new TranscriptionServerUnavailableError("Transcription server is unreachable");
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
|
||||
@ -7,6 +7,7 @@ import {
|
||||
OPENAI_COMPATIBLE_TRANSCRIPTION_DEFAULT_MODEL,
|
||||
OPENAI_COMPATIBLE_TRANSCRIPTION_MAX_BYTES,
|
||||
parseTranscriptionResponse,
|
||||
TranscriptionServerUnavailableError,
|
||||
} from "./openai-compatible-transcription.js";
|
||||
import {
|
||||
EMULATED_TRANSCRIPT,
|
||||
@ -209,7 +210,11 @@ describe("OpenAI-compatible transcription provider", () => {
|
||||
},
|
||||
});
|
||||
await expect(unreachable.transcribe({ audio: AUDIO, mimeType: "audio/wav" })).rejects.toThrow(
|
||||
"Transcription server is unreachable",
|
||||
TranscriptionServerUnavailableError,
|
||||
);
|
||||
server.failNextStatus = 503;
|
||||
await expect(provider.transcribe({ audio: AUDIO, mimeType: "audio/wav" })).rejects.toThrow(
|
||||
TranscriptionServerUnavailableError,
|
||||
);
|
||||
});
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user