Examples

export RABEL_API_KEY=<your_api_key_here>

pip install websockets

export DEEPGRAM_API_KEY=<your_deepgram_api_key_here>

Copy

// Bell beside your own speech-to-text: Deepgram's streaming, fed your WAV file. Each interim result goes to Bell as `heard`
// (the turn's words so far), the turn's final words as `end_of_turn`; Bell's guesses are printed as they come.
// Node 22; RABEL_API_KEY and DEEPGRAM_API_KEY in your environment; then: node with_deepgram.ts
import { readFileSync } from "node:fs";

const API = "https://api.rabellabs.com";
const KEY = process.env.RABEL_API_KEY ?? "YOUR_API_KEY";
const DEEPGRAM_KEY = process.env.DEEPGRAM_API_KEY ?? "YOUR_DEEPGRAM_API_KEY";
const DEEPGRAM = "wss://api.deepgram.com/v1/listen?model=nova-3&encoding=linear16&sample_rate=16000&interim_results=true";
const RECORDING = "recording.wav"; // your WAV file: 16 kHz, mono, 16-bit
// Another format? ffmpeg -i in.mp3 -ar 16000 -ac 1 -sample_fmt s16 recording.wav

const wav = readFileSync(RECORDING);
if (wav.readUInt32LE(24) !== 16000 || wav.readUInt16LE(22) !== 1 || wav.readUInt16LE(34) !== 16) {
  throw new Error(`${RECORDING} must be 16 kHz, mono, 16-bit (see the ffmpeg line above)`);
}
const audio = wav.subarray(wav.indexOf("data") + 8);

const ids = new Set();
let t0 = 0, turns = 0, ends = 0, deepgramDone = false;
const at = () => ((performance.now() - t0) / 1000).toFixed(2).padStart(5);
const bell = new WebSocket(`${API.replace("https", "wss")}/v1/conversation`);
bell.onopen = () => {
  bell.send(JSON.stringify({ key: KEY }));
  console.log("Connecting to Bell...");
};
bell.onmessage = (e) => {
  const m = JSON.parse(e.data);
  if (m.type === "ready") { t0 = performance.now(); listen(); }
  if (m.type === "guess") { ids.add(m.id); console.log(`${at()}s  guess:       ${m.text}`); }
  if (m.type === "end_of_turn") {
    const matched = ids.has(m.guess_id) ? `(matched guess ${m.guess_id})` : "(no guess matched)";
    console.log(`${at()}s  end of turn: ${m.text} ${matched}`);
    ends++;
    if (deepgramDone && ends >= turns) bell.close(); // Bell's end of turn for each turn sent
  }
  if (m.type === "error") console.log("error:", m.message);
};

function listen() {
  const dg = new WebSocket(DEEPGRAM, ["token", DEEPGRAM_KEY]);
  let final = ""; // the turn's finalized words (Deepgram's is_final results, appended)
  dg.onopen = async () => { // at the pace of speech, 80 ms at a time, then Deepgram's CloseStream
    for (let i = 0; i < audio.length; i += 2560) {
      dg.send(audio.subarray(i, i + 2560));
      await new Promise((r) => setTimeout(r, 80));
    }
    dg.send(JSON.stringify({ type: "CloseStream" }));
  };
  dg.onmessage = (e) => {
    const r = JSON.parse(e.data);
    if (r.type !== "Results") return;
    const words = [final, r.channel.alternatives[0].transcript].filter(Boolean).join(" ");
    if (r.is_final) final = words;
    if (r.speech_final && final) { // Deepgram's end of the speaker's turn
      bell.send(JSON.stringify({ type: "end_of_turn", text: final }));
      final = "";
      turns++;
    } else if (words) bell.send(JSON.stringify({ type: "heard", text: words }));
  };
  dg.onclose = () => {
    if (final) { bell.send(JSON.stringify({ type: "end_of_turn", text: final })); turns++; } // the file ended inside a turn
    deepgramDone = true;
    if (ends >= turns) bell.close();
  };
}

# Bell beside your own speech-to-text: Deepgram's streaming, fed your WAV file. Each interim result goes to Bell as `heard`
# (the turn's words so far), the turn's final words as `end_of_turn`; Bell's guesses are printed as they come.
# pip install websockets; RABEL_API_KEY and DEEPGRAM_API_KEY in your environment; then: python with_deepgram.py
import asyncio, json, os, time, wave
import websockets

