A conversational AI as one pipeline

Add your own node to a model pipeline put one custom node beside the zoo's text trio. This guide composes a complete application the same way: a conversational AI as one ClikaRT::runtime::Pipeline, a spoken question in as wav bytes, the spoken answer out. Eight steps, three of them the zoo's serving nodes, five of them yours:

step	node	in -> out
listen	yours	wav bytes -> whisper features
transcribe	yours	features -> question text
template	yours	question -> the chat-templated prompt
tokenize	zoo TokenizerNode	prompt -> input_ids
decode	zoo GenerativeDecoderModel	input_ids -> tokens, continuous-batched
detokenize	zoo DetokenizerNode	tokens -> answer text
speak	yours	answer text -> waveform and its sample rate
post	yours	waveform -> the peak-normalized waveform

The complete program is examples/cpp/modelverse/03_conversational_pipeline.cpp in the examples tree.

The models

Three registry loads. The TTS is a voice-cloning family and requires a voice reference; any short spoken wav works. max_seq sizes the decoder's KV pool, and a conversation turn needs nowhere near a 131072-token context window, so cap it:

constexpr const char* kSttRepo = "openai/whisper-large-v3-turbo";
constexpr const char* kLlmRepo = "meta-llama/Llama-3.2-1B-Instruct";
constexpr const char* kTtsRepo = "ResembleAI/chatterbox-flash";
constexpr std::int64_t kLlmMaxSeq = 8192;

const mv::ModelRegistry& registry = mv::ModelRegistry::builtin();
mv::LoadOptions load;
load.device = device.value();
mv::LoadOptions llm_load = load;
llm_load.max_seq = kLlmMaxSeq;
const mv::SttModel        stt = unwrap(registry.load_stt(kSttRepo, load));
const mv::GenerativeModel llm = unwrap(registry.load_generative(kLlmRepo, llm_load));
const mv::TtsModel        tts = unwrap(registry.load_tts(kTtsRepo, load));

Placement. --device puts all three models on one device (the CPU by default, cuda on request). A single request runs the steps in sequence, so the models share that device without contention.

The custom nodes

A node is the same three-member runtime::Model contract the custom-node guide walks: schema() names the input and output tensors, phases() declares one RunOnce phase, Phase_RunOnce does the work. The speech nodes wrap the model handles they are given; the tensor specs they share are three small helpers over spec::TensorSpec (a byte string, a Float32 tensor of dynamic dims, an Int32 scalar):

spec::TensorSpec bytes_spec(const char* name) {
    return spec::TensorSpec{name, DataType::UInt8, {spec::TensorSpec::kDynamicDim},
                            /*optional=*/false};
}
spec::TensorSpec f32_spec(const char* name, std::size_t rank) {
    return spec::TensorSpec{name, DataType::Float32,
                            std::vector<std::int64_t>(rank, spec::TensorSpec::kDynamicDim),
                            /*optional=*/false};
}
spec::TensorSpec i32_scalar_spec(const char* name) {
    return spec::TensorSpec{name, DataType::Int32, {}, /*optional=*/false};
}

The listen node turns the request's encoded wav bytes into the whisper feature tensor through the model's own preprocessor configuration:

class ListenNode final : public crt::Model {
 public:
    explicit ListenNode(const mv::SttModel& stt) : stt_(&stt) {
        this->schema_.inputs.push_back(bytes_spec(kAudioIn));
        this->schema_.outputs.push_back(f32_spec(kFeatures, 3));
        this->phases_.push_back(crt::PhaseSpec{"listen", crt::PhaseKind::RunOnce});
    }
    const crt::ModelSchema&             schema() const override { return this->schema_; }
    ClikaRT::Span<const crt::PhaseSpec> phases() const override { return this->phases_; }
    void Phase_RunOnce(const crt::PhaseSpec&, crt::PhaseContext& ctx) override {
        const std::vector<std::uint8_t> wav =
            unwrap(ctx.inputs->get(kAudioIn)).item_as_vec<std::uint8_t>();
        ctx.outputs->set(kFeatures, this->stt_->features_from_bytes(wav));
    }

 private:
    const mv::SttModel*         stt_;
    crt::ModelSchema            schema_;
    std::vector<crt::PhaseSpec> phases_;
};

The transcribe node is the same shape over SttModel::transcribe_features, and the speak node wraps TtsModel::synthesize with the required voice reference, emitting the waveform plus its sample rate as two outputs:

