Add your own node to a model pipeline

Your application needs logic the model does not have: redaction, templating, routing, scoring, any transform of what goes in or comes out. In Modelverse that logic is not a wrapper around the pipeline; it is a node inside it. The zoo's serving nodes and your code meet on one contract, ClikaRT::runtime::Model, and ClikaRT::runtime::Pipeline::create composes any mix of them into one pipeline with one request/response surface.

A node implements three members: schema() declares its input and output tensors by name, phases() declares its execution phases, and Phase_RunOnce does the work. The zoo's tokenizer, generative decoder and detokenizer implement exactly the same three, which is why yours can stand beside them without an adapter layer.

The program

Same two-file project as part 4 of the tutorial; only main.cpp changes. The custom node here uppercases the reply, standing in for any post-processing you own:

main.cpp
#include <algorithm>
#include <cctype>
#include <cstdio>
#include <string>
#include <utility>
#include <vector>

#include <ClikaRT/clika_rt.h>

#include "clika_modelverse/generation/generation_config.h"
#include "clika_modelverse/hub/hub.h"
#include "clika_modelverse/registry/model_registry.h"
#include "clika_modelverse/runtime/decoder_node.h"
#include "clika_modelverse/runtime/pipeline_io.h"
#include "clika_modelverse/runtime/text_nodes.h"

namespace mv  = clika_modelverse;
namespace mvr = clika_modelverse::runtime;
namespace crt = ClikaRT::runtime;
using ClikaRT::DataType;

// The detokenizer's reply routes here under this name instead of going
// straight out.
constexpr const char* kDraftText = "draft_text";

// Everything a user writes: declare I/O, declare a phase, do the work.
class ShoutNode final : public crt::Model {
 public:
    ShoutNode() {
        schema_.inputs.push_back(ClikaRT::spec::TensorSpec{
            kDraftText, DataType::UInt8, {ClikaRT::spec::TensorSpec::kDynamicDim},
            /*optional=*/false});
        schema_.outputs.push_back(ClikaRT::spec::TensorSpec{
            mvr::kText, DataType::UInt8, {ClikaRT::spec::TensorSpec::kDynamicDim},
            /*optional=*/false});
        phases_.push_back(crt::PhaseSpec{"shout", crt::PhaseKind::RunOnce});
    }

    const crt::ModelSchema&             schema() const override { return schema_; }
    ClikaRT::Span<const crt::PhaseSpec> phases() const override { return phases_; }

    void Phase_RunOnce(const crt::PhaseSpec&, crt::PhaseContext& ctx) override {
        std::string text = mvr::byte_tensor_to_string(ctx.inputs->get(kDraftText));
        std::transform(text.begin(), text.end(), text.begin(), [](unsigned char c) {
            return static_cast<char>(std::toupper(c));
        });
        ctx.outputs->set(mvr::kText, mvr::string_to_byte_tensor(text));
    }

 private:
    crt::ModelSchema            schema_;
    std::vector<crt::PhaseSpec> phases_;
};

int main() {
    // The zoo half, exactly as in the tutorial: snapshot, registry, model.
    mv::hub::SnapshotOptions snap;
    snap.cache_dir = "models";
    const mv::hub::SnapshotResult snapped = mv::hub::snapshot(
        "meta-llama/Llama-3.2-1B-Instruct", snap);
    mv::LoadOptions load;
    load.max_seq = 4096;  // the CLI's default context cap, as in part 4
    mv::GenerativeModel model =
        mv::ModelRegistry::builtin().load_generative(snapped.local_dir, load);

    mv::generation::GenerationConfig gen = model.defaults();
    gen.max_new_tokens = 24;
    gen.stop = {"."};  // raw completion: one sentence is the demo

    // The zoo's three serving nodes, constructed directly.
    mvr::TokenizerNode tok(model.tokenizer(), model.device(),
                           /*add_special_tokens=*/false);
    mvr::GenerativeDecoderModel dec(model.provider(), gen, /*max_active=*/1,
                                    mv::KVCacheMode::Continuous,
                                    &model.tokenizer(),
                                    /*prefill_chunk_tokens=*/0);
    mvr::DetokenizerNode detok(model.tokenizer());
    ShoutNode            shout;  // yours

    // One pipeline over all four. Edges wire by name; the one rename (the
    // detokenizer's `text` becomes `draft_text`) routes the reply through
    // the custom node instead of straight out.
    std::vector<crt::PipelineStep> steps(4);
    steps[0].name = "tokenize";
    steps[0].model = &tok;
    steps[1].name = "decode";
    steps[1].model = &dec;
    steps[2].name = "detokenize";
    steps[2].model = &detok;
    steps[2].output_map.emplace_back(mvr::kText, kDraftText);
    steps[3].name = "shout";
    steps[3].model = &shout;

    // The pipeline's public request surface: what callers set and read.
    crt::ModelSchema ext;
    ext.inputs.push_back(ClikaRT::spec::TensorSpec{
        mvr::kPrompt, DataType::UInt8, {ClikaRT::spec::TensorSpec::kDynamicDim},
        /*optional=*/false});
    ext.outputs.push_back(ClikaRT::spec::TensorSpec{
        mvr::kText, DataType::UInt8, {ClikaRT::spec::TensorSpec::kDynamicDim},
        /*optional=*/false});
    crt::Pipeline pipeline = crt::Pipeline::create(std::move(steps), std::move(ext));

    // The same Request/Response surface every pipeline speaks.
    crt::Request req;
    req.inputs.set(mvr::kPrompt, mvr::string_to_byte_tensor(
        "Once upon a time, in a port town by a cold sea,"));
    const crt::SessionId sid  = pipeline.enqueue(std::move(req));
    const crt::Response  resp = pipeline.await(sid);

    std::printf("%s\n", mvr::byte_tensor_to_string(resp.outputs.get(mvr::kText)).c_str());
    pipeline.shutdown();
    return 0;
}

 THERE LIVED A YOUNG GIRL NAMED KAITO.

One build flag matters here: the custom node derives from the serving runtime's Model, so its translation unit compiles with -fno-rtti, matching how the library builds; without it the link fails on typeinfo for ClikaRT::runtime::Model. The ClikaRT integration guide covers the flag and which consumers need it.

What the composition rests on

Edges wire by tensor name. Each step's outputs feed the next step's same-named inputs; output_map renames one edge where names must differ. The single rename above is the whole routing change: without it the detokenizer's text would leave the pipeline directly.
The external schema is the caller's contract. Only what it declares is settable and readable from outside; everything between the nodes stays internal. Requests and responses are exactly the ones part 4 used, so a caller cannot tell a customized pipeline from a stock one.
Placement in the chain is yours. A node before the tokenizer transforms the prompt (templating, redaction); a node after the detokenizer transforms the reply, as here; a fully custom model runs beside a zoo model in the same pipeline.

The 02_custom_node program in Additional examples is this guide's offline twin: it synthesizes a toy checkpoint at run time, so the composition runs in moments with no download and no arguments. When HTTP is the goal, mount an engine on the built-in server instead (Serve an OpenAI-compatible endpoint); a pipeline node changes what a model computes, an engine changes what the server serves. And when the model itself is yours rather than the catalog's, that is the other half of the custom track: the Adding your own model tutorial series, with Register your own model family as its full contract.

The program​

What the composition rests on​

The program

What the composition rests on