Molecule.dev

@molecule/api-ai-speech-confucius4

Provider bond · ai-speech · API (Node) · v1.0.1 · Apache-2.0

Confucius4-R2T2 speech provider for molecule.dev — self-hosted append-only streaming speech-to-text

npm install @molecule/api-ai-speech-confucius4

npm · Source on GitHub · Implements @molecule/api-ai-speech

How it works

@molecule/api-ai-speech-confucius4 is a provider bond on the API (Node) side: it implements the ai-speech core interface (@molecule/api-ai-speech) with a concrete vendor or library behind it.

Your code calls the core; you wire this provider once at startup. Swapping vendors later is one line in that wiring, not a rewrite.

import { requireProvider, setProvider } from '@molecule/api-ai-speech'
import { provider } from '@molecule/api-ai-speech-confucius4'

// Reads CONFUCIUS4_WS_URL (default ws://127.0.0.1:8272/asr_stream_api_v1) per call.
setProvider(provider)

const speech = requireProvider()
// micChunks: AsyncIterable<Uint8Array> of PCM16 mono audio forwarded by your API.
for await (const event of speech.transcribeStream!(micChunks, { sampleRate: 48000 })) {
  if (event.type === 'delta') process.stdout.write(event.text) // never revised
  if (event.type === 'turn-end') process.stdout.write('\n')
}

Works with: @molecule/api-ai-speech, @molecule/api-secrets

Secrets: CONFUCIUS4_WS_URL (optional), CONFUCIUS4_SECRET_KEY (optional), CONFUCIUS4_SYSTEM_PROMPT (optional)

Reference

Confucius4-R2T2 provider for @molecule/api-ai-speech.