void Phase_RunOnce(const crt::PhaseSpec&, crt::PhaseContext& ctx) override {
    mv::SynthesizeOptions opts;
    opts.voice = this->voice_;
    const mv::SynthesizedAudio audio = this->tts_->synthesize(
        mvr::byte_tensor_to_string(unwrap(ctx.inputs->get(kAnswerText))), opts);
    const std::int32_t rate = audio.sample_rate;
    ctx.outputs->set(kSpeech, unwrap(Tensor::from_data(
        audio.samples.data(), {static_cast<std::int64_t>(audio.samples.size())},
        DataType::Float32)));
    ctx.outputs->set(kSpeechRate, unwrap(Tensor::from_data(&rate, {}, DataType::Int32)));
}

The post node peak-normalizes the waveform in tensor math, so it runs wherever the waveform lives; the sample rate rides from speak straight to the pipeline output. Request inputs are read-only on the node side, so the two ops that read speech allocate their results and only those are written in place:

void Phase_RunOnce(const crt::PhaseSpec&, crt::PhaseContext& ctx) override {
    const Tensor speech = unwrap(ctx.inputs->get(kSpeech));
    Tensor peak = unwrap(ops::amax(unwrap(ops::abs(speech))));
    CLIKA_CHECK(ops::clamp_(peak, kPeakFloor));
    Tensor normalized = unwrap(ops::div(speech, peak));
    CLIKA_CHECK(ops::mul_(normalized, kPeakTarget));
    ctx.outputs->set(kAudioOut, std::move(normalized));
}

One build flag: a translation unit deriving from runtime::Model compiles with -fno-rtti, matching how the library builds.

The chat template is the caller's job

A pipeline consumes prompt bytes as-is; there is no hidden templating between nodes. The template node renders the transcript into the checkpoint's own chat framing, generation turn appended, so the decoder answers the question instead of continuing it:

void Phase_RunOnce(const crt::PhaseSpec&, crt::PhaseContext& ctx) override {
    ClikaRT::json::Json messages = ClikaRT::json::Json::array();
    ClikaRT::json::Json turn     = ClikaRT::json::Json::object();
    turn["role"]    = ClikaRT::json::Json("user");
    turn["content"] = ClikaRT::json::Json(
        mvr::byte_tensor_to_string(unwrap(ctx.inputs->get(kTranscript))));
    messages.push_back(std::move(turn));
    ctx.outputs->set(mvr::kPrompt, mvr::string_to_byte_tensor(unwrap(
        this->tokenizer_->apply_chat_template(messages, /*add_generation_prompt=*/true))));
}

Skip this step and an instruction-tuned model greedy-decodes an untemplated instruction straight to its end-of-turn token; the reply is empty and nothing errors. The template node is where that goes right.

Wiring

The zoo trio constructs directly. A PipelineStep names the step, points at its node (non-owning; you keep the node alive), carries the step's execution knobs, and two rename maps: input_map from a pipeline name to a node input, output_map from a node output to a pipeline name. Edges otherwise wire by name, so one rename routes the detokenizer's generic text onto the conversational answer_text edge, where the speak node picks it up, and the decode step's scheduler admits kMaxActive sessions at a time:

crt::PipelineStep step(const char* name, crt::Model& model) {
    crt::PipelineStep st;
    st.name  = name;
    st.model = &model;
    return st;
}

mvr::TokenizerNode tokenize(llm.tokenizer(), llm.device(),
                            /*add_special_tokens=*/false);
mvr::GenerativeDecoderModel decode(llm.provider(), gen, kMaxActive,
                                   mv::KVCacheMode::Continuous, &llm.tokenizer(),
                                   /*prefill_chunk_tokens=*/0);
mvr::DetokenizerNode detokenize(llm.tokenizer());
SpeakNode            speak(tts, voice_path);
PostNode             post;

std::vector<crt::PipelineStep> steps;
steps.push_back(step("listen", listen));
steps.push_back(step("transcribe", transcribe));
steps.push_back(step("template", prompt));
steps.push_back(step("tokenize", tokenize));
steps.push_back(step("decode", decode));
steps.back().exec_config.scheduler.max_active_sessions = kMaxActive;
steps.push_back(step("detokenize", detokenize));
steps.back().output_map.emplace_back(mvr::kText, kAnswerText);
steps.push_back(step("speak", speak));
steps.push_back(step("post", post));