API = "https://api.rabellabs.com"
KEY = os.environ.get("RABEL_API_KEY", "YOUR_API_KEY")
DEEPGRAM_KEY = os.environ.get("DEEPGRAM_API_KEY", "YOUR_DEEPGRAM_API_KEY")
DEEPGRAM = "wss://api.deepgram.com/v1/listen?model=nova-3&encoding=linear16&sample_rate=16000&interim_results=true"
RECORDING = "recording.wav"  # your WAV file: 16 kHz, mono, 16-bit
# Another format? ffmpeg -i in.mp3 -ar 16000 -ac 1 -sample_fmt s16 recording.wav

async def main():
    w = wave.open(RECORDING)
    if (w.getframerate(), w.getnchannels(), w.getsampwidth()) != (16000, 1, 2):
        raise SystemExit(f"{RECORDING} must be 16 kHz, mono, 16-bit (see the ffmpeg line above)")
    audio = w.readframes(w.getnframes())
    async with websockets.connect(API.replace("https", "wss") + "/v1/conversation") as bell:
        await bell.send(json.dumps({"key": KEY}))
        print("Connecting to Bell...", flush=True)
        m = json.loads(await bell.recv())
        if m["type"] != "ready":
            raise SystemExit(f"error: {m['message']}")
        t0, ids, turns, ends = time.monotonic(), set(), 0, asyncio.Queue()

        async def from_bell():
            async for msg in bell:
                m = json.loads(msg)
                if m["type"] == "guess":
                    ids.add(m["id"])
                    print(f"{time.monotonic() - t0:5.2f}s  guess:       {m['text']}")
                elif m["type"] == "end_of_turn":
                    matched = f"(matched guess {m['guess_id']})" if m["guess_id"] in ids else "(no guess matched)"
                    print(f"{time.monotonic() - t0:5.2f}s  end of turn: {m['text']} {matched}")
                    ends.put_nowait(m)
                elif m["type"] == "error":
                    print("error:", m["message"])
        listener = asyncio.create_task(from_bell())

        async with websockets.connect(DEEPGRAM, additional_headers={"Authorization": "Token " + DEEPGRAM_KEY}) as dg:
            async def send_audio():  # at the pace of speech, 80 ms at a time, then Deepgram's CloseStream
                for i in range(0, len(audio), 2560):
                    await dg.send(audio[i:i + 2560])
                    await asyncio.sleep(0.08)
                await dg.send(json.dumps({"type": "CloseStream"}))
            sender = asyncio.create_task(send_audio())

            final = ""  # the turn's finalized words (Deepgram's is_final results, appended)
            async for msg in dg:
                r = json.loads(msg)
                if r.get("type") != "Results":
                    continue
                words = " ".join(x for x in (final, r["channel"]["alternatives"][0]["transcript"]) if x)
                if r["is_final"]:
                    final = words
                if r["speech_final"] and final:  # Deepgram's end of the speaker's turn
                    await bell.send(json.dumps({"type": "end_of_turn", "text": final}))
                    final, turns = "", turns + 1
                elif words:
                    await bell.send(json.dumps({"type": "heard", "text": words}))
        if final:  # the file ended inside a turn
            await bell.send(json.dumps({"type": "end_of_turn", "text": final}))
            turns += 1
        for _ in range(turns):  # Bell's end of turn for each turn sent
            await asyncio.wait_for(ends.get(), 10)
        listener.cancel()

asyncio.run(main())

python with_deepgram.py      # or: node with_deepgram.ts

Connecting to Bell...
 2.62s  guess:       please turn off only
 2.63s  guess:       please turn off only one light
 2.63s  guess:       please turn off only one
 3.66s  guess:       please turn off only right side line
 3.66s  guess:       please turn off only right side line lights
 3.66s  guess:       please turn off only right side line light
 3.73s  end of turn: please turn off only right side light (no guess matched)

Copy

// Start your reply on Bell's guess, before the speaker finishes.
// Node 22; RABEL_API_KEY in your environment; then: node reply_on_guess.ts
const API = "https://api.rabellabs.com";
const KEY = process.env.RABEL_API_KEY ?? "YOUR_API_KEY";

// Your reply model goes here (an LLM call, then your voice). This stand-in takes 0.8 s.
// `signal` tells it the reply was dropped, so it can stop its own call.
async function myReply(words, signal) {
  await new Promise((r) => setTimeout(r, 800));
  return `(your reply to: ${words})`;
}

