--- title: A conversational AI as one pipeline description: "Compose speech-to-text, a chat model and text-to-speech into a single runtime::Pipeline: wav bytes in, the spoken answer out." --- [Add your own node to a model pipeline](add-a-pipeline-node.md) 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 32768-token context window, so cap it: ```cpp constexpr const char* kSttRepo = "openai/whisper-large-v3-turbo"; constexpr const char* kLlmRepo = "Qwen/Qwen2.5-1.5B-Instruct"; constexpr const char* kTtsRepo = "ResembleAI/chatterbox-flash"; constexpr std::int64_t kLlmMaxSeq = 8192; ``` ```cpp 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](add-a-pipeline-node.md) 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): ```cpp 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(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: ```cpp 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 phases() const override { return this->phases_; } void Phase_RunOnce(const crt::PhaseSpec&, crt::PhaseContext& ctx) override { const std::vector wav = unwrap(ctx.inputs->get(kAudioIn)).item_as_vec(); ctx.outputs->set(kFeatures, this->stt_->features_from_bytes(wav)); } private: const mv::SttModel* stt_; crt::ModelSchema schema_; std::vector 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: ```cpp 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(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: ```cpp 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: ```cpp 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: ```cpp crt::PipelineStep step(const char* name, crt::Model& model) { crt::PipelineStep st; st.name = name; st.model = &model; return st; } ``` ```cpp mvr::TokenizerNode tokenize(llm.tokenizer(), /*add_special_tokens=*/false); mvr::GenerativeDecoderModel decode(llm, gen, kMaxActive, mv::KVCacheMode::Continuous, /*prefill_chunk_tokens=*/0); mvr::DetokenizerNode detokenize(llm.tokenizer()); SpeakNode speak(tts, voice_path); PostNode post; std::vector 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): ```cpp 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(); 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: ```text 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`: ```cpp 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) { ++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: ```cpp std::vector 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(sid), error_detail(e).c_str()); } } ``` Where a session silently returning a wrong answer under a burst would be costly, verify batched answers or serve one session at a time (`max_active` 1, the example's setting); Clika/ClikaRT#2058 is the tracker for that failure on the CUDA paged cache. ## 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.