Append-only streaming speech-to-text on your own GPU (NetEase Youdao's 2B Confucius4-R2T2, built on Qwen3-ASR): text it has sent never changes, which suits live captions that must not rewrite themselves. Tuned for Chinese and English; supports context/hotword prompts. Implements transcribeStream (streamingAppendOnly: true) and a buffer transcribe built on it.

Auto-generated, AI-first package reference for the molecule.dev ecosystem. It is written to be read by coding agents as much as by people, and is generated from this package's source — edit src/index.ts JSDoc, not this file.

Quick Start

import { requireProvider, setProvider } from '@molecule/api-ai-speech'
import { provider } from '@molecule/api-ai-speech-confucius4'

// Reads CONFUCIUS4_WS_URL (default ws://127.0.0.1:8272/asr_stream_api_v1) per call.
setProvider(provider)

const speech = requireProvider()
// micChunks: AsyncIterable<Uint8Array> of PCM16 mono audio forwarded by your API.
for await (const event of speech.transcribeStream!(micChunks, { sampleRate: 48000 })) {
  if (event.type === 'delta') process.stdout.write(event.text) // never revised
  if (event.type === 'turn-end') process.stdout.write('\n')
}

Type

provider

Installation

npm install @molecule/api-ai-speech-confucius4 @molecule/api-ai-speech @molecule/api-secrets

API

Interfaces

DecodedWav

Decoded PCM16 audio.

interface DecodedWav {
  /** Sample rate of the decoded audio in Hz. */
  sampleRate: number
  /** Number of interleaved channels in the source file. */
  channels: number
  /** Mono PCM16 samples (multi-channel input is averaged down). */
  samples: Int16Array
}

RealtimeSocket

An open (or opening) realtime socket.

interface RealtimeSocket {
  /**
   * Send a text or binary frame.
   *
   * @param data - JSON text or raw bytes.
   */
  send(data: string | Uint8Array): void
  /**
   * Close the socket.
   *
   * @param code - Close code (default 1000).
   * @param reason - Close reason.
   */
  close(code?: number, reason?: string): void
}

RealtimeSocketHandlers

Callbacks a RealtimeSocketFactory reports socket activity to.

interface RealtimeSocketHandlers {
  /** The connection is open. */
  onOpen(): void
  /** A frame arrived (text frames as strings, binary frames as bytes). */
  onMessage(data: string | Uint8Array): void
  /** The connection closed. */
  onClose(code: number, reason: string): void
  /** The connection failed. */
  onError(message: string): void
}

Confucius4SpeechConfig

Configuration for the Confucius4-R2T2 speech provider. Every field falls back to its env var, read on each call (never at import time).

interface Confucius4SpeechConfig {
  /**
   * WebSocket URL of `ws_server.py`. Defaults to the `CONFUCIUS4_WS_URL` env
   * var, then `ws://127.0.0.1:8272/asr_stream_api_v1`.
   */
  wsUrl?: string
  /**
   * The `secret_key` sent in the header message. Defaults to
   * `CONFUCIUS4_SECRET_KEY`, then `test0102` (the server's hardcoded whitelist).
   */
  secretKey?: string
  /**
   * Context / hotword prompt (`system_prompt`, at most 4000 characters) used
   * when a call passes no `prompt`. Defaults to `CONFUCIUS4_SYSTEM_PROMPT`.
   */
  systemPrompt?: string
  /** `language` header value when a call passes none. Defaults to `'zhen'` (auto Chinese/English). */
  defaultLanguage?: string
  /** `use_vad` header value: let the server's Stream-VAD cut segments. Defaults to `false`. */
  useVad?: boolean
  /** `mode` header value. Defaults to `'slow'` (the reference client's value). */
  mode?: string
  /** `smooth` header value. Omitted unless set. */
  smooth?: boolean
  /**
   * How long a stream waits, after sending end-of-stream, for the server to
   * close the socket before it gives up. Defaults to 30000 ms.
   */
  finalTimeoutMs?: number
  /** WebSocket factory. Defaults to the runtime's global `WebSocket`. */
  socketFactory?: RealtimeSocketFactory
}

Types

RealtimeSocketFactory

Opens a realtime socket.

type RealtimeSocketFactory = (
  url: string,
  headers: Record<string, string>,
  handlers: RealtimeSocketHandlers,
) => RealtimeSocket
  • url — ws:// or wss:// URL.
  • headers — Extra handshake headers (e.g. Authorization).
  • handlers — Activity callbacks.

Returns: The socket.

Classes

Pcm16Resampler

Streaming PCM16 little-endian resampler (linear interpolation). Keeps the state a chunked stream needs: an odd trailing byte and the fractional read position carry over to the next chunk, so chunk boundaries add no clicks.

class Pcm16Resampler {
  private readonly step: number
  private tail = new Int16Array(0)
  private position = 0
  private carry: number | null = null

  /**
   * Create a resampler.
   *
   * @param fromRate - Input sample rate in Hz.
   * @param toRate - Output sample rate in Hz.
   */
  constructor(
    private readonly fromRate: number,
    private readonly toRate: number,
  ) {
    if (!Number.isFinite(fromRate) || fromRate <= 0 || !Number.isFinite(toRate) || toRate <= 0) {
      throw new Error(`Invalid sample rate: ${fromRate} → ${toRate}`)
    }
    this.step = fromRate / toRate
  }

  /**
   * Resample one chunk of PCM16 LE bytes.
   *
   * @param chunk - Raw bytes (may split a sample; the odd byte is carried).
   * @returns Resampled PCM16 LE bytes (possibly empty).
   */
  push(chunk: Uint8Array): Uint8Array {
    return samplesToBytes(this.pushSamples(this.decode(chunk)))
  }

  /**
   * Emit whatever the last chunk left pending.
   *
   * @returns Remaining PCM16 LE bytes.
   */
  flush(): Uint8Array {
    return samplesToBytes(this.flushSamples())
  }

  /**
   * Resample decoded samples.
   *
   * @param samples - Mono samples.
   * @returns Resampled samples.
   */
  pushSamples(samples: Int16Array): Int16Array {
    if (this.fromRate === this.toRate) return samples
    const buffer = new Int16Array(this.tail.length + samples.length)
    buffer.set(this.tail)
    buffer.set(samples, this.tail.length)
    const out: number[] = []
    for (;;) {
      const index = Math.floor(this.position)
      if (index + 1 >= buffer.length) break
      const fraction = this.position - index
      out.push(Math.round(buffer[index] * (1 - fraction) + buffer[index + 1] * fraction))
      this.position += this.step
    }
    const drop = Math.min(Math.floor(this.position), Math.max(buffer.length - 1, 0))
    this.tail = buffer.slice(drop)
    this.position -= drop
    return Int16Array.from(out)
  }

  /**
   * Emit the final samples held back for interpolation.
   *
   * @returns Remaining samples.
   */
  flushSamples(): Int16Array {
    if (this.fromRate === this.toRate) return new Int16Array(0)
    const out: number[] = []
    while (this.tail.length > 0 && this.position <= this.tail.length - 1) {
      out.push(this.tail[Math.floor(this.position)])
      this.position += this.step
    }
    this.tail = new Int16Array(0)
    this.position = 0
    return Int16Array.from(out)
  }

  /**
   * Decode LE bytes into samples, carrying an odd trailing byte.
   *
   * @param chunk - Raw bytes.
   * @returns Decoded samples.
   */
  private decode(chunk: Uint8Array): Int16Array {
    let bytes = chunk
    if (this.carry !== null) {
      bytes = new Uint8Array(chunk.length + 1)
      bytes[0] = this.carry
      bytes.set(chunk, 1)
      this.carry = null
    }
    if (bytes.length % 2 === 1) {
      this.carry = bytes[bytes.length - 1]
      bytes = bytes.subarray(0, bytes.length - 1)
    }
    const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength)
    const samples = new Int16Array(bytes.length / 2)
    for (let i = 0; i < samples.length; i++) samples[i] = view.getInt16(i * 2, true)
    return samples
  }
}

