Cogitator
Voice

Pipeline Mode

Process voice through STT -> Agent -> TTS with VoicePipeline for batch processing and PipelineSession for continuous streaming conversations with VAD-driven turn detection.

Overview

Pipeline mode routes audio through a three-stage loop: Speech-to-Text, Agent processing, Text-to-Speech. This works with any Cogitator agent regardless of the underlying LLM backend.

Two classes handle different use cases:

  • VoicePipeline — batch processing: send a complete audio buffer, get back transcript + response + audio
  • PipelineSession — streaming: push audio chunks continuously, get events as things happen

VoicePipeline

One-shot processing for complete audio buffers.

import { VoicePipeline, OpenAISTT, OpenAITTS, createCogitatorRunner } from '@cogitator-ai/voice';

const pipeline = new VoicePipeline({
  stt: new OpenAISTT({ apiKey: process.env.OPENAI_API_KEY! }),
  tts: new OpenAITTS({ apiKey: process.env.OPENAI_API_KEY! }),
  agent: createCogitatorRunner(cogitator, agent),
});

The agent is any VoiceAgentRunner — { run(input, context?) => Promise<{ content }> }. context carries the sessionId and an abort signal. createCogitatorRunner(cogitator, agent) builds one from a Cogitator runtime and agent (memory thread voice:<sessionId> per session).

process(audio, options?)

Transcribes the audio, runs the agent, synthesizes the response. Silent input (empty transcript) skips the agent and TTS and returns empty strings and an empty buffer; an empty agent reply skips TTS. Pass { sessionId } to forward a session id to the agent.

const result = await pipeline.process(audioBuffer);

console.log(result.transcript); // what the user said
console.log(result.response); // what the agent replied
// result.audio — synthesized response as Buffer

TTS options

ttsOptions is passed to every synthesize / streamSynthesize call made by the pipeline and its sessions. Use it to pick the output format or a per-pipeline voice, speed or instructions:

const pipeline = new VoicePipeline({
  stt: new OpenAISTT({ apiKey: process.env.OPENAI_API_KEY! }),
  tts: new OpenAITTS({ apiKey: process.env.OPENAI_API_KEY! }),
  agent: createCogitatorRunner(cogitator, agent),
  ttsOptions: { format: 'pcm16' },
});

Without ttsOptions the TTS provider's defaults apply (MP3 output). format: 'pcm16' returns raw PCM16 mono at 24kHz. VoiceAgent accepts the same ttsOptions in pipeline mode.

openai is an optional peer dependency: install it only when you use OpenAISTT / OpenAITTS. They load it on the first transcribe / synthesize call, which throws with an install hint if it is missing.

createSession(options?)

Creates a PipelineSession for streaming use. Pass { sessionId } to set the id the agent sees (default: a random id). See below.

PipelineSession

Streaming mode for continuous conversations. Push audio chunks as they arrive, receive events for each stage of processing.

const session = pipeline.createSession();

session.id; // passed to the agent as context.sessionId
session.currentState; // 'idle' | 'listening' | 'processing' | 'speaking'

Pushing Audio

// push PCM16 audio chunks as they arrive from the microphone
session.pushAudio(pcm16Chunk);

If the pipeline has a VAD provider, the session automatically detects when speech starts and ends. Without VAD, it streams to STT immediately and the turn is committed with endAudio().

session.endAudio(); // commit buffered audio (no-VAD mode)
await session.sendText('Hi there'); // run a turn from text, bypassing STT

Events

session.on('speech_start', () => {
  console.log('User started speaking');
});

session.on('speech_end', () => {
  console.log('User stopped speaking, transcribing...');
});

session.on('transcript', (text: string, isFinal: boolean) => {
  if (isFinal) {
    console.log('Final transcript:', text);
  } else {
    console.log('Interim:', text);
  }
});

session.on('agent_response', (text: string) => {
  console.log('Agent:', text);
});

session.on('audio', (chunk: Buffer) => {
  // streamed TTS audio — play it back to the user
  playAudio(chunk);
});

session.on('turn_end', () => {
  console.log('Agent finished speaking');
});

session.on('error', (error: Error) => {
  console.error('Session error:', error);
});
EventPayloadDescription
speech_start—VAD detected speech beginning (VAD only)
speech_end—VAD detected speech ending (VAD only)
transcript(text, isFinal)STT result (interim or final)
agent_response(text)Agent's text response
audio(chunk: Buffer)Streamed TTS audio chunk
turn_end—Response fully spoken (or empty)
error(error: Error)Processing error (STT, VAD, agent, TTS)

Interruption

Interrupt the current processing at any stage — discards the STT stream, aborts the agent run (its signal fires) and stops TTS streaming. A stale turn never emits events or changes state after an interruption:

session.interrupt();

After interruption, the session returns to idle and is ready for new audio input.

Closing

await session.close();

Aborts any in-progress turn (without waiting for hung providers), releases the STT stream and removes event listeners.

Pipeline with VAD

When a VAD provider is set, the session handles turn detection automatically:

