enne2/qwentts.cpp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

codec: stateful frame by frame streaming decode on a static graph

Browse Source 
Every causal conv carries its left context in a persistent backend
tensor and every transposed conv its overlap tail, so a T=1 frame
decode reproduces the offline full decode exactly with zero re decoded
context. The tokenizer transformer attends over a sliding window KV
ring written through set_rows. The frame graph builds and allocates
once, then every frame is input uploads, one direct backend compute,
and one readback. The quantizer conts each codebook id view so the
Vulkan get_rows path accepts the direct compute. Each generated frame
emits its samples immediately and ICL priming feeds the full reference
through the same state. The buffered path keeps the chunked decode and
both codec framing knobs now apply to it alone.
This commit is contained in:
Pascal
2026-07-05 10:25:21 +02:00
parent bcac46352e
commit 2700d4c746
9 changed files with 712 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/causal-trans-conv.h
+89
View File
@@ -200,3 +200,92 @@ static struct ggml_tensor * qwen_causal_conv1d(struct ggml_context * ctx,
}
return y;
}
// Streaming causal Conv1d, stride 1. The offline zero left pad is
// replaced by a persistent state tensor carrying the last (k-1)*d input
// rows across calls: the graph concats the state ahead of the fresh
// rows, runs a pad free conv, and refreshes the state in graph with the
// tail of the extended input. The state is [(k-1)*d, IC] f32, backend
// resident and zero cleared at stream reset, so the first call matches
// the offline zero pad bit for bit. The state write depends on the
// concat output, so it always executes after the read.
// w: [k, IC, OC] f32, x: [T, IC] f32 T-first
// Returns [T, OC] f32 T-first.
static struct ggml_tensor * qwen_causal_conv1d_stream(struct ggml_context * ctx,
struct ggml_cgraph * gf,
struct ggml_tensor * w,
struct ggml_tensor * b,
struct ggml_tensor * x,
int k,
int d,
struct ggml_tensor * state) {
int OC = (int) w->ne[2];
int L = (k - 1) * d;
struct ggml_tensor * x_ext = ggml_concat(ctx, state, x, 0); // [L + T, IC]
// State refresh: the last L rows of x_ext feed the next call.
struct ggml_tensor * tail =
ggml_view_2d(ctx, x_ext, L, x_ext->ne[1], x_ext->nb[1], (size_t) (x_ext->ne[0] - L) * x_ext->nb[0]);
ggml_build_forward_expand(gf, ggml_cpy(ctx, tail, state));
struct ggml_tensor * y = ggml_reshape_3d(ctx, x_ext, x_ext->ne[0], x_ext->ne[1], 1);
y = ggml_conv_1d(ctx, w, y, 1, 0, d);
y = ggml_reshape_2d(ctx, y, y->ne[0], y->ne[1]);
if (b) {
struct ggml_tensor * b2d = ggml_reshape_2d(ctx, b, 1, OC);
y = ggml_add(ctx, y, b2d);
}
return y;
}
// Streaming causal ConvTranspose1d. The raw col2im output spans
// (T-1)*stride + K rows; the offline path right trims K - stride of
// them, the streaming path instead carries that tail into the next
// call: the persistent carry [K - stride, OC] adds onto the head of the
// raw output and refreshes with the raw tail, bias free (the bias
// applies once, on the emitted rows). The carry consumer expands before
// the carry write so the read always precedes the overwrite.
// w_perm: [IC, K*OC] f32 from qwen_load_ctw_f32, x: [T, IC] f32 T-first
// Returns [T*stride, OC] f32 T-first.
static struct ggml_tensor * qwen_causal_trans_conv1d_stream(struct ggml_context * ctx,
struct ggml_cgraph * gf,
struct ggml_tensor * w_perm,
struct ggml_tensor * b,
struct ggml_tensor * x,
int stride,
int kernel,
int oc,
struct ggml_tensor * carry) {
int T = (int) x->ne[0];
int trim = kernel - stride;
int emit = T * stride;
struct ggml_tensor * xt = ggml_cont(ctx, ggml_transpose(ctx, x));
struct ggml_tensor * col = ggml_mul_mat(ctx, w_perm, xt);
// Raw scatter [(T-1)*stride + K, OC] = [emit + trim, OC]
struct ggml_tensor * raw = ggml_col2im_1d(ctx, col, stride, oc, 0);
// Head rows [0, trim) receive the previous call's tail.
struct ggml_tensor * head = ggml_view_2d(ctx, raw, trim, raw->ne[1], raw->nb[1], 0);
struct ggml_tensor * y = ggml_add(ctx, head, carry);
if (emit > trim) {
struct ggml_tensor * mid =
ggml_view_2d(ctx, raw, emit - trim, raw->ne[1], raw->nb[1], (size_t) trim * raw->nb[0]);
y = ggml_concat(ctx, y, mid, 0);
}
ggml_build_forward_expand(gf, y);
// Carry refresh with the raw tail [emit, emit + trim), expanded
// after the head sum so the carry read wins the ordering.
struct ggml_tensor * tail = ggml_view_2d(ctx, raw, trim, raw->ne[1], raw->nb[1], (size_t) emit * raw->nb[0]);
ggml_build_forward_expand(gf, ggml_cpy(ctx, tail, carry));
if (b) {
struct ggml_tensor * b2d = ggml_reshape_2d(ctx, b, 1, oc);
y = ggml_add(ctx, y, b2d);
}
return y;
}
src/codec-chunked-decode.h
+41 -87
View File
@@ -20,7 +20,7 @@
// fits in a single chunk_frames sized window. Bounds VRAM beyond
// that, mirrors the upstream chunked_decode loop frame for frame.
//
// codec_chunked_decoder_stream : rolling state for AR streaming.
// codec_stream_decoder : stateful frame by frame AR streaming.
// The pipeline pushes one frame at a time as the talker produces
// them ; push_frame decodes and emits a fresh chunk_frames sized
// audio block through the on_chunk callback as soon as enough new
@@ -84,106 +84,60 @@ static inline std::vector<float> codec_chunked_decode(PipelineCodec * pc,
return out;
}
// Rolling streaming decoder. Stores codes K major as K parallel vectors
// (by_k[k][t]) so emit_one can memcpy a contiguous K major slice into
// pipeline_codec_decode without a transpose. push_frame triggers as
// many emits as possible after appending one frame ; flush emits the
// tail at EOS.
struct codec_chunked_decoder_stream {
std::vector<std::vector<int32_t>> by_k;
int K;
int T_so_far;
int chunk_frames;
int left_ctx_frames;
int emit_start_frame;
// Set true when an emit returned false because the on_chunk callback
// requested a cancel. Stays false on decode failures so the caller
// can route to QT_STATUS_CANCELLED vs QT_STATUS_GENERATE_FAILED on
// a push_frame / flush negative return.
bool cancelled;
// Stateful streaming decoder over pipeline_codec_decode_stream: every
// pushed frame decodes immediately through the persistent codec state
// and emits its TOKENIZER_HOP_LENGTH samples, no buffering, no left
// context re-decode, no tail to drain at EOS. ICL priming feeds the
// full reference through the same state with the audio discarded,
// which matches the upstream reference plus generated decode exactly.
struct codec_stream_decoder {
int K;
// Set true when push_frame returned false because the on_chunk
// callback requested a cancel. Stays false on decode failures so
// the caller can route to QT_STATUS_CANCELLED vs
// QT_STATUS_GENERATE_FAILED on a negative return.
bool cancelled;
void init(int K_, int chunk_frames_, int left_ctx_frames_) {
K = K_;
T_so_far = 0;
chunk_frames = chunk_frames_ < 1 ? 1 : chunk_frames_;
left_ctx_frames = left_ctx_frames_ < 0 ? 0 : left_ctx_frames_;
emit_start_frame = 0;
cancelled = false;
by_k.assign((size_t) K, {});
std::vector<float> frame;
// Reset the persistent codec state to the zero context. Returns
// false when the state allocation fails.
bool init(PipelineCodec * pc, int K_) {
K = K_;
cancelled = false;
frame.assign((size_t) TOKENIZER_HOP_LENGTH, 0.0f);
return pipeline_codec_stream_reset(pc);
}
// Seed the left context with the tail of the ICL reference codes so
// the first emitted chunk draws causal context from the reference
// instead of an empty decoder state, matching the upstream pipeline
// which decodes reference plus generated then trims. ref_kt is K
// major [K, ref_T]; the last min(ref_T, left_ctx_frames) frames are
// kept. Call once, after init and before any push_frame; the seeded
// frames sit below emit_start_frame so they are never emitted.
void seed_reference(const int32_t * ref_kt, int ref_T) {
int seed = ref_T < left_ctx_frames ? ref_T : left_ctx_frames;
if (seed <= 0) {
return;
}
for (int k = 0; k < K; k++) {
const int32_t * row = ref_kt + (size_t) k * (size_t) ref_T + (size_t) (ref_T - seed);
by_k[(size_t) k].insert(by_k[(size_t) k].end(), row, row + seed);
}
T_so_far = seed;
emit_start_frame = seed;
}
// Append one frame (K int32 codes, one per codebook). Drain any
// chunks that became emittable. Returns false on decode failure or
// when cb returns false (cancellation).
bool push_frame(PipelineCodec * pc, const int32_t * frame_codes, qt_audio_chunk_cb cb, void * cb_ud) {
for (int k = 0; k < K; k++) {
by_k[(size_t) k].push_back(frame_codes[k]);
}
T_so_far++;
while (T_so_far - emit_start_frame >= chunk_frames) {
if (!emit_one(pc, emit_start_frame + chunk_frames, cb, cb_ud)) {
// Prime the codec state with the full ICL reference: every frame
// runs through the streaming decode with the audio discarded, so
// the first generated frame sees the reference's exact causal
// state. ref_kt is K major [K, ref_T]. Call once, after init and
// before any push_frame.
bool seed_reference(PipelineCodec * pc, const int32_t * ref_kt, int ref_T) {
std::vector<int32_t> codes((size_t) K);
for (int t = 0; t < ref_T; t++) {
for (int k = 0; k < K; k++) {
codes[(size_t) k] = ref_kt[(size_t) k * (size_t) ref_T + (size_t) t];
}
if (!pipeline_codec_decode_stream(pc, codes.data(), NULL)) {
return false;
}
}
return true;
}
// Drain the tail. If frames remain past emit_start_frame, decode
// them with left context and emit one final short chunk. Idempotent
// on empty tail.
bool flush(PipelineCodec * pc, qt_audio_chunk_cb cb, void * cb_ud) {
if (T_so_far > emit_start_frame) {
return emit_one(pc, T_so_far, cb, cb_ud);
}
return true;
}
private:
// Decode [emit_start_frame - ctx .. end_frame] with left context
// stripped from the emitted samples, then advance emit_start_frame.
bool emit_one(PipelineCodec * pc, int end_frame, qt_audio_chunk_cb cb, void * cb_ud) {
int ctx = (emit_start_frame - left_ctx_frames > 0) ? left_ctx_frames : emit_start_frame;
int slice_start = emit_start_frame - ctx;
int slice_T = end_frame - slice_start;
std::vector<int32_t> slice((size_t) K * (size_t) slice_T);
for (int k = 0; k < K; k++) {
std::memcpy(slice.data() + (size_t) k * (size_t) slice_T, by_k[(size_t) k].data() + (size_t) slice_start,
(size_t) slice_T * sizeof(int32_t));
}
std::vector<float> wav = pipeline_codec_decode(pc, slice.data(), K, slice_T);
if (wav.empty()) {
// Decode one frame (K int32 codes, one per codebook) and emit its
// samples through the callback. Returns false on decode failure or
// when cb returns false (cancellation).
bool push_frame(PipelineCodec * pc, const int32_t * frame_codes, qt_audio_chunk_cb cb, void * cb_ud) {
if (!pipeline_codec_decode_stream(pc, frame_codes, frame.data())) {
return false;
}
const size_t drop = (size_t) ctx * (size_t) TOKENIZER_HOP_LENGTH;
const float * emit_first = wav.data() + drop;
int emit_n = (int) (wav.size() - drop);
if (emit_n > 0 && !cb(emit_first, emit_n, cb_ud)) {
if (!cb(frame.data(), TOKENIZER_HOP_LENGTH, cb_ud)) {
cancelled = true;
return false;
}
emit_start_frame = end_frame;
return true;
}
};
src/convnext-block.h
+75
View File
@@ -189,3 +189,78 @@ static struct ggml_tensor * upsample_stage_forward(struct ggml_context * ctx
}
return x;
}
// Streaming state for the upsample stage: one depthwise conv left
// context per ConvNeXt block, [dwconv_kernel - 1, C] f32 at the block's
// own rate. The transposed convs have kernel == stride, so they carry
// nothing and run the offline helper unchanged.
struct QwenUpsampleStreamState {
struct ggml_tensor * dw[UPSAMPLE_MAX_BLOCKS];
};
// Streaming ConvNeXt block: the depthwise causal conv reads its left
// context from the persistent state instead of a zero pad. Everything
// else is pointwise and stateless.
static struct ggml_tensor * convnext_block_forward_stream(struct ggml_context * ctx,
struct ggml_cgraph * gf,
const QwenConvNeXtBlock & block,
struct ggml_tensor * x,
int kernel,
struct ggml_tensor * dw_state) {
int C = (int) x->ne[1];
struct ggml_tensor * residual = x;
// dwconv: concat the state ahead of the fresh rows, refresh it with
// the tail, run the depthwise conv pad free.
struct ggml_tensor * x_ext = ggml_concat(ctx, dw_state, x, 0); // [k-1 + T, C]
struct ggml_tensor * tail =
ggml_view_2d(ctx, x_ext, kernel - 1, C, x_ext->nb[1], (size_t) (x_ext->ne[0] - (kernel - 1)) * x_ext->nb[0]);
ggml_build_forward_expand(gf, ggml_cpy(ctx, tail, dw_state));
struct ggml_tensor * y = ggml_reshape_3d(ctx, x_ext, x_ext->ne[0], C, 1);
y = ggml_conv_1d_dw(ctx, block.dwconv_w, y, 1, 0, 1); // [T, C, 1]
y = ggml_reshape_2d(ctx, y, y->ne[0], C);
if (block.dwconv_b) {
struct ggml_tensor * b2d = ggml_reshape_2d(ctx, block.dwconv_b, 1, C);
y = ggml_add(ctx, y, b2d);
}
// LayerNorm wants the channel dim on ne[0]: transpose to [C, T].
y = ggml_cont(ctx, ggml_transpose(ctx, y));
y = ggml_norm(ctx, y, 1e-6f);
y = ggml_mul(ctx, y, block.norm_w);
y = ggml_add(ctx, y, block.norm_b);
y = ggml_mul_mat(ctx, block.pwconv1_w, y);
y = ggml_add(ctx, y, block.pwconv1_b);
y = ggml_gelu(ctx, y);
y = ggml_mul_mat(ctx, block.pwconv2_w, y);
y = ggml_add(ctx, y, block.pwconv2_b);
y = ggml_mul(ctx, y, block.gamma);
y = ggml_cont(ctx, ggml_transpose(ctx, y));
y = ggml_add(ctx, y, residual);
return y;
}
// Streaming upsample stage: transposed convs run the offline helper
// (kernel == stride, exact under chunking), ConvNeXt blocks thread
// their depthwise state.
static struct ggml_tensor * upsample_stage_forward_stream(struct ggml_context * ctx,
struct ggml_cgraph * gf,
const QwenUpsampleStage * stage,
struct ggml_tensor * x,
const QwenUpsampleStreamState * st) {
int kernel = stage->upsample_ratio;
for (int i = 0; i < stage->num_blocks; i++) {
x = qwen_causal_trans_conv1d(ctx, stage->transconv_w[i], stage->transconv_b[i], x, stage->upsample_ratio,
kernel, stage->channels);
x = convnext_block_forward_stream(ctx, gf, stage->convnext[i], x, stage->dwconv_kernel, st->dw[i]);
}
return x;
}
src/dac-decoder-v2.h
+51
View File
@@ -261,3 +261,54 @@ static struct ggml_tensor * dac_decoder_forward(struct ggml_context * ctx,
x = qwen_causal_conv1d(ctx, d->conv_post_w, d->conv_post_b, x, 7, 1);
return x;
}
// Streaming state for the DAC decoder: one left context per stride 1
// causal conv ([(k-1)*d, IC] f32 at the conv's own rate) and one
// overlap carry per transposed conv ([kernel - stride, out_ch] f32,
// bias free). The k=1 res unit convs are pointwise and stateless.
struct QwenDACStreamState {
struct ggml_tensor * pre; // conv_pre k=7, [6, 1024]
struct ggml_tensor * carry[DAC_NUM_BLOCKS]; // transconv tails, [stride, out_ch]
struct ggml_tensor * ru[DAC_NUM_BLOCKS][DAC_RES_UNITS]; // conv1 k=7 dilated, [6*d, ch]
struct ggml_tensor * post; // conv_post k=7, [6, 96]
};
// Streaming residual unit: conv1 reads its left context from the
// persistent state, conv2 is pointwise.
static struct ggml_tensor * dac_res_unit_stream(struct ggml_context * ctx,
struct ggml_cgraph * gf,
const QwenDACResUnit * ru,
struct ggml_tensor * x,
struct ggml_tensor * state) {
struct ggml_tensor * skip = x;
x = dac_snake(ctx, x, ru->act1);
x = qwen_causal_conv1d_stream(ctx, gf, ru->c1w, ru->c1b, x, 7, ru->dilation, state);
x = dac_snake(ctx, x, ru->act2);
x = qwen_causal_conv1d(ctx, ru->c2w, ru->c2b, x, 1, 1);
return ggml_add(ctx, skip, x);
}
// Streaming DAC forward graph: same stack as dac_decoder_forward with
// every stateful op threaded through the persistent stream state.
// x: [T, 1024] f32 T-first
// returns [T * 1920, 1] f32 T-first, unclamped.
static struct ggml_tensor * dac_decoder_forward_stream(struct ggml_context * ctx,
struct ggml_cgraph * gf,
const QwenDACDecoder * d,
struct ggml_tensor * x,
const QwenDACStreamState * st) {
x = qwen_causal_conv1d_stream(ctx, gf, d->conv_pre_w, d->conv_pre_b, x, 7, 1, st->pre);
for (int i = 0; i < DAC_NUM_BLOCKS; i++) {
const QwenDACBlock & b = d->blk[i];
x = dac_snake(ctx, x, b.snake1);
x = qwen_causal_trans_conv1d_stream(ctx, gf, b.tcw, b.tcb, x, b.stride, b.kernel, b.out_ch, st->carry[i]);
for (int r = 0; r < DAC_RES_UNITS; r++) {
x = dac_res_unit_stream(ctx, gf, &b.ru[r], x, st->ru[i][r]);
}
}
x = dac_snake(ctx, x, d->snake_post);
x = qwen_causal_conv1d_stream(ctx, gf, d->conv_post_w, d->conv_post_b, x, 7, 1, st->post);
return x;
}
src/pipeline-codec.cpp
+224 -3
View File
@@ -21,9 +21,16 @@
#include <vector>
bool pipeline_codec_load(PipelineCodec * pc, const char * gguf_path, BackendPair bp) {
pc->bp = bp;
pc->backend = bp.backend;
pc->qenc_host_ready = false;
pc->bp = bp;
pc->backend = bp.backend;
pc->qenc_host_ready = false;
pc->stream_ready = false;
pc->stream_ctx = NULL;
pc->stream_buf = NULL;
pc->stream_pos = 0;
pc->stream_graph_ctx = NULL;
pc->stream_gf = NULL;
pc->stream_galloc = NULL;
if (!gf_load(&pc->gguf, gguf_path)) {
qt_log(QT_LOG_ERROR, "[Pipeline] failed to load %s", gguf_path);
@@ -178,6 +185,207 @@ std::vector<float> pipeline_codec_decode(PipelineCodec * pc, const int32_t * cod
return audio;
}
// Ring width of the streaming transformer KV cache: covers the 72 frame
// sliding window plus the fresh frame with headroom, constant so the
// step graph shape never changes.
static const int CODEC_STREAM_RING = 128;
// Allocate the streaming state on first use: one tensor per causal conv
// left context and per transposed conv carry, plus the transformer KV
// ring and the dedicated graph arena. Idempotent.
static bool pipeline_codec_stream_ensure(PipelineCodec * pc) {
if (pc->stream_ready) {
return true;
}
const int n_state = 2 + UPSAMPLE_MAX_BLOCKS + DAC_NUM_BLOCKS * (1 + DAC_RES_UNITS) + 4;
struct ggml_init_params gp = { ggml_tensor_overhead() * (size_t) n_state, NULL, true };
pc->stream_ctx = ggml_init(gp);
if (!pc->stream_ctx) {
qt_log(QT_LOG_ERROR, "[Pipeline] stream state ggml_init failed");
return false;
}
char name[64];
auto tensor2d = [&](int t, int c, const char * n) {
struct ggml_tensor * x = ggml_new_tensor_2d(pc->stream_ctx, GGML_TYPE_F32, t, c);
ggml_set_name(x, n);
return x;
};
// pre_conv k=3 over the 512 wide quantizer latents
pc->stream_pre_conv = tensor2d(2, 512, "stream_pre_conv");
// upsample ConvNeXt depthwise contexts, k=7 over the stage width
for (int i = 0; i < pc->upsample.num_blocks; i++) {
snprintf(name, sizeof(name), "stream_up_dw_%d", i);
pc->stream_up.dw[i] = tensor2d(pc->upsample.dwconv_kernel - 1, pc->upsample.channels, name);
}
// DAC contexts: conv_pre, per block transconv carry + res unit conv1, conv_post.
// conv_pre reads the 1024 wide upsample output.
pc->stream_dac.pre = tensor2d(6, 1024, "stream_dac_pre");
for (int i = 0; i < DAC_NUM_BLOCKS; i++) {
const QwenDACBlock & b = pc->dac.blk[i];
snprintf(name, sizeof(name), "stream_dac_carry_%d", i);
pc->stream_dac.carry[i] = tensor2d(b.kernel - b.stride, b.out_ch, name);
for (int r = 0; r < DAC_RES_UNITS; r++) {
snprintf(name, sizeof(name), "stream_dac_ru_%d_%d", i, r);
pc->stream_dac.ru[i][r] = tensor2d(6 * b.ru[r].dilation, b.out_ch, name);
}
}
pc->stream_dac.post = tensor2d(6, pc->dac.channels[DAC_NUM_BLOCKS], "stream_dac_post");
pc->stream_buf = ggml_backend_alloc_ctx_tensors(pc->stream_ctx, pc->backend);
if (!pc->stream_buf) {
qt_log(QT_LOG_ERROR, "[Pipeline] stream state backend allocation failed");
ggml_free(pc->stream_ctx);
pc->stream_ctx = NULL;
return false;
}
if (!kv_cache_init(&pc->stream_kv, pc->transformer.num_layers, pc->transformer.num_kv_heads,
pc->transformer.head_dim, CODEC_STREAM_RING, pc->backend)) {
ggml_backend_buffer_free(pc->stream_buf);
pc->stream_buf = NULL;
ggml_free(pc->stream_ctx);
pc->stream_ctx = NULL;
return false;
}
// Build the static T=1 frame graph once: fixed topology, fixed
// tensor addresses, computed directly on the backend every frame.
{
const int max_nodes = 4096;
struct ggml_init_params gp2 = {
ggml_tensor_overhead() * (size_t) max_nodes + ggml_graph_overhead_custom(max_nodes, false),
NULL,
true,
};
pc->stream_graph_ctx = ggml_init(gp2);
if (!pc->stream_graph_ctx) {
qt_log(QT_LOG_ERROR, "[Pipeline] stream graph ggml_init failed");
kv_cache_free(&pc->stream_kv);
ggml_backend_buffer_free(pc->stream_buf);
pc->stream_buf = NULL;
ggml_free(pc->stream_ctx);
pc->stream_ctx = NULL;
return false;
}
struct ggml_context * gctx = pc->stream_graph_ctx;
struct ggml_cgraph * gf = ggml_new_graph_custom(gctx, max_nodes, false);
const int ring = CODEC_STREAM_RING;
const int K = TOKENIZER_NUM_CODEBOOKS;
struct ggml_tensor * codes_in = ggml_new_tensor_2d(gctx, GGML_TYPE_I32, 1, K);
struct ggml_tensor * pos_in = ggml_new_tensor_1d(gctx, GGML_TYPE_I32, 1);
struct ggml_tensor * rows_in = ggml_new_tensor_1d(gctx, GGML_TYPE_I64, 1);
struct ggml_tensor * mask_in = ggml_new_tensor_2d(gctx, GGML_TYPE_F32, ring, 1);
ggml_set_name(codes_in, "codes_in");
ggml_set_name(pos_in, "positions");
ggml_set_name(rows_in, "kv_rows");
ggml_set_name(mask_in, "ring_mask");
ggml_set_input(codes_in);
ggml_set_input(pos_in);
ggml_set_input(rows_in);
ggml_set_input(mask_in);
// Same module chain as pipeline_codec_decode with the stateful
// variants threaded through the persistent stream tensors. The
// quantizer uses the alignment safe variant: no scheduler input
// duplication happens on the direct backend compute path.
struct ggml_tensor * h = quant_decode_stream(gctx, &pc->qdec, codes_in); // [512, 1] C-first
h = ggml_cont(gctx, ggml_transpose(gctx, h)); // [1, 512] T-first
h = qwen_causal_conv1d_stream(gctx, gf, pc->pre_conv_w, pc->pre_conv_b, h, 3, 1, pc->stream_pre_conv);
h = ggml_cont(gctx, ggml_transpose(gctx, h)); // [1024, 1] C-first
h = tok_trans_forward_stream(gctx, gf, &pc->transformer, h, pos_in, mask_in, rows_in, &pc->stream_kv);
h = ggml_cont(gctx, ggml_transpose(gctx, h)); // [1, 1024] T-first
h = upsample_stage_forward_stream(gctx, gf, &pc->upsample, h, &pc->stream_up);
h = dac_decoder_forward_stream(gctx, gf, &pc->dac, h, &pc->stream_dac);
h = ggml_clamp(gctx, h, -1.0f, 1.0f);
ggml_set_name(h, "audio_out");
ggml_set_output(h);
ggml_build_forward_expand(gf, h);
pc->stream_galloc = ggml_gallocr_new(ggml_backend_get_default_buffer_type(pc->backend));
if (!pc->stream_galloc || !ggml_gallocr_alloc_graph(pc->stream_galloc, gf)) {
qt_log(QT_LOG_ERROR, "[Pipeline] stream graph allocation failed");
if (pc->stream_galloc) {
ggml_gallocr_free(pc->stream_galloc);
pc->stream_galloc = NULL;
}
ggml_free(pc->stream_graph_ctx);
pc->stream_graph_ctx = NULL;
kv_cache_free(&pc->stream_kv);
ggml_backend_buffer_free(pc->stream_buf);
pc->stream_buf = NULL;
ggml_free(pc->stream_ctx);
pc->stream_ctx = NULL;
return false;
}
pc->stream_gf = gf;
pc->stream_in_codes = codes_in;
pc->stream_in_pos = pos_in;
pc->stream_in_rows = rows_in;
pc->stream_in_mask = mask_in;
pc->stream_out = h;
}
pc->stream_ready = true;
qt_log(QT_LOG_INFO, "[Pipeline] Codec stream state ready: %d conv contexts, KV ring %d", n_state - 4,
CODEC_STREAM_RING);
return true;
}
bool pipeline_codec_stream_reset(PipelineCodec * pc) {
if (!pipeline_codec_stream_ensure(pc)) {
return false;
}
// Zero contexts reproduce the offline zero left pads bit for bit;
// the zeroed KV ring stays hidden behind the sliding window mask.
ggml_backend_buffer_clear(pc->stream_buf, 0);
ggml_backend_buffer_clear(pc->stream_kv.buffer, 0);
kv_cache_reset(&pc->stream_kv);
pc->stream_pos = 0;
return true;
}
bool pipeline_codec_decode_stream(PipelineCodec * pc, const int32_t * codes, float * audio_out) {
const int K = TOKENIZER_NUM_CODEBOOKS;
const int ring = CODEC_STREAM_RING;
ggml_backend_tensor_set(pc->stream_in_codes, codes, 0, (size_t) K * sizeof(int32_t));
int32_t pos = pc->stream_pos;
ggml_backend_tensor_set(pc->stream_in_pos, &pos, 0, sizeof(int32_t));
int64_t row = (int64_t) (pc->stream_pos % ring);
ggml_backend_tensor_set(pc->stream_in_rows, &row, 0, sizeof(int64_t));
std::vector<float> mask_buf;
tok_trans_build_stream_mask(pc->stream_pos, 1, ring, pc->transformer.sliding_window, mask_buf);
ggml_backend_tensor_set(pc->stream_in_mask, mask_buf.data(), 0, mask_buf.size() * sizeof(float));
enum ggml_status st = ggml_backend_graph_compute(pc->backend, pc->stream_gf);
if (st != GGML_STATUS_SUCCESS) {
qt_log(QT_LOG_ERROR, "[Pipeline] stream graph_compute status=%d", (int) st);
return false;
}
if (audio_out) {
ggml_backend_tensor_get(pc->stream_out, audio_out, 0, (size_t) TOKENIZER_HOP_LENGTH * sizeof(float));
}
// The state tensors and the graph allocation persist into the next
// frame.
pc->stream_pos++;
return true;
}
bool pipeline_codec_ensure_encoder(PipelineCodec * pc) {
if (pc->enc_loaded) {
return true;
@@ -430,6 +638,19 @@ void pipeline_codec_free(PipelineCodec * pc) {
pc->sched = NULL;
}
graph_arena_free(&pc->dec_arena);
if (pc->stream_ready) {
ggml_gallocr_free(pc->stream_galloc);
pc->stream_galloc = NULL;
ggml_free(pc->stream_graph_ctx);
pc->stream_graph_ctx = NULL;
pc->stream_gf = NULL;
kv_cache_free(&pc->stream_kv);
ggml_backend_buffer_free(pc->stream_buf);
pc->stream_buf = NULL;
ggml_free(pc->stream_ctx);
pc->stream_ctx = NULL;
pc->stream_ready = false;
}
if (pc->enc_loaded) {
quant_encode_free(&pc->qenc);
enc_down_free(&pc->enc_downsample);
src/pipeline-codec.h
+42
View File
@@ -31,6 +31,7 @@
#include "ggml-backend.h"
#include "gguf-weights.h"
#include "graph-arena.h"
#include "kv-cache.h"
#include "quantizer-decode.h"
#include "quantizer-encode.h"
#include "seanet-encoder.h"
@@ -76,6 +77,34 @@ struct PipelineCodec {
// so constant size streaming slices replay a captured executable.
GraphArena dec_arena;
// Stateful streaming decoder: every causal conv left context, every
// transposed conv overlap carry, and the transformer sliding window
// KV ring live as backend resident tensors, so a T=1 frame decode
// reproduces the offline full decode exactly with zero re-decoded
// context. Loaded lazily on the first pipeline_codec_stream_reset:
// the buffered chunked path never pays for it.
bool stream_ready;
struct ggml_context * stream_ctx;
ggml_backend_buffer_t stream_buf;
struct ggml_tensor * stream_pre_conv; // pre_conv k=3, [2, 512]
QwenUpsampleStreamState stream_up;
QwenDACStreamState stream_dac;
KVCache stream_kv; // tok transformer ring, [hd, ring, n_kv] per layer
int stream_pos; // absolute frame position, drives RoPE and ring slots
// Static frame graph: the T=1 topology and every tensor address are
// constant, so the graph builds and allocates once and every frame
// is input uploads + one backend compute + one readback. The single
// backend runs the whole graph, no scheduler involved.
struct ggml_context * stream_graph_ctx;
struct ggml_cgraph * stream_gf;
ggml_gallocr_t stream_galloc;
struct ggml_tensor * stream_in_codes;
struct ggml_tensor * stream_in_pos;
struct ggml_tensor * stream_in_rows;
struct ggml_tensor * stream_in_mask;
struct ggml_tensor * stream_out;
// CPU mirror of the RVQ encode side, lazy-loaded on first encode call.
QwenQuantizerEncodeHost qenc_sem_host;
QwenQuantizerEncodeHost qenc_aco_host;
@@ -99,6 +128,19 @@ bool pipeline_codec_ensure_encoder(PipelineCodec * pc);
// Returns audio of length T * TOKENIZER_HOP_LENGTH, empty on failure.
std::vector<float> pipeline_codec_decode(PipelineCodec * pc, const int32_t * codes, int K, int T);
// Reset the stateful streaming decoder to the zero context: allocates
// the state tensors on first call, clears every conv left context,
// transposed conv carry, and the transformer KV ring, and rewinds the
// absolute position. Call once before each streamed utterance.
bool pipeline_codec_stream_reset(PipelineCodec * pc);
// Decode one frame through the stateful streaming path. codes holds the
// K codebook entries of a single frame; the persistent state advances
// as a side effect. When audio_out is non NULL the frame's
// TOKENIZER_HOP_LENGTH samples copy into it; a NULL audio_out primes
// the state without a readback (ICL reference priming).
bool pipeline_codec_decode_stream(PipelineCodec * pc, const int32_t * codes, float * audio_out);
// Encode a 24 kHz mono waveform into RVQ codes.
// audio : [n_samples] f32 mono 24 kHz. Must be a multiple of
// TOKENIZER_HOP_LENGTH (1920); the caller is expected
src/pipeline-tts.cpp
+26 -31
View File
@@ -609,13 +609,13 @@ qt_status pipeline_tts_synthesize(PipelineTTS * pt,
const float subtk_T = params->subtalker_do_sample ? params->subtalker_temperature : 0.0f;
const float talker_rp = params->repetition_penalty;
// Codec decode framing. Both the streaming path and the buffered
// path route through codec_chunked_decode, with a rolling left
// context window that mirrors the upstream Qwen3-TTS 12 Hz tokenizer
// chunked_decode rule : every chunk re uses up to left_ctx_frames
// previously decoded frames as left context, then the matching audio
// samples are stripped from the head of the decoded chunk. The first
// chunk has its left context collapsed to whatever is available.
// Codec decode framing. The buffered path routes through
// codec_chunked_decode with a rolling left context window mirroring
// the upstream Qwen3-TTS 12 Hz tokenizer chunked_decode rule: every
// chunk re uses up to left_ctx_frames previously decoded frames as
// left context, then the matching audio samples are stripped from
// the head of the decoded chunk. The streaming path decodes frame
// by frame through the stateful codec and ignores both knobs.
const bool streaming = (params->on_chunk != NULL);
const float chunk_sec = params->codec_chunk_sec > 0.0f ? params->codec_chunk_sec : 24.0f;
const float left_ctx_sec = params->codec_left_context_sec >= 0.0f ? params->codec_left_context_sec : 2.0f;
@@ -639,17 +639,24 @@ qt_status pipeline_tts_synthesize(PipelineTTS * pt,
std::vector<int32_t> prev_ids((size_t) num_codebooks, 0);
const float * prev_overlay = NULL;
// Streaming rolling decoder. Holds the K major codes buffer, the
// emit cursor and the left context window. push_frame triggers an
// emit as soon as chunk_frames new frames have accumulated since
// the previous emit boundary ; flush drains the tail at EOS.
codec_chunked_decoder_stream stream;
// Stateful streaming decoder: every generated frame decodes
// immediately through the persistent codec state and emits its
// samples, so the first audio callback fires with the first frame.
// ICL clone priming runs the full reference through the same state,
// matching the upstream reference plus generated decode exactly.
codec_stream_decoder stream;
if (streaming) {
stream.init(num_codebooks, chunk_frames, left_ctx_frames);
// ICL clone: the reference tail seeds the decoder left context
// so the onset is voiced with the reference's causal state.
if (!stream.init(&pt->codec, num_codebooks)) {
qt_set_error("pipeline_tts_synthesize: codec stream state init failed");
return QT_STATUS_GENERATE_FAILED;
}
if (ref_codes_ptr != NULL) {
stream.seed_reference(ref_codes_ptr, ref_codes_T);
Timer t_seed;
if (!stream.seed_reference(&pt->codec, ref_codes_ptr, ref_codes_T)) {
qt_set_error("pipeline_tts_synthesize: codec stream reference priming failed");
return QT_STATUS_GENERATE_FAILED;
}
perf.codec_ms += t_seed.ms();
}
}
@@ -819,22 +826,10 @@ qt_status pipeline_tts_synthesize(PipelineTTS * pt,
debug_dump_i32_as_f32(&d, "codes-full", flat.data(), shape, 2);
}
// Streaming tail: flush remaining frames through the callback. The
// buffered output stays empty in this branch; the caller already
// received every sample through on_chunk.
// Streaming tail: nothing to drain, every frame already emitted at
// generation time through the stateful decoder. The buffered output
// stays empty in this branch.
if (streaming) {
Timer t_flush;
bool flushed = stream.flush(&pt->codec, params->on_chunk, params->on_chunk_user_data);
perf.codec_ms += t_flush.ms();
if (!flushed) {
if (stream.cancelled) {
qt_log(QT_LOG_INFO, "[Pipeline] on_chunk callback aborted the synthesis on tail flush");
return QT_STATUS_CANCELLED;
}
qt_set_error("pipeline_tts_synthesize: streaming codec decode failed on tail flush");
qt_log(QT_LOG_ERROR, "[Pipeline] streaming codec decode failed on tail flush");
return QT_STATUS_GENERATE_FAILED;
}
out->samples = NULL;
out->n_samples = 0;
out->sample_rate = TOKENIZER_SAMPLE_RATE;
src/quantizer-decode.h
+33
View File
@@ -164,3 +164,36 @@ static struct ggml_tensor * quant_decode(struct ggml_context * ctx,
return ggml_add(ctx, h_sem, h_aco);
}
// Streaming variant for the static frame graph computed directly on the
// backend, without the scheduler's per view input duplication: every
// codebook id passes through a ggml_cont so the get_rows sources land
// on allocator aligned tensors, which the Vulkan get_rows path
// requires. Same math and structure as quant_decode.
static struct ggml_tensor * rvq_group_decode_stream(struct ggml_context * ctx,
const QwenRVQGroup & g,
struct ggml_tensor * codes_split,
int T) {
struct ggml_tensor * sum = NULL;
for (int k = 0; k < g.num_codebooks; k++) {
struct ggml_tensor * idx = ggml_cont(ctx, ggml_view_1d(ctx, codes_split, T, (size_t) k * codes_split->nb[1]));
struct ggml_tensor * emb = ggml_get_rows(ctx, g.embed[(size_t) k], idx);
sum = (sum == NULL) ? emb : ggml_add(ctx, sum, emb);
}
return ggml_mul_mat(ctx, g.out_proj_w, sum);
}
static struct ggml_tensor * quant_decode_stream(struct ggml_context * ctx,
const QwenQuantizerDecoder * dec,
struct ggml_tensor * codes) {
int T = (int) codes->ne[0];
struct ggml_tensor * codes_sem = ggml_view_2d(ctx, codes, T, dec->num_semantic_quantizers, codes->nb[1], 0);
size_t aco_off = (size_t) dec->num_semantic_quantizers * codes->nb[1];
struct ggml_tensor * codes_aco = ggml_view_2d(ctx, codes, T, dec->num_acoustic_quantizers, codes->nb[1], aco_off);
struct ggml_tensor * h_sem = rvq_group_decode_stream(ctx, dec->semantic, codes_sem, T);
struct ggml_tensor * h_aco = rvq_group_decode_stream(ctx, dec->acoustic, codes_aco, T);
return ggml_add(ctx, h_sem, h_aco);
}
src/tokenizer-transformer.h
+131
View File
@@ -13,6 +13,7 @@
#include "ggml-backend.h"
#include "ggml.h"
#include "gguf-weights.h"
#include "kv-cache.h"
#include "weight-ctx.h"
#include <cmath>
@@ -291,3 +292,133 @@ static struct ggml_tensor * tok_trans_forward(struct ggml_context * c
return h;
}
// Streaming layer forward: the fresh K and V rows write into a
// persistent ring cache via set_rows at the slots carried by kv_rows,
// and the attention reads the whole ring with the mask killing every
// slot outside the sliding window. Ring slots hold RoPE rotated keys at
// absolute positions, so relative attention falls out as usual. The
// graph topology is constant across steps: pure CUDA graph replay.
static struct ggml_tensor * tok_trans_layer_forward_stream(struct ggml_context * ctx,
struct ggml_cgraph * gf,
const QwenTokenizerTransformer * tr,
const QwenTransformerLayer & layer,
struct ggml_tensor * x,
struct ggml_tensor * positions,
struct ggml_tensor * mask,
struct ggml_tensor * kv_rows,
struct ggml_tensor * k_cache,
struct ggml_tensor * v_cache,
int T,
int ring) {
int n_q_heads = tr->num_attention_heads;
int n_kv = tr->num_kv_heads;
int hd = tr->head_dim;
struct ggml_tensor * ln1 = ggml_rms_norm(ctx, x, tr->rms_norm_eps);
ln1 = ggml_mul(ctx, ln1, layer.input_norm_w);
struct ggml_tensor * q = ggml_mul_mat(ctx, layer.attn.q_proj_w, ln1);
struct ggml_tensor * k = ggml_mul_mat(ctx, layer.attn.k_proj_w, ln1);
struct ggml_tensor * v = ggml_mul_mat(ctx, layer.attn.v_proj_w, ln1);
q = ggml_reshape_3d(ctx, q, hd, n_q_heads, T);
k = ggml_reshape_3d(ctx, k, hd, n_kv, T);
v = ggml_reshape_3d(ctx, v, hd, n_kv, T);
q = ggml_rope_ext(ctx, q, positions, NULL, hd, GGML_ROPE_TYPE_NEOX, 0, tr->rope_theta, 1.0f, 0.0f, 1.0f, 0.0f,
0.0f);
k = ggml_rope_ext(ctx, k, positions, NULL, hd, GGML_ROPE_TYPE_NEOX, 0, tr->rope_theta, 1.0f, 0.0f, 1.0f, 0.0f,
0.0f);
// Ring write: [hd, T, n_kv] rows land at kv_rows, ids broadcast
// across the head dim.
struct ggml_tensor * k_perm = ggml_cont(ctx, ggml_permute(ctx, k, 0, 2, 1, 3));
struct ggml_tensor * v_perm = ggml_cont(ctx, ggml_permute(ctx, v, 0, 2, 1, 3));
ggml_build_forward_expand(gf, ggml_set_rows(ctx, k_cache, k_perm, kv_rows));
ggml_build_forward_expand(gf, ggml_set_rows(ctx, v_cache, v_perm, kv_rows));
struct ggml_tensor * q_p = ggml_cont(ctx, ggml_permute(ctx, q, 0, 2, 1, 3));
struct ggml_tensor * k_full = ggml_view_3d(ctx, k_cache, hd, ring, n_kv, k_cache->nb[1], k_cache->nb[2], 0);
struct ggml_tensor * v_full = ggml_view_3d(ctx, v_cache, hd, ring, n_kv, v_cache->nb[1], v_cache->nb[2], 0);
struct ggml_tensor * scores = ggml_mul_mat(ctx, k_full, q_p); // [ring, T, n_q_heads]
float scale = 1.0f / sqrtf((float) hd);
scores = ggml_soft_max_ext(ctx, scores, mask, scale, 0.0f);
struct ggml_tensor * vt = ggml_cont(ctx, ggml_transpose(ctx, v_full)); // [ring, hd, n_kv]
struct ggml_tensor * attn = ggml_mul_mat(ctx, vt, scores); // [hd, T, n_q_heads]
attn = ggml_cont(ctx, ggml_permute(ctx, attn, 0, 2, 1, 3));
attn = ggml_reshape_2d(ctx, attn, n_q_heads * hd, T);
struct ggml_tensor * o = ggml_mul_mat(ctx, layer.attn.o_proj_w, attn);
o = ggml_mul(ctx, o, layer.attn_scale);
x = ggml_add(ctx, x, o);
struct ggml_tensor * ln2 = ggml_rms_norm(ctx, x, tr->rms_norm_eps);
ln2 = ggml_mul(ctx, ln2, layer.post_attn_norm_w);
struct ggml_tensor * gate = ggml_mul_mat(ctx, layer.mlp.gate_proj_w, ln2);
struct ggml_tensor * up = ggml_mul_mat(ctx, layer.mlp.up_proj_w, ln2);
gate = ggml_silu(ctx, gate);
struct ggml_tensor * gu = ggml_mul(ctx, gate, up);
struct ggml_tensor * mlp = ggml_mul_mat(ctx, layer.mlp.down_proj_w, gu);
mlp = ggml_mul(ctx, mlp, layer.mlp_scale);
x = ggml_add(ctx, x, mlp);
return x;
}
// Streaming transformer forward over a persistent KV ring. kv holds one
// [hd, ring, n_kv] pair per layer; kv_rows carries the ring slots the T
// fresh positions land in; mask is [ring, T] f32 with 0 on the slots
// inside the causal sliding window and neg inf elsewhere.
static struct ggml_tensor * tok_trans_forward_stream(struct ggml_context * ctx,
struct ggml_cgraph * gf,
const QwenTokenizerTransformer * tr,
struct ggml_tensor * x,
struct ggml_tensor * positions,
struct ggml_tensor * mask,
struct ggml_tensor * kv_rows,
KVCache * kv) {
int T = (int) x->ne[1];
int ring = kv->max_seq_len;
struct ggml_tensor * h = ggml_mul_mat(ctx, tr->input_proj_w, x);
h = ggml_add(ctx, h, tr->input_proj_b);
for (int l = 0; l < tr->num_layers; l++) {
h = tok_trans_layer_forward_stream(ctx, gf, tr, tr->layers[l], h, positions, mask, kv_rows, kv->k[(size_t) l],
kv->v[(size_t) l], T, ring);
}
h = ggml_rms_norm(ctx, h, tr->rms_norm_eps);
h = ggml_mul(ctx, h, tr->norm_w);
h = ggml_mul_mat(ctx, tr->output_proj_w, h);
h = ggml_add(ctx, h, tr->output_proj_b);
return h;
}
// Ring mask for the streaming path: [ring, T] f32, row q carries 0 on
// the ring slots holding positions inside [pos_q - window + 1, pos_q]
// and neg inf everywhere else, padded and future slots included.
static void tok_trans_build_stream_mask(int pos0, int T, int ring, int sliding_window, std::vector<float> & dst) {
dst.assign((size_t) ring * (size_t) T, -INFINITY);
for (int q = 0; q < T; q++) {
int pos_q = pos0 + q;
int k_min = pos_q - sliding_window + 1;
if (k_min < 0) {
k_min = 0;
}
for (int p = k_min; p <= pos_q; p++) {
dst[(size_t) q * (size_t) ring + (size_t) (p % ring)] = 0.0f;
}
}
}