Confucius4SpeechError

Error thrown when a Confucius4 stream fails during a buffer transcribe. status is the WebSocket close code (4401 = secret key rejected; 0 when the failure was not a close). Deliberately NOT tagged with statusCode/ errorKey, so API middleware answers its generic 500.

class Confucius4SpeechError extends Error {
  /** WebSocket close code, or 0. */
  readonly status: number

  /**
   * Create the error.
   *
   * @param message - Human-readable message.
   * @param status - WebSocket close code, or 0.
   */
  constructor(message: string, status = 0) {
    super(message)
    this.name = 'Confucius4SpeechError'
    this.status = status
  }
}

AsyncQueue

A single-consumer async queue: producers push, the consumer iterates, and end() finishes the iteration once everything pushed has been read.

class AsyncQueue<T> implements AsyncIterable<T> {
  private readonly items: T[] = []
  private waiting: ((result: IteratorResult<T>) => void) | null = null
  private ended = false

  /**
   * Add an item.
   *
   * @param item - The item (ignored after `end()`).
   */
  push(item: T): void {
    if (this.ended) return
    if (this.waiting) {
      const resolve = this.waiting
      this.waiting = null
      resolve({ value: item, done: false })
    } else {
      this.items.push(item)
    }
  }

  /** Finish the queue; the consumer drains what is left, then stops. */
  end(): void {
    if (this.ended) return
    this.ended = true
    if (this.waiting) {
      const resolve = this.waiting
      this.waiting = null
      resolve({ value: undefined, done: true })
    }
  }

  /**
   * Whether `end()` has been called.
   *
   * @returns True once ended.
   */
  get isEnded(): boolean {
    return this.ended
  }

  /**
   * Iterate the queue.
   *
   * @returns An async iterator over pushed items.
   */
  [Symbol.asyncIterator](): AsyncIterator<T> {
    return {
      next: (): Promise<IteratorResult<T>> => {
        if (this.items.length > 0) {
          return Promise.resolve({ value: this.items.shift() as T, done: false })
        }
        if (this.ended) return Promise.resolve({ value: undefined, done: true })
        return new Promise((resolve) => {
          this.waiting = resolve
        })
      },
    }
  }
}

