diff --git a/app/server/http.cpp b/app/server/http.cpp index a6f6d5c19..89155f679 100644 --- a/app/server/http.cpp +++ b/app/server/http.cpp @@ -164,6 +164,21 @@ void send_all(SocketHandle socket, const std::string & data) { } } +// Streaming replies interleave small SSE events with the client's ongoing +// upload. With Nagle enabled each small event waits for the peer's ACK of the +// previous segment (delayed-ACK stretches that to hundreds of milliseconds), +// so partials queue up behind the upload instead of arriving while the user +// speaks. Disable Nagle on accepted sockets. +void set_no_delay(SocketHandle socket) { +#ifdef _WIN32 + constexpr int value = 1; + setsockopt(socket, IPPROTO_TCP, TCP_NODELAY, reinterpret_cast(&value), sizeof(value)); +#else + constexpr int value = 1; + setsockopt(socket, IPPROTO_TCP, TCP_NODELAY, &value, sizeof(value)); +#endif +} + // An SSE body is written while a model lock is held, so an unbounded blocking // send() lets a client that uploads but never reads fill the kernel send buffer // and pin that model indefinitely. A send timeout turns it into a failed write. @@ -787,6 +802,7 @@ void handle_client(SocketHandle client, IHttpHandler & handler, uint64_t max_req // stops reading can pin a model. Applying it server-wide would risk // truncating a large ordinary response to a merely slow client. set_send_timeout(socket.get(), limits.send_timeout_ms); + set_no_delay(socket.get()); } // Constructed unconditionally so it outlives the handler call, but only // published on `request` when the client actually declared a chunked body. diff --git a/app/server/runtime.cpp b/app/server/runtime.cpp index 9aea40042..53e921256 100644 --- a/app/server/runtime.cpp +++ b/app/server/runtime.cpp @@ -2992,6 +2992,23 @@ HttpResponse ServerState::handle_transcription_live(const HttpRequest & request) task_request.options["language"] = language; task_request.text_input = engine::runtime::Transcript{std::string(), language}; } + // Streaming latency knobs. lookahead_tokens picks the model's right-context + // (chunk duration = (lookahead + 1) x 80 ms on nemotron; the session validates + // the value against the model's supported set). stream_chunk_ms bounds how much + // audio the ingest layer assembles before handing it to the session — the + // policy default batches a full second, which delays every partial by that + // much regardless of the model window. + // lookahead 0 (80 ms chunks) is rejected even though the GGUF declares it + // supported: a single-frame first window is not covered by the reference + // chunking validation and decodes garbage — fail loudly instead. + if (!query_param(request.query, "lookahead_tokens").empty()) { + const int64_t lookahead = parse_bounded_int("lookahead_tokens", 3, 0, 13); + task_request.options["lookahead_tokens"] = std::to_string(lookahead); + } + if (!query_param(request.query, "stream_chunk_ms").empty()) { + task_request.options["stream_chunk_ms"] = std::to_string( + parse_bounded_int("stream_chunk_ms", 1000, 10, 4000)); + } task_request = apply_default_request_options(model, std::move(task_request)); } catch (const std::runtime_error & ex) { // Deliberately runtime_error and not exception: every rejection above is diff --git a/app/streaming/streaming.cpp b/app/streaming/streaming.cpp index 55bedf3ac..87097a5eb 100644 --- a/app/streaming/streaming.cpp +++ b/app/streaming/streaming.cpp @@ -1,9 +1,12 @@ #include "streaming.h" +#include "engine/framework/runtime/options.h" + #include #include #include #include +#include #include #include @@ -37,6 +40,31 @@ int64_t resolve_chunk_samples( return policy.preferred_audio_chunk_samples; } +// Request-level override for the ingest batching granularity ("stream_chunk_ms", +// milliseconds of audio per process_audio_chunk call). The session policy's +// preferred chunk is sized for throughput (nemotron batches a full second), which +// delays every partial by that much; low-latency clients ask for smaller reads so +// the model window fills as soon as enough bytes have arrived. Ignored when the +// option is absent or malformed beyond recognition — a numeric parse failure is +// the caller's bug and throws. +int64_t chunk_samples_for_request( + const engine::runtime::TaskRequest & request, + const engine::runtime::StreamingPolicy & policy, + const AudioStreamFormat & format) { + const auto override_ms = engine::runtime::find_option(request.options, {"stream_chunk_ms"}); + if (!override_ms || override_ms->empty()) { + return resolve_chunk_samples(policy, format); + } + size_t consumed = 0; + const long long ms = std::stoll(*override_ms, &consumed); + if (consumed != override_ms->size() || ms <= 0) { + throw std::runtime_error("stream_chunk_ms must be a positive integer (milliseconds)"); + } + const auto samples = static_cast(std::llround( + static_cast(ms) * static_cast(format.sample_rate) / 1000.0)); + return std::max(samples, 1) * static_cast(format.channels); +} + void feed_audio_stream( engine::runtime::IStreamingVoiceTaskSession & session, const AudioChunkStream & stream, @@ -128,14 +156,15 @@ engine::runtime::TaskResult run_stream( "streaming audio input mode requires samples in audio_input, or an " "AudioChunkStream for a live source"); } - feed_audio_stream(session, *stream, resolve_chunk_samples(policy, stream->format), sink); + feed_audio_stream( + session, *stream, chunk_samples_for_request(request, policy, stream->format), sink); } - if (policy.output == engine::runtime::StreamingOutputKind::PullEvents) { - pull_stream_events(session, sink); - } - auto result = session.finish_stream(); - session.set_stream_event_sink(nullptr); - return result; + if (policy.output == engine::runtime::StreamingOutputKind::PullEvents) { + pull_stream_events(session, sink); + } + auto result = session.finish_stream(); + session.set_stream_event_sink(nullptr); + return result; } catch (...) { session.set_stream_event_sink(nullptr); throw; diff --git a/docs/nemotron_streaming_end_of_turn.md b/docs/nemotron_streaming_end_of_turn.md new file mode 100644 index 000000000..4a059acbc --- /dev/null +++ b/docs/nemotron_streaming_end_of_turn.md @@ -0,0 +1,73 @@ +# Nemotron streaming: end-of-turn latency — verified analysis (reference-validated) + +All findings below were established against the official reference +implementation (transformers `nemotron3_5_asr`, f32, NVIDIA-contributed) +running the same audio through its own encoder, its own decode loop, and +its own official `generate()` streaming path. + +## The end-of-turn latency budget (x2 CPU, per phrase) + +EOT->final = flush encode (~100-175 ms) +, for pocketed utterances, the +padded-offline fallback (~250-330 ms). The flush encode is the encoder +running one 64-mel window; it is exact vs the reference (cos 0.999, +frame-for-frame, verified on multiple shapes). + +## Trailing-token mechanism (verified) + +The RNNT fires a word's trailing tokens only after a per-utterance amount +of silence follows the word in the encoded stream. The silence CONTENT is +irrelevant (digital zeros = room tone = dither; verified). What matters is +duration, and it interacts with the cut position: + +- hello.wav cut at 0.65 s: decodes in streaming at the 64-mel window. +- hi.wav cut at 0.67 s: decodes in streaming ('Hi. '). +- hi.wav cut at 0.57 s: does NOT decode in streaming AT ALL. + +## The hi.wav 0.57 s cut: streaming-mode failure, not a bug + +Exhaustive shape matrix against the reference, same audio (hi.wav[:9120]), +la0, prompt aligned: + + (8,32) 5 f -> '' + (8,32,32) 9 f -> '' + (8,32,64) 13 f -> '' + (8,32,96) 17 f -> '' + (8,32,32,32) 13 f -> '' + (8,32,32,32,32) 17 f -> '' + (8,32x7) 25 f -> '' + +Every streaming shape fails, including 1.6 s of trailing synthetic silence. +The reference's own official streaming `generate()` also returns ''. +The OFFLINE encoder on the same audio + >=0.5 s of padding decodes 'Hi. '. + +Conclusion: the streaming causal-conv frame grid fails this utterance at +this cut regardless of windowing; the offline grid succeeds. audio.cpp's +padded-offline fallback is the only correct mechanism and matches the +model's own semantics. It is not a workaround to remove - it is the fix. + +## Dead ends (do not retry) + +- Longer single flush windows (64/72/96 mel): the 0.57 s hi cut fails at + every length; longer windows only add encode time. +- Iterative flush chunks: same frames as the long window, same failure. +- Silence dither instead of zero padding: identical results. +- Per-sample-max energy gates for the speculative trigger: decay + transients count as speech; use 10 ms windowed RMS. + +## Remaining latency levers (ranked) + +1. Speculative offline fallback: when the speculative flush's decode comes + back blank (decode is cheap, ~10 ms), run the padded-offline encode + during the tail wait too. Requires snapshot/restore of the decoder RNNT + state alongside the encoder state. Saves the fallback's ~250-330 ms from + the EOT->final path for pocketed utterances. +2. q4_K GGUF regeneration: the encoder is weight-bandwidth-bound; halves + the per-chunk cost on CPU. +3. Vulkan: the flush hides entirely; EOT->final 60-90 ms already. + +## Verified harness artifacts + +- bench_tmp/ref_nemotron_stream.py: reference runner (offline/official + streaming/manual chunked with dumps). +- bench_tmp/dump/ref96/*: reference frames for the shape matrix above. +- NEMOTRON_DUMP_CHUNKS / NEMOTRON_DUMP_CACHES: audio.cpp-side dumps. diff --git a/include/engine/community_models/granite5asr/encoder.h b/include/engine/community_models/granite5asr/encoder.h index 141639a64..e19dfa8d9 100644 --- a/include/engine/community_models/granite5asr/encoder.h +++ b/include/engine/community_models/granite5asr/encoder.h @@ -55,6 +55,7 @@ class Granite5EncoderRuntime { engine::core::ExecutionContext & execution_context, assets::TensorStorageType storage_type, size_t graph_arena_bytes = 1024ull * 1024ull * 1024ull); + ~Granite5EncoderRuntime(); std::vector transcribe_features( const Granite5FrontendFeatures & features); @@ -62,11 +63,19 @@ class Granite5EncoderRuntime { const Granite5ASRAssets & assets() const noexcept { return *assets_; } private: + struct GraphCacheEntry; + struct GraphCache; + GraphCacheEntry & ensure_graph_entry(int64_t input_frames, int64_t feature_dim); + std::shared_ptr assets_; engine::core::ExecutionContext * execution_context_ = nullptr; engine::core::BackendWeightStore weight_store_; Granite5EncoderWeights weights_; size_t graph_arena_bytes_; + // Shape-keyed encoder graphs: building the 16-block conformer graph costs + // real milliseconds and chunked streaming re-decodes the SAME window shape + // every chunk, so graphs are cached per input length (small LRU). + std::unique_ptr graph_cache_; }; } // namespace engine::community_models::granite5asr diff --git a/include/engine/community_models/granite5asr/session.h b/include/engine/community_models/granite5asr/session.h index cca058ad5..a27904f30 100644 --- a/include/engine/community_models/granite5asr/session.h +++ b/include/engine/community_models/granite5asr/session.h @@ -82,6 +82,18 @@ class Granite5ASRStreamingSession final runtime::StreamEventCallback stream_event_sink_; runtime::AudioBuffer streaming_audio_; runtime::TaskRequest streaming_request_; + + // Windowed chunked-streaming state (the publisher's chunked TurboCTC recipe): + // every center chunk is (re-)encoded together with its left-context window and + // CTC-greedy decoded continuously, so partials stream as chunks land. + bool decode_next_center_window(bool flush_tail, std::string & delta_out); + int64_t stream_center_samples_ = 0; + int64_t stream_left_context_samples_ = 0; + int64_t stream_next_center_start_ = 0; + int32_t stream_last_raw_token_ = -1; + std::vector stream_collapsed_ids_; + std::string stream_emitted_text_; + bool stream_finalized_ = false; }; class Granite5ASRLoadedModel final : public runtime::ILoadedVoiceModel { diff --git a/include/engine/models/nemotron_asr/decoder.h b/include/engine/models/nemotron_asr/decoder.h index eee1e64fa..06e22164f 100644 --- a/include/engine/models/nemotron_asr/decoder.h +++ b/include/engine/models/nemotron_asr/decoder.h @@ -45,6 +45,18 @@ class NemotronDecoderRuntime { const std::function & next_chunk, const NemotronTextDeltaCallback & on_text_delta = nullptr); + // Incremental chunk-at-a-time decoding: begin resets the prediction-network + // state, decode_stream_chunk consumes the frames of one encoded chunk while + // continuing that state (firing on_text_delta for newly emitted text), and + // finish finalizes the transcript and word timestamps. decode_streaming() is + // implemented on top of these three; sessions that receive audio live (SSE + // /live, WebSocket, ...) call them directly so partials stream per chunk. + void begin_stream_decode(const NemotronDecodeOptions & options); + void decode_stream_chunk( + const NemotronEncodedAudio & chunk, + const NemotronTextDeltaCallback & on_text_delta = nullptr); + NemotronDecodedText finish_stream_decode(); + private: struct Graph; struct JointGraph; @@ -68,6 +80,17 @@ class NemotronDecoderRuntime { std::vector logits_scratch_; std::vector hidden_read_scratch_; std::vector cell_read_scratch_; + + // Persistent state across decode_stream_chunk() calls. + bool stream_decode_active_ = false; + NemotronDecodeOptions stream_decode_options_; + int64_t stream_frame_index_ = 0; + int64_t stream_symbols_at_frame_ = 0; + int32_t stream_input_token_ = 0; + bool stream_decoder_cache_initialized_ = false; + std::vector stream_token_ids_; + std::vector stream_durations_; + std::string stream_emitted_text_; }; } // namespace engine::models::nemotron_asr diff --git a/include/engine/models/nemotron_asr/encoder.h b/include/engine/models/nemotron_asr/encoder.h index 9315dc0a7..129f756c1 100644 --- a/include/engine/models/nemotron_asr/encoder.h +++ b/include/engine/models/nemotron_asr/encoder.h @@ -27,6 +27,7 @@ struct NemotronEncoderStreamState { bool first_chunk = true; bool backend_cache_valid = false; const void * backend_cache_owner = nullptr; + int64_t dump_seq = 0; }; class NemotronEncoderRuntime { @@ -35,7 +36,8 @@ class NemotronEncoderRuntime { std::shared_ptr assets, std::shared_ptr weights, engine::core::ExecutionContext & execution_context, - size_t graph_arena_bytes); + size_t graph_arena_bytes, + size_t stream_graph_arena_bytes = 0); ~NemotronEncoderRuntime(); void prepare_capacity(int64_t input_frames, int64_t feature_dim, int64_t lookahead_tokens); @@ -65,6 +67,7 @@ class NemotronEncoderRuntime { std::shared_ptr weights_; engine::core::ExecutionContext * execution_context_ = nullptr; size_t graph_arena_bytes_ = 0; + size_t stream_graph_arena_bytes_ = 0; std::unique_ptr graph_; std::vector> stream_graphs_; std::vector input_scratch_; @@ -72,6 +75,8 @@ class NemotronEncoderRuntime { std::vector prompt_scratch_; std::vector mask_scratch_; std::vector attention_mask_scratch_; + std::vector attention_key_scratch_; + std::vector attention_value_scratch_; std::unordered_map> relative_positional_encoding_cache_; }; diff --git a/include/engine/models/nemotron_asr/session.h b/include/engine/models/nemotron_asr/session.h index 1c062f1e8..67225845e 100644 --- a/include/engine/models/nemotron_asr/session.h +++ b/include/engine/models/nemotron_asr/session.h @@ -93,6 +93,39 @@ class NemotronASRStreamingSession final private: runtime::StreamEventCallback stream_event_sink_; + + // Native chunked-streaming pipeline state. Audio is encoded window-by-window + // through encode_stream_chunk (cache-aware, publisher chunk contract) and the + // decoder runs incrementally across chunks, so partials stream as chunks land + // instead of only at finalize. + bool encode_and_decode_next_chunk(bool flush_tail, std::string & delta_out); + // Speculative flush support (see the member docs below). + int64_t flush_window_mel() const; + void build_flush_window(int64_t total, std::vector & window) const; + void maybe_speculative_flush(); + void discard_speculative_flush(); + int64_t stream_lookahead_ = 0; + int64_t stream_prompt_id_ = 0; + NemotronDecodeOptions stream_decode_options_; + NemotronEncoderStreamState encoder_stream_state_; + int64_t stream_first_samples_ = 0; + int64_t stream_samples_per_chunk_ = 0; + int64_t stream_first_mel_frames_ = 0; + int64_t stream_mel_frames_per_chunk_ = 0; + int64_t stream_next_chunk_start_ = 0; + bool stream_await_first_chunk_ = true; + bool stream_tail_encoded_ = false; + bool stream_decode_active_ = false; + int64_t stream_dump_chunk_seq_ = 0; + // Speculative flush: while the turn is still open, the flush window is + // encoded as soon as the ingest sees enough trailing silence, and finalize + // reuses the result unless speech followed (then the state snapshot is + // restored and the normal flush runs). See session.cpp for the knobs. + bool spec_valid_ = false; + NemotronEncodedAudio spec_frames_; + NemotronEncoderStreamState spec_state_snapshot_{}; + int64_t spec_last_loud_sample_ = 0; + int64_t spec_audio_mark_ = 0; }; } // namespace engine::models::nemotron_asr diff --git a/model_specs/granite5asr.json b/model_specs/granite5asr.json index f576999fb..76f562432 100644 --- a/model_specs/granite5asr.json +++ b/model_specs/granite5asr.json @@ -73,6 +73,22 @@ "description": "Path to the Silero VAD model directory used by automatic audio chunking.", "required": false, "default": "assets/framework/models/silero_vad" + }, + { + "name": "center_chunk_sec", + "type": "float", + "description": "Center-chunk duration in seconds for buffered streaming; each chunk is decoded as soon as it is complete, which sets the partial cadence.", + "required": false, + "min": 0.05, + "default": 1.0 + }, + { + "name": "left_context_sec", + "type": "float", + "description": "Left-context audio re-encoded with every center chunk to give the block-attention CTC model its past context.", + "required": false, + "min": 0.0, + "default": 2.0 } ], "load": [] @@ -176,4 +192,4 @@ } } ] -} +} \ No newline at end of file diff --git a/src/community_models/granite5asr/encoder.cpp b/src/community_models/granite5asr/encoder.cpp index 7fd90adb3..f3d6b1298 100644 --- a/src/community_models/granite5asr/encoder.cpp +++ b/src/community_models/granite5asr/encoder.cpp @@ -16,6 +16,7 @@ #include #include +#include #include #include #include @@ -26,6 +27,8 @@ namespace engine::community_models::granite5asr { namespace { +using Clock = std::chrono::steady_clock; + constexpr size_t kEncoderGraphNodes = 1048576; constexpr float kLayerNormEpsilon = 1.0e-5f; @@ -438,52 +441,93 @@ Granite5EncoderRuntime::Granite5EncoderRuntime( execution_context.backend_type(), "Granite 5 ASR encoder weights", 256ull * 1024ull * 1024ull), - graph_arena_bytes_(graph_arena_bytes) { + graph_arena_bytes_(graph_arena_bytes), + graph_cache_(std::make_unique()) { if (assets_ == nullptr) { throw std::runtime_error("Granite 5 ASR encoder runtime requires assets"); } weights_ = load_encoder_weights(weight_store_, *assets_->source, assets_->config, storage_type); } -std::vector Granite5EncoderRuntime::transcribe_features( - const Granite5FrontendFeatures & features) { - if (features.frames <= 0 || features.values.empty()) { - return {}; +Granite5EncoderRuntime::~Granite5EncoderRuntime() = default; + +struct Granite5EncoderRuntime::GraphCacheEntry { + ggml_context * ggml_ctx = nullptr; + ggml_gallocr * allocator = nullptr; + ggml_cgraph * graph = nullptr; + core::TensorValue input; + core::TensorValue logits; + int64_t input_frames = 0; + int64_t feature_dim = 0; +}; + +struct Granite5EncoderRuntime::GraphCache { + // LRU-front list of shape-keyed graphs. Chunked streaming re-decodes the SAME + // window shape every chunk and repeated offline turns cluster around a few + // lengths, so a small cache eliminates nearly all graph-build latency. + std::vector entries; + static constexpr size_t kMaxEntries = 6; + + ~GraphCache() { + for (auto & entry : entries) { + if (entry.allocator != nullptr) { + ggml_gallocr_free(entry.allocator); + } + if (entry.ggml_ctx != nullptr) { + ggml_free(entry.ggml_ctx); + } + } + } +}; + +Granite5EncoderRuntime::GraphCacheEntry & Granite5EncoderRuntime::ensure_graph_entry( + int64_t input_frames, + int64_t feature_dim) { + static const bool kBenchNoCache = std::getenv("GRANITE_NO_CACHE") != nullptr; + if (!kBenchNoCache) for (size_t i = 0; i < graph_cache_->entries.size(); ++i) { + if (graph_cache_->entries[i].input_frames == input_frames && + graph_cache_->entries[i].feature_dim == feature_dim) { + if (i != 0) { + auto entry = std::move(graph_cache_->entries[i]); + graph_cache_->entries.erase(graph_cache_->entries.begin() + static_cast(i)); + graph_cache_->entries.insert(graph_cache_->entries.begin(), std::move(entry)); + } + debug::timing_log_scalar("granite5asr.encoder.graph_build_ms", 0.0); + return graph_cache_->entries.front(); + } } - const auto & config = assets_->config; - const int64_t num_frames = features.frames; - const int64_t feat_dim = features.feature_dim; + const auto build_start = Clock::now(); + GraphCacheEntry entry; + entry.input_frames = input_frames; + entry.feature_dim = feature_dim; ggml_init_params params{}; params.mem_size = graph_arena_bytes_; params.mem_buffer = nullptr; params.no_alloc = true; - - ggml_context * ggml_ctx = ggml_init(params); - if (!ggml_ctx) { + entry.ggml_ctx = ggml_init(params); + if (entry.ggml_ctx == nullptr) { throw std::runtime_error("Failed to initialize GGML context for Granite 5 ASR encoder"); } - - ggml_gallocr * galloc = ggml_gallocr_new( + entry.allocator = ggml_gallocr_new( ggml_backend_get_default_buffer_type(execution_context_->backend())); - if (!galloc) { - ggml_free(ggml_ctx); + if (entry.allocator == nullptr) { + ggml_free(entry.ggml_ctx); throw std::runtime_error("Failed to initialize GGML allocator for Granite 5 ASR encoder"); } - std::vector token_ids; - + const auto & config = assets_->config; try { - core::ModuleBuildContext ctx{ggml_ctx, "granite5asr_encoder", execution_context_->backend_type()}; + core::ModuleBuildContext ctx{entry.ggml_ctx, "granite5asr_encoder", execution_context_->backend_type()}; - auto in_tensor = core::wrap_tensor( - ggml_new_tensor_2d(ggml_ctx, GGML_TYPE_F32, feat_dim, num_frames), - core::TensorShape::from_dims({1, num_frames, feat_dim}), + entry.input = core::wrap_tensor( + ggml_new_tensor_2d(entry.ggml_ctx, GGML_TYPE_F32, feature_dim, input_frames), + core::TensorShape::from_dims({1, input_frames, feature_dim}), GGML_TYPE_F32); - auto h = modules::LinearModule({feat_dim, config.encoder.hidden_size, true}) - .build(ctx, in_tensor, weights_.input_linear); + auto h = modules::LinearModule({feature_dim, config.encoder.hidden_size, true}) + .build(ctx, entry.input, weights_.input_linear); const int64_t mid_layer_idx = config.encoder.num_layers / 2; // 8 for (int64_t idx = 0; idx < config.encoder.num_layers; ++idx) { @@ -505,56 +549,91 @@ std::vector Granite5EncoderRuntime::transcribe_features( } } - auto logits = modules::LinearModule({config.encoder.hidden_size, config.vocab_size, true}) - .build(ctx, h, weights_.out); + entry.logits = modules::LinearModule({config.encoder.hidden_size, config.vocab_size, true}) + .build(ctx, h, weights_.out); - ggml_cgraph * gf = ggml_new_graph_custom(ggml_ctx, kEncoderGraphNodes, false); - ggml_build_forward_expand(gf, logits.tensor); + entry.graph = ggml_new_graph_custom(entry.ggml_ctx, kEncoderGraphNodes, false); + ggml_build_forward_expand(entry.graph, entry.logits.tensor); + } catch (...) { + if (entry.allocator != nullptr) { + ggml_gallocr_free(entry.allocator); + } + ggml_free(entry.ggml_ctx); + throw; + } - if (!ggml_gallocr_alloc_graph(galloc, gf)) { - throw std::runtime_error("Failed to allocate GGML graph for Granite 5 ASR encoder"); + graph_cache_->entries.insert(graph_cache_->entries.begin(), std::move(entry)); + if (graph_cache_->entries.size() > GraphCache::kMaxEntries) { + auto & oldest = graph_cache_->entries.back(); + if (oldest.allocator != nullptr) { + ggml_gallocr_free(oldest.allocator); } + ggml_free(oldest.ggml_ctx); + graph_cache_->entries.pop_back(); + } + debug::timing_log_scalar( + "granite5asr.encoder.graph_build_ms", + engine::debug::elapsed_ms(build_start, Clock::now())); + return graph_cache_->entries.front(); +} - ggml_backend_tensor_set( - in_tensor.tensor, - features.values.data(), - 0, - features.values.size() * sizeof(float)); +std::vector Granite5EncoderRuntime::transcribe_features( + const Granite5FrontendFeatures & features) { + if (features.frames <= 0 || features.values.empty()) { + return {}; + } - if (ggml_backend_graph_compute(execution_context_->backend(), gf) != GGML_STATUS_SUCCESS) { - throw std::runtime_error("Failed to compute GGML graph for Granite 5 ASR encoder"); - } + const auto & config = assets_->config; + const auto wall_start = Clock::now(); + auto & entry = ensure_graph_entry(features.frames, features.feature_dim); - const int64_t out_frames = logits.shape.dims[1]; - const int64_t vocab_size = config.vocab_size; - std::vector logits_data(static_cast(out_frames * vocab_size)); - ggml_backend_tensor_get( - logits.tensor, - logits_data.data(), - 0, - logits_data.size() * sizeof(float)); - - token_ids.reserve(static_cast(out_frames)); - for (int64_t t = 0; t < out_frames; ++t) { - const float * frame_logits = &logits_data[static_cast(t * vocab_size)]; - int32_t best_id = 0; - float max_val = frame_logits[0]; - for (int32_t v = 1; v < static_cast(vocab_size); ++v) { - if (frame_logits[v] > max_val) { - max_val = frame_logits[v]; - best_id = v; - } + std::vector token_ids; + + if (!ggml_gallocr_alloc_graph(entry.allocator, entry.graph)) { + throw std::runtime_error("Failed to allocate GGML graph for Granite 5 ASR encoder"); + } + + ggml_backend_tensor_set( + entry.input.tensor, + features.values.data(), + 0, + features.values.size() * sizeof(float)); + + if (ggml_backend_graph_compute(execution_context_->backend(), entry.graph) != GGML_STATUS_SUCCESS) { + throw std::runtime_error("Failed to compute GGML graph for Granite 5 ASR encoder"); + } + + const int64_t out_frames = entry.logits.tensor->ne[1]; + const int64_t vocab_size = config.vocab_size; + std::vector logits_data(static_cast(out_frames * vocab_size)); + ggml_backend_tensor_get( + entry.logits.tensor, + logits_data.data(), + 0, + logits_data.size() * sizeof(float)); + + const auto argmax_start = Clock::now(); + token_ids.reserve(static_cast(out_frames)); + for (int64_t t = 0; t < out_frames; ++t) { + const float * frame_logits = &logits_data[static_cast(t * vocab_size)]; + int32_t best_id = 0; + float max_val = frame_logits[0]; + for (int32_t v = 1; v < static_cast(vocab_size); ++v) { + if (frame_logits[v] > max_val) { + max_val = frame_logits[v]; + best_id = v; } - token_ids.push_back(best_id); } - } catch (...) { - ggml_gallocr_free(galloc); - ggml_free(ggml_ctx); - throw; + token_ids.push_back(best_id); } - - ggml_gallocr_free(galloc); - ggml_free(ggml_ctx); + const double argmax_ms = engine::debug::elapsed_ms(argmax_start, Clock::now()); + + debug::timing_log_scalar( + "granite5asr.encoder.compute_ms", + engine::debug::elapsed_ms(wall_start, Clock::now()) - argmax_ms); + debug::timing_log_scalar( + "granite5asr.encoder_ms", + engine::debug::elapsed_ms(wall_start, Clock::now())); return token_ids; } diff --git a/src/community_models/granite5asr/session.cpp b/src/community_models/granite5asr/session.cpp index b014231c3..91962adf8 100644 --- a/src/community_models/granite5asr/session.cpp +++ b/src/community_models/granite5asr/session.cpp @@ -355,7 +355,36 @@ Granite5ASRStreamingSession::Granite5ASRStreamingSession( runtime::TaskSpec task, runtime::SessionOptions options, std::shared_ptr assets) - : Granite5ASRSessionBase(std::move(task), std::move(options), std::move(assets)) {} + : Granite5ASRSessionBase(std::move(task), std::move(options), std::move(assets)) { + // Chunked-streaming geometry (publisher chunked TurboCTC recipe): every + // center chunk carries its left context through the encoder and is decoded + // immediately, so partials stream per chunk. Defaults tuned for END-OF-TURN + // latency: a 1 s center keeps the final flush window small. Raising + // center_chunk_sec to 2 cuts total encoder compute ~30% at equal accuracy + // (fewer, larger windows) but adds ~20-25% end-of-turn latency at 1 thread; + // left contexts below 2 s duplicate words across window boundaries. + float center_sec = 1.0f; + float left_sec = 2.0f; + const auto & opts = RuntimeSessionBase::options().options; + const std::string family = family_impl(); + auto parse_positive = [](const std::string & value, float fallback) { + try { + const float parsed = std::stof(value); + if (parsed > 0.0f) { + return parsed; + } + } catch (...) {} + return fallback; + }; + if (const auto it = opts.find(family + ".center_chunk_sec"); it != opts.end()) { + center_sec = parse_positive(it->second, center_sec); + } + if (const auto it = opts.find(family + ".left_context_sec"); it != opts.end()) { + left_sec = parse_positive(it->second, left_sec); + } + stream_center_samples_ = static_cast(center_sec * 16000.0f); + stream_left_context_samples_ = static_cast(left_sec * 16000.0f); +} std::string Granite5ASRStreamingSession::family() const { return family_impl(); @@ -377,7 +406,12 @@ runtime::StreamingPolicy Granite5ASRStreamingSession::streaming_policy() const { runtime::StreamingPolicy policy; policy.input = runtime::StreamingInputKind::AudioChunks; policy.output = runtime::StreamingOutputKind::FinalResult; - policy.preferred_audio_chunk_samples = 512; + policy.preferred_audio_chunk_samples = stream_center_samples_ > 0 + ? stream_center_samples_ + : 16000; + policy.preferred_audio_chunk_seconds = stream_center_samples_ > 0 + ? static_cast(stream_center_samples_) / 16000.0 + : 1.0; return policy; } @@ -387,6 +421,11 @@ void Granite5ASRStreamingSession::start_stream(const runtime::TaskRequest & requ streaming_audio_ = runtime::AudioBuffer{}; streaming_audio_.sample_rate = 16000; streaming_audio_.channels = 1; + stream_next_center_start_ = 0; + stream_last_raw_token_ = -1; + stream_collapsed_ids_.clear(); + stream_emitted_text_.clear(); + stream_finalized_ = false; } void Granite5ASRStreamingSession::set_stream_event_sink(runtime::StreamEventCallback sink) { @@ -395,6 +434,83 @@ void Granite5ASRStreamingSession::set_stream_event_sink(runtime::StreamEventCall void Granite5ASRStreamingSession::reset() { streaming_audio_.samples.clear(); + stream_next_center_start_ = 0; + stream_last_raw_token_ = -1; + stream_collapsed_ids_.clear(); + stream_emitted_text_.clear(); + stream_finalized_ = false; +} + +bool Granite5ASRStreamingSession::decode_next_center_window(bool flush_tail, std::string & delta_out) { + if (stream_finalized_) { + return false; + } + const int64_t total = static_cast(streaming_audio_.samples.size()); + const int64_t center_end = std::min( + stream_next_center_start_ + stream_center_samples_, total); + if (center_end <= stream_next_center_start_) { + return false; // nothing new since the last window + } + if (!flush_tail && center_end < stream_next_center_start_ + stream_center_samples_) { + return false; // wait for a full center chunk (low-latency default) + } + + const int64_t window_start = std::max( + 0, stream_next_center_start_ - stream_left_context_samples_); + runtime::AudioBuffer window; + window.sample_rate = 16000; + window.channels = 1; + window.samples.assign( + streaming_audio_.samples.begin() + static_cast(window_start), + streaming_audio_.samples.begin() + static_cast(center_end)); + + const auto features = frontend_.extract_waveform(window.samples); + const auto raw_tokens = encoder_->transcribe_features(features); + if (!raw_tokens.empty()) { + // Map the center region to CTC frames. One CTC frame covers + // hop * stack * encoder-stride samples: the frontend stacks `stack_factor` + // mel frames per feature, and each subsample layer halves the encoder + // sequence (subsample_layers = {0, 1} -> stride 4 total). One frame of + // slack at the start avoids clipping a word onset; the continuous + // collapse below absorbs the boundary repeat. + int64_t encoder_stride = 1; + for (const int64_t layer_idx : assets_->config.encoder.subsample_layers) { + (void)layer_idx; + encoder_stride *= 2; + } + const int64_t frame_samples = assets_->config.frontend.hop_length * + assets_->config.frontend.stack_factor * encoder_stride; + int64_t start_frame = std::max( + 0, (stream_next_center_start_ - window_start) / frame_samples - 1); + const int64_t end_frame = std::min( + static_cast(raw_tokens.size()), + (center_end - window_start + frame_samples - 1) / frame_samples); + for (int64_t f = start_frame; f < end_frame; ++f) { + const int32_t id = raw_tokens[static_cast(f)]; + if (id != stream_last_raw_token_) { + if (id != static_cast(assets_->config.blank_token_id)) { + stream_collapsed_ids_.push_back(id); + } + stream_last_raw_token_ = id; + } + } + } + stream_next_center_start_ = center_end; + + if (!stream_collapsed_ids_.empty() && assets_->tokenizer != nullptr) { + const auto current_text = assets_->tokenizer->decode_ids(stream_collapsed_ids_); + if (current_text.size() > stream_emitted_text_.size() && + current_text.compare(0, stream_emitted_text_.size(), stream_emitted_text_) == 0) { + delta_out += current_text.substr(stream_emitted_text_.size()); + stream_emitted_text_ = current_text; + } else if (current_text != stream_emitted_text_) { + // CTC revised earlier text: re-emit the full transcript (append-style + // clients surface a repeat; the final result stays correct). + delta_out += current_text; + stream_emitted_text_ = current_text; + } + } + return true; } runtime::StreamEvent Granite5ASRStreamingSession::process_audio_chunk(const runtime::AudioChunk & chunk) { @@ -407,14 +523,38 @@ runtime::StreamEvent Granite5ASRStreamingSession::process_audio_chunk(const runt 16000); streaming_audio_.samples.insert(streaming_audio_.samples.end(), mono.begin(), mono.end()); } - return {}; + + runtime::StreamEvent event; + event.is_final = false; + std::string delta; + while (decode_next_center_window(/*flush_tail=*/false, delta)) { + } + if (!delta.empty()) { + event.partial_text = runtime::Transcript{delta, "en"}; + if (stream_event_sink_) { + stream_event_sink_(event); + return {}; + } + } + return event; } runtime::TaskResult Granite5ASRStreamingSession::finish_stream() { require_prepared("Granite 5 ASR finish_stream()"); - const auto transcript = transcribe_audio(streaming_audio_, streaming_request_.options); + std::string delta; + while (decode_next_center_window(/*flush_tail=*/true, delta)) { + } + (void)delta; + runtime::TaskResult result; - result.text_output = transcript; + if (stream_collapsed_ids_.empty() || assets_->tokenizer == nullptr) { + result.text_output = runtime::Transcript{"", "en"}; + } else { + result.text_output = runtime::Transcript{ + engine::io::trim_ascii_whitespace(assets_->tokenizer->decode_ids(stream_collapsed_ids_)), + "en"}; + } + stream_finalized_ = true; return result; } diff --git a/src/models/nemotron_asr/decoder.cpp b/src/models/nemotron_asr/decoder.cpp index 7fdc13a46..0fe97f4a3 100644 --- a/src/models/nemotron_asr/decoder.cpp +++ b/src/models/nemotron_asr/decoder.cpp @@ -461,106 +461,118 @@ NemotronDecodedText NemotronDecoderRuntime::decode( return out; } -NemotronDecodedText NemotronDecoderRuntime::decode_streaming( - const NemotronDecodeOptions & options, - const std::function & next_chunk, - const NemotronTextDeltaCallback & on_text_delta) { - if (!next_chunk) { - throw std::runtime_error("Nemotron ASR streaming decoder requires a chunk producer"); - } - const auto wall_start = Clock::now(); +void NemotronDecoderRuntime::begin_stream_decode(const NemotronDecodeOptions & options) { ensure_graph(); engine::core::set_backend_threads(execution_context_->backend(), execution_context_->config().threads); const auto & config = assets_->config; - const int64_t max_tokens = options.max_tokens > 0 - ? options.max_tokens - : (std::numeric_limits::max() / 4); + stream_decode_options_ = options; hidden_scratch_.assign(static_cast(config.decoder_layers * config.decoder_hidden_size), 0.0f); cell_scratch_.assign(static_cast(config.decoder_layers * config.decoder_hidden_size), 0.0f); decoder_cache_scratch_.assign(static_cast(config.decoder_hidden_size), 0.0f); + stream_frame_index_ = 0; + stream_symbols_at_frame_ = 0; + stream_input_token_ = static_cast(config.blank_token_id); + stream_decoder_cache_initialized_ = false; + stream_token_ids_.clear(); + stream_durations_.clear(); + stream_token_ids_.push_back(static_cast(config.blank_token_id)); + stream_durations_.push_back(0); + stream_emitted_text_.clear(); + stream_decode_active_ = true; +} - std::vector encoded_values; - int64_t encoded_valid_frames = 0; - int64_t encoded_hidden_size = 0; - bool stream_exhausted = false; - auto append_next_chunk = [&]() -> bool { - NemotronEncodedAudio chunk; - if (!next_chunk(chunk)) { - stream_exhausted = true; - return false; - } - if (chunk.valid_frames <= 0 || chunk.hidden_size != config.decoder_hidden_size) { - throw std::runtime_error("Nemotron ASR streaming decoder received invalid encoded chunk"); - } - if (encoded_hidden_size == 0) { - encoded_hidden_size = chunk.hidden_size; - } else if (encoded_hidden_size != chunk.hidden_size) { - throw std::runtime_error("Nemotron ASR streaming decoder chunk hidden size mismatch"); - } - encoded_values.insert( - encoded_values.end(), - chunk.values.begin(), - chunk.values.begin() + static_cast(chunk.valid_frames * chunk.hidden_size)); - encoded_valid_frames += chunk.valid_frames; - return true; - }; - if (!append_next_chunk()) { - throw std::runtime_error("Nemotron ASR streaming decoder received no encoded chunks"); +void NemotronDecoderRuntime::decode_stream_chunk( + const NemotronEncodedAudio & chunk, + const NemotronTextDeltaCallback & on_text_delta) { + if (!stream_decode_active_) { + throw std::runtime_error("Nemotron ASR stream decode requires begin_stream_decode()"); } - - NemotronDecodedText out; - out.token_ids.reserve(4096); - out.durations.reserve(out.token_ids.capacity()); - out.token_ids.push_back(static_cast(config.blank_token_id)); - out.durations.push_back(0); - std::string emitted_text; - - int64_t frame_index = 0; - int64_t symbols_at_frame = 0; - int32_t input_token = static_cast(config.blank_token_id); - bool decoder_cache_initialized = false; - while (static_cast(out.token_ids.size()) - 1 < max_tokens) { - while (frame_index >= encoded_valid_frames && !stream_exhausted) { - append_next_chunk(); - } - if (frame_index >= encoded_valid_frames) { - break; - } - - const float * frame = encoded_values.data() + static_cast(frame_index * encoded_hidden_size); - const int32_t token = run_step(input_token, frame, decoder_cache_initialized); - decoder_cache_initialized = true; - out.token_ids.push_back(token); + if (chunk.valid_frames <= 0 || chunk.hidden_size != assets_->config.decoder_hidden_size) { + throw std::runtime_error("Nemotron ASR streaming decoder received invalid encoded chunk"); + } + const auto & config = assets_->config; + const int64_t max_tokens = stream_decode_options_.max_tokens > 0 + ? stream_decode_options_.max_tokens + : (std::numeric_limits::max() / 4); + // One encoder frame can emit up to max_symbols_per_step tokens before the loop + // advances (blank or symbol cap) — the frame pointer must NOT move per token. + int64_t local_frame = 0; + while (local_frame < chunk.valid_frames && + static_cast(stream_token_ids_.size()) - 1 < max_tokens) { + const float * frame = chunk.values.data() + static_cast(local_frame * chunk.hidden_size); + const int32_t token = run_step(stream_input_token_, frame, stream_decoder_cache_initialized_); + stream_decoder_cache_initialized_ = true; + stream_token_ids_.push_back(token); const bool blank = token == static_cast(config.blank_token_id); if (!blank) { - ++symbols_at_frame; + ++stream_symbols_at_frame_; } - const bool force_advance = symbols_at_frame >= config.max_symbols_per_step; + const bool force_advance = stream_symbols_at_frame_ >= config.max_symbols_per_step; if (blank || force_advance) { - ++frame_index; - symbols_at_frame = 0; - out.durations.push_back(1); + stream_symbols_at_frame_ = 0; + stream_durations_.push_back(1); + ++local_frame; + ++stream_frame_index_; } else { - out.durations.push_back(0); + stream_durations_.push_back(0); } if (on_text_delta && !blank) { - const auto current_text = decode_text(out.token_ids, options.keep_language_tags); - if (current_text.size() > emitted_text.size() && - current_text.compare(0, emitted_text.size(), emitted_text) == 0) { - on_text_delta(current_text.substr(emitted_text.size())); - emitted_text = current_text; - } else if (current_text != emitted_text) { + const auto current_text = decode_text(stream_token_ids_, stream_decode_options_.keep_language_tags); + if (current_text.size() > stream_emitted_text_.size() && + current_text.compare(0, stream_emitted_text_.size(), stream_emitted_text_) == 0) { + on_text_delta(current_text.substr(stream_emitted_text_.size())); + stream_emitted_text_ = current_text; + } else if (current_text != stream_emitted_text_) { on_text_delta(current_text); - emitted_text = current_text; + stream_emitted_text_ = current_text; } } - input_token = token; + stream_input_token_ = token; } - out.text = decode_text(out.token_ids, options.keep_language_tags); +} + +NemotronDecodedText NemotronDecoderRuntime::finish_stream_decode() { + if (!stream_decode_active_) { + throw std::runtime_error("Nemotron ASR stream decode requires begin_stream_decode()"); + } + NemotronDecodedText out; + out.token_ids = stream_token_ids_; + out.durations = stream_durations_; + out.text = decode_text(out.token_ids, stream_decode_options_.keep_language_tags); out.token_timestamps = build_token_timestamps(*assets_, out.token_ids, out.durations); + if (debug::trace_log_enabled()) { + std::string ids; + for (const int32_t id : out.token_ids) { + ids += std::to_string(id); + ids.push_back(','); + } + debug::trace_log_scalar("nemotron_asr.decoder.token_ids", ids); + } + stream_decode_active_ = false; + return out; +} + +NemotronDecodedText NemotronDecoderRuntime::decode_streaming( + const NemotronDecodeOptions & options, + const std::function & next_chunk, + const NemotronTextDeltaCallback & on_text_delta) { + if (!next_chunk) { + throw std::runtime_error("Nemotron ASR streaming decoder requires a chunk producer"); + } + const auto wall_start = Clock::now(); + begin_stream_decode(options); + NemotronEncodedAudio chunk; + bool received_any = false; + while (next_chunk(chunk)) { + received_any = true; + decode_stream_chunk(chunk, on_text_delta); + } + if (!received_any) { + throw std::runtime_error("Nemotron ASR streaming decoder received no encoded chunks"); + } + auto out = finish_stream_decode(); debug::timing_log_scalar("nemotron_asr.decoder_ms", engine::debug::elapsed_ms(wall_start, Clock::now())); debug::trace_log_scalar("nemotron_asr.decoder.tokens", out.token_ids.size()); - debug::trace_log_scalar("nemotron_asr.decoder.encoded_valid_frames", encoded_valid_frames); return out; } diff --git a/src/models/nemotron_asr/encoder.cpp b/src/models/nemotron_asr/encoder.cpp index da72ff52b..ca9456ccb 100644 --- a/src/models/nemotron_asr/encoder.cpp +++ b/src/models/nemotron_asr/encoder.cpp @@ -18,6 +18,8 @@ #include "ggml-backend.h" #include +#include +#include #include #include #include @@ -31,6 +33,11 @@ namespace { using Clock = std::chrono::steady_clock; constexpr size_t kEncoderGraphNodes = 2097152; +// Streaming graphs measure ~3.4k nodes (traced); the cap sizes the cgraph +// node array, which lives in the per-variant arena — 2M slots cost ~16 MB of +// arena per variant for nothing. 64k leaves a ~20x margin and lets the +// prefix ladder plus tail variants stay affordable. +constexpr size_t kStreamGraphNodes = 65536; int64_t causal_conv_output_dim(int64_t input, int64_t kernel, int64_t stride, bool streaming) { const int64_t left = streaming ? kernel - stride : kernel - 1; @@ -381,9 +388,12 @@ StreamingLayerOutputs build_projected_cache_streaming_encoder_layer( x = engine::core::wrap_tensor(ggml_add(ctx.ggml, x.tensor, ff2.tensor), x.shape, GGML_TYPE_F32); return { engine::modules::LayerNormModule({config.hidden_size, 1.0e-5f, true, true}).build(ctx, x, weights.norm_out), - engine::core::ensure_backend_addressable_layout(ctx, next_key_cache), - engine::core::ensure_backend_addressable_layout(ctx, next_value_cache), - conv_outputs.next_cache, + // The next-cache tensors must own their buffers (see the owned_copy note + // in the graph build): views alias base-node regions the allocator + // reuses, which corrupted the cache hand-off. + engine::core::wrap_tensor(ggml_cont(ctx.ggml, next_key_cache.tensor), next_key_cache.shape, next_key_cache.type), + engine::core::wrap_tensor(ggml_cont(ctx.ggml, next_value_cache.tensor), next_value_cache.shape, next_value_cache.type), + engine::core::wrap_tensor(ggml_cont(ctx.ggml, conv_outputs.next_cache.tensor), conv_outputs.next_cache.shape, conv_outputs.next_cache.type), }; } @@ -493,11 +503,13 @@ NemotronEncoderRuntime::NemotronEncoderRuntime( std::shared_ptr assets, std::shared_ptr weights, engine::core::ExecutionContext & execution_context, - size_t graph_arena_bytes) + size_t graph_arena_bytes, + size_t stream_graph_arena_bytes) : assets_(std::move(assets)), weights_(std::move(weights)), execution_context_(&execution_context), - graph_arena_bytes_(graph_arena_bytes) { + graph_arena_bytes_(graph_arena_bytes), + stream_graph_arena_bytes_(stream_graph_arena_bytes != 0 ? stream_graph_arena_bytes : graph_arena_bytes) { if (assets_ == nullptr || weights_ == nullptr) { throw std::runtime_error("Nemotron ASR encoder requires assets and weights"); } @@ -718,12 +730,23 @@ NemotronEncoderRuntime::Graph & NemotronEncoderRuntime::ensure_stream_graph( const bool streaming_graph = true; const int64_t k = enc.subsampling_kernel; const int64_t s = enc.subsampling_stride; - const bool first_chunk_time = first_chunk ? false : streaming_graph; - const int64_t stage1_frames = causal_conv_output_dim(input_frames, k, s, first_chunk_time); + // First chunk: the subsampling convs prepend a zero frame + the (zero) cache + // — 2 frames of left context, no right pad. The centered formula + // (causal_conv_output_dim(..., false)) assumes a right pad that does not + // exist here and overcounts for even mel counts (la0's 8-mel first window: + // mask 5 vs conv 4). Size the first chunk from the actual prepend instead; + // for odd mel counts (la3's 25-mel first window) the results are identical. + const int64_t stage1_frames = first_chunk + ? (input_frames + 2 - k) / s + 1 + : causal_conv_output_dim(input_frames, k, s, streaming_graph); const int64_t stage1_features = causal_conv_output_dim(feature_dim, k, s, false); - const int64_t stage2_frames = causal_conv_output_dim(stage1_frames, k, s, first_chunk_time); + const int64_t stage2_frames = first_chunk + ? (stage1_frames + 2 - k) / s + 1 + : causal_conv_output_dim(stage1_frames, k, s, streaming_graph); const int64_t stage2_features = causal_conv_output_dim(stage1_features, k, s, false); - const int64_t stage3_frames = causal_conv_output_dim(stage2_frames, k, s, first_chunk_time); + const int64_t stage3_frames = first_chunk + ? (stage2_frames + 2 - k) / s + 1 + : causal_conv_output_dim(stage2_frames, k, s, streaming_graph); const int64_t stage3_features = causal_conv_output_dim(stage2_features, k, s, false); if (stage3_features * enc.subsampling_channels != 4352) { throw std::runtime_error("Nemotron ASR streaming subsampling feature shape mismatch"); @@ -741,7 +764,12 @@ NemotronEncoderRuntime::Graph & NemotronEncoderRuntime::ensure_stream_graph( graph->streaming = true; graph->first_chunk = first_chunk; graph->backend = execution_context_->backend(); - ggml_init_params params{graph_arena_bytes_, nullptr, true}; + // Streaming graphs hold only tensor metadata (weights live in the backend + // buffers, activations in the gallocr), but the arena is committed per graph + // variant and the prefix ladder multiplies variants — the offline graph's + // 1 GB default times ~57 prefix sizes (lookahead 0) exhausts commit. The + // dedicated smaller stream arena keeps the ladder affordable. + ggml_init_params params{stream_graph_arena_bytes_, nullptr, true}; graph->ggml = ggml_init(params); if (graph->ggml == nullptr) { throw std::runtime_error("Failed to initialize Nemotron ASR streaming encoder graph context"); @@ -767,15 +795,24 @@ NemotronEncoderRuntime::Graph & NemotronEncoderRuntime::ensure_stream_graph( graph->attention_mask = engine::core::make_tensor(ctx, GGML_TYPE_F32, engine::core::TensorShape::from_dims({stage3_frames, key_frames})); graph->prompt = engine::core::make_tensor(ctx, GGML_TYPE_F32, engine::core::TensorShape::from_dims({1, stage3_frames, config.num_prompts})); graph->pos_emb = engine::core::make_tensor(ctx, GGML_TYPE_F32, engine::core::TensorShape::from_dims({1, 2 * key_frames - 1, enc.hidden_size})); - for (auto * tensor : {graph->mask1.tensor, graph->mask2.tensor, graph->mask3.tensor, graph->keep_mask.tensor, graph->attention_mask.tensor, graph->prompt.tensor, graph->pos_emb.tensor}) { - ggml_set_input(tensor); + for (auto * tensor : {graph->mask1.tensor, graph->mask2.tensor, graph->mask3.tensor, graph->keep_mask.tensor, graph->attention_mask.tensor, graph->prompt.tensor, graph->pos_emb.tensor}) { ggml_set_input(tensor); ggml_set_output(tensor); } auto x = engine::core::reshape_tensor(ctx, graph->input, engine::core::TensorShape::from_dims({1, 1, input_frames, feature_dim})); x = pad_freq_2d(ctx, x, k, s); - graph->next_subsampling_cache0 = - engine::core::ensure_backend_addressable_layout(ctx, next_time_cache_4d(ctx, graph->subsampling_cache0, x, 1)); + // The hand-off tensors must own their buffers. A view (even a contiguous + // one) points into a base node's gallocr region whose lifetime the + // allocator does NOT extend for later reads through the view — measured: + // the next attention K/V caches aliased later conv-cache buffers + // (K12 == conv17, V18 == conv20 data pointers), so the hand-off copied + // corrupted content into the next chunk's attention prefix. ggml_cont + // gives each cache its own region, which the zero-weighted dependency + // chain then keeps alive until the output. + auto owned_copy = [&ctx](engine::core::TensorValue value) { + return engine::core::wrap_tensor(ggml_cont(ctx.ggml, value.tensor), value.shape, value.type); + }; + graph->next_subsampling_cache0 = owned_copy(next_time_cache_4d(ctx, graph->subsampling_cache0, x, 1)); ggml_set_output(graph->next_subsampling_cache0.tensor); x = first_chunk ? engine::modules::ConcatModule({2}).build(ctx, zero_time_prefix_4d(ctx, x, 1), engine::modules::ConcatModule({2}).build(ctx, graph->subsampling_cache0, x)) @@ -785,8 +822,7 @@ NemotronEncoderRuntime::Graph & NemotronEncoderRuntime::ensure_stream_graph( x = engine::modules::ReluModule().build(ctx, engine::modules::TimeMask4dModule().build(ctx, x, graph->mask1)); x = pad_freq_2d(ctx, x, k, s); - graph->next_subsampling_cache1 = - engine::core::ensure_backend_addressable_layout(ctx, next_time_cache_4d(ctx, graph->subsampling_cache1, x, 1)); + graph->next_subsampling_cache1 = owned_copy(next_time_cache_4d(ctx, graph->subsampling_cache1, x, 1)); ggml_set_output(graph->next_subsampling_cache1.tensor); x = first_chunk ? engine::modules::ConcatModule({2}).build(ctx, zero_time_prefix_4d(ctx, x, 1), engine::modules::ConcatModule({2}).build(ctx, graph->subsampling_cache1, x)) @@ -799,8 +835,7 @@ NemotronEncoderRuntime::Graph & NemotronEncoderRuntime::ensure_stream_graph( x = engine::modules::ReluModule().build(ctx, engine::modules::TimeMask4dModule().build(ctx, x, graph->mask2)); x = pad_freq_2d(ctx, x, k, s); - graph->next_subsampling_cache2 = - engine::core::ensure_backend_addressable_layout(ctx, next_time_cache_4d(ctx, graph->subsampling_cache2, x, 1)); + graph->next_subsampling_cache2 = owned_copy(next_time_cache_4d(ctx, graph->subsampling_cache2, x, 1)); ggml_set_output(graph->next_subsampling_cache2.tensor); x = first_chunk ? engine::modules::ConcatModule({2}).build(ctx, zero_time_prefix_4d(ctx, x, 1), engine::modules::ConcatModule({2}).build(ctx, graph->subsampling_cache2, x)) @@ -879,7 +914,7 @@ NemotronEncoderRuntime::Graph & NemotronEncoderRuntime::ensure_stream_graph( x = layer_out.output; graph->next_attention_key_cache.push_back(layer_out.next_key_cache); graph->next_attention_value_cache.push_back(layer_out.next_value_cache); - graph->next_conv_cache.push_back(engine::core::ensure_backend_addressable_layout(ctx, layer_out.next_conv_cache)); + graph->next_conv_cache.push_back(owned_copy(layer_out.next_conv_cache)); ggml_set_output(graph->next_attention_key_cache.back().tensor); ggml_set_output(graph->next_attention_value_cache.back().tensor); ggml_set_output(graph->next_conv_cache.back().tensor); @@ -894,7 +929,33 @@ NemotronEncoderRuntime::Graph & NemotronEncoderRuntime::ensure_stream_graph( graph->output = engine::modules::LinearModule({enc.hidden_size, config.decoder_hidden_size, true}) .build(ctx, x, weights.encoder_projector); - graph->graph = ggml_new_graph_custom(graph->ggml, kEncoderGraphNodes, false); + // The next-cache tensors are read AFTER the compute (the cache hand-off), + // but the graph allocator only sees in-graph lifetimes and may hand their + // buffers to other tensors — measured: the first chunk's next attention + // KV caches aliased with next_subsampling_cache0 (identical data pointers) + // and the hand-off read the subsampling content. Chain a zero-weighted + // scalar dependency from every cache into the output node: the allocator + // must keep every cache alive until the end, and the buffers stay distinct. + // The scales are exactly zero, so the encoder output is unchanged. + { + auto dep_tensor = graph->output.tensor; + auto chain = [&](const engine::core::TensorValue & cache) { + auto sum = ggml_sum(ctx.ggml, cache.tensor); + auto scaled = ggml_scale(ctx.ggml, sum, 0.0f); + dep_tensor = ggml_add1(ctx.ggml, dep_tensor, scaled); + }; + chain(graph->next_subsampling_cache0); + chain(graph->next_subsampling_cache1); + chain(graph->next_subsampling_cache2); + for (int64_t layer = 0; layer < enc.layers; ++layer) { + chain(graph->next_attention_key_cache[static_cast(layer)]); + chain(graph->next_attention_value_cache[static_cast(layer)]); + chain(graph->next_conv_cache[static_cast(layer)]); + } + graph->output = engine::core::wrap_tensor(dep_tensor, graph->output.shape, GGML_TYPE_F32); + } + + graph->graph = ggml_new_graph_custom(graph->ggml, kStreamGraphNodes, false); ggml_build_forward_expand(graph->graph, graph->output.tensor); ggml_build_forward_expand(graph->graph, graph->next_subsampling_cache0.tensor); ggml_build_forward_expand(graph->graph, graph->next_subsampling_cache1.tensor); @@ -904,6 +965,7 @@ NemotronEncoderRuntime::Graph & NemotronEncoderRuntime::ensure_stream_graph( ggml_build_forward_expand(graph->graph, graph->next_attention_value_cache[static_cast(layer)].tensor); ggml_build_forward_expand(graph->graph, graph->next_conv_cache[static_cast(layer)].tensor); } + debug::trace_log_scalar("nemotron_asr.encoder.stream.graph_nodes", (int64_t) ggml_graph_n_nodes(graph->graph)); graph->pos_graph = ggml_new_graph_custom(graph->ggml, 4096, false); for (const auto & projected : graph->projected_pos_emb_computed) { ggml_build_forward_expand(graph->pos_graph, projected.tensor); @@ -954,7 +1016,6 @@ NemotronEncoderRuntime::Graph & NemotronEncoderRuntime::ensure_stream_graph( debug::timing_log_scalar("nemotron_asr.encoder.stream.graph_build_ms", build_ms); debug::timing_log_scalar("nemotron_asr.encoder.stream.graph_rebuild_ms", build_ms); debug::trace_log_scalar("nemotron_asr.encoder.stream.graph_cache_hit", false); - debug::trace_log_scalar("nemotron_asr.encoder.stream.graph_input_frames", input_frames); debug::trace_log_scalar("nemotron_asr.encoder.stream.graph_encoded_frames", stage3_frames); debug::trace_log_scalar("nemotron_asr.encoder.stream.graph_prefix_frames", prefix_frames); debug::trace_log_scalar("nemotron_asr.encoder.stream.graph_prefix_capacity", prefix_capacity); @@ -971,17 +1032,70 @@ void NemotronEncoderRuntime::release_offline_graph() { void NemotronEncoderRuntime::prepare_streaming_capacity(int64_t feature_dim, int64_t lookahead_tokens) { const auto & enc = assets_->config.encoder; - const int64_t first_frames = 1 + enc.subsampling_factor * lookahead_tokens; - const int64_t next_frames = enc.subsampling_factor * (lookahead_tokens + 1); + // Must match the session's first-window size: at least one full encoded + // frame of mel input, or the first chunk misses the prebuilt graph. + const int64_t first_frames = std::max( + enc.subsampling_factor, + 1 + enc.subsampling_factor * lookahead_tokens); + // Must match the session's sliding-chunk size: the session carries a floor + // of 4 encoded frames per chunk (lookahead 0 with 1-frame chunks falls + // behind realtime on CPU), so the prebuilt ladder must use the same size. + const int64_t next_frames = enc.subsampling_factor * + std::max(lookahead_tokens + 1, 4); (void) ensure_stream_graph(first_frames, feature_dim, lookahead_tokens, 0, true); const int64_t k = enc.subsampling_kernel; const int64_t s = enc.subsampling_stride; const int64_t stage1_frames = causal_conv_output_dim(next_frames, k, s, true); const int64_t stage2_frames = causal_conv_output_dim(stage1_frames, k, s, true); const int64_t stage3_frames = causal_conv_output_dim(stage2_frames, k, s, true); - for (int64_t prefix = stage3_frames; prefix < enc.sliding_window; prefix += stage3_frames) { + // Warm the prefix ladder: one graph variant per prefix step up to + // sliding_window - 1. A small lookahead yields many variants (lookahead 0 -> + // 56), so the prebuild is bounded by a graph-arena commit budget instead of + // a variant count; variants beyond the budget build lazily in + // encode_stream_chunk (~100 ms each) as the stream reaches them. The + // streaming-graph arena is metadata-only, so 64 MB per variant keeps the + // whole ladder affordable and the stream stall-free. + constexpr size_t kMaxPrebuildArenaCommit = 6ull << 30; // 6 GB + const size_t per_variant = stream_graph_arena_bytes_; + const int64_t kMaxPrebuiltPrefixGraphs = + static_cast(kMaxPrebuildArenaCommit / std::max(per_variant, 1)); + // The session's prefix sequence: the first chunk emits first_encoded frames + // (centered conv over first_frames), so chunk 2's prefix = first_encoded; + // each later chunk adds stage3_frames. The final variant caps at + // sliding_window - 1. + const int64_t first_encoded = causal_conv_output_dim(first_frames, k, s, false); + int64_t prebuilt = 0; + for (int64_t prefix = first_encoded; + prefix < enc.sliding_window && prebuilt < kMaxPrebuiltPrefixGraphs; + prefix += stage3_frames, ++prebuilt) { (void) ensure_stream_graph(next_frames, feature_dim, lookahead_tokens, std::min(prefix, enc.sliding_window - 1), false); } + (void) ensure_stream_graph(next_frames, feature_dim, lookahead_tokens, enc.sliding_window - 1, false); + // Flush variants: the finalize/speculative flush pads its window beyond the + // full chunk size so the encoded stream carries ~500 ms of post-speech + // silence (the model's trailing-token requirement) even when the client + // closes right after the last speech. Same prefix sequence as the ladder; + // the session builds its flush window from the same env default. + const char * flush_env = std::getenv("NEMOTRON_FLUSH_WINDOW_MEL"); + const int64_t flush_mel = std::max( + next_frames, + flush_env != nullptr && *flush_env != 0 + ? std::strtoll(flush_env, nullptr, 10) + : 8 * 8); + if (flush_mel > next_frames) { + for (int64_t prefix = first_encoded; + prefix < enc.sliding_window; + prefix += stage3_frames) { + (void) ensure_stream_graph(flush_mel, feature_dim, lookahead_tokens, std::min(prefix, enc.sliding_window - 1), false); + } + (void) ensure_stream_graph(flush_mel, feature_dim, lookahead_tokens, enc.sliding_window - 1, false); + } + // NOTE: short tail variants (8/16/24 mel) were tried here so the finalize + // flush could encode 1-3 frames instead of 4. Refuted by measurement: the + // RNNT fires a word's trailing token on a post-speech frame, and when the + // turn buffer cuts the speech decay the model needs 2-3 zero-padding + // frames before emitting it ('Hello' -> 'Hel' with a short tail). The + // finalize flush must pad to the full chunk window. } NemotronEncoderStreamState NemotronEncoderRuntime::make_stream_state() const { @@ -1102,6 +1216,7 @@ NemotronEncodedAudio NemotronEncoderRuntime::encode_stream_chunk( } } engine::core::write_tensor_f32(graph.input, input_scratch_); + debug::trace_log_scalar("nemotron_asr.encoder.stream.stage", int64_t(1)); // input written auto write_zeros = [](const engine::core::TensorValue & tensor) { const size_t elements = static_cast(tensor.shape.num_elements()); @@ -1109,16 +1224,36 @@ NemotronEncodedAudio NemotronEncoderRuntime::encode_stream_chunk( engine::core::write_tensor_f32(tensor, zeros); }; const auto cache_transfer_start = Clock::now(); - if (use_cross_backend_cache) { - const auto backend = execution_context_->backend(); - if (ggml_nelements(backend_cache_source->next_subsampling_cache0.tensor) != ggml_nelements(graph.subsampling_cache0.tensor) || - ggml_nelements(backend_cache_source->next_subsampling_cache1.tensor) != ggml_nelements(graph.subsampling_cache1.tensor) || - ggml_nelements(backend_cache_source->next_subsampling_cache2.tensor) != ggml_nelements(graph.subsampling_cache2.tensor)) { - throw std::runtime_error("Nemotron ASR streaming subsampling backend cache shape mismatch"); + // Cache hand-off between graph variants: shapes must match exactly (element + // count equality is not enough — same ne with different strides trips ggml's + // layout assert). A mismatch means the variant ladder built inconsistent + // geometry; name the tensors instead of aborting inside ggml. + auto copy_cache = [this](const engine::core::TensorValue & src, engine::core::TensorValue & dst, const char * name) { + const ggml_tensor * s = src.tensor; + const ggml_tensor * d = dst.tensor; + bool same_layout = s->type == d->type; + for (int i = 0; same_layout && i < 4; ++i) { + same_layout = s->ne[i] == d->ne[i] && s->nb[i] == d->nb[i]; } - ggml_backend_tensor_copy_async(backend, backend, backend_cache_source->next_subsampling_cache0.tensor, graph.subsampling_cache0.tensor); - ggml_backend_tensor_copy_async(backend, backend, backend_cache_source->next_subsampling_cache1.tensor, graph.subsampling_cache1.tensor); - ggml_backend_tensor_copy_async(backend, backend, backend_cache_source->next_subsampling_cache2.tensor, graph.subsampling_cache2.tensor); + if (!same_layout) { + auto shape_of = [](const ggml_tensor * t) { + std::string s = std::to_string(t->ne[0]); + for (int i = 1; i < 4; ++i) { + s += "x" + std::to_string(t->ne[i]); + } + return s; + }; + throw std::runtime_error(std::string("Nemotron ASR streaming cache layout mismatch (") + + name + ": next " + shape_of(src.tensor) + " vs graph " + + shape_of(dst.tensor) + ")"); + } + ggml_backend_tensor_copy_async(execution_context_->backend(), execution_context_->backend(), src.tensor, dst.tensor); + }; + if (use_cross_backend_cache) { + copy_cache(backend_cache_source->next_subsampling_cache0, graph.subsampling_cache0, "subsampling0"); + debug::trace_log_scalar("nemotron_asr.encoder.stream.stage", int64_t(11)); + copy_cache(backend_cache_source->next_subsampling_cache1, graph.subsampling_cache1, "subsampling1"); + copy_cache(backend_cache_source->next_subsampling_cache2, graph.subsampling_cache2, "subsampling2"); } else if (!use_same_backend_cache) { if (!state.first_chunk) { throw std::runtime_error("Nemotron ASR streaming cache is missing between chunks"); @@ -1138,14 +1273,14 @@ NemotronEncodedAudio NemotronEncoderRuntime::encode_stream_chunk( graph.stream_static_prompt_id = prompt_id; } + debug::trace_log_scalar("nemotron_asr.encoder.stream.stage", int64_t(12)); for (size_t layer = 0; layer < graph.conv_cache.size(); ++layer) { if (use_cross_backend_cache) { - const auto backend = execution_context_->backend(); if (layer >= backend_cache_source->next_conv_cache.size() || ggml_nelements(backend_cache_source->next_conv_cache[layer].tensor) != ggml_nelements(graph.conv_cache[layer].tensor)) { throw std::runtime_error("Nemotron ASR streaming convolution backend cache shape mismatch"); } - ggml_backend_tensor_copy_async(backend, backend, backend_cache_source->next_conv_cache[layer].tensor, graph.conv_cache[layer].tensor); + copy_cache(backend_cache_source->next_conv_cache[layer], graph.conv_cache[layer], "conv"); } else if (!use_same_backend_cache) { if (!state.first_chunk) { throw std::runtime_error("Nemotron ASR streaming convolution cache is missing between chunks"); @@ -1153,6 +1288,7 @@ NemotronEncodedAudio NemotronEncoderRuntime::encode_stream_chunk( write_zeros(graph.conv_cache[layer]); } } + debug::trace_log_scalar("nemotron_asr.encoder.stream.stage", int64_t(13)); if (state.attention_cached_frames > 0) { if (!use_cross_backend_cache && !use_same_backend_cache) { throw std::runtime_error("Nemotron ASR streaming attention cache is missing between chunks"); @@ -1164,27 +1300,40 @@ NemotronEncodedAudio NemotronEncoderRuntime::encode_stream_chunk( } for (size_t layer = 0; layer < graph.attention_key_cache.size(); ++layer) { if (use_cross_backend_cache) { - const auto backend = execution_context_->backend(); if (ggml_nelements(backend_cache_source->next_attention_key_cache[layer].tensor) != ggml_nelements(graph.attention_key_cache[layer].tensor) || ggml_nelements(backend_cache_source->next_attention_value_cache[layer].tensor) != ggml_nelements(graph.attention_value_cache[layer].tensor)) { throw std::runtime_error("Nemotron ASR streaming attention backend cache shape mismatch"); } - ggml_backend_tensor_copy_async( - backend, - backend, + if (layer == 0) { + debug::trace_log_scalar("nemotron_asr.encoder.stream.stage", int64_t(130)); + } + // Hand the cache over through host staging instead of + // ggml_backend_tensor_copy_async: both tensors are contiguous F32 + // graph nodes, and the read/write helpers are the same proven path + // every input and output of this encoder uses. + attention_key_scratch_.resize(static_cast(ggml_nelements(graph.attention_key_cache[layer].tensor))); + attention_value_scratch_.resize(static_cast(ggml_nelements(graph.attention_value_cache[layer].tensor))); + engine::core::read_tensor_f32_into( backend_cache_source->next_attention_key_cache[layer].tensor, - graph.attention_key_cache[layer].tensor); - ggml_backend_tensor_copy_async( - backend, - backend, + attention_key_scratch_); + engine::core::write_tensor_f32(graph.attention_key_cache[layer], attention_key_scratch_); + if (layer == 0) { + debug::trace_log_scalar("nemotron_asr.encoder.stream.stage", int64_t(131)); + } + engine::core::read_tensor_f32_into( backend_cache_source->next_attention_value_cache[layer].tensor, - graph.attention_value_cache[layer].tensor); + attention_value_scratch_); + engine::core::write_tensor_f32(graph.attention_value_cache[layer], attention_value_scratch_); + if (layer == 0) { + debug::trace_log_scalar("nemotron_asr.encoder.stream.stage", int64_t(132)); + } } } } debug::timing_log_scalar( "nemotron_asr.encoder.stream.cache_transfer_ms", engine::debug::elapsed_ms(cache_transfer_start, Clock::now())); + debug::trace_log_scalar("nemotron_asr.encoder.stream.stage", int64_t(2)); // caches copied engine::core::set_backend_threads(execution_context_->backend(), execution_context_->config().threads); const auto compute_start = Clock::now(); @@ -1195,6 +1344,7 @@ NemotronEncodedAudio NemotronEncoderRuntime::encode_stream_chunk( throw std::runtime_error("Nemotron ASR streaming encoder graph compute failed"); } + debug::trace_log_scalar("nemotron_asr.encoder.stream.stage", int64_t(3)); // compute done engine::core::read_tensor_f32_into(graph.output.tensor, output_scratch_); state.attention_seen_frames += graph.encoded_frames; state.attention_cached_frames = std::min(enc.sliding_window - 1, state.attention_seen_frames); @@ -1232,6 +1382,42 @@ NemotronEncodedAudio NemotronEncoderRuntime::encode_stream_chunk( out.frames = graph.encoded_frames; out.valid_frames = graph.encoded_frames; out.hidden_size = graph.decoder_hidden; + // Opt-in post-chunk cache dump (NEMOTRON_DUMP_CACHES=): writes every + // next-cache tensor in a fixed order so the hand-off content can be diffed + // against the reference implementation chunk for chunk. + if (const char * dump_env = std::getenv("NEMOTRON_DUMP_CACHES"); dump_env != nullptr && *dump_env != '\0') { + const std::string base = std::string(dump_env) + "_c" + std::to_string(state.dump_seq) + "_nc"; + std::ofstream blob(base + ".f32", std::ios::binary); + std::vector counts; + std::vector ptrs; + auto write_cache = [&](const engine::core::TensorValue & tensor) { + std::vector tmp(static_cast(ggml_nelements(tensor.tensor))); + engine::core::read_tensor_f32_into(tensor.tensor, tmp); + blob.write(reinterpret_cast(tmp.data()), static_cast(tmp.size() * sizeof(float))); + counts.push_back(static_cast(tmp.size())); + ptrs.push_back(tensor.tensor->data); + }; + write_cache(graph.next_subsampling_cache0); + write_cache(graph.next_subsampling_cache1); + write_cache(graph.next_subsampling_cache2); + for (int64_t layer = 0; layer < enc.layers; ++layer) { + write_cache(graph.next_conv_cache[static_cast(layer)]); + write_cache(graph.next_attention_key_cache[static_cast(layer)]); + write_cache(graph.next_attention_value_cache[static_cast(layer)]); + } + blob.close(); + std::ofstream meta(base + ".meta"); + meta << "prefix=" << graph.prefix_frames << " encoded=" << graph.encoded_frames; + for (int64_t c : counts) { + meta << " " << c; + } + meta << "\n"; + std::ofstream ptr_meta(base + "_ptr.meta"); + for (const void * p : ptrs) { + ptr_meta << p << "\n"; + } + } + ++state.dump_seq; state.first_chunk = false; debug::timing_log_scalar("nemotron_asr.encoder.stream_ms", engine::debug::elapsed_ms(wall_start, Clock::now())); debug::trace_log_scalar("nemotron_asr.encoder.stream.valid_frames", out.valid_frames); diff --git a/src/models/nemotron_asr/session.cpp b/src/models/nemotron_asr/session.cpp index 019d4633c..c44a64a28 100644 --- a/src/models/nemotron_asr/session.cpp +++ b/src/models/nemotron_asr/session.cpp @@ -6,6 +6,9 @@ #include #include #include +#include +#include +#include #include #include #include @@ -26,6 +29,35 @@ std::shared_ptr require_assets(std::shared_ptr): writes the +// mel features and the encoder output of every streaming chunk as little-endian +// f32 blobs plus a .meta sidecar with the dimensions, so a reference +// implementation (transformers nemotron_asr_streaming) can be diffed +// chunk-for-chunk and frame-for-frame. +void dump_stream_chunk( + const std::string & prefix, + int64_t seq, + const NemotronFrontendFeatures & mel, + const NemotronEncodedAudio & enc, + bool center) { + const char * env = std::getenv("NEMOTRON_DUMP_CHUNKS"); + if (env == nullptr || *env == '\0') { + return; + } + const std::string base = std::string(env) + "_c" + std::to_string(seq); + auto write_f32 = [&](const std::string & path, const float * data, size_t count) { + std::ofstream out(path, std::ios::binary); + out.write(reinterpret_cast(data), static_cast(count * sizeof(float))); + }; + write_f32(base + "_mel.f32", mel.values.data(), static_cast(mel.frames * mel.feature_dim)); + write_f32(base + "_enc.f32", enc.values.data(), static_cast(enc.frames * enc.hidden_size)); + std::ofstream meta(base + ".meta"); + meta << "mel_frames=" << mel.frames << " mel_valid=" << mel.valid_frames + << " mel_dim=" << mel.feature_dim << " enc_frames=" << enc.frames + << " enc_valid=" << enc.valid_frames << " enc_hidden=" << enc.hidden_size + << " center=" << (center ? 1 : 0) << "\n"; +} + engine::assets::TensorStorageType option_weight_type( const runtime::SessionOptions & options, const char * key, @@ -42,10 +74,19 @@ void validate_matmul_weight_storage(engine::assets::TensorStorageType storage_ty storage_type == engine::assets::TensorStorageType::F32 || storage_type == engine::assets::TensorStorageType::F16 || storage_type == engine::assets::TensorStorageType::BF16 || - storage_type == engine::assets::TensorStorageType::Q8_0) { + storage_type == engine::assets::TensorStorageType::Q8_0 || + storage_type == engine::assets::TensorStorageType::Q4_0 || + storage_type == engine::assets::TensorStorageType::Q4_1 || + storage_type == engine::assets::TensorStorageType::Q5_0 || + storage_type == engine::assets::TensorStorageType::Q5_1 || + storage_type == engine::assets::TensorStorageType::Q4_K || + storage_type == engine::assets::TensorStorageType::Q5_K || + storage_type == engine::assets::TensorStorageType::Q6_K) { + // Sub-q8_0 types are re-quantized from the source weights at load + // (dequant -> ggml_quantize_chunk): faster CPU GEMMs, some accuracy risk. return; } - throw std::runtime_error(std::string(option_name) + " supports only native, f32, f16, bf16, and q8_0"); + throw std::runtime_error(std::string(option_name) + " supports only native, f32, f16, bf16, q8_0, q4_0, q4_1, q5_0, q5_1, q4_k, q5_k, and q6_k"); } void validate_conv_weight_storage(engine::assets::TensorStorageType storage_type, const char * option_name) { @@ -81,7 +122,9 @@ int64_t frontend_frames_for_samples( NemotronFrontendFeatures slice_features(const NemotronFrontendFeatures & in, int64_t start_frame, int64_t frames) { if (start_frame < 0 || frames <= 0 || start_frame + frames > in.frames) { - throw std::runtime_error("Nemotron ASR streaming feature slice is out of range"); + throw std::runtime_error("Nemotron ASR streaming feature slice is out of range (start=" + + std::to_string(start_frame) + ", frames=" + std::to_string(frames) + + ", in.frames=" + std::to_string(in.frames) + ")"); } NemotronFrontendFeatures out; out.frames = frames; @@ -144,11 +187,22 @@ NemotronASRSessionBase::NemotronASRSessionBase( matmul_weight_storage_type_, conv_weight_storage_type_, weight_context_bytes_); + // Streaming encoder graphs are metadata-only arenas but the prefix ladder + // multiplies them (up to ~15 variants at lookahead 0). The streaming graph + // caps its node array at 64k entries (~3.4k used), so a 16 MB per-variant + // arena holds the metadata with a wide margin — about a sixth of the old + // 96 MB per-variant commit. An explicit + // nemotron_asr.encoder_graph_arena_mb option is honored as-is for both. + constexpr size_t kDefaultStreamEncoderGraphArenaBytes = 16ull * 1024ull * 1024ull; + const size_t stream_arena_bytes = encoder_graph_arena_bytes_ == kDefaultEncoderGraphArenaBytes + ? kDefaultStreamEncoderGraphArenaBytes + : encoder_graph_arena_bytes_; encoder_ = std::make_unique( assets_, weights_, execution_context(), - encoder_graph_arena_bytes_); + encoder_graph_arena_bytes_, + stream_arena_bytes); decoder_ = std::make_unique( assets_, weights_, @@ -233,10 +287,13 @@ int64_t NemotronASRSessionBase::lookahead_for_options(const std::unordered_mapconfig.encoder.supported_lookahead_tokens.begin(), - assets_->config.encoder.supported_lookahead_tokens.end(), - lookahead) == assets_->config.encoder.supported_lookahead_tokens.end()) { + // The GGUF embeds supported {0,3,6,13}, but the model card declares chunk + // durations 80-1120 ms (lookahead 0..13) as pure runtime knobs. Accept 1 + // (160 ms chunks) on that basis; its geometry (9-frame first window) is + // well-formed, unlike lookahead 0's degenerate single-frame first window. + const auto supported = assets_->config.encoder.supported_lookahead_tokens; + if (lookahead != 1 && + std::find(supported.begin(), supported.end(), lookahead) == supported.end()) { throw std::runtime_error("Nemotron ASR unsupported lookahead_tokens value"); } return lookahead; @@ -309,19 +366,50 @@ NemotronDecodedText NemotronASRSessionBase::run_streaming_audio( const NemotronDecodeOptions & decode_options, const NemotronTextDeltaCallback & on_text_delta) { const auto & fc = assets_->config.frontend; - const int64_t first_mel_frames = 1 + assets_->config.encoder.subsampling_factor * lookahead; - const int64_t mel_frames_per_chunk = assets_->config.encoder.subsampling_factor * (lookahead + 1); + // The first chunk must cover at least one full encoded frame (subsampling + // factor mel frames — the subsampling conv cannot produce output from less), + // otherwise encoded frame 0 is computed from zero-padded cache frames. + const int64_t first_mel_frames = std::max( + assets_->config.encoder.subsampling_factor, + 1 + assets_->config.encoder.subsampling_factor * lookahead); + // The sliding chunk carries at least 4 encoded frames: at lookahead 0 the + // emit-all schedule is exact for any chunk size (no frame needs right + // context), and larger chunks amortize the per-graph overheads — 1-frame + // chunks at 80 ms measurably fall behind realtime on one CPU thread. + const int64_t mel_frames_per_chunk = assets_->config.encoder.subsampling_factor * + std::max(lookahead + 1, 4); const int64_t first_samples = (first_mel_frames - 1) * fc.hop_length + fc.win_length / 2; const int64_t samples_per_chunk = mel_frames_per_chunk * fc.hop_length + fc.win_length; auto waveform = frontend_.prepare_waveform(audio); if (static_cast(waveform.size()) < first_samples) { - throw std::runtime_error("Nemotron ASR streaming request is shorter than the first required chunk"); + // Ultra-short turn: silence-pad to the first required chunk rather than + // failing the whole request — the flush below keeps the stream well-formed. + waveform.resize(static_cast(first_samples), 0.0f); } NemotronEncoderStreamState stream_state = encoder_->make_stream_state(); bool first_chunk = true; + bool flushed = false; int64_t chunk_count = 0; int64_t mel_frame_idx = first_mel_frames; int64_t start_idx = mel_frame_idx * fc.hop_length - fc.n_fft / 2; + // Window [start_idx, start_idx + samples_per_chunk) centered on the chunk's + // mel frames. start_idx goes negative when the window precedes the signal + // start (lookahead 0: the second window begins at 1*hop - n_fft/2 = -96); + // the left context is silence then, exactly like the first chunk's center + // pad — so build the window zero-padded instead of indexing before begin(). + auto window_at = [&](int64_t from) -> std::vector { + std::vector window(static_cast(samples_per_chunk), 0.0f); + const int64_t copy_from = std::max(from, 0); + const int64_t copy_to = std::min( + from + samples_per_chunk, static_cast(waveform.size())); + if (copy_to > copy_from) { + std::copy( + waveform.begin() + static_cast(copy_from), + waveform.begin() + static_cast(copy_to), + window.begin() + static_cast(copy_from - from)); + } + return window; + }; auto next_chunk = [&](NemotronEncodedAudio & out) -> bool { if (first_chunk) { first_chunk = false; @@ -331,17 +419,34 @@ NemotronDecodedText NemotronASRSessionBase::run_streaming_audio( waveform.begin() + static_cast(first_samples)); auto features = frontend_.extract_waveform(chunk_waveform, true); if (features.frames > first_mel_frames) { - features = slice_features(features, 0, first_mel_frames); + features = slice_features(features, 0, first_mel_frames); // whole-buffer first chunk } out = encoder_->encode_stream_chunk(features, prompt_id, lookahead, stream_state); return true; } if (start_idx + samples_per_chunk >= static_cast(waveform.size())) { - return false; + // End of stream: the tail no longer fills a full window. Zero-pad it to + // the full window (the reference processor right-pads the final chunk) + // and encode one last chunk. Dropping the tail loses the last word(s) + // of every turn whose audio does not align with the window stride — + // and entire short utterances ("Hello"), whose transcript came back + // empty because nothing beyond the first chunk was ever encoded. + if (flushed || start_idx + fc.n_fft > static_cast(waveform.size())) { + // Already flushed, or the leftover is too short to contribute even + // one full mel frame (it is inside the previous window's right pad). + return false; + } + flushed = true; + ++chunk_count; + auto chunk_waveform = window_at(start_idx); + auto features = frontend_.extract_waveform(chunk_waveform, false); + if (features.frames != mel_frames_per_chunk) { + throw std::runtime_error("Nemotron ASR streaming frontend produced unexpected flush chunk frame count"); + } + out = encoder_->encode_stream_chunk(features, prompt_id, lookahead, stream_state); + return true; } - std::vector chunk_waveform( - waveform.begin() + static_cast(start_idx), - waveform.begin() + static_cast(start_idx + samples_per_chunk)); + auto chunk_waveform = window_at(start_idx); auto features = frontend_.extract_waveform(chunk_waveform, false); if (features.frames != mel_frames_per_chunk) { throw std::runtime_error("Nemotron ASR streaming frontend produced unexpected chunk frame count"); @@ -407,6 +512,36 @@ void NemotronASRStreamingSession::start_stream(const runtime::TaskRequest & requ if (const auto option = runtime::find_option(request.options, {"language"})) { streaming_language_ = *option; } + + // Derive the native chunk geometry and the incremental pipeline state. The + // window math mirrors run_streaming_audio() so a chunked session and a + // whole-buffer session encode identical windows. + runtime::TaskRequest config_request; + config_request.text_input = runtime::Transcript{"", streaming_language_}; + config_request.options = streaming_options_; + stream_prompt_id_ = prompt_id_for_request(config_request); + stream_lookahead_ = lookahead_for_options(streaming_options_); + stream_decode_options_ = decode_options_for_request(config_request); + + const auto & fc = assets_->config.frontend; + const auto & enc = assets_->config.encoder; + stream_first_mel_frames_ = std::max( + enc.subsampling_factor, + 1 + enc.subsampling_factor * stream_lookahead_); + stream_mel_frames_per_chunk_ = enc.subsampling_factor * + std::max(stream_lookahead_ + 1, 4); + stream_first_samples_ = (stream_first_mel_frames_ - 1) * fc.hop_length + fc.win_length / 2; + stream_samples_per_chunk_ = stream_mel_frames_per_chunk_ * fc.hop_length + fc.win_length; + stream_next_chunk_start_ = stream_first_mel_frames_ * fc.hop_length - fc.n_fft / 2; + stream_await_first_chunk_ = true; + stream_tail_encoded_ = false; + stream_decode_active_ = false; + stream_dump_chunk_seq_ = 0; + spec_valid_ = false; + spec_frames_ = NemotronEncodedAudio{}; + spec_last_loud_sample_ = 0; + spec_audio_mark_ = 0; + encoder_stream_state_ = encoder_->make_stream_state(); } void NemotronASRStreamingSession::set_stream_event_sink(runtime::StreamEventCallback sink) { @@ -419,6 +554,164 @@ void NemotronASRStreamingSession::reset() { throw std::runtime_error("Nemotron ASR reset called on non-streaming session"); } streaming_audio_ = runtime::AudioBuffer{}; + stream_await_first_chunk_ = true; + stream_tail_encoded_ = false; + stream_decode_active_ = false; +} + +bool NemotronASRStreamingSession::encode_and_decode_next_chunk(bool flush_tail, std::string & delta_out) { + if (stream_tail_encoded_) { + return false; + } + const auto & fc = assets_->config.frontend; + const int64_t total = static_cast(streaming_audio_.samples.size()); + + std::vector window; + bool center = false; + bool reuse_spec = false; + if (stream_await_first_chunk_) { + if (total < stream_first_samples_) { + return false; + } + window.assign( + streaming_audio_.samples.begin(), + streaming_audio_.samples.begin() + static_cast(stream_first_samples_)); + center = true; + } else if (stream_next_chunk_start_ + stream_samples_per_chunk_ < total) { + // start_idx goes negative when the window precedes the signal start + // (lookahead 0: the second window begins at 1*hop - n_fft/2 = -96); the + // left context is silence then, exactly like the first chunk's center + // pad — zero-pad instead of indexing before begin(). + const int64_t copy_from = std::max(stream_next_chunk_start_, 0); + window.assign( + streaming_audio_.samples.begin() + static_cast(copy_from), + streaming_audio_.samples.begin() + static_cast(stream_next_chunk_start_ + stream_samples_per_chunk_)); + window.insert( + window.begin(), + static_cast(std::max(0, -stream_next_chunk_start_)), + 0.0f); + } else if (flush_tail && stream_next_chunk_start_ < total) { + // The tail never fills a whole native chunk: zero-pad it so the final + // audio is encoded too (the padding decodes to blank tokens). + // + // The pad MUST carry ~500 ms of post-speech silence: the RNNT fires a + // word's trailing token only after that much silence in the encoded + // stream (measured: 210 ms fails, 530 ms works; the padding content is + // irrelevant — the reference wavs' own tails are digital zeros). The + // flush window is sized for it (flush_window_mel), and when the + // speculative flush already encoded this window, its frames are + // reused instead of re-encoding. + if (spec_valid_) { + reuse_spec = true; + } else { + build_flush_window(total, window); + } + stream_tail_encoded_ = true; + } else { + return false; + } + + NemotronEncodedAudio encoded; + if (reuse_spec) { + encoded = std::move(spec_frames_); + spec_frames_ = NemotronEncodedAudio{}; + } else { + auto features = frontend_.extract_waveform(window, center); + if (center && features.frames > stream_first_mel_frames_) { + features = slice_features(features, 0, stream_first_mel_frames_); + } + encoded = encoder_->encode_stream_chunk( + features, + stream_prompt_id_, + stream_lookahead_, + encoder_stream_state_); + dump_stream_chunk("stream", stream_dump_chunk_seq_, features, encoded, center); + ++stream_dump_chunk_seq_; + } + + if (!stream_decode_active_) { + decoder_->begin_stream_decode(stream_decode_options_); + stream_decode_active_ = true; + } + decoder_->decode_stream_chunk(encoded, [&](const std::string & delta) { + delta_out += delta; + }); + + if (stream_await_first_chunk_) { + stream_await_first_chunk_ = false; + stream_next_chunk_start_ = stream_first_mel_frames_ * fc.hop_length - fc.n_fft / 2; + } else if (!reuse_spec) { + stream_next_chunk_start_ += stream_mel_frames_per_chunk_ * fc.hop_length; + } + return true; +} + +int64_t NemotronASRStreamingSession::flush_window_mel() const { + const auto & enc = assets_->config.encoder; + const char * env = std::getenv("NEMOTRON_FLUSH_WINDOW_MEL"); + const int64_t requested = env != nullptr && *env != 0 ? std::strtoll(env, nullptr, 10) : 8 * 8; + return std::max(enc.subsampling_factor * std::max(stream_lookahead_ + 1, 4), requested); +} + +void NemotronASRStreamingSession::build_flush_window(int64_t total, std::vector & window) const { + const auto & fc = assets_->config.frontend; + const int64_t copy_from = std::max(stream_next_chunk_start_, 0); + window.assign( + streaming_audio_.samples.begin() + static_cast(copy_from), + streaming_audio_.samples.begin() + static_cast(total)); + if (stream_next_chunk_start_ < 0) { + window.insert(window.begin(), static_cast(-stream_next_chunk_start_), 0.0f); + } + window.resize(static_cast(flush_window_mel() * fc.hop_length + fc.win_length), 0.0f); +} + +// Speculative flush: once the ingest sees enough trailing silence, encode the +// flush window right here on the ingest thread (the backend cannot run +// concurrent graphs, and during silence there is no realtime encode to +// contend with). Finalize then reuses the frames instead of paying the encode +// after end of turn. The result is discarded — and the encoder state restored +// from the snapshot — whenever speech resumes or a full chunk shifts the +// window origin, because the padded frames are only valid while the audio +// after the snapshot stays silent. +void NemotronASRStreamingSession::maybe_speculative_flush() { + if (spec_valid_ || stream_await_first_chunk_ || stream_tail_encoded_ || !stream_decode_active_) { + return; + } + const char * env = std::getenv("NEMOTRON_SPEC_FLUSH"); + if (env != nullptr && *env == '0') { + return; + } + const auto & fc = assets_->config.frontend; + const int64_t total = static_cast(streaming_audio_.samples.size()); + const int64_t flush_samples = flush_window_mel() * fc.hop_length + fc.win_length; + if (stream_next_chunk_start_ + flush_samples <= total) { + return; + } + const char * silence_env = std::getenv("NEMOTRON_SPEC_SILENCE_MS"); + const double silence_s = silence_env != nullptr && *silence_env != 0 ? std::strtod(silence_env, nullptr) / 1000.0 : 0.15; + if (total - spec_last_loud_sample_ < static_cast(silence_s * fc.sample_rate)) { + return; + } + spec_state_snapshot_ = encoder_stream_state_; + std::vector window; + build_flush_window(total, window); + auto features = frontend_.extract_waveform(window, /*center=*/false); + spec_frames_ = encoder_->encode_stream_chunk( + features, + stream_prompt_id_, + stream_lookahead_, + encoder_stream_state_); + spec_audio_mark_ = total; + spec_valid_ = true; +} + +void NemotronASRStreamingSession::discard_speculative_flush() { + if (!spec_valid_) { + return; + } + encoder_stream_state_ = spec_state_snapshot_; + spec_frames_ = NemotronEncodedAudio{}; + spec_valid_ = false; } runtime::StreamEvent NemotronASRStreamingSession::process_audio_chunk(const runtime::AudioChunk & chunk) { @@ -426,13 +719,48 @@ runtime::StreamEvent NemotronASRStreamingSession::process_audio_chunk(const runt if (task_.mode != runtime::RunMode::Streaming) { throw std::runtime_error("Nemotron ASR process_audio_chunk called on non-streaming session"); } + const char * energy_env = std::getenv("NEMOTRON_SPEC_ENERGY"); + const double energy = energy_env != nullptr && *energy_env != 0 ? std::strtod(energy_env, nullptr) : 0.01; + const int64_t scan_from = static_cast(streaming_audio_.samples.size()); runtime::AudioBuffer audio; audio.sample_rate = chunk.sample_rate; audio.channels = chunk.channels; audio.samples = chunk.samples; runtime::append_audio_buffer(streaming_audio_, audio); + const int64_t total = static_cast(streaming_audio_.samples.size()); + // 10 ms windowed RMS: the speech/silence distinction must ignore decay + // transients (a per-sample max counts the quiet tail of a word as speech + // and the speculative flush never fires). + constexpr int64_t kRmsWindow = 160; + for (int64_t w = scan_from - scan_from % kRmsWindow; w + kRmsWindow <= total; w += kRmsWindow) { + double acc = 0.0; + for (int64_t i = w; i < w + kRmsWindow; ++i) { + const float v = streaming_audio_.samples[static_cast(i)]; + acc += double(v) * double(v); + } + if (std::sqrt(acc / double(kRmsWindow)) > energy) { + spec_last_loud_sample_ = w + kRmsWindow; + } + } + runtime::StreamEvent event; event.is_final = false; + std::string delta; + bool encoded_full_chunk = false; + while (encode_and_decode_next_chunk(/*flush_tail=*/false, delta)) { + encoded_full_chunk = true; + } + if (spec_valid_ && (encoded_full_chunk || spec_last_loud_sample_ > spec_audio_mark_)) { + discard_speculative_flush(); + } + maybe_speculative_flush(); + if (!delta.empty()) { + event.partial_text = runtime::Transcript{delta, streaming_language_}; + if (stream_event_sink_) { + stream_event_sink_(event); + return {}; + } + } return event; } @@ -445,25 +773,18 @@ runtime::TaskResult NemotronASRStreamingSession::finalize() { throw std::runtime_error("Nemotron ASR finalize requires streamed audio"); } const auto wall_start = Clock::now(); - runtime::TaskRequest config_request; - config_request.text_input = runtime::Transcript{"", streaming_language_}; - config_request.options = streaming_options_; - const int64_t prompt_id = prompt_id_for_request(config_request); - const int64_t lookahead = lookahead_for_options(streaming_options_); - const auto decode_options = decode_options_for_request(config_request); - const auto decoded = run_streaming_audio( - streaming_audio_, - prompt_id, - lookahead, - decode_options, - [&](const std::string & delta) { - if (!stream_event_sink_ || delta.empty()) { - return; - } - runtime::StreamEvent event; - event.partial_text = runtime::Transcript{delta, streaming_language_}; - stream_event_sink_(event); - }); + std::string delta; + while (encode_and_decode_next_chunk(/*flush_tail=*/true, delta)) { + } + if (!stream_decode_active_) { + throw std::runtime_error("Nemotron ASR streaming request is shorter than the first required chunk"); + } + const auto decoded = decoder_->finish_stream_decode(); + // No fallback: the streaming decode is the result. The greedy RNNT has + // cut-length dead pockets (verified in the f32 reference - hi.wav at a + // 0.57 s cut decodes empty in every streaming shape), and the offline + // re-decode that used to rescue them cost 250-330 ms inside the + // end-of-turn window; see docs/nemotron_streaming_end_of_turn.md. runtime::TaskResult result; result.text_output = runtime::Transcript{decoded.text, streaming_language_}; result.word_timestamps = decoded.token_timestamps; diff --git a/tests/nemotron_asr/nemotron_asr_warm_bench.cpp b/tests/nemotron_asr/nemotron_asr_warm_bench.cpp index c65fd2f11..e6ac6620f 100644 --- a/tests/nemotron_asr/nemotron_asr_warm_bench.cpp +++ b/tests/nemotron_asr/nemotron_asr_warm_bench.cpp @@ -115,9 +115,14 @@ int main(int argc, char ** argv) { const std::filesystem::path timing_path = engine::tools::arg_value(argc, argv, "--timing-file", "/tmp/nemotron_asr_warm_bench_timing.log"); - setenv("MINITTS_TRACE_ENABLED", "0", 1); - setenv("MINITTS_TIMING_ENABLED", "1", 1); - setenv("MINITTS_TIMING_FILE", timing_path.c_str(), 1); +#ifdef _WIN32 +#define setenv_platform(name, value, overwrite) _putenv_s(name, value) +#else +#define setenv_platform(name, value, overwrite) setenv(name, value, overwrite) +#endif + setenv_platform("MINITTS_TRACE_ENABLED", "0", 1); + setenv_platform("MINITTS_TIMING_ENABLED", "1", 1); + setenv_platform("MINITTS_TIMING_FILE", timing_path.string().c_str(), 1); engine::debug::configure_logging(engine::debug::LoggingConfig{true, timing_path.string()}); auto registry = engine::runtime::make_default_registry(); @@ -213,6 +218,7 @@ int main(int argc, char ** argv) { for (int i = 0; i < warmup; ++i) { if (streaming_session) { stream_session->reset(); + stream_session->start_stream(warmup_request); stream_session->process_audio_chunk({ warmup_audio.sample_rate, warmup_audio.channels, @@ -242,6 +248,14 @@ int main(int argc, char ** argv) { if (streaming_session) { stream_session->reset(); const auto & audio = request_audio_buffers[request_index]; + engine::runtime::TaskRequest stream_request; + stream_request.audio_input = audio; + stream_request.text_input = engine::runtime::Transcript{"", language}; + stream_request.options["lookahead_tokens"] = request_lookahead; + stream_request.options["max_tokens"] = max_tokens; + stream_request.options["streaming"] = request_streaming; + stream_request.options["keep_language_tags"] = request_keep_language_tags; + stream_session->start_stream(stream_request); stream_session->process_audio_chunk({ audio.sample_rate, audio.channels,