Add private realtime Talk for OpenClaw

2026-09-03 22:52:15 +02:00
parent 42ec28c9f6
commit 0a68df22eb
10 changed files with 829 additions and 11 deletions
@@ -0,0 +1,309 @@
+import { randomUUID } from "node:crypto";
+import { definePluginEntry } from "openclaw/plugin-sdk/plugin-entry";
+
+const AUDIO_FORMAT = { encoding: "pcm16", sampleRateHz: 24000, channels: 1 } as const;
+
+type ProviderConfig = {
+  baseUrl?: string;
+  apiKey?: string;
+  voice?: string;
+  language?: string;
+  vadThreshold?: number;
+  silenceDurationMs?: number;
+  prefixPaddingMs?: number;
+  maxSpeechSeconds?: number;
+};
+
+function record(value: unknown): Record<string, any> {
+  return value && typeof value === "object" && !Array.isArray(value)
+    ? value as Record<string, any>
+    : {};
+}
+
+function resolveConfig(req: any): Required<ProviderConfig> {
+  const raw = record(req.providerConfig);
+  const modelProviders = record(record(req.cfg).models).providers;
+  const athena = record(record(modelProviders).athena);
+  return {
+    baseUrl: String(raw.baseUrl || athena.baseUrl || "http://192.168.1.212:8081/v1").replace(/\/$/, ""),
+    apiKey: String(raw.apiKey || athena.apiKey || ""),
+    voice: String(raw.voice || req.voice || "alloy"),
+    language: String(raw.language || req.language || "de"),
+    vadThreshold: Number(raw.vadThreshold ?? 0.018),
+    silenceDurationMs: Number(raw.silenceDurationMs ?? 750),
+    prefixPaddingMs: Number(raw.prefixPaddingMs ?? 300),
+    maxSpeechSeconds: Number(raw.maxSpeechSeconds ?? 45),
+  };
+}
+
+function wavFromPcm16(pcm: Buffer, sampleRate = 24000): Buffer {
+  const header = Buffer.alloc(44);
+  header.write("RIFF", 0);
+  header.writeUInt32LE(36 + pcm.length, 4);
+  header.write("WAVEfmt ", 8);
+  header.writeUInt32LE(16, 16);
+  header.writeUInt16LE(1, 20);
+  header.writeUInt16LE(1, 22);
+  header.writeUInt32LE(sampleRate, 24);
+  header.writeUInt32LE(sampleRate * 2, 28);
+  header.writeUInt16LE(2, 32);
+  header.writeUInt16LE(16, 34);
+  header.write("data", 36);
+  header.writeUInt32LE(pcm.length, 40);
+  return Buffer.concat([header, pcm]);
+}
+
+function pcmRms(pcm: Buffer): number {
+  if (pcm.length < 2) return 0;
+  let sum = 0;
+  const count = Math.floor(pcm.length / 2);
+  for (let i = 0; i < count; i += 1) {
+    const value = pcm.readInt16LE(i * 2) / 32768;
+    sum += value * value;
+  }
+  return Math.sqrt(sum / count);
+}
+
+function readWavPcm24k(wav: Buffer): Buffer {
+  if (wav.length < 44 || wav.toString("ascii", 0, 4) !== "RIFF") {
+    throw new Error("Athena TTS did not return PCM WAV audio");
+  }
+  let offset = 12;
+  let sampleRate = 0;
+  let channels = 0;
+  let bits = 0;
+  let data: Buffer | undefined;
+  while (offset + 8 <= wav.length) {
+    const id = wav.toString("ascii", offset, offset + 4);
+    const size = wav.readUInt32LE(offset + 4);
+    const start = offset + 8;
+    if (id === "fmt " && size >= 16) {
+      if (wav.readUInt16LE(start) !== 1) throw new Error("Athena TTS WAV is not PCM");
+      channels = wav.readUInt16LE(start + 2);
+      sampleRate = wav.readUInt32LE(start + 4);
+      bits = wav.readUInt16LE(start + 14);
+    } else if (id === "data") {
+      data = wav.subarray(start, Math.min(start + size, wav.length));
+    }
+    offset = start + size + (size % 2);
+  }
+  if (!data || !sampleRate || bits !== 16 || channels < 1) {
+    throw new Error("Unsupported Athena TTS WAV format");
+  }
+
+  const frames = Math.floor(data.length / (2 * channels));
+  const mono = new Int16Array(frames);
+  for (let i = 0; i < frames; i += 1) mono[i] = data.readInt16LE(i * channels * 2);
+  if (sampleRate === 24000) return Buffer.from(mono.buffer);
+
+  const outFrames = Math.max(1, Math.round(frames * 24000 / sampleRate));
+  const out = Buffer.alloc(outFrames * 2);
+  for (let i = 0; i < outFrames; i += 1) {
+    const source = i * sampleRate / 24000;
+    const left = Math.min(frames - 1, Math.floor(source));
+    const right = Math.min(frames - 1, left + 1);
+    const fraction = source - left;
+    const value = Math.round(mono[left] * (1 - fraction) + mono[right] * fraction);
+    out.writeInt16LE(Math.max(-32768, Math.min(32767, value)), i * 2);
+  }
+  return out;
+}
+
+class AthenaTalkBridge {
+  readonly supportsToolResultContinuation = false;
+  readonly supportsToolResultSuppression = false;
+  private connected = false;
+  private closed = false;
+  private speaking = false;
+  private speech: Buffer[] = [];
+  private prefix: Buffer[] = [];
+  private prefixBytes = 0;
+  private silenceTimer?: NodeJS.Timeout;
+  private active?: AbortController;
+  private generation = 0;
+
+  constructor(private readonly req: any, private readonly cfg: Required<ProviderConfig>) {}
+
+  async connect(): Promise<void> {
+    this.connected = true;
+    this.req.onEvent?.({ direction: "server", type: "session.created" });
+    this.req.onReady?.();
+  }
+
+  isConnected(): boolean { return this.connected && !this.closed; }
+
+  setMediaTimestamp(_timestamp: number): void {}
+
+  sendAudio(chunk: Buffer): void {
+    if (!this.isConnected() || chunk.length === 0) return;
+    const voiced = pcmRms(chunk) >= this.cfg.vadThreshold;
+    const maxPrefix = Math.round(24000 * 2 * this.cfg.prefixPaddingMs / 1000);
+
+    if (!this.speaking) {
+      this.prefix.push(Buffer.from(chunk));
+      this.prefixBytes += chunk.length;
+      while (this.prefixBytes > maxPrefix && this.prefix.length > 1) {
+        this.prefixBytes -= this.prefix.shift()!.length;
+      }
+      if (!voiced) return;
+      if (this.active) {
+        this.active.abort();
+        this.active = undefined;
+        this.generation += 1;
+        this.req.onClearAudio?.("barge-in");
+      }
+      this.speaking = true;
+      this.speech = this.prefix;
+      this.prefix = [];
+      this.prefixBytes = 0;
+    } else {
+      this.speech.push(Buffer.from(chunk));
+    }
+
+    const maxBytes = this.cfg.maxSpeechSeconds * 24000 * 2;
+    if (this.speech.reduce((sum, part) => sum + part.length, 0) >= maxBytes) {
+      void this.finishSpeech();
+      return;
+    }
+    if (voiced) {
+      if (this.silenceTimer) clearTimeout(this.silenceTimer);
+      this.silenceTimer = setTimeout(() => void this.finishSpeech(), this.cfg.silenceDurationMs);
+    }
+  }
+
+  sendUserMessage(text: string): void {
+    const trimmed = text.trim();
+    if (trimmed) void this.answer(trimmed);
+  }
+
+  triggerGreeting(instructions?: string): void {
+    void this.answer(instructions?.trim() || "Begrüße mich kurz auf Deutsch.");
+  }
+
+  submitToolResult(_callId: string, _result: unknown): void {}
+  acknowledgeMark(_markName: string): void {}
+
+  close(): void {
+    this.closed = true;
+    this.connected = false;
+    this.generation += 1;
+    this.active?.abort();
+    if (this.silenceTimer) clearTimeout(this.silenceTimer);
+    this.req.onClose?.("completed");
+  }
+
+  private async finishSpeech(): Promise<void> {
+    if (!this.speaking) return;
+    this.speaking = false;
+    if (this.silenceTimer) clearTimeout(this.silenceTimer);
+    this.silenceTimer = undefined;
+    const pcm = Buffer.concat(this.speech);
+    this.speech = [];
+    if (pcm.length < 24000 * 2 * 0.25) return;
+
+    try {
+      const text = await this.transcribe(pcm);
+      if (!text || !this.isConnected()) return;
+      this.req.onTranscript?.("user", text, true);
+      await this.answer(text);
+    } catch (error) {
+      if ((error as Error).name !== "AbortError") this.req.onError?.(error as Error);
+    }
+  }
+
+  private async transcribe(pcm: Buffer): Promise<string> {
+    const form = new FormData();
+    form.append("file", new Blob([wavFromPcm16(pcm)], { type: "audio/wav" }), "talk.wav");
+    form.append("model", "whisper-1");
+    form.append("language", this.cfg.language);
+    const response = await fetch(`${this.cfg.baseUrl}/audio/transcriptions`, {
+      method: "POST",
+      headers: this.authHeaders(),
+      body: form,
+    });
+    if (!response.ok) throw new Error(`Athena STT failed (HTTP ${response.status})`);
+    const payload = record(await response.json());
+    return String(payload.text || "").trim();
+  }
+
+  private async answer(prompt: string): Promise<void> {
+    if (!this.req.runAgentConsult) throw new Error("OpenClaw agent-consult is unavailable");
+    this.active?.abort();
+    const controller = new AbortController();
+    this.active = controller;
+    const generation = ++this.generation;
+    const responseId = `athena-${randomUUID()}`;
+    try {
+      const result = await this.req.runAgentConsult({ prompt, signal: controller.signal });
+      if (generation !== this.generation || !this.isConnected()) return;
+      const text = String(result?.text || "").trim();
+      if (!text) return;
+      this.req.onTranscript?.("assistant", text, true);
+      const pcm = await this.synthesize(text, controller.signal);
+      if (generation !== this.generation || !this.isConnected()) return;
+      const frameBytes = 24000 * 2 / 50;
+      for (let offset = 0; offset < pcm.length; offset += frameBytes) {
+        this.req.onAudio?.(pcm.subarray(offset, Math.min(offset + frameBytes, pcm.length)));
+      }
+      this.req.onResponseDone?.({ status: "completed", responseId });
+      this.req.onEvent?.({ direction: "server", type: "response.done", responseId });
+    } catch (error) {
+      if ((error as Error).name === "AbortError") {
+        this.req.onResponseDone?.({ status: "cancelled", responseId });
+      } else {
+        this.req.onError?.(error as Error);
+        this.req.onResponseDone?.({ status: "failed", responseId, error: String(error) });
+      }
+    } finally {
+      if (this.active === controller) this.active = undefined;
+    }
+  }
+
+  private async synthesize(text: string, signal: AbortSignal): Promise<Buffer> {
+    const response = await fetch(`${this.cfg.baseUrl}/audio/speech`, {
+      method: "POST",
+      headers: { "Content-Type": "application/json", ...this.authHeaders() },
+      // Let Athena select its currently active local backend (XTTS or Piper).
+      body: JSON.stringify({ voice: this.cfg.voice, input: text, response_format: "wav" }),
+      signal,
+    });
+    if (!response.ok) throw new Error(`Athena TTS failed (HTTP ${response.status})`);
+    return readWavPcm24k(Buffer.from(await response.arrayBuffer()));
+  }
+
+  private authHeaders(): Record<string, string> {
+    return this.cfg.apiKey ? { Authorization: `Bearer ${this.cfg.apiKey}` } : {};
+  }
+}
+
+export default definePluginEntry({
+  id: "athena-talk",
+  name: "Athena Local Talk",
+  description: "Private voice loop using Athena STT and TTS with the normal OpenClaw agent.",
+  register(api) {
+    api.registerRealtimeVoiceProvider({
+      id: "athena-talk",
+      label: "Athena Local Talk",
+      aliases: ["athena", "local-athena"],
+      defaultModel: "athena-local",
+      voices: ["alloy", "claribel"],
+      autoSelectOrder: 1,
+      capabilities: {
+        transports: ["gateway-relay"],
+        inputAudioFormats: [AUDIO_FORMAT],
+        outputAudioFormats: [AUDIO_FORMAT],
+        supportsBargeIn: true,
+        handlesInputAudioBargeIn: true,
+        supportsToolCalls: true,
+        supportsSessionResumption: false,
+      },
+      resolveConfig: ({ rawConfig }) => record(rawConfig),
+      isConfigured: ({ providerConfig, cfg }) => {
+        const raw = record(providerConfig);
+        const athena = record(record(record(cfg).models).providers).athena;
+        return Boolean(raw.baseUrl || record(athena).baseUrl);
+      },
+      createBridge: (req) => new AthenaTalkBridge(req, resolveConfig(req)),
+    } as any);
+  },
+});