Functions

isWav(bytes)

Checks whether the bytes start with a RIFF/WAVE header.

function isWav(bytes: Uint8Array<ArrayBufferLike>): boolean
  • bytes — Audio bytes.

Returns: True for a WAV container.

decodeWav(bytes)

Decodes an uncompressed 16-bit PCM WAV file to mono samples.

function decodeWav(bytes: Uint8Array<ArrayBufferLike>): DecodedWav
  • bytes — The WAV file bytes.

Returns: The sample rate, source channel count and mono samples.

resample(samples, fromRate, toRate)

Resamples mono PCM16 with linear interpolation.

function resample(
  samples: Int16Array<ArrayBufferLike>,
  fromRate: number,
  toRate: number,
): Int16Array<ArrayBufferLike>
  • samples — Mono input samples.
  • fromRate — Input sample rate in Hz.
  • toRate — Output sample rate in Hz.

Returns: Resampled samples (the input itself when the rates match).

encodeWav(samples, sampleRate)

Encodes mono PCM16 samples as a WAV file.

function encodeWav(
  samples: Int16Array<ArrayBufferLike>,
  sampleRate: number,
): Uint8Array<ArrayBufferLike>
  • samples — Mono samples.
  • sampleRate — Sample rate in Hz.

Returns: The WAV file bytes.

toTargetWav(bytes)

Converts any 16-bit PCM WAV into a 16 kHz mono WAV.

function toTargetWav(bytes: Uint8Array<ArrayBufferLike>): Uint8Array<ArrayBufferLike>
  • bytes — The source WAV bytes.

Returns: A 16 kHz mono 16-bit PCM WAV.

wavToTargetPcm(bytes)

Decodes a WAV file into 16 kHz mono PCM16 little-endian bytes.

function wavToTargetPcm(bytes: Uint8Array<ArrayBufferLike>): Uint8Array<ArrayBufferLike>
  • bytes — The source WAV bytes.

Returns: Raw 16 kHz mono PCM16 LE bytes.

samplesToBytes(samples)

Serializes samples as PCM16 little-endian bytes.

function samplesToBytes(samples: Int16Array<ArrayBufferLike>): Uint8Array<ArrayBufferLike>
  • samples — PCM16 samples.

Returns: Little-endian bytes.

toTargetBlocks(audio, sampleRate, blockBytes)

Converts a PCM16 LE stream at any sample rate to 16 kHz and re-chunks it into fixed-size blocks (the last block may be shorter).

function toTargetBlocks(
  audio: AsyncIterable<Uint8Array<ArrayBufferLike>>,
  sampleRate: number,
  blockBytes: number,
): AsyncGenerator<Uint8Array<ArrayBufferLike>, any, any>
  • audio — Raw PCM16 LE mono chunks.
  • sampleRate — Input sample rate in Hz.
  • blockBytes — Output block size in bytes (even).

toBase64(bytes)

Encodes bytes as base64.

function toBase64(bytes: Uint8Array<ArrayBufferLike>): string
  • bytes — Raw bytes.

Returns: Base64 text.

createProvider(config)

Creates a Confucius4-R2T2 speech provider instance.

function createProvider(config?: Confucius4SpeechConfig): AISpeechProvider
  • config — Optional overrides (URL, secret key, prompt, header options, socket factory).

Returns: An AISpeechProvider backed by a self-hosted Confucius4 server.

openGlobalWebSocket(url, headers, handlers)

Default socket factory over the runtime's global WebSocket. Handshake headers use the Node (undici) WebSocketInit.headers extension.