import { VoicePipeline, OpenAISTT, OpenAITTS, EnergyVAD } from '@cogitator-ai/voice';

const pipeline = new VoicePipeline({
  stt: new OpenAISTT({ apiKey: process.env.OPENAI_API_KEY! }),
  tts: new OpenAITTS({ apiKey: process.env.OPENAI_API_KEY! }),
  vad: new EnergyVAD({ threshold: 0.01, silenceDuration: 500 }),
  agent: myAgent,
});

const session = pipeline.createSession();

session.on('speech_start', () => {
  // VAD detected speech — session opens an STT stream
});

session.on('speech_end', () => {
  // VAD detected silence — session closes STT, runs agent, streams TTS
});

// just keep pushing audio (PCM16 at the VAD's sample rate), VAD handles the rest
microphoneStream.on('data', (chunk) => {
  session.pushAudio(chunk);
});

VoiceAgent

VoiceAgent is the high-level class that combines pipeline/realtime processing with WebSocket transport and session management. For pipeline mode, it creates a VoicePipeline internally and manages per-client sessions.

import { VoiceAgent, OpenAISTT, OpenAITTS, EnergyVAD } from '@cogitator-ai/voice';

const voiceAgent = new VoiceAgent({
  agent: createCogitatorRunner(cogitator, agent),
  mode: 'pipeline',
  stt: new OpenAISTT({ apiKey: process.env.OPENAI_API_KEY! }),
  tts: new OpenAITTS({ apiKey: process.env.OPENAI_API_KEY! }),
  vad: new EnergyVAD(),
  transport: {
    path: '/voice',
    maxConnections: 50,
    verifyClient: async (req) => isValidToken(req.headers.authorization), // false → 401; { code, message } for another status
  },
});

voiceAgent.on('session_start', (id) => console.log(`Session: ${id}`));
voiceAgent.on('session_end', (id) => console.log(`Ended: ${id}`));
voiceAgent.on('error', (err) => console.error(err));

await voiceAgent.listen(8080);

console.log(`Active sessions: ${voiceAgent.activeSessions}`);
await voiceAgent.close();

Use voiceAgent.attach(httpServer) instead of listen() to share an existing HTTP server; upgrades for other paths are left to other handlers (with listen() they get 404). voiceAgent.port is the bound port. Errors are only emitted when an error listener is attached — a misbehaving client cannot crash the process.

Transport options

OptionDefaultDescription
path'/voice'WebSocket path
maxConnections100Further upgrades are rejected with 503
verifyClient—(req) => boolean | { code, message } (may be async): true accepts, false rejects with 401, { code, message } rejects with that status; a throw rejects with 500
transport: {
  verifyClient: (req) => {
    const token = new URL(req.url ?? '/', 'http://localhost').searchParams.get('token');
    if (!token) return false;
    return token === process.env.VOICE_TOKEN ? true : { code: 403, message: 'Forbidden' };
  },
},

Browsers cannot set headers on a WebSocket, so pass credentials in the query string or a cookie. Idle clients that stop answering pings (every 30s) are dropped. WebSocketTransport is exported too, for building your own server around VoiceClient connections (audio / message events, sendAudio(), sendMessage()).

WebSocket protocol

DirectionFormatContent
Client -> ServerBinaryPCM16 mono (16kHz pipeline / Gemini, 24kHz OpenAI realtime)
Client -> ServerJSON{ type: 'interrupt' }, { type: 'end_of_speech' } (pipeline), { type: 'text', text }
Server -> ClientBinaryResponse audio (TTS output format in pipeline mode, MP3 by default; PCM16 24kHz in realtime mode)
Server -> ClientJSON{ type: 'transcript', text, isFinal }, agent_response, speech_start, speech_end, turn_end

In realtime mode transcript carries role ('user' or 'assistant') instead of isFinal, and there is no agent_response or speech_end. Invalid JSON and unknown control types are reported through the error event and otherwise ignored.

VoicePipelineConfig

FieldTypeRequiredDescription
sttSTTProviderYesSpeech-to-text provider
ttsTTSProviderYesText-to-speech provider
vadVADProviderNoVoice activity detection
ttsOptionsTTSOptionsNoOptions for every TTS call (format, voice, speed, instructions)
agentVoiceAgentRunnerYescreateCogitatorRunner(...) or any { run(input, ctx) }
sampleRatenumberNoDeprecated, unused (configure the STT/VAD providers instead)

Voice Activity Detection

Detect when users start and stop speaking with @cogitator-ai/voice VAD providers — zero-dependency EnergyVAD for simple environments and neural Silero VAD for noisy conditions.

Realtime Mode

Native speech-to-speech with RealtimeSession — connect to OpenAI Realtime API or Gemini Live API for low-latency voice conversations with tool calling support.

On this page

OverviewVoicePipelineprocess(audio, options?)TTS optionscreateSession(options?)PipelineSessionPushing AudioEventsInterruptionClosingPipeline with VADVoiceAgentTransport optionsWebSocket protocolVoicePipelineConfig