feat(stream): add append-only streaming mode with flat cost

Add an append-only streaming path that feeds each audio chunk as a new
conversation turn, so nothing is ever recomputed: the KV grows by
append, no prefix is re-decoded and no audio is re-encoded. Per-step
cost becomes flat instead of growing with the segment (RTF 1.06 -> 0.26
on 62.6s of continuous speech), but accuracy requires ~8s chunks because
the audio encoder needs its 8s attention window, so text only appears
every 8s. The incremental mode stays the default.

新增追加式流式:每块音频作为一个新的对话轮追加,前缀不重算、旧音频不重
编码,单步成本不再随段落增长(连续 62.6s 语音 RTF 从 1.06 降到 0.26)。
但准确率要求 chunk≈8s,否则会被截断,所以出字粒度是 8s;增量式仍为默认。
This commit is contained in:
blumia committed 2026-10-12 14:22:29 +08:00
1 parent 7e5257429a
commit 85b206f1cd
10 files changed
+330 -18

No files matched your search

+11 -2
View File
@@ -24,8 +24,10 @@ struct AsrStreamStep {
std::string newText; // 本步新生成的文本(不含前缀,未 trim)
int newTokens = 0;
int audioTokens = 0;
double encodeSeconds = 0.0;
double inferSeconds = 0.0;
double encodeSeconds = 0.0; // 音频编码 + 写入 KV
double prefillSeconds = 0.0; // 文本前缀(tail + prefix)prefill
double genSeconds = 0.0; // 贪心生成
double inferSeconds = 0.0; // 以上合计
};
// 把推理后端抽象出来,好处是流式策略(StreamingAsr)和实时管线(LivePipeline)
@@ -44,6 +46,13 @@ public:
// 流式:只编码新音频,已解码的音频留在 KV cache 里复用
virtual AsrStreamStep streamAppend(const float * pcm, size_t nSamples,
const std::string & prefix, int maxTokens) = 0;
// 追加式轮次(append-only):把这一小段音频当作一个新的对话轮,
// 追加到已有 KV 的末尾,不重算任何前缀、也不重编码旧音频。
// 代价是每块只用自己的长度做编码(没有 8s 上下文),准确率需要实测确认。
virtual AsrStreamStep streamAppendTurn(const float * pcm, size_t nSamples,
int maxTokens) = 0;
virtual void streamReset() = 0;
// 用词表把 text 编码成 token、丢掉最后 dropLast 个、再还原成文本。
+160 -4
View File
@@ -38,6 +38,13 @@ struct Qwen3Asr::Impl {
size_t stableSamples = 0; // 已固化进 KV 的音频样本数(8s 对齐)
llama_pos nPastStable = 0; // stable 音频末尾的 KV 位置
// 追加式轮次(append-only)状态
bool turnReady = false;
std::vector<llama_token> firstHead; // 第 1 轮的轮首文本(含 <|audio_start|>)
std::vector<llama_token> contHead; // 后续轮的轮首文本(<|im_end|>\n<|im_start|>user\n<|audio_start|>)
llama_pos nPastTurn = 0;
int turnCount = 0;
std::string buildPrompt(const std::string & marker) const
{
const char * tmpl = llama_model_chat_template(model, nullptr);
@@ -112,6 +119,48 @@ struct Qwen3Asr::Impl {
nPast = np;
return true;
}
// 把带 media marker 的 prompt 拆成「音频前文本」和「音频后文本」两组 token
bool splitPrompt(const std::string & skeleton,
std::vector<llama_token> & head,
std::vector<llama_token> & tail,
std::string * err)
{
std::vector<float> dummy(1600, 0.0f); // 0.1s 占位,只为触发音频分块
mtmd_bitmap * bmp = mtmd_bitmap_init_from_audio(dummy.size(), dummy.data());
mtmd_input_chunks * chunks = mtmd_input_chunks_init();
const mtmd_input_text itext{ skeleton.c_str(), true, true };
const mtmd_bitmap * bms[1] = { bmp };
if (mtmd_tokenize(mctx, chunks, &itext, bms, 1) != 0) {
mtmd_bitmap_free(bmp);
mtmd_input_chunks_free(chunks);
if (err) *err = "mtmd_tokenize 失败";
return false;
}
mtmd_bitmap_free(bmp);
bool seenAudio = false;
for (size_t i = 0; i < mtmd_input_chunks_size(chunks); ++i) {
const mtmd_input_chunk * c = mtmd_input_chunks_get(chunks, i);
if (mtmd_input_chunk_get_type(c) == MTMD_INPUT_CHUNK_TYPE_AUDIO) {
seenAudio = true;
continue;
}
size_t n = 0;
const llama_token * toks = mtmd_input_chunk_get_tokens_text(c, &n);
if (!toks || n == 0) continue;
std::vector<llama_token> & dst = seenAudio ? tail : head;
dst.insert(dst.end(), toks, toks + n);
}
mtmd_input_chunks_free(chunks);
if (head.empty()) {
if (err) *err = "prompt 里没有音频之前的文本";
return false;
}
return true;
}
};
Qwen3Asr::Qwen3Asr() : d(new Impl) {}
@@ -313,6 +362,109 @@ void Qwen3Asr::streamReset()
d->streamAudio.clear();
d->stableSamples = 0;
d->nPastStable = 0;
d->turnReady = false;
d->firstHead.clear();
d->contHead.clear();
d->nPastTurn = 0;
d->turnCount = 0;
}
// 追加式轮次:把每小段音频当作一个新的对话轮追加到 KV 末尾。
// 序列形如 [user: audio_1][assistant: text_1][user: audio_2][assistant: text_2]...
// 好处是纯追加:不重算前缀、不重编码旧音频,单步成本与音频总长无关。
bool Qwen3Asr::prepareTurn(std::string * error)
{
if (d->turnReady) return true;
const std::string marker = mtmd_default_marker();
// 第 1 轮是完整的 user/assistant 结构;后续轮接在上一轮文本之后开新一轮。
const std::string s1 = "<|im_start|>user\n" + marker +
"<|im_end|>\n<|im_start|>assistant\n";
const std::string sc = "<|im_end|>\n<|im_start|>user\n" + marker +
"<|im_end|>\n<|im_start|>assistant\n";
std::vector<llama_token> tail1, tailC;
if (!d->splitPrompt(s1, d->firstHead, tail1, error)) return false;
if (!d->splitPrompt(sc, d->contHead, tailC, error)) return false;
// 两种轮尾应当一致
d->tailTokens = tail1;
llama_memory_clear(llama_get_memory(d->ctx), true);
d->nPastTurn = 0;
d->turnCount = 0;
d->turnReady = true;
return true;
}
AsrStreamStep Qwen3Asr::streamAppendTurn(const float * pcm, size_t nSamples, int maxTokens)
{
AsrStreamStep res;
if (!d->model || !d->ctx || !d->mctx) { res.error = "模型未加载"; return res; }
if (nSamples == 0) { res.error = "音频为空"; return res; }
std::string err;
if (!prepareTurn(&err)) { res.error = err; return res; }
const auto t0 = std::chrono::steady_clock::now();
// 1) 轮首文本(纯追加,位置接在 KV 末尾)
const std::vector<llama_token> & head = (d->turnCount == 0) ? d->firstHead : d->contHead;
llama_pos p = d->nPastTurn;
if (!d->decodeTokens(head.data(), (int32_t) head.size(), p, false)) {
res.error = "解码轮首失败";
return res;
}
p += (llama_pos) head.size();
// 2) 这一小段音频:只编码一次,写进 KV 就永久保留
const llama_pos beforeAudio = p;
if (!d->appendAudio(pcm, nSamples, p, &err)) {
res.error = "音频编码/解码失败: " + err;
return res;
}
res.audioTokens = (int) (p - beforeAudio);
const auto tEnc = std::chrono::steady_clock::now();
// 3) 轮尾文本 + 开始生成
if (!d->decodeTokens(d->tailTokens.data(), (int32_t) d->tailTokens.size(), p, true)) {
res.error = "解码轮尾失败";
return res;
}
p += (llama_pos) d->tailTokens.size();
const auto tPre = std::chrono::steady_clock::now();
llama_sampler * smpl = llama_sampler_chain_init(llama_sampler_chain_default_params());
llama_sampler_chain_add(smpl, llama_sampler_init_greedy());
std::string out;
for (int i = 0; i < maxTokens; ++i) {
llama_token id = llama_sampler_sample(smpl, d->ctx, -1);
if (llama_vocab_is_eog(d->vocab, id)) break;
char buf[256];
const int32_t n = llama_token_to_piece(d->vocab, id, buf, sizeof(buf), 0, true);
if (n > 0) out.append(buf, (size_t) n);
++res.newTokens;
if (!d->decodeTokens(&id, 1, p, true)) { res.error = "生成时解码失败"; break; }
++p;
}
llama_sampler_free(smpl);
d->nPastTurn = p;
++d->turnCount;
const auto t1 = std::chrono::steady_clock::now();
res.encodeSeconds = std::chrono::duration<double>(tEnc - t0).count();
res.prefillSeconds = std::chrono::duration<double>(tPre - tEnc).count();
res.genSeconds = std::chrono::duration<double>(t1 - tPre).count();
res.inferSeconds = std::chrono::duration<double>(t1 - t0).count();
res.newText = out;
res.ok = res.error.empty();
return res;
}
bool Qwen3Asr::prepareStream(std::string * error)
@@ -452,6 +604,8 @@ AsrStreamStep Qwen3Asr::streamAppend(const float * pcm, size_t nSamples,
llama_sampler * smpl = llama_sampler_chain_init(llama_sampler_chain_default_params());
llama_sampler_chain_add(smpl, llama_sampler_init_greedy());
const auto tPre = std::chrono::steady_clock::now();
std::string out;
for (int i = 0; i < maxTokens; ++i) {
llama_token id = llama_sampler_sample(smpl, d->ctx, -1);
@@ -468,10 +622,12 @@ AsrStreamStep Qwen3Asr::streamAppend(const float * pcm, size_t nSamples,
llama_sampler_free(smpl);
const auto t1 = std::chrono::steady_clock::now();
res.encodeSeconds = std::chrono::duration<double>(tEnc - t0).count();
res.inferSeconds = std::chrono::duration<double>(t1 - t0).count();
res.newText = out;
res.ok = res.error.empty();
res.encodeSeconds = std::chrono::duration<double>(tEnc - t0).count();
res.prefillSeconds = std::chrono::duration<double>(tPre - tEnc).count();
res.genSeconds = std::chrono::duration<double>(t1 - tPre).count();
res.inferSeconds = std::chrono::duration<double>(t1 - t0).count();
res.newText = out;
res.ok = res.error.empty();
return res;
}
+2
View File
@@ -38,12 +38,14 @@ public:
AsrStreamStep streamAppend(const float * pcm, size_t nSamples,
const std::string & prefix, int maxTokens) override;
AsrStreamStep streamAppendTurn(const float * pcm, size_t nSamples, int maxTokens) override;
void streamReset() override;
std::string dropLastTokens(const std::string & text, int dropLast) const override;
private:
bool prepareStream(std::string * error);
bool prepareTurn(std::string * error);
struct Impl;
Impl * d = nullptr;
+36 -1
View File
@@ -94,15 +94,33 @@ void StreamingAsr::runStep(const std::vector<float> & audio, bool finalStep,
s.chunkIndex = chunkIndex_;
std::string newText;
if (opt_.incremental) {
if (opt_.appendOnly) {
// 追加式:这一块作为一个新的对话轮,文本随即定稿,不做前缀回滚
AsrStreamStep r = asr_.streamAppendTurn(audio.data(), audio.size(),
opt_.maxTokensPerStep);
s.ok = r.ok;
s.error = r.error;
s.encodeSeconds = r.encodeSeconds;
s.prefillSeconds = r.prefillSeconds;
s.genSeconds = r.genSeconds;
s.inferSeconds = r.inferSeconds;
s.audioTokens = r.audioTokens;
s.newTokens = r.newTokens;
newText = r.newText;
totalSeconds_ += (double) audio.size() / asr_.sampleRate();
s.windowSeconds = totalSeconds_;
} else if (opt_.incremental) {
// 只把这一块交给增量引擎;音频总长不影响单步成本
AsrStreamStep r =
asr_.streamAppend(audio.data(), audio.size(), prefix, opt_.maxTokensPerStep);
s.ok = r.ok;
s.error = r.error;
s.encodeSeconds = r.encodeSeconds;
s.prefillSeconds = r.prefillSeconds;
s.genSeconds = r.genSeconds;
s.inferSeconds = r.inferSeconds;
s.audioTokens = r.audioTokens;
s.newTokens = r.newTokens;
newText = r.newText;
totalSeconds_ += (double) audio.size() / asr_.sampleRate();
s.windowSeconds = totalSeconds_;
@@ -126,6 +144,23 @@ void StreamingAsr::runStep(const std::vector<float> & audio, bool finalStep,
}
if (s.ok) {
if (opt_.appendOnly) {
// 追加式:每轮文本直接定稿,追加进历史,没有"不稳定尾部"
const ParsedTranscript pTurn = parseAsrOutput(newText);
if (!pTurn.language.empty()) language_ = pTurn.language;
historyText_ += pTurn.text;
s.rawFull = newText;
s.full = historyText_;
s.committed = historyText_;
s.tail.clear();
s.language = language_;
lastRawFull_.clear();
committedRaw_.clear();
++chunkIndex_;
if (onStep) onStep(s);
return;
}
// 拼接要用未经 trim 的文本:流式续写时开头那个空格是有意义的
s.rawFull = prefix + newText;
lastRawFull_ = s.rawFull;
+8
View File
@@ -21,6 +21,11 @@ struct StreamOptions {
// false = 每次重编码整个窗口(Phase 2 早期的做法,仅用于对比)
bool incremental = true;
// true = 追加式轮次(append-only):每块音频作为新一轮追加,前缀完全不重算、
// 旧音频不重编码,单步成本与总长无关。代价是每块只用自己的长度编码
// (没有 8s 上下文),准确率需实测确认。
bool appendOnly = false;
// 单段最长时长(秒):超过就提交并重置上下文(0 = 不限)。
// 真实场景应由静音检测决定分段;这里是一个兜底上界,
// 因为解码器对 KV 的注意力随总长增长,段太长单步会超预算。
@@ -42,8 +47,11 @@ public:
std::string rawFull; // 未 trim 的完整文本(含 language 头)
double windowSeconds = 0.0;
double encodeSeconds = 0.0;
double prefillSeconds = 0.0;
double genSeconds = 0.0;
double inferSeconds = 0.0;
int audioTokens = 0;
int newTokens = 0;
bool ok = true;
std::string error;
};