function openGlobalWebSocket(
  url: string,
  headers: Record<string, string>,
  handlers: RealtimeSocketHandlers,
): {
  send: (data: string | Uint8Array<ArrayBufferLike>) => void
  close: (code: number | undefined, reason: string | undefined) => void
}
  • url — ws:// or wss:// URL.
  • headers — Extra handshake headers.
  • handlers — Activity callbacks.

Returns: The socket.

toWebSocketUrl(baseUrl)

Turns an http(s):// base URL into its ws(s):// equivalent.

function toWebSocketUrl(baseUrl: string): string
  • baseUrl — HTTP base URL.

Returns: The WebSocket base URL.

parseFrame(data)

Decodes a text frame as JSON.

function parseFrame(data: string | Uint8Array<ArrayBufferLike>): Record<string, unknown> | null
  • data — The frame.

Returns: The parsed object, or null for binary/non-JSON frames.

Constants

TARGET_SAMPLE_RATE

The sample rate every self-hosted speech server in this family expects.

const TARGET_SAMPLE_RATE: 16000

DEFAULT_CONFUCIUS4_WS_URL

Default WebSocket URL of ws_server.py.

const DEFAULT_CONFUCIUS4_WS_URL: 'ws://127.0.0.1:8272/asr_stream_api_v1'

DEFAULT_CONFUCIUS4_SECRET_KEY

The server's hardcoded secret-key whitelist entry.

const DEFAULT_CONFUCIUS4_SECRET_KEY: 'test0102'

CONFUCIUS4_EOS

End-of-audio text frame.

const CONFUCIUS4_EOS: 'YOUDAO_ONETIME_ASR_STREAM_EOS'

MAX_SYSTEM_PROMPT_CHARS

Longest system_prompt the server accepts.

const MAX_SYSTEM_PROMPT_CHARS: 4000

CLOSE_BAD_SECRET_KEY

Close code the server uses for a rejected secret key.

const CLOSE_BAD_SECRET_KEY: 4401

DEFAULT_FINAL_TIMEOUT_MS

Default wait for the server to close after end-of-stream.

const DEFAULT_FINAL_TIMEOUT_MS: 30000

provider

The provider implementation.

const provider: AISpeechProvider

aiSpeechConfucius4SecretDefinitions

Secret definitions used by the Confucius4-R2T2 speech bond.

const aiSpeechConfucius4SecretDefinitions: SecretDefinition[]

Core Interface

Implements @molecule/api-ai-speech interface.

Bond Wiring

Setup function to register this provider with the core interface:

import { setProvider } from '@molecule/api-ai-speech'
import { provider } from '@molecule/api-ai-speech-confucius4'

export function setupAiSpeechConfucius4(): void {
  setProvider(provider)
}

Injection Notes

Requirements

Peer dependencies:

  • @molecule/api-ai-speech >=1.1.0
  • @molecule/api-secrets ^1.0.1

Environment Variables

  • CONFUCIUS4_WS_URL (optional) — Confucius4-R2T2 WebSocket URL
    • Setup: The URL of your own ws_server.py (./run_start_server.sh start … --port 8272). Defaults to ws://127.0.0.1:8272/asr_stream_api_v1.
    • Get it here: https://github.com/netease-youdao/Confucius4-R2T2
    • Example: ws://127.0.0.1:8272/asr_stream_api_v1
  • CONFUCIUS4_SECRET_KEY (optional) — Confucius4-R2T2 secret key
  • CONFUCIUS4_SYSTEM_PROMPT (optional) — Confucius4-R2T2 context prompt