// the same words, ignoring case and punctuation
const norm = (s) => (s ?? "").toLowerCase().replace(/[^\w\s']/g, "").split(/\s+/).filter(Boolean).join(" ");
const same = (a, b) => !!a && norm(a) === norm(b);

// the sample turn's words as a speech-to-text sends them; in your agent, your speech-to-text's own
const sample = await (await fetch(`${API}/v1/sample.json`)).json();

let reply = null, startedOn = null, abort = null, heard = "", guesses = [], t0 = 0;
const at = () => ((performance.now() - t0) / 1000).toFixed(2).padStart(5);
function start(words) { // start the reply on these words, dropping the one in progress
  abort?.abort();
  abort = new AbortController();
  reply = myReply(words, abort.signal);
  startedOn = words;
  console.log(`${at()}s  reply started on: ${words}`);
}

const ws = new WebSocket(`${API.replace("https", "wss")}/v1/conversation`);
ws.onopen = () => {
  ws.send(JSON.stringify({ key: KEY }));
  console.log("Connecting to Bell...");
};
ws.onmessage = async (e) => {
  const m = JSON.parse(e.data);
  if (m.type === "ready") {
    t0 = performance.now();
    let t = 0;
    for (const w of sample.words) { // the turn's words so far, each time your speech-to-text has new words
      await new Promise((r) => setTimeout(r, w.ms - t));
      t = w.ms;
      heard = w.text;
      ws.send(JSON.stringify({ type: "heard", text: heard }));
      const earlier = guesses.find((g) => same(g, heard));
      if (earlier && !same(earlier, startedOn)) start(earlier); // the speaker said a guess Bell made earlier
    }
    await new Promise((r) => setTimeout(r, 200));
    ws.send(JSON.stringify({ type: "end_of_turn", text: sample.final })); // its final words
  } else if (m.type === "guess") {
    guesses.push(m.text);
    if (!same(startedOn, heard)) start(m.text); // keep a reply whose words the speaker has already said
  } else if (m.type === "end_of_turn") {
    const tEnd = performance.now();
    console.log(`${at()}s  end of turn: ${m.text}`);
    if (same(startedOn, m.text)) console.log("        the guess was right: keeping the reply started early");
    else { console.log("        the guess was wrong: dropping it and starting again"); start(m.text); }
    const text = await reply;
    console.log(`${at()}s  reply ready ${((performance.now() - tEnd) / 1000).toFixed(2)}s after the end of turn: ${text}`);
    ws.close();
  } else if (m.type === "error") console.log("error:", m.message);
};

# Start your reply on Bell's guess, before the speaker finishes.
# pip install websockets; RABEL_API_KEY in your environment; then: python reply_on_guess.py
import asyncio, json, os, re, time, urllib.request
import websockets

API = "https://api.rabellabs.com"
KEY = os.environ.get("RABEL_API_KEY", "YOUR_API_KEY")

async def my_reply(words):
    """Your reply model goes here (an LLM call, then your voice). This stand-in takes 0.8 s."""
    await asyncio.sleep(0.8)
    return f"(your reply to: {words})"

def same(a, b):  # the same words, ignoring case and punctuation
    norm = lambda s: re.sub(r"[^\w\s']", "", (s or "").lower()).split()
    return bool(a) and norm(a) == norm(b)

async def main():
    # the sample turn's words as a speech-to-text sends them; in your agent, your speech-to-text's own
    sample = json.load(urllib.request.urlopen(urllib.request.Request(API + "/v1/sample.json", headers={"User-Agent": "rabel-api-example"})))
    reply, started_on, heard, guesses = None, None, "", []

    def start(words):  # start the reply on these words, dropping the one in progress
        nonlocal reply, started_on
        if reply:
            reply.cancel()
        reply, started_on = asyncio.create_task(my_reply(words)), words
        print(f"{time.monotonic() - t0:5.2f}s  reply started on: {words}")

    async with websockets.connect(API.replace("https", "wss") + "/v1/conversation") as ws:
        await ws.send(json.dumps({"key": KEY}))
        print("Connecting to Bell...", flush=True)

        async def send():
            nonlocal heard
            t = 0
            for w in sample["words"]:  # the turn's words so far, each time your speech-to-text has new words
                await asyncio.sleep((w["ms"] - t) / 1000)
                t, heard = w["ms"], w["text"]
                await ws.send(json.dumps({"type": "heard", "text": heard}))
                earlier = next((g for g in guesses if same(g, heard)), None)
                if earlier and not same(earlier, started_on):  # the speaker said a guess Bell made earlier
                    start(earlier)
            await asyncio.sleep(0.2)
            await ws.send(json.dumps({"type": "end_of_turn", "text": sample["final"]}))  # its final words

        async for msg in ws:
            m = json.loads(msg)
            if m["type"] == "ready":
                t0 = time.monotonic()
                sender = asyncio.create_task(send())
            elif m["type"] == "guess":
                guesses.append(m["text"])
                if not same(started_on, heard):  # keep a reply whose words the speaker has already said
                    start(m["text"])
            elif m["type"] == "end_of_turn":
                t_end = time.monotonic()
                print(f"{t_end - t0:5.2f}s  end of turn: {m['text']}")
                if same(started_on, m["text"]):
                    print("        the guess was right: keeping the reply started early")
                else:
                    print("        the guess was wrong: dropping it and starting again")
                    start(m["text"])
                text = await reply
                print(f"{time.monotonic() - t0:5.2f}s  reply ready {time.monotonic() - t_end:.2f}s after the end of turn: {text}")
                break
            elif m["type"] == "error":
                print("error:", m["message"])

asyncio.run(main())

python reply_on_guess.py      # or: node reply_on_guess.ts

Connecting to Bell...
 2.12s  reply started on: please
 2.81s  reply started on: please turn it off
 2.82s  reply started on: please turn it on
 2.82s  reply started on: please turn
 3.06s  reply started on: please turn off the lights
...
 3.28s  reply started on: please turn off the only
 3.28s  reply started on: please turn off the only one
 3.49s  reply started on: please turn off the only right side light
 3.49s  reply started on: please turn off the only right side
 3.72s  reply started on: please turn off the only right side light
 3.94s  end of turn: Please turn off the only right side light.
        the guess was right: keeping the reply started early
 4.52s  reply ready 0.59s after the end of turn: (your reply to: please turn off the only right side light)

Copy

// Handle the API's refusals: wait and try again on 429 and 529, stop on 402.
// This opens three conversations at once with one key; a key has at most two open, so the third is refused and tries again.
// Node 22; RABEL_API_KEY in your environment; then: node refusals.ts
const API = "https://api.rabellabs.com";
const KEY = process.env.RABEL_API_KEY ?? "YOUR_API_KEY";
const MAX_WAIT_S = 120; // keep trying for at most this long (a key's day limit asks you to wait until 00:00 UTC: stop)

const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
// the sample turn's words as a speech-to-text sends them; in your agent, your speech-to-text's own
const sample = await (await fetch(`${API}/v1/sample.json`)).json();

// One try: resolves {words} at the end of the turn, or {refused} with the API's error message.
function tryOnce(name) {
  return new Promise((resolve) => {
    const ws = new WebSocket(`${API.replace("https", "wss")}/v1/conversation`);
    let started = false;
    ws.onopen = () => ws.send(JSON.stringify({ key: KEY }));
    ws.onclose = () => resolve({});
    ws.onmessage = async (e) => {
      const m = JSON.parse(e.data);
      if (m.type === "error") {
        if (!started) resolve({ refused: m });
        else { console.log(`${name}: stopped ${m.status} ${m.error}: ${m.message}`); resolve({}); }
      } else if (m.type === "end_of_turn") {
        resolve({ words: m.text });
        ws.close();
      } else if (m.type === "ready") {
        started = true;
        console.log(`${name}: ready`);
        let t = 0;
        for (const w of sample.words) { // the turn's words so far, each time your speech-to-text has new words
          await sleep(w.ms - t);
          t = w.ms;
          ws.send(JSON.stringify({ type: "heard", text: w.text }));
        }
        await sleep(200);
        ws.send(JSON.stringify({ type: "end_of_turn", text: sample.final })); // its final words
      }
    };
  });
}

// One conversation, tried again while the API asks to wait. Returns the turn's final words, or null.
async function conversation(name) {
  const deadline = performance.now() + MAX_WAIT_S * 1000;
  for (let attempt = 1; ; attempt++) {
    const r = await tryOnce(name);
    if (r.words) { console.log(`${name}: end of turn: ${r.words}`); return r.words; }
    if (!r.refused) return null;
    const { status, error, message, retry_after_s: wait = 0 } = r.refused;
    console.log(`${name}: refused ${status} ${error}: ${message}`);
    if (status === 402) return null; // free seconds and credit used up: add credit, retrying will not help
    if (![429, 529, 503].includes(status) || wait <= 0 || performance.now() + wait * 1000 > deadline) return null;
    console.log(`${name}: waiting ${wait} s, then trying again (try ${attempt + 1})`);
    await sleep(wait * 1000);
  }
}

console.log("Connecting to Bell...");
const results = await Promise.all([1, 2, 3].map((i) => conversation(`conversation ${i}`)));
console.log(`${results.filter((r) => r !== null).length} of 3 conversations finished`);

# Handle the API's refusals: wait and try again on 429 and 529, stop on 402.
# This opens three conversations at once with one key; a key has at most two open, so the third is refused and tries again.
# pip install websockets; RABEL_API_KEY in your environment; then: python refusals.py
import asyncio, json, os, time, urllib.request
import websockets

API = "https://api.rabellabs.com"
KEY = os.environ.get("RABEL_API_KEY", "YOUR_API_KEY")
MAX_WAIT_S = 120  # keep trying for at most this long (a key's day limit asks you to wait until 00:00 UTC: stop)

async def conversation(name, sample):
    """One conversation, tried again while the API asks to wait. Returns the turn's final words, or None."""
    deadline, attempt = time.monotonic() + MAX_WAIT_S, 1
    while True:
        async with websockets.connect(API.replace("https", "wss") + "/v1/conversation") as ws:
            await ws.send(json.dumps({"key": KEY}))
            m = json.loads(await ws.recv())
            if m["type"] == "error":
                status, wait = m.get("status"), m.get("retry_after_s", 0)
                print(f"{name}: refused {status} {m['error']}: {m['message']}")
                if status == 402:  # free seconds and credit used up: add credit, retrying will not help
                    return None
                if status in (429, 529, 503) and 0 < wait and time.monotonic() + wait <= deadline:
                    attempt += 1
                    print(f"{name}: waiting {wait} s, then trying again (try {attempt})")
                    await asyncio.sleep(wait)
                    continue
                return None  # a wrong key, or a wait too long for this script
            print(f"{name}: ready")
            t = 0
            for w in sample["words"]:  # the turn's words so far, each time your speech-to-text has new words
                await asyncio.sleep((w["ms"] - t) / 1000)
                t = w["ms"]
                await ws.send(json.dumps({"type": "heard", "text": w["text"]}))
            await asyncio.sleep(0.2)
            await ws.send(json.dumps({"type": "end_of_turn", "text": sample["final"]}))  # its final words
            async for msg in ws:
                m = json.loads(msg)
                if m["type"] == "end_of_turn":
                    print(f"{name}: end of turn: {m['text']}")
                    return m["text"]
                if m["type"] == "error":
                    print(f"{name}: stopped {m.get('status')} {m['error']}: {m['message']}")
                    return None
            return None  # the socket closed without an end of turn

async def main():
    # the sample turn's words as a speech-to-text sends them; in your agent, your speech-to-text's own
    sample = json.load(urllib.request.urlopen(urllib.request.Request(API + "/v1/sample.json", headers={"User-Agent": "rabel-api-example"})))
    print("Connecting to Bell...", flush=True)
    results = await asyncio.gather(*(conversation(f"conversation {i}", sample) for i in (1, 2, 3)))
    print(f"{sum(r is not None for r in results)} of 3 conversations finished")

asyncio.run(main())

python refusals.py      # or: node refusals.ts

Connecting to Bell...
conversation 3: refused 429 key_at_once: This key already has 2 conversations open. Close one to start another.
conversation 3: waiting 1 s, then trying again (try 2)
conversation 1: ready
conversation 2: ready
conversation 3: refused 429 key_at_once: This key already has 2 conversations open. Close one to start another.
conversation 3: waiting 1 s, then trying again (try 3)
conversation 3: refused 429 key_at_once: This key already has 2 conversations open. Close one to start another.
conversation 3: waiting 1 s, then trying again (try 4)
conversation 3: refused 429 key_at_once: This key already has 2 conversations open. Close one to start another.
conversation 3: waiting 1 s, then trying again (try 5)
conversation 1: end of turn: Please turn off the only right side light.
conversation 2: end of turn: Please turn off the only right side light.
conversation 3: ready
conversation 3: end of turn: Please turn off the only right side light.
3 of 3 conversations finished

pip install -r requirements.txt

from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair, LLMUserAggregatorParams
from pipecat.turns.user_turn_strategies import EagerUserTurnStrategies

from rabel_bell_stt import RabelBellProcessor

bell = RabelBellProcessor(
    api_key=os.environ.get("RABEL_API_KEY"),
)

user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
    context,
    user_params=LLMUserAggregatorParams(
        user_turn_strategies=EagerUserTurnStrategies(),
    ),
)

pipeline = Pipeline(
    [
        transport.input(),
        stt,
        bell,
        user_aggregator,
        llm,
        tts,
        transport.output(),
        assistant_aggregator,
    ]
)

python compare.py