crt::ModelSchema ext;
ext.inputs.push_back(bytes_spec(kAudioIn));
ext.outputs.push_back(f32_spec(kAudioOut, 1));
ext.outputs.push_back(i32_scalar_spec(kSpeechRate));
ext.outputs.push_back(bytes_spec(kAnswerText));
crt::Pipeline pipeline = crt::Pipeline::create(std::move(steps), std::move(ext));

The external schema is the caller's whole contract: only audio is settable from outside; audio_out, speech_rate and answer_text are readable; every edge between the nodes stays internal.

One spoken question

The request surface is the one every pipeline speaks. The response carries the answer's text, its normalized waveform and the waveform's sample rate; io::save_audio writes the wav file (io::encode_audio yields the same bytes in memory, for a serving payload):

crt::Request request;
request.inputs.set(kAudioIn, bytes_to_tensor(question));
const crt::Response response =
    unwrap(pipeline.await(unwrap(pipeline.enqueue(std::move(request)))));

const std::string answer =
    mvr::byte_tensor_to_string(unwrap(response.outputs.get(kAnswerText)));
const Tensor reply = unwrap(response.outputs.get(kAudioOut));
const std::int32_t rate =
    unwrap(response.outputs.get(kSpeechRate)).item<std::int32_t>();
CLIKA_CHECK(ClikaRT::io::save_audio(reply, rate, reply_path));

With a spoken "What is the capital of France?" as question.wav and any short spoken clip as voice.wav (espeak-ng -v en-us -w question.wav "What is the capital of France?" synthesizes one when no recording is at hand), 03_conversational_pipeline question.wav voice.wav reply.wav --device cuda prints:

answer: The capital of France is Paris.
spoken: 61440 samples at 24000 Hz -> reply.wav

The sample count varies from run to run; the TTS decodes with sampling.

Streaming

enqueue takes RequestCallbacks for event-driven consumption. The contract from runtime/serving.h: on_chunk runs per streamed chunk and the last clean-finish chunk carries final = true; on_complete runs once, at finalize, with the response; both are contained at the boundary, so a throw from one never unwinds into the engine; with callbacks set, no blocking await is needed. The request opts in with streaming = true:

crt::RequestCallbacks cbs;
cbs.on_chunk = [&](const crt::ResponseChunk& c) {
    ++chunks;
    if (c.final) saw_final_marker = true;
    events.push_back(c.final ? 1 : 0);
};
cbs.on_complete = [&](ClikaRT::Result<crt::Response>) {
    ++completes;
    events.push_back(2);
};
crt::Request r = audio_request(question_wav);
r.streaming    = true;
crt::SessionId sid = 0;
sid = pipe.enqueue(std::move(r), std::move(cbs));

Callbacks run on the engine's pump thread; do not block in them.

Concurrent sessions

The decode step's max_active_sessions sizes its continuous batch; sessions beyond it queue rather than fail. With max_active 4, eight requests enqueued before any is awaited:

std::vector<crt::SessionId> ids;
for (std::size_t i = 0; i < args.burst_n; ++i) {
    ids.push_back(unwrap(pipe.enqueue(audio_request(question_wav))));
}
std::size_t good = 0;
for (const crt::SessionId sid : ids) {
    try {
        if (carries_answer(text_of(unwrap(pipe.await(sid)), convo::kAnswerText))) ++good;
    } catch (const ClikaRT::Error& e) {
        std::printf("[burst ] session %llu FAILED: %s\n",
                    static_cast<unsigned long long>(sid), error_detail(e).c_str());
    }
}

On CUDA the burst answers in about 2 to 3 seconds, about 0.3 seconds per session. Known limitation (Clika/ClikaRT#2058): under a burst on the CUDA paged cache a session can return a wrong answer with no error, about one burst in eight. Until it closes, verify batched answers or serve one session at a time (max_active 1, the example's setting).

Rebuilding, and the KV pool

A live GenerativeDecoderModel holds its full KV pool on the device; two of them hold two pools, an out-of-memory at the second build on a 16 GB card. Scope the first pipeline and its nodes so they destruct before a rebuild (the example calls pipeline.shutdown() before it returns), and let max_seq size the pool for the conversation you serve.

What the composition buys

The composed pipeline builds its executors, its KV pool and its warmed kernels once and reuses them every turn: about 7 seconds per session, against about 22 seconds for the same three models driven in sequence with a fresh pipeline per turn. What a manual driver that kept its pipeline still would not get is the continuous batching.