Runtime Dependencies

  • @molecule/api-ai-speech

  • @molecule/api-secrets

  • WEIGHT LICENCE: the NetEase Model Use License Agreement, not Apache-2.0 (only the code is). Commercial use is allowed, but a company with >100M MAU or >RMB 1B annual revenue needs a separate licence; derivative works must carry a no-endorsement notice; it is banned in "high-risk scenarios" including large-scale biometric surveillance and automated decision-making; you may not use it to improve other AI models (except non-commercial ones); PRC law and CIETAC arbitration apply. Check the automated-decision-making clause before offering it as a hosted service.

  • The "secret key" is not auth. ws_server.py checks secret_key against a hardcoded ["test0102"] (a mismatch closes with code 4401), so anyone who can reach port 8272 uses your GPU. Keep it on a private network or behind an authenticating reverse proxy; never expose it publicly.

  • You run the server, on a CUDA GPU (vLLM with strict CUDA/PyTorch versions, single process, one GPU): ./run_start_server.sh start --model_path <ckpt> --vad_model_path <Stream-VAD dir> --port 8272 --gpu 0.

  • 16 kHz only, and the server does NOT resample — 48 kHz browser audio produces garbage with no error. The bond resamples PCM16 input from sampleRate to 16 kHz for you; never bypass it with raw frames. transcribe accepts 16-bit PCM WAV only and throws for m4a/webm/mp3.

  • Finality has no flag. Every server message is an append-only increment (delta events). reset: true (VAD end of speech, a detected hallucination, or the 60 s soft / 90 s hard reset timeout) is a segment boundary, NOT an error: the bond emits final (that segment's text, equal to its deltas) then turn-end. The stream ends when the server closes the socket after end-of-stream — never wait for a final flag. Set createProvider({ useVad: true }) for VAD-driven segments.

  • language is sent verbatim; only 'zhen' (auto Chinese/English, the default) is documented for the WebSocket server. prompt (or CONFUCIUS4_SYSTEM_PROMPT) becomes system_prompt, at most 4000 characters — longer throws. No word timestamps or speakers over the WebSocket (diarize is ignored).

  • Server-side only. A browser microphone streams to YOUR API, which calls transcribeStream; never connect the browser to the model host. The bond needs a global WebSocket (Node 22+) or a socketFactory.

  • A failed buffer transcribe throws Confucius4SpeechError (status = close code; 4401 = secret key rejected); streams report failures as an error event and end.

E2E Tests

Integration checklist — drive the real UI (live preview, no mocks), adapt each item to this app's actual screens/flows, and check every box off one by one. A box you can't check is an integration bug to fix — not a skip. The wired provider implements a subset (TTS, STT, or both) — feature-detect and run only the direction(s) it actually exposes:

  • TTS: each read-aloud / narration / voice-note flow the app defines returns real PLAYABLE audio — the response carries an audio Content-Type (e.g. audio/mpeg) and non-trivial bytes (not a 0-byte file, not a JSON-encoded blob), and the UI's <audio>/player actually plays it. The spoken audio reflects the input text you sent.
  • TTS: synthesizing two different strings yields two audibly different clips (distinct byte payloads) — proves the text drives synthesis and you are not hearing a cached/placeholder file.
  • TTS: any exposed voice/format/speed option takes effect — switching voice (voice/voiceId) or output format (responseFormat/ outputFormat) changes the returned audio (audible voice, Content-Type, or file extension); an unknown voice/format surfaces a visible error rather than silently falling back.
  • STT: transcribing a KNOWN spoken clip returns text that MATCHES the words spoken (not empty, not garbage) and renders in the UI; translating a non-English clip returns English text. Any timestamps the UI shows line up with the audio.
  • STT: a silent / empty / too-short clip is handled gracefully — an empty or "no speech detected" result shown in the UI, never a crash or a spinner that hangs forever.
  • The audio artifact is served from the app's OWN origin (a stored upload or a route the app serves), NOT a raw expiring provider URL — copy the audio src, reload it later, and confirm it still plays.
  • A provider error (bad key, quota, unsupported format/voice, or a capability the wired provider lacks) surfaces as a readable message in the UI, not an unhandled crash or a blank/broken player.
  • Synthesis/transcription runs SERVER-SIDE and authorized: the provider key never reaches the browser (check the network tab / client bundle), and no unauthenticated or unbounded endpoint lets a caller fire arbitrary synthesize/transcribe calls — each is billed per character / per minute of audio, so an open endpoint is a cost-DoS vector.