From f85087f992316d8409b1e443043253b34061cb14 Mon Sep 17 00:00:00 2001 From: mudler-agent Date: Mon, 19 Oct 2026 12:30:52 +0200 Subject: [PATCH] feat(tts): stream Qwen text and add assistant preset (#12770) * feat(tts): stream Qwen text through audio pipelines Feed cleaned LLM deltas into one persistent Qwen synthesis session. Return audio before input ends and preserve the live RPC through model cache, staging and in-flight wrappers. Stop generation on sink failures or cancellation without replaying speech through a fallback. Pin the incremental native fork and verify early audio, late text, finalization and cancellation with lifecycle and real-model checks. Assisted-by: Codex:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto * build(tts): use the organization Qwen fork Fetch the reviewed incremental synthesis commit from localai-org so backend builds use the organization-maintained fork. Assisted-by: Codex:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto * feat(gallery): add a Qwen streaming assistant preset Expose an English pipeline using existing VAD, ASR, LLM and named-voice TTS models. Enable incremental text and audio delivery, and document separate component installs because the gallery has no dependency installer. Assisted-by: Codex:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto * fix(gallery): use Parakeet TDT v3 for the assistant Select the existing Parakeet TDT 0.6B v3 transcription entry and update its prerequisite install command. Keep the preset's English voice and incremental Qwen synthesis configuration. Assisted-by: Codex:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto * fix(kokoros): reject unsupported backend RPCs Implement the MotionStream and SlotCache methods required by the shared protobuf service. Return Unimplemented so Kokoros compiles without advertising support for either operation. Assisted-by: Codex:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto * fix(tts): mark cleanup and logging results ignored Explicitly ignore connection shutdown, test server shutdown and Ginkgo diagnostic writer errors. These operations do not replace the stream's inference or sink errors, and errcheck requires their return values to be acknowledged. Assisted-by: Codex:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto * feat(tts): stream buffered clauses in realtime pipelines Start clause synthesis while LLM text is still arriving when both streaming flags are enabled. Keep explicit clause opt-out and native Qwen sessions, bound automatic queued work, and choose streaming support on resolved failover targets. Assisted-by: Codex:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto * test(tts): bound synthetic voice clone generation The sine-wave reference checks speaker conditioning and WAV output, but does not provide a meaningful speech EOS target. Limit that smoke request to 128 frames and verify the request limit reaches native synthesis. Leave production sampling defaults unchanged. Assisted-by: Codex:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto * fix(tts): require explicit clause buffering opt-in Keep whole-reply synthesis for nonnative engines when clause_chunking is unset or false. Use the bounded cancellable worker when explicitly enabled, with unary clauses when TTS streaming is disabled. Preserve native Qwen precedence for every clause setting. Assisted-by: Codex:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto * feat(gallery): add VITS and Piper streaming assistants Provide comparable full pipeline configurations with shared VAD, Parakeet transcription, and Qwen chat components. Enable clause buffering explicitly and document the separate prerequisite model installs and Talk selection. Assisted-by: Codex:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto --------- Signed-off-by: Ettore Di Giacinto Co-authored-by: Ettore Di Giacinto --- backend/backend.proto | 3 + backend/go/qwen3-tts-cpp/CMakeLists.txt | 2 + backend/go/qwen3-tts-cpp/Makefile | 16 +- backend/go/qwen3-tts-cpp/README.md | 30 +++ .../go/qwen3-tts-cpp/cpp/goqwen3ttscpp.cpp | 32 +++ backend/go/qwen3-tts-cpp/cpp/goqwen3ttscpp.h | 8 + backend/go/qwen3-tts-cpp/e2e_test.go | 7 +- backend/go/qwen3-tts-cpp/live.go | 127 ++++++++++++ backend/go/qwen3-tts-cpp/live_native_test.go | 99 ++++++++++ backend/go/qwen3-tts-cpp/live_state.go | 38 ++++ backend/go/qwen3-tts-cpp/live_state_test.go | 46 +++++ backend/go/qwen3-tts-cpp/live_test.go | 69 +++++++ backend/go/qwen3-tts-cpp/main.go | 1 + backend/go/qwen3-tts-cpp/qwen3ttscpp_test.go | 18 ++ backend/rust/kokoros/src/service.rs | 25 +++ core/backend/tts_live.go | 30 +++ core/config/model_config.go | 4 +- .../openai/realtime_failover_test.go | 18 ++ core/http/endpoints/openai/realtime_model.go | 56 ++++++ core/http/endpoints/openai/realtime_speech.go | 25 ++- .../endpoints/openai/realtime_speech_test.go | 52 +++++ core/http/endpoints/openai/realtime_stream.go | 102 ++++++---- .../endpoints/openai/realtime_stream_test.go | 59 +++--- .../openai/realtime_tts_buffered_test.go | 182 ++++++++++++++++++ .../endpoints/openai/realtime_tts_live.go | 106 ++++++++++ .../openai/realtime_tts_live_test.go | 140 ++++++++++++++ core/services/nodes/file_staging_client.go | 27 +++ core/services/nodes/inflight.go | 10 + docs/content/features/openai-realtime.md | 69 ++++++- gallery/index.yaml | 98 ++++++++++ pkg/grpc/tts_live.go | 167 ++++++++++++++++ pkg/grpc/tts_live_test.go | 150 +++++++++++++++ pkg/model/connection_evicting_client.go | 11 ++ pkg/model/tts_live_test.go | 43 +++++ 34 files changed, 1780 insertions(+), 90 deletions(-) create mode 100644 backend/go/qwen3-tts-cpp/live.go create mode 100644 backend/go/qwen3-tts-cpp/live_native_test.go create mode 100644 backend/go/qwen3-tts-cpp/live_state.go create mode 100644 backend/go/qwen3-tts-cpp/live_state_test.go create mode 100644 backend/go/qwen3-tts-cpp/live_test.go create mode 100644 core/backend/tts_live.go create mode 100644 core/http/endpoints/openai/realtime_tts_buffered_test.go create mode 100644 core/http/endpoints/openai/realtime_tts_live.go create mode 100644 core/http/endpoints/openai/realtime_tts_live_test.go create mode 100644 pkg/grpc/tts_live.go create mode 100644 pkg/grpc/tts_live_test.go create mode 100644 pkg/model/tts_live_test.go diff --git a/backend/backend.proto b/backend/backend.proto index 1c8b273f8..9e5c459a9 100644 --- a/backend/backend.proto +++ b/backend/backend.proto @@ -36,6 +36,9 @@ service Backend { rpc AudioTranscriptionLive(stream TranscriptLiveRequest) returns (stream TranscriptLiveResponse) {} rpc TTS(TTSRequest) returns (Result) {} rpc TTSStream(TTSRequest) returns (stream Reply) {} + // First frame configures a persistent utterance; later frames carry only text. + // Half-close commits end of text, while audio continues until EOF. + rpc TTSLive(stream TTSRequest) returns (stream Reply) {} rpc SoundGeneration(SoundGenerationRequest) returns (Result) {} rpc TokenizeString(PredictOptions) returns (TokenizationResponse) {} rpc Detokenize(DetokenizeRequest) returns (DetokenizeResponse) {} diff --git a/backend/go/qwen3-tts-cpp/CMakeLists.txt b/backend/go/qwen3-tts-cpp/CMakeLists.txt index 6cf7d626e..1ffbe2ff2 100644 --- a/backend/go/qwen3-tts-cpp/CMakeLists.txt +++ b/backend/go/qwen3-tts-cpp/CMakeLists.txt @@ -4,6 +4,8 @@ set(CMAKE_POSITION_INDEPENDENT_CODE ON) set(CMAKE_EXPORT_COMPILE_COMMANDS ON) set(QWENTTS_DIR ${CMAKE_CURRENT_SOURCE_DIR}/sources/qwentts.cpp) +# Upstream uses the top-level source directory for this cache default. +set(GGML_SOURCE_DIR ${QWENTTS_DIR}/ggml CACHE PATH "ggml source tree" FORCE) # Override upstream's CMAKE_CUDA_ARCHITECTURES before add_subdirectory. if(NOT DEFINED CMAKE_CUDA_ARCHITECTURES) diff --git a/backend/go/qwen3-tts-cpp/Makefile b/backend/go/qwen3-tts-cpp/Makefile index e75eb6cca..5a627af36 100644 --- a/backend/go/qwen3-tts-cpp/Makefile +++ b/backend/go/qwen3-tts-cpp/Makefile @@ -6,19 +6,9 @@ GOCMD?=go GO_TAGS?= JOBS?=$(shell nproc --ignore=1) -# qwentts.cpp version -# -# Held at 35ebe537 rather than tracking latest: abab6b3 hangs in synthesis. -# TTS() never returns from the native call, so tests-qwen3-tts-cpp goes from -# ~5 minutes to the 20 minute Go test timeout. Reproduced on master on -# 2026-08-01 and again on re-run, and the bump PR (#11241) was merged with -# this same check already red. -# -# The regression is in 35ebe537..abab6b3, three upstream commits whose only -# functional change is 26dd8adb, "predictor: unroll the frame into one cgraph -# and sample in standard ops". Restore the bump once that is fixed upstream. -QWEN3TTS_REPO?=https://github.com/ServeurpersoCom/qwentts.cpp -QWEN3TTS_CPP_VERSION?=35ebe5376b82a0a59d008586d55bbe623d449011 +# Incremental text input uses the persistent-session ABI in this fork. +QWEN3TTS_REPO?=https://github.com/localai-org/qwentts.cpp +QWEN3TTS_CPP_VERSION?=7cdb17cdaf6ef88961365e9b5177a0284501beb9 SO_TARGET?=libgoqwen3ttscpp.so CMAKE_ARGS+=-DBUILD_SHARED_LIBS=OFF diff --git a/backend/go/qwen3-tts-cpp/README.md b/backend/go/qwen3-tts-cpp/README.md index 2c0f256af..247c39c3c 100644 --- a/backend/go/qwen3-tts-cpp/README.md +++ b/backend/go/qwen3-tts-cpp/README.md @@ -83,3 +83,33 @@ QWEN3TTS_CODEC=/path/to/qwen-tokenizer-12hz-Q8_0.gguf \ QWEN3TTS_LIBRARY=backend/go/qwen3-tts-cpp/libgoqwen3ttscpp-fallback.so \ go test ./backend/go/qwen3-tts-cpp -ginkgo.label-filter=e2e ``` + +### Incremental text and audio + +The `TTSLive` backend RPC accepts text fragments while emitting audio from one +persistent utterance. The first `TTSRequest` carries voice, language and sampling +options; subsequent requests carry only `text`. Each fragment must contain valid UTF-8; +the protobuf string boundary does not accept incomplete Unicode characters. Closing the request stream commits +end of input. Continue reading responses until EOF to receive the codec tail. +Responses contain one streaming WAV header followed by mono PCM16 chunks at 24 kHz. + +Realtime audio pipelines select this path when their TTS model uses +`qwen3-tts-cpp` and `pipeline.streaming.llm` and `pipeline.streaming.tts` are +`true`. Cleaned LLM content feeds the same synthesis session; clause splitting is +bypassed. WebSocket and WebRTC transports receive audio as it becomes available. +The text queue is bounded, so slow audio consumers apply backpressure to text +generation. Audio emits before the full response is available; WebSocket +conversation history still retains the final audio, as in the existing path. Cancellation stops input and audio. A failed live session is not +retried because consumed text and emitted speech cannot be replayed safely. + +This requires the incremental-input native fork. Its initial implementation uses +one native batch slot and commits whitespace-delimited words or individual CJK, Kana and +Hangul characters. Incomplete words remain buffered until a stable boundary. Reference audio without `ref_text` is supported; transcript-based ICL +cloning is rejected for incremental sessions. Other TTS backends retain their +existing whole-text or clause synthesis paths. The standard speech HTTP request +continues to accept a complete JSON input. + +To run the real native streaming check, provide `QWEN3TTS_LIBRARY`, +`QWEN3TTS_MODEL`, `QWEN3TTS_CODEC`, and `QWEN3TTS_LIVE_TEST=1`, then run +`go test ./backend/go/qwen3-tts-cpp -ginkgo.focus TTSLiveNative -v` from the +repository root. It withholds late text and end of input until real PCM arrives. diff --git a/backend/go/qwen3-tts-cpp/cpp/goqwen3ttscpp.cpp b/backend/go/qwen3-tts-cpp/cpp/goqwen3ttscpp.cpp index 0cd62293d..d1b49883c 100644 --- a/backend/go/qwen3-tts-cpp/cpp/goqwen3ttscpp.cpp +++ b/backend/go/qwen3-tts-cpp/cpp/goqwen3ttscpp.cpp @@ -189,3 +189,35 @@ int qt3_n_speakers(void) { return g_ctx ? qt_n_speakers(g_ctx) : 0; } const char *qt3_speaker_name(int i) { return g_ctx ? qt_speaker_name(g_ctx, i) : nullptr; } + +namespace { +struct live_callbacks { + qt3_text_cb text; + qt3_chunk_cb audio; + void *user; +}; +static int live_next_text(void *opaque, const char **text, size_t *n_bytes) { + auto *callbacks = static_cast(opaque); + return callbacks->text(callbacks->user, text, n_bytes); +} +static bool live_audio_chunk(const float *samples, int n_samples, void *opaque) { + auto *callbacks = static_cast(opaque); + return callbacks->audio(samples, n_samples, callbacks->user) != 0; +} +} + +int qt3_tts_live(const char *lang, const char *instruct, const char *speaker, + const float *ref_samples, int ref_n, const char *ref_text, + long long seed, float temperature, int top_k, float top_p, + float repetition_penalty, int max_new_tokens, + qt_text_input_cb next_text, qt3_chunk_cb cb, void *user_data) { + if (!g_ctx || !cb || !next_text) return 1; + qt_tts_params tp; + fill_params(&tp, nullptr, lang, instruct, speaker, ref_samples, ref_n, + ref_text, seed, temperature, top_k, top_p, repetition_penalty, + max_new_tokens); + live_callbacks callbacks{next_text, cb, user_data}; + tp.on_chunk = live_audio_chunk; + tp.on_chunk_user_data = &callbacks; + return (int) qt_synthesize_incremental(g_ctx, &tp, live_next_text, &callbacks); +} diff --git a/backend/go/qwen3-tts-cpp/cpp/goqwen3ttscpp.h b/backend/go/qwen3-tts-cpp/cpp/goqwen3ttscpp.h index 5e7dd4823..351592ba3 100644 --- a/backend/go/qwen3-tts-cpp/cpp/goqwen3ttscpp.h +++ b/backend/go/qwen3-tts-cpp/cpp/goqwen3ttscpp.h @@ -1,4 +1,5 @@ #pragma once +#include extern "C" { @@ -34,6 +35,13 @@ int qt3_tts_stream(const char *text, const char *lang, const char *instruct, int top_k, float top_p, float repetition_penalty, int max_new_tokens, qt3_chunk_cb cb, void *user_data); +typedef int (*qt3_text_cb)(void *user_data, const char **text, size_t *n_bytes); +int qt3_tts_live(const char *lang, const char *instruct, const char *speaker, + const float *ref_samples, int ref_n, const char *ref_text, + long long seed, float temperature, int top_k, float top_p, + float repetition_penalty, int max_new_tokens, + qt3_text_cb next_text, qt3_chunk_cb cb, void *user_data); + // Free a buffer returned by qt3_tts. void qt3_pcm_free(float *p); diff --git a/backend/go/qwen3-tts-cpp/e2e_test.go b/backend/go/qwen3-tts-cpp/e2e_test.go index eea64d18c..8c2cd8d29 100644 --- a/backend/go/qwen3-tts-cpp/e2e_test.go +++ b/backend/go/qwen3-tts-cpp/e2e_test.go @@ -87,7 +87,12 @@ var _ = Describe("qwen3-tts-cpp e2e", Label("e2e"), func() { dst := GinkgoT().TempDir() + "/clone.wav" lang := "english" // Empty Voice -> the config audio_path is used as the clone reference. - Expect(b.TTS(ttsReq("Cloned voice test.", "", &lang, dst))).To(Succeed()) + // The sine reference exercises speaker conditioning, not speech quality. + // It has no meaningful EOS guarantee; bound this smoke independently + // of the production generation limit. + req := ttsReq("Cloned voice test.", "", &lang, dst) + req.Params = map[string]string{"max_new_tokens": "128"} + Expect(b.TTS(req)).To(Succeed()) fi, err := os.Stat(dst) Expect(err).ToNot(HaveOccurred()) Expect(fi.Size()).To(BeNumerically(">", int64(44))) diff --git a/backend/go/qwen3-tts-cpp/live.go b/backend/go/qwen3-tts-cpp/live.go new file mode 100644 index 000000000..644e974e4 --- /dev/null +++ b/backend/go/qwen3-tts-cpp/live.go @@ -0,0 +1,127 @@ +package main + +import ( + "context" + "fmt" + "io" + "runtime" + "sync" + "unsafe" + + "github.com/ebitengine/purego" + pb "github.com/mudler/LocalAI/pkg/grpc/proto" +) + +var CppTTSLive func(lang, instruct, speaker string, refSamples unsafe.Pointer, refN int, + refText string, seed int64, temperature float32, topK int, topP, repPen float32, maxNew int, + nextText, audio, user uintptr) int + +var liveOnce sync.Once +var liveTextCB, liveAudioCB uintptr + +// Native synthesis invokes both callbacks synchronously on its worker thread. +// The backend model lock and streamMu permit only one active native utterance. +var liveRecv func() (*pb.TTSRequest, error) +var liveSend func([]byte) error +var liveText []byte +var liveError error +var liveInput liveInputState + +func nextLiveText(_ uintptr, textOut unsafe.Pointer, sizeOut unsafe.Pointer) uintptr { + if liveInput == liveInputEnded { + return 0 + } + if liveInput == liveInputFailed { + return ^uintptr(0) + } + req, err := liveRecv() + if err == io.EOF { + liveInput, _ = reduceLiveInput(liveInput, liveInputEOF) + return 0 + } + if err != nil { + liveError = err + liveInput, _ = reduceLiveInput(liveInput, liveInputError) + return ^uintptr(0) + } + liveInput, _ = reduceLiveInput(liveInput, liveInputChunk) + liveText = append(liveText[:0], []byte(req.Text)...) + if len(liveText) == 0 { + liveText = append(liveText, 0) + } + *(*unsafe.Pointer)(textOut) = unsafe.Pointer(&liveText[0]) + *(*uintptr)(sizeOut) = uintptr(len(req.Text)) + return 1 +} +func sendLiveAudio(samples *float32, n int32, _ uintptr) uintptr { + if n <= 0 { + return 1 + } + pcm := floatToPCM16LE(unsafe.Slice(samples, int(n))) + if err := liveSend(pcm); err != nil { + liveError = err + liveInput, _ = reduceLiveInput(liveInput, liveInputError) + return 0 + } + return 1 +} + +func (q *Qwen3TtsCpp) TTSLive(ctx context.Context, recv func() (*pb.TTSRequest, error), send func([]byte) error) error { + req, err := recv() + if err != nil { + return err + } + lang, instruct, speaker, refText, ref, s, err := q.resolveRequest(req) + if err != nil { + return err + } + if refText != "" { + return fmt.Errorf("qwen3-tts: incremental input does not support ref_text; use speaker or reference audio without transcription") + } + if CppTTSLive == nil { + return fmt.Errorf("qwen3-tts: incremental synthesis unavailable in native library") + } + if err := send(wavHeader24k()); err != nil { + return err + } + liveOnce.Do(func() { liveTextCB = purego.NewCallback(nextLiveText); liveAudioCB = purego.NewCallback(sendLiveAudio) }) + streamMu.Lock() + defer streamMu.Unlock() + first := req.Text != "" + liveRecv = func() (*pb.TTSRequest, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + if first { + first = false + return req, nil + } + return recv() + } + liveSend = func(b []byte) error { + if err := ctx.Err(); err != nil { + return err + } + return send(b) + } + liveError = nil + liveInput = liveInputOpen + defer func() { liveRecv = nil; liveSend = nil; liveText = nil; liveError = nil }() + var refPtr unsafe.Pointer + if len(ref) > 0 { + refPtr = unsafe.Pointer(&ref[0]) + } + rc := CppTTSLive(lang, instruct, speaker, refPtr, len(ref), refText, s.seed, s.temperature, s.topK, s.topP, s.repPen, s.maxNew, liveTextCB, liveAudioCB, 0) + runtime.KeepAlive(ref) + runtime.KeepAlive(liveText) + if liveError != nil { + return liveError + } + if err := ctx.Err(); err != nil { + return err + } + if rc != 0 { + return fmt.Errorf("qwen3-tts: incremental synthesis failed (rc=%d)", rc) + } + return nil +} diff --git a/backend/go/qwen3-tts-cpp/live_native_test.go b/backend/go/qwen3-tts-cpp/live_native_test.go new file mode 100644 index 000000000..a1b4fb37a --- /dev/null +++ b/backend/go/qwen3-tts-cpp/live_native_test.go @@ -0,0 +1,99 @@ +package main + +import ( + "context" + "encoding/binary" + "fmt" + . "github.com/onsi/ginkgo/v2" + "io" + "os" + "sync" + "time" + + "github.com/ebitengine/purego" + pb "github.com/mudler/LocalAI/pkg/grpc/proto" +) + +// Opt-in real-model check. The input producer cannot send late text or EOF +// until it observes actual PCM, so a whole-text implementation deadlocks here. +var _ = Describe("TTSLiveNative", func() { + It("preserves the live synthesis contract", func() { + if os.Getenv("QWEN3TTS_LIVE_TEST") != "1" { + Skip(fmt.Sprint("set QWEN3TTS_LIVE_TEST=1 and model/codec/library paths")) + } + lib, e := purego.Dlopen(os.Getenv("QWEN3TTS_LIBRARY"), purego.RTLD_NOW|purego.RTLD_GLOBAL) + if e != nil { + Fail(fmt.Sprint(e)) + } + purego.RegisterLibFunc(&CppLoad, lib, "qt3_load") + purego.RegisterLibFunc(&CppUnload, lib, "qt3_unload") + purego.RegisterLibFunc(&CppTTSLive, lib, "qt3_tts_live") + if rc := CppLoad(os.Getenv("QWEN3TTS_MODEL"), os.Getenv("QWEN3TTS_CODEC"), 1, 0); rc != 0 { + Fail(fmt.Sprint("load", rc)) + } + defer CppUnload() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + defer cancel() + input := make(chan *pb.TTSRequest) + early := make(chan struct{}) + var once sync.Once + go func() { + select { + case input <- &pb.TTSRequest{Voice: "ryan", Language: func() *string { s := "english"; return &s }(), Text: "Hello there. ", Params: map[string]string{"max_new_tokens": "160", "seed": "42"}}: + case <-ctx.Done(): + return + } + select { + case <-early: + case <-ctx.Done(): + return + } + select { + case input <- &pb.TTSRequest{Text: "This is a streaming test. "}: + case <-ctx.Done(): + return + } + close(input) + }() + chunks, bytes := 0, 0 + var output []byte + outputPath := os.Getenv("QWEN3TTS_LIVE_OUTPUT") + b := &Qwen3TtsCpp{} + e = b.TTSLive(ctx, func() (*pb.TTSRequest, error) { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case req, ok := <-input: + if !ok { + return nil, io.EOF + } + return req, nil + } + }, func(audio []byte) error { + if outputPath != "" { + output = append(output, audio...) + } + // The first callback is only a WAV header and cannot open the audio gate. + if chunks > 0 { + bytes += len(audio) + once.Do(func() { close(early) }) + } + chunks++ + return nil + }) + if e != nil { + Fail(fmt.Sprint(e)) + } + if bytes == 0 { + Fail(fmt.Sprint("no PCM")) + } + if outputPath != "" { + binary.LittleEndian.PutUint32(output[4:8], uint32(len(output)-8)) + binary.LittleEndian.PutUint32(output[40:44], uint32(len(output)-44)) + if err := os.WriteFile(outputPath, output, 0600); err != nil { + Fail(err.Error()) + } + } + _, _ = fmt.Fprintf(GinkgoWriter, "PCM bytes=%d callbacks=%d; late text and EOS sent after first PCM", bytes, chunks) + }) +}) diff --git a/backend/go/qwen3-tts-cpp/live_state.go b/backend/go/qwen3-tts-cpp/live_state.go new file mode 100644 index 000000000..03a8f4bea --- /dev/null +++ b/backend/go/qwen3-tts-cpp/live_state.go @@ -0,0 +1,38 @@ +package main + +// Native may ask for text again after EOF while flushing its codec. Keep input +// terminal so that request cannot block waiting for a frame that will never come. +type liveInputState uint8 + +const ( + liveInputOpen liveInputState = iota + liveInputEnded + liveInputFailed +) + +type liveInputEvent uint8 + +const ( + liveInputChunk liveInputEvent = iota + liveInputEOF + liveInputError +) + +func reduceLiveInput(state liveInputState, event liveInputEvent) (liveInputState, string) { + switch state { + case liveInputOpen: + switch event { + case liveInputChunk: + return liveInputOpen, "R1-chunk" + case liveInputEOF: + return liveInputEnded, "R2-finish" + case liveInputError: + return liveInputFailed, "R3-fail" + } + case liveInputEnded: + return state, "R4-ended-ignore" + case liveInputFailed: + return state, "R5-failed-ignore" + } + panic("invalid live input transition") +} diff --git a/backend/go/qwen3-tts-cpp/live_state_test.go b/backend/go/qwen3-tts-cpp/live_state_test.go new file mode 100644 index 000000000..7c66c0ba7 --- /dev/null +++ b/backend/go/qwen3-tts-cpp/live_state_test.go @@ -0,0 +1,46 @@ +package main + +import ( + "fmt" + . "github.com/onsi/ginkgo/v2" +) + +var _ = Describe("LiveInputTraces", func() { + It("preserves the live synthesis contract", func() { + for _, tc := range []struct { + name string + events []liveInputEvent + state liveInputState + rules []string + }{ + {"late input then EOF", []liveInputEvent{liveInputChunk, liveInputChunk, liveInputEOF, liveInputChunk}, liveInputEnded, []string{"R1-chunk", "R1-chunk", "R2-finish", "R4-ended-ignore"}}, + {"disconnect before text", []liveInputEvent{liveInputError, liveInputEOF}, liveInputFailed, []string{"R3-fail", "R5-failed-ignore"}}, + {"failure after audio", []liveInputEvent{liveInputChunk, liveInputError, liveInputChunk}, liveInputFailed, []string{"R1-chunk", "R3-fail", "R5-failed-ignore"}}, + } { + By(tc.name) + func() { + state := liveInputOpen + for i, e := range tc.events { + next, rule := reduceLiveInput(state, e) + if rule != tc.rules[i] { + Fail(fmt.Sprint(rule)) + } + state = next + } + if state != tc.state { + Fail(fmt.Sprint(state)) + } + }() + } + // Exhaust the finite graph: no terminal state can reopen, and every open + // state admits both a normal completion and a cancellation. + for s := liveInputOpen; s <= liveInputFailed; s++ { + for e := liveInputChunk; e <= liveInputError; e++ { + next, _ := reduceLiveInput(s, e) + if s != liveInputOpen && next != s { + Fail(fmt.Sprint("terminal input reopened")) + } + } + } + }) +}) diff --git a/backend/go/qwen3-tts-cpp/live_test.go b/backend/go/qwen3-tts-cpp/live_test.go new file mode 100644 index 000000000..dddea727f --- /dev/null +++ b/backend/go/qwen3-tts-cpp/live_test.go @@ -0,0 +1,69 @@ +package main + +import ( + "context" + "errors" + "fmt" + . "github.com/onsi/ginkgo/v2" + "io" + "reflect" + "unsafe" + + pb "github.com/mudler/LocalAI/pkg/grpc/proto" +) + +var _ = Describe("TTSLiveAdapter", func() { + It("preserves the live synthesis contract", func() { + old := CppTTSLive + defer func() { CppTTSLive = old }() + var got []string + CppTTSLive = func(_, _, _ string, _ unsafe.Pointer, _ int, _ string, _ int64, _ float32, _ int, _, _ float32, _ int, _, _, _ uintptr) int { + for { + var text unsafe.Pointer + var size uintptr + result := nextLiveText(0, unsafe.Pointer(&text), unsafe.Pointer(&size)) + if result == 0 { + return 0 + } + if result != 1 { + return 1 + } + got = append(got, string(unsafe.Slice((*byte)(text), int(size)))) + samples := []float32{0.1, 0.2} + if sendLiveAudio(&samples[0], 2, 0) == 0 { + return 2 + } + } + } + input := []*pb.TTSRequest{{Text: "early "}, {Text: "late"}} + chunks := 0 + err := (&Qwen3TtsCpp{}).TTSLive(context.Background(), func() (*pb.TTSRequest, error) { + if len(input) == 0 { + return nil, io.EOF + } + r := input[0] + input = input[1:] + return r, nil + }, func([]byte) error { chunks++; return nil }) + if err != nil { + Fail(fmt.Sprint(err)) + } + if !reflect.DeepEqual(got, []string{"early ", "late"}) || chunks != 3 { + Fail(fmt.Sprintf("input=%v chunks=%d", got, chunks)) + } + // A downstream failure must stop the native callback and preserve the exact + // error, rather than reporting successful synthesis or consuming more input. + boom := errors.New("audio consumer failed") + chunks = 0 + err = (&Qwen3TtsCpp{}).TTSLive(context.Background(), func() (*pb.TTSRequest, error) { return &pb.TTSRequest{Text: "text"}, nil }, func([]byte) error { + chunks++ + if chunks > 1 { + return boom + } + return nil + }) + if !errors.Is(err, boom) { + Fail(fmt.Sprint(err)) + } + }) +}) diff --git a/backend/go/qwen3-tts-cpp/main.go b/backend/go/qwen3-tts-cpp/main.go index 041a23ad0..7a978be70 100644 --- a/backend/go/qwen3-tts-cpp/main.go +++ b/backend/go/qwen3-tts-cpp/main.go @@ -38,6 +38,7 @@ func main() { {&CppLoad, "qt3_load"}, {&CppTTS, "qt3_tts"}, {&CppTTSStream, "qt3_tts_stream"}, + {&CppTTSLive, "qt3_tts_live"}, {&CppPCMFree, "qt3_pcm_free"}, {&CppUnload, "qt3_unload"}, } diff --git a/backend/go/qwen3-tts-cpp/qwen3ttscpp_test.go b/backend/go/qwen3-tts-cpp/qwen3ttscpp_test.go index 5da2b01a0..79fffc68f 100644 --- a/backend/go/qwen3-tts-cpp/qwen3ttscpp_test.go +++ b/backend/go/qwen3-tts-cpp/qwen3ttscpp_test.go @@ -3,7 +3,9 @@ package main import ( "bytes" "encoding/binary" + pb "github.com/mudler/LocalAI/pkg/grpc/proto" "testing" + "unsafe" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -104,6 +106,22 @@ var _ = Describe("parseSampling", func() { }) }) +var _ = Describe("TTS request sampling", func() { + It("forwards a request generation bound to the native synthesizer", func() { + original := CppTTS + DeferCleanup(func() { CppTTS = original }) + var received int + CppTTS = func(_, _, _, _ string, _ unsafe.Pointer, _ int, _ string, _ int64, _ float32, _ int, _, _ float32, maxNew int, _ unsafe.Pointer) uintptr { + received = maxNew + return 0 + } + b := &Qwen3TtsCpp{} + err := b.TTS(&pb.TTSRequest{Text: "Bounded synthesis", Dst: "unused.wav", Params: map[string]string{"max_new_tokens": "128"}}) + Expect(err).To(MatchError("qwen3-tts: synthesis failed")) + Expect(received).To(Equal(128)) + }) +}) + var _ = Describe("wavHeader24k", func() { It("emits a 44-byte streaming WAV header at 24 kHz mono 16-bit", func() { h := wavHeader24k() diff --git a/backend/rust/kokoros/src/service.rs b/backend/rust/kokoros/src/service.rs index 512cc003b..6761a3d26 100644 --- a/backend/rust/kokoros/src/service.rs +++ b/backend/rust/kokoros/src/service.rs @@ -201,6 +201,15 @@ impl Backend for KokorosService { } } + type TTSLiveStream = ReceiverStream>; + + async fn tts_live( + &self, + _req: Request>, + ) -> Result, Status> { + Err(Status::unimplemented("incremental text TTS is not supported")) + } + type TTSStreamStream = ReceiverStream>; async fn tts_stream( @@ -302,6 +311,22 @@ impl Backend for KokorosService { // --- Unimplemented RPCs --- + type MotionStreamStream = ReceiverStream>; + + async fn motion_stream( + &self, + _: Request>, + ) -> Result, Status> { + Err(Status::unimplemented("Not supported")) + } + + async fn slot_cache( + &self, + _: Request, + ) -> Result, Status> { + Err(Status::unimplemented("Not supported")) + } + async fn predict( &self, _: Request, diff --git a/core/backend/tts_live.go b/core/backend/tts_live.go new file mode 100644 index 000000000..d81d2b01a --- /dev/null +++ b/core/backend/tts_live.go @@ -0,0 +1,30 @@ +package backend + +import ( + "context" + "fmt" + + "github.com/mudler/LocalAI/core/config" + grpcbackend "github.com/mudler/LocalAI/pkg/grpc" + "github.com/mudler/LocalAI/pkg/model" +) + +// ModelTTSLive streams one persistent utterance. Closing text commits input EOS; +// this call returns only after the final audio or a cancellation/error. +func ModelTTSLive(ctx context.Context, text <-chan string, voice, language string, params map[string]string, loader *model.ModelLoader, appConfig *config.ApplicationConfig, cfg config.ModelConfig, onAudio func([]byte) error) error { + tts, err := loader.Load(ModelOptions(cfg, appConfig, model.WithContext(ctx))...) + if err != nil { + return err + } + live, ok := tts.(grpcbackend.TTSLiveBackend) + if !ok { + return fmt.Errorf("backend %q does not support incremental text TTS", cfg.Backend) + } + release, err := AcquireGlobalBackendSlot() + if err != nil { + return err + } + defer release() + req := newTTSRequest("", cfg.Model, voice, "", language, "", params, cfg.Model) + return live.TTSLive(ctx, req, text, onAudio) +} diff --git a/core/config/model_config.go b/core/config/model_config.go index 247b97093..18c45df78 100644 --- a/core/config/model_config.go +++ b/core/config/model_config.go @@ -1204,8 +1204,8 @@ type PipelineStreaming struct { // ClauseChunking splits the streamed LLM reply into speakable clauses and // synthesizes each as soon as it completes, instead of buffering the whole // message before TTS. Script-aware (CJK/Thai), so it does not rely on - // whitespace sentence boundaries. Requires LLM streaming; unset buffers the - // whole message (today's default). + // whitespace sentence boundaries. Requires explicit true; unset and false + // retain whole-reply synthesis for engines without native live text input. ClauseChunking *bool `yaml:"clause_chunking,omitempty" json:"clause_chunking,omitempty"` } diff --git a/core/http/endpoints/openai/realtime_failover_test.go b/core/http/endpoints/openai/realtime_failover_test.go index 06dfda2b3..3caf22753 100644 --- a/core/http/endpoints/openai/realtime_failover_test.go +++ b/core/http/endpoints/openai/realtime_failover_test.go @@ -14,6 +14,8 @@ import ( "github.com/mudler/LocalAI/pkg/system" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) type rtSource map[string]config.ModelConfig @@ -43,6 +45,22 @@ var _ = Describe("realtime failover", func() { stageTargetConfig: func(name string) (*config.ModelConfig, error) { return &config.ModelConfig{Name: name}, nil }}} } + It("checks streaming support on resolved chain targets before backend loading", func() { + m := chainModel() + m.TTSConfig = &config.ModelConfig{Backend: "sherpa-onnx"} + var resolved []string + m.stageTargetConfig = func(name string) (*config.ModelConfig, error) { + resolved = append(resolved, name) + return &config.ModelConfig{Name: name, Backend: "piper"}, nil + } + Expect(m.SupportsTTSStream()).To(BeTrue()) + err := m.TTSStream(context.Background(), "Hello", "", "", func([]byte, int) error { return nil }) + Expect(status.Code(err)).To(Equal(codes.Unimplemented)) + Expect(resolved).To(Equal([]string{"a", "b"})) + m.TTSConfig.Backend = "piper" + Expect(m.SupportsTTSStream()).To(BeTrue()) + }) + It("routes a chain stage through the plan and retries before commit", func() { m := chainModel() var tried []string diff --git a/core/http/endpoints/openai/realtime_model.go b/core/http/endpoints/openai/realtime_model.go index 0e66e7751..f8de16dce 100644 --- a/core/http/endpoints/openai/realtime_model.go +++ b/core/http/endpoints/openai/realtime_model.go @@ -9,6 +9,7 @@ import ( "errors" "fmt" "maps" + "slices" "strings" "sync" "time" @@ -28,6 +29,8 @@ import ( "github.com/mudler/LocalAI/pkg/grpc/proto" model "github.com/mudler/LocalAI/pkg/model" "github.com/mudler/xlog" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) var ( @@ -534,6 +537,9 @@ func (m *wrappedModel) setTTSParams(params map[string]string) { func (m *wrappedModel) TTSStream(ctx context.Context, text, voice, language string, onAudio func(pcm []byte, sampleRate int) error) error { return m.stageCall(ctx, config.PipelineStageTTS, m.TTSConfig, func(cfg *config.ModelConfig, commit func()) error { + if !ttsBackendSupportsStreaming(cfg.Backend) { + return status.Error(codes.Unimplemented, "TTS backend does not support streaming audio") + } // Audio that reached the client cannot be taken back, so the first // chunk ends the retries. return ttsStream(ctx, m.modelLoader, m.appConfig, *cfg, text, voice, language, maps.Clone(m.ttsParams), func(pcm []byte, sr int) error { @@ -1228,3 +1234,53 @@ func liveVoiceOptions(ctx context.Context, registry voicerecognition.Registry, c } return []backend.LiveOption{backend.WithKnownVoices(voices)} } + +// SupportsTTSLive selects the native persistent-input path explicitly. Other +// engines retain their existing whole-text or clause synthesis behaviour. +func (m *wrappedModel) SupportsTTSLive() bool { + return m.TTSConfig != nil && m.TTSConfig.Backend == "qwen3-tts-cpp" +} +func (m *wrappedModel) TTSLive(ctx context.Context, text <-chan string, voice, language string, onAudio func([]byte, int) error) error { + return m.stageCall(ctx, config.PipelineStageTTS, m.TTSConfig, func(cfg *config.ModelConfig, commit func()) error { + // Text is consumed once. Retrying a session would omit consumed input or + // duplicate speech already heard, so fallback ends before consuming it. + commit() + var header []byte + rate := 0 + return backend.ModelTTSLive(ctx, text, voice, language, maps.Clone(m.ttsParams), m.modelLoader, m.appConfig, *cfg, func(b []byte) error { + if rate != 0 { + return onAudio(b, rate) + } + header = append(header, b...) + if len(header) < wavStreamHeaderBytes { + return nil + } + if string(header[:4]) != "RIFF" || string(header[8:12]) != "WAVE" { + return fmt.Errorf("invalid live TTS WAV header") + } + rate = int(binary.LittleEndian.Uint32(header[24:28])) + if rate == 0 { + return fmt.Errorf("invalid live TTS sample rate") + } + if len(header) > wavStreamHeaderBytes { + return onAudio(header[wavStreamHeaderBytes:], rate) + } + header = nil + return nil + }) + }) +} + +// SupportsTTSStream avoids probing unary-only engines through the legacy +// streaming RPC. Unknown backends retain their existing streaming probe. +func (m *wrappedModel) SupportsTTSStream() bool { + if m.isChainStage(config.PipelineStageTTS) || m.TTSConfig == nil { + return true + } + return ttsBackendSupportsStreaming(m.TTSConfig.Backend) +} + +func ttsBackendSupportsStreaming(backend string) bool { + capability := config.GetBackendCapability(backend) + return capability == nil || slices.Contains(capability.GRPCMethods, config.MethodTTSStream) +} diff --git a/core/http/endpoints/openai/realtime_speech.go b/core/http/endpoints/openai/realtime_speech.go index ec4bbc4b0..1039b0d79 100644 --- a/core/http/endpoints/openai/realtime_speech.go +++ b/core/http/endpoints/openai/realtime_speech.go @@ -10,6 +10,8 @@ import ( "github.com/mudler/LocalAI/core/http/endpoints/openai/types" laudio "github.com/mudler/LocalAI/pkg/audio" "github.com/mudler/LocalAI/pkg/sound" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) // emitSpeech synthesizes text and sends the audio to the client. When the @@ -31,6 +33,7 @@ func emitSpeech(ctx context.Context, t Transport, session *Session, responseID, _, isWebRTC := t.(*WebRTCTransport) + audioCommitted := false var wsAudio []byte // PCM at the session output rate, accumulated for the item record // sendChunk hands one PCM buffer to the transport: WebRTC consumes the raw @@ -40,6 +43,10 @@ func emitSpeech(ctx context.Context, t Transport, session *Session, responseID, if len(pcm) == 0 { return nil } + if err := ctx.Err(); err != nil { + return err + } + audioCommitted = true if err := t.SendAudio(ctx, pcm, sampleRate); err != nil { return err } @@ -68,11 +75,23 @@ func emitSpeech(ctx context.Context, t Transport, session *Session, responseID, language = session.InputAudioTranscription.Language } - if session.ModelConfig != nil && session.ModelConfig.Pipeline.StreamTTS() { - if err := session.ModelInterface.TTSStream(ctx, text, session.Voice, language, sendChunk); err != nil { + streamingAvailable := true + if capability, ok := session.ModelInterface.(interface{ SupportsTTSStream() bool }); ok { + streamingAvailable = capability.SupportsTTSStream() + } + if session.ModelConfig != nil && session.ModelConfig.Pipeline.StreamTTS() && streamingAvailable { + err := session.ModelInterface.TTSStream(ctx, text, session.Voice, language, sendChunk) + if err == nil { + return wsAudio, nil + } + // Only an unsupported RPC before audio may safely fall back. Once a + // transport sees PCM, replaying this clause could repeat heard speech. + if status.Code(err) != codes.Unimplemented || audioCommitted || ctx.Err() != nil { return nil, err } - return wsAudio, nil + } + if err := ctx.Err(); err != nil { + return nil, err } // Unary fallback: synthesize the whole utterance to a file, then emit once. diff --git a/core/http/endpoints/openai/realtime_speech_test.go b/core/http/endpoints/openai/realtime_speech_test.go index a501f946c..4ffc9a468 100644 --- a/core/http/endpoints/openai/realtime_speech_test.go +++ b/core/http/endpoints/openai/realtime_speech_test.go @@ -2,7 +2,12 @@ package openai import ( "context" + "fmt" + proto "github.com/mudler/LocalAI/pkg/grpc/proto" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "os" + "path/filepath" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -68,3 +73,50 @@ var _ = Describe("emitSpeech", func() { Expect(t.countEvents(types.ServerEventTypeResponseOutputAudioDelta)).To(Equal(1)) }) }) + +type unsupportedStreamingSpeech struct { + fakeModel + emitBeforeError bool + unaryCalls int +} + +func (m *unsupportedStreamingSpeech) TTSStream(_ context.Context, _, _, _ string, onAudio func([]byte, int) error) error { + if m.emitBeforeError { + if err := onAudio([]byte{1, 2}, 24000); err != nil { + return err + } + } + return status.Error(codes.Unimplemented, "streaming unavailable") +} +func (m *unsupportedStreamingSpeech) TTS(ctx context.Context, text, voice, language string) (string, *proto.Result, error) { + m.unaryCalls++ + return m.fakeModel.TTS(ctx, text, voice, language) +} + +var _ = Describe("TTS streaming capability fallback", func() { + for _, partial := range []bool{false, true} { + It(fmt.Sprintf("replays only unsupported streams without committed PCM (partial=%t)", partial), func() { + path := filepath.Join(GinkgoT().TempDir(), "speech.wav") + f, err := os.Create(path) + Expect(err).NotTo(HaveOccurred()) + header := laudio.NewWAVHeader(2) + Expect(header.Write(f)).To(Succeed()) + _, err = f.Write([]byte{3, 4}) + Expect(err).NotTo(HaveOccurred()) + Expect(f.Close()).To(Succeed()) + model := &unsupportedStreamingSpeech{fakeModel: fakeModel{ttsFile: path}, emitBeforeError: partial} + on := true + session := &Session{OutputSampleRate: 24000, ModelInterface: model, ModelConfig: &config.ModelConfig{Pipeline: config.Pipeline{Streaming: config.PipelineStreaming{TTS: &on}}}} + transport := &fakeTransport{} + _, err = emitSpeech(context.Background(), transport, session, "response", "item", "Hello") + if partial { + Expect(status.Code(err)).To(Equal(codes.Unimplemented)) + Expect(model.unaryCalls).To(BeZero()) + } else { + Expect(err).NotTo(HaveOccurred()) + Expect(model.unaryCalls).To(Equal(1)) + } + Expect(transport.countEvents(types.ServerEventTypeResponseOutputAudioDelta)).To(Equal(1)) + }) + } +}) diff --git a/core/http/endpoints/openai/realtime_stream.go b/core/http/endpoints/openai/realtime_stream.go index debe2de2b..1983a7461 100644 --- a/core/http/endpoints/openai/realtime_stream.go +++ b/core/http/endpoints/openai/realtime_stream.go @@ -16,10 +16,8 @@ import ( // transcriptStreamer turns streamed LLM tokens into the assistant's spoken // transcript: it strips reasoning incrementally and sends one // response.output_audio_transcript.delta per content fragment. It does NOT -// synthesize audio — the caller buffers the full message and synthesizes it -// once (streaming the audio chunks when the TTS backend supports TTSStream), -// which works uniformly for streaming and non-streaming TTS and for languages -// without sentence or word boundaries. +// synthesize audio. The caller sends cleaned content to a persistent live TTS +// session when supported, or uses the existing whole-text or clause path. type transcriptStreamer struct { ctx context.Context t Transport @@ -78,15 +76,18 @@ func (s *transcriptStreamer) content() string { } // streamLLMResponse drives a streamed realtime reply. It streams the assistant -// transcript as the LLM generates, then synthesizes the whole buffered message -// once (streaming the audio chunks when the TTS backend supports it, otherwise a -// single unary delta). Tool calls parsed from the autoparser ChatDeltas are +// transcript as the LLM generates and feeds a persistent live TTS session when +// available. Other engines keep whole-text or clause synthesis. Tool calls +// parsed from the autoparser ChatDeltas are // emitted after the spoken content. The assistant content item is created lazily // on the first content delta, so a content-less tool-call turn emits only the // tool calls. It returns true when it has fully handled the response so the // caller can return; callers must only invoke it for an audio modality, and with // tools only when the model uses its tokenizer template (see triggerResponseAtTurn). func streamLLMResponse(ctx context.Context, session *Session, conv *Conversation, t Transport, r *liveResponse, history schema.Messages, images []string, llmCfg *config.ModelConfig, tools []types.ToolUnion, toolChoice *types.ToolChoiceUnion, toolTurn int) bool { + callerCtx := ctx + ctx, cancelWork := context.WithCancel(ctx) + defer cancelWork() responseID := r.id itemID := generateItemID() item := types.MessageItemUnion{ @@ -163,37 +164,40 @@ func streamLLMResponse(ctx context.Context, session *Session, conv *Conversation streamer := newTranscriptStreamer(ctx, t, responseID, itemID, thinkingStartToken, reasoningCfg) streamer.announce = announce - // Clause chunking (opt-in): synthesize each clause as soon as it completes - // instead of buffering the whole reply. Synthesis runs on a worker goroutine - // (ttsPipeline) rather than inline in the token callback: emitSpeech blocks - // until the whole clause is synthesized (and, for WebRTC, played back at - // real time), and the callback runs on the goroutine that drains the LLM - // gRPC stream — so speaking inline stalls generation and freezes the - // assistant transcript at every clause boundary. The worker lets generation - // and the transcript stream keep flowing while audio is produced behind them. + // Synthesize completed clauses on a bounded worker so audio playback can + // apply backpressure without running synthesis inside the token callback. var chunker *clauseChunker - var ttsPipe *ttsPipeline - if session.ModelConfig != nil && session.ModelConfig.Pipeline.ChunkClauses() { + var livePipe *liveSpeechPipeline + liveModel, liveCapable := session.ModelInterface.(liveSpeechModel) + useLive := liveCapable && liveModel.SupportsTTSLive() && session.ModelConfig != nil && session.ModelConfig.Pipeline.StreamTTS() + defer func() { + if livePipe != nil { + livePipe.abort() + } + }() + if !useLive && session.ModelConfig != nil && session.ModelConfig.Pipeline.ChunkClauses() { chunker = newClauseChunker(defaultClauseMinRunes, defaultClauseMaxRunes) - ttsPipe = newTTSPipeline(func(clause string) ([]byte, error) { - return emitSpeech(ctx, t, session, responseID, itemID, clause) - }) + } + // Start synthesis only when the first speakable clause is available. + enqueueClause := func(clause string) bool { + if livePipe == nil { + livePipe = newLiveSpeechPipeline(ctx, func(workerCtx context.Context, input <-chan string) ([]byte, error) { + audio, err := emitBufferedSpeech(workerCtx, input, t, session, responseID, itemID) + if err != nil { + cancelWork() + } + return audio, err + }) + } + return livePipe.push(clause) } var streamedAudio []byte var ttsErr error - // Backstop: always join the TTS worker, even on an unexpected early return. - // wait() is idempotent, so the explicit drain below (which captures the - // streamed audio and first error) stays authoritative; this only guarantees - // the goroutine can never leak if a new return path is added. - if ttsPipe != nil { - defer func() { _, _ = ttsPipe.wait() }() - } - // fail reports a mid-stream failure. A cancelled context means the client // interrupted (barge-in), so roll the turn back instead of erroring. fail := func(code, msg string, err error) bool { - if ctx.Err() != nil { + if callerCtx.Err() != nil { cancel() } else { sendError(t, code, fmt.Sprintf("%s: %v", msg, err), "", itemID) @@ -215,13 +219,26 @@ func streamLLMResponse(ctx context.Context, session *Session, conv *Conversation text = functions.ContentFromChatDeltas(usage.ChatDeltas) } delta := streamer.onToken(text) + if useLive && delta != "" { + if livePipe == nil { + livePipe = newLiveSpeechPipeline(ctx, func(liveCtx context.Context, input <-chan string) ([]byte, error) { + audio, err := emitLiveSpeech(liveCtx, liveModel, input, t, session, responseID, itemID) + if err != nil { + cancelWork() + } + return audio, err + }) + } + if !livePipe.push(delta) { + return false + } + } if chunker != nil && delta != "" { for _, clause := range chunker.push(delta) { - // Hand the clause to the worker and keep going — never block the - // recv loop on synthesis. A false return means a prior clause - // already failed; stop the prediction (the error is collected - // from the pipeline after predFunc returns). - if !ttsPipe.enqueue(clause) { + // The clause queue applies backpressure when audio falls behind. + // A false return means cancellation or a prior synthesis failure; + // collect the worker error after prediction stops. + if !enqueueClause(clause) { return false } } @@ -239,17 +256,22 @@ func streamLLMResponse(ctx context.Context, session *Session, conv *Conversation // Drain the TTS worker. On a clean finish, enqueue the trailing clause(s) the // chunker was still holding; on an error or barge-in, stop synthesizing. - // wait() runs on every path so the worker goroutine never leaks, and it - // returns the audio streamed so far plus the first synthesis failure. - if ttsPipe != nil { + // The pipeline is joined below to collect audio and the first failure. + if chunker != nil { if err == nil && ctx.Err() == nil { for _, clause := range chunker.flush() { - if !ttsPipe.enqueue(clause) { + if !enqueueClause(clause) { break } } } - streamedAudio, ttsErr = ttsPipe.wait() + } + + if livePipe != nil { + if err != nil || ctx.Err() != nil { + livePipe.cancel() + } + streamedAudio, ttsErr = livePipe.finish() } // A clause synthesis failed mid-stream (the callback stopped the prediction); @@ -282,7 +304,7 @@ func streamLLMResponse(ctx context.Context, session *Session, conv *Conversation // synthesize it once now — emitSpeech streams the audio chunks when the // TTS backend supports TTSStream, otherwise it sends a single unary delta. var audio []byte - if chunker != nil { + if chunker != nil || livePipe != nil { audio = streamedAudio } else { audio, ttsErr = emitSpeech(ctx, t, session, responseID, itemID, content) diff --git a/core/http/endpoints/openai/realtime_stream_test.go b/core/http/endpoints/openai/realtime_stream_test.go index 5f160027e..32cefe342 100644 --- a/core/http/endpoints/openai/realtime_stream_test.go +++ b/core/http/endpoints/openai/realtime_stream_test.go @@ -2,6 +2,7 @@ package openai import ( "context" + "fmt" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -83,37 +84,39 @@ var _ = Describe("transcriptStreamer", func() { // streamLLMResponse drives a full streamed realtime turn: live transcript // deltas while the LLM generates, then the whole message is synthesized once. var _ = Describe("streamLLMResponse", func() { - It("streams transcript deltas then synthesizes the whole message once", func() { - on := true - m := &fakeModel{ - predictTokens: []string{"Hello", " world.", " How are you?"}, - predictResp: backend.LLMResponse{Response: "Hello world. How are you?"}, - ttsStreamChunks: [][]byte{{9}}, - ttsStreamRate: 24000, - } - session := &Session{ - OutputSampleRate: 24000, - ModelInterface: m, - ModelConfig: &config.ModelConfig{ - Pipeline: config.Pipeline{Streaming: config.PipelineStreaming{LLM: &on, TTS: &on}}, - }, - } - conv := &Conversation{} - t := &fakeTransport{} - llmCfg := &config.ModelConfig{} + for _, clauseFlag := range []*bool{nil, new(false)} { + It(fmt.Sprintf("buffers the whole reply with clause chunking %v", clauseFlag), func() { + on := true + m := &fakeModel{ + predictTokens: []string{"Hello", " world.", " How are you?"}, + predictResp: backend.LLMResponse{Response: "Hello world. How are you?"}, + ttsStreamChunks: [][]byte{{9}}, + ttsStreamRate: 24000, + } + session := &Session{ + OutputSampleRate: 24000, + ModelInterface: m, + ModelConfig: &config.ModelConfig{ + Pipeline: config.Pipeline{Streaming: config.PipelineStreaming{LLM: &on, TTS: &on, ClauseChunking: clauseFlag}}, + }, + } + conv := &Conversation{} + t := &fakeTransport{} + llmCfg := &config.ModelConfig{} - r := &liveResponse{id: "resp1"} - handled := streamLLMResponse(context.Background(), session, conv, t, r, nil, nil, llmCfg, nil, nil, 0) + r := &liveResponse{id: "resp1"} + handled := streamLLMResponse(context.Background(), session, conv, t, r, nil, nil, llmCfg, nil, nil, 0) - Expect(handled).To(BeTrue()) - // One live transcript delta per streamed token. - Expect(t.countEvents(types.ServerEventTypeResponseOutputAudioTranscriptDelta)).To(Equal(3)) - // The whole message is synthesized ONCE (not per sentence): a single - // emitSpeech replays the one TTS stream chunk. - Expect(t.countEvents(types.ServerEventTypeResponseOutputAudioDelta)).To(Equal(1)) - Expect(t.transcriptDeltaText()).To(Equal("Hello world. How are you?")) - }) + Expect(handled).To(BeTrue()) + // One live transcript delta per streamed token. + Expect(t.countEvents(types.ServerEventTypeResponseOutputAudioTranscriptDelta)).To(Equal(3)) + // The whole message is synthesized ONCE (not per sentence): a single + // emitSpeech replays the one TTS stream chunk. + Expect(t.countEvents(types.ServerEventTypeResponseOutputAudioDelta)).To(Equal(1)) + Expect(t.transcriptDeltaText()).To(Equal("Hello world. How are you?")) + }) + } It("synthesizes each clause as it completes when clause chunking is enabled", func() { on := true m := &fakeModel{ diff --git a/core/http/endpoints/openai/realtime_tts_buffered_test.go b/core/http/endpoints/openai/realtime_tts_buffered_test.go new file mode 100644 index 000000000..742b06ea1 --- /dev/null +++ b/core/http/endpoints/openai/realtime_tts_buffered_test.go @@ -0,0 +1,182 @@ +package openai + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "sync" + "time" + + "github.com/mudler/LocalAI/core/backend" + "github.com/mudler/LocalAI/core/config" + "github.com/mudler/LocalAI/core/http/endpoints/openai/types" + "github.com/mudler/LocalAI/core/schema" + laudio "github.com/mudler/LocalAI/pkg/audio" + pb "github.com/mudler/LocalAI/pkg/grpc/proto" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +type bufferedSpeechModel struct { + fakeModel + engine, dir string + clauses []string + early chan struct{} + fail error + cancel context.CancelFunc + streamCalls, unaryCalls int +} + +func (m *bufferedSpeechModel) SupportsTTSStream() bool { + return (&wrappedModel{TTSConfig: &config.ModelConfig{Backend: m.engine}}).SupportsTTSStream() +} +func (m *bufferedSpeechModel) Predict(ctx context.Context, _ schema.Messages, _, _, _ []string, cb func(string, backend.TokenUsage) bool, _ []types.ToolUnion, _ *types.ToolChoiceUnion, _, _ *int, _ map[string]float64) (func() (backend.LLMResponse, error), error) { + return func() (backend.LLMResponse, error) { + if !cb("First complete sentence. Next", backend.TokenUsage{}) { + return backend.LLMResponse{}, ctx.Err() + } + select { + case <-m.early: + case <-ctx.Done(): + return backend.LLMResponse{}, ctx.Err() + } + if !cb(" complete sentence. Final tail.", backend.TokenUsage{}) { + return backend.LLMResponse{}, ctx.Err() + } + return backend.LLMResponse{Response: "First complete sentence. Next complete sentence. Final tail."}, nil + }, nil +} +func (m *bufferedSpeechModel) TTSStream(ctx context.Context, text, _, _ string, onAudio func([]byte, int) error) error { + m.streamCalls++ + m.clauses = append(m.clauses, text) + if m.fail != nil { + return m.fail + } + if err := onAudio([]byte{byte(len(m.clauses)), 0}, 24000); err != nil { + return err + } + if len(m.clauses) == 1 { + if m.cancel != nil { + m.cancel() + } + } + return onAudio([]byte{byte(len(m.clauses)), 1}, 24000) +} +func (m *bufferedSpeechModel) TTS(ctx context.Context, text, _, _ string) (string, *pb.Result, error) { + m.unaryCalls++ + m.clauses = append(m.clauses, text) + if m.fail != nil { + return "", nil, m.fail + } + file, err := os.CreateTemp(m.dir, "clause-*.wav") + if err != nil { + return "", nil, err + } + header := laudio.NewWAVHeader(2) + if err := header.Write(file); err != nil { + _ = file.Close() + return "", nil, err + } + if _, err := file.Write([]byte{byte(len(m.clauses)), 0}); err != nil { + _ = file.Close() + return "", nil, err + } + if err := file.Close(); err != nil { + return "", nil, err + } + return filepath.Clean(file.Name()), &pb.Result{Success: true}, nil +} + +type bufferedAudioTransport struct { + fakeTransport + ready chan struct{} + once sync.Once +} + +func (t *bufferedAudioTransport) SendAudio(ctx context.Context, pcm []byte, rate int) error { + if err := t.fakeTransport.SendAudio(ctx, pcm, rate); err != nil { + return err + } + t.once.Do(func() { close(t.ready) }) + return nil +} + +var _ = Describe("Buffered progressive TTS", func() { + for _, tc := range []struct { + engine string + stream bool + }{{"sherpa-onnx", true}, {"piper", true}, {"sherpa-onnx", false}} { + engine := tc.engine + It(fmt.Sprintf("emits ordered clauses before LLM EOF with %s (stream=%t)", engine, tc.stream), func() { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + on := true + m := &bufferedSpeechModel{engine: engine, dir: GinkgoT().TempDir(), early: make(chan struct{})} + session := &Session{OutputSampleRate: 24000, ModelInterface: m, ModelConfig: &config.ModelConfig{Pipeline: config.Pipeline{Streaming: config.PipelineStreaming{LLM: &on, TTS: &tc.stream, ClauseChunking: &on}}}} + transport := &bufferedAudioTransport{ready: m.early} + r := &liveResponse{id: "buffered"} + streamLLMResponse(ctx, session, &Conversation{}, transport, r, nil, nil, &config.ModelConfig{}, nil, nil, 0) + Expect(ctx.Err()).NotTo(HaveOccurred(), "audio must reach the transport before the predictor can finish") + Expect(m.clauses).To(Equal([]string{"First complete sentence.", "Next complete sentence.", "Final tail."})) + if engine == "piper" || !tc.stream { + Expect(m.streamCalls).To(BeZero()) + Expect(m.unaryCalls).To(Equal(3)) + Expect(transport.audio).To(HaveLen(3)) + } else { + Expect(m.streamCalls).To(Equal(3)) + Expect(transport.audio).To(HaveLen(6)) + } + var sequence []byte + for _, chunk := range transport.audio { + sequence = append(sequence, chunk.pcm[0]) + } + if engine == "piper" || !tc.stream { + Expect(sequence).To(Equal([]byte{1, 2, 3})) + } else { + Expect(sequence).To(Equal([]byte{1, 1, 2, 2, 3, 3})) + } + }) + } + It("cancels a waiting predictor immediately on synthesis error", func() { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + on := true + m := &bufferedSpeechModel{engine: "sherpa-onnx", early: make(chan struct{}), fail: errors.New("synthesis failed")} + session := &Session{OutputSampleRate: 24000, ModelInterface: m, ModelConfig: &config.ModelConfig{Pipeline: config.Pipeline{Streaming: config.PipelineStreaming{LLM: &on, TTS: &on, ClauseChunking: &on}}}} + r := &liveResponse{id: "error"} + streamLLMResponse(ctx, session, &Conversation{}, &fakeTransport{}, r, nil, nil, &config.ModelConfig{}, nil, nil, 0) + Expect(ctx.Err()).NotTo(HaveOccurred()) + Expect(r.outcome).To(Equal(outcomeFailed)) + Expect(m.clauses).To(HaveLen(1)) + }) + It("stops queued speech on caller cancellation", func() { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + on := true + m := &bufferedSpeechModel{engine: "sherpa-onnx", early: make(chan struct{}), cancel: cancel} + session := &Session{OutputSampleRate: 24000, ModelInterface: m, ModelConfig: &config.ModelConfig{Pipeline: config.Pipeline{Streaming: config.PipelineStreaming{LLM: &on, TTS: &on, ClauseChunking: &on}}}} + r := &liveResponse{id: "cancel"} + streamLLMResponse(ctx, session, &Conversation{}, &fakeTransport{}, r, nil, nil, &config.ModelConfig{}, nil, nil, 0) + Expect(r.outcome).To(Equal(outcomeCancelled)) + Expect(m.clauses).To(HaveLen(1)) + }) +}) + +var _ = Describe("Buffered speech backpressure", func() { + It("bounds pending clauses and wakes a blocked producer on cancellation", func() { + p := newLiveSpeechPipeline(context.Background(), func(ctx context.Context, _ <-chan string) ([]byte, error) { <-ctx.Done(); return nil, ctx.Err() }) + defer p.abort() + for i := 0; i < cap(p.text); i++ { + Expect(p.push("queued clause")).To(BeTrue()) + } + pushed := make(chan bool, 1) + go func() { pushed <- p.push("blocked clause") }() + Consistently(pushed, 20*time.Millisecond).ShouldNot(Receive()) + p.cancel() + Eventually(pushed).Should(Receive(BeFalse())) + _, err := p.finish() + Expect(err).To(MatchError(context.Canceled)) + }) +}) diff --git a/core/http/endpoints/openai/realtime_tts_live.go b/core/http/endpoints/openai/realtime_tts_live.go new file mode 100644 index 000000000..652f27916 --- /dev/null +++ b/core/http/endpoints/openai/realtime_tts_live.go @@ -0,0 +1,106 @@ +package openai + +import ( + "context" + "encoding/base64" + "sync" + + "github.com/mudler/LocalAI/core/http/endpoints/openai/types" + "github.com/mudler/LocalAI/pkg/sound" +) + +type liveSpeechModel interface { + SupportsTTSLive() bool + TTSLive(context.Context, <-chan string, string, string, func([]byte, int) error) error +} + +// liveSpeechPipeline bounds pending text and couples producer lifetime to the +// audio sink: cancellation/error wakes producers and the native input callback. +type liveSpeechPipeline struct { + text chan string + done chan struct{} + cancel context.CancelFunc + once sync.Once + stop <-chan struct{} + audio []byte + err error +} + +func newLiveSpeechPipeline(ctx context.Context, run func(context.Context, <-chan string) ([]byte, error)) *liveSpeechPipeline { + ctx, cancel := context.WithCancel(ctx) + p := &liveSpeechPipeline{text: make(chan string, 8), done: make(chan struct{}), stop: ctx.Done(), cancel: cancel} + go func() { defer close(p.done); defer cancel(); p.audio, p.err = run(ctx, p.text) }() + return p +} +func (p *liveSpeechPipeline) push(text string) bool { + select { + case <-p.stop: + return false + default: + } + select { + case <-p.stop: + return false + case p.text <- text: + return true + } +} +func (p *liveSpeechPipeline) finish() ([]byte, error) { + p.once.Do(func() { close(p.text) }) + <-p.done + return p.audio, p.err +} +func (p *liveSpeechPipeline) abort() { p.cancel(); _, _ = p.finish() } + +func emitLiveSpeech(ctx context.Context, model liveSpeechModel, text <-chan string, t Transport, session *Session, responseID, itemID string) ([]byte, error) { + var audio []byte + language := "" + if session.InputAudioTranscription != nil { + language = session.InputAudioTranscription.Language + } + _, webRTC := t.(*WebRTCTransport) + err := model.TTSLive(ctx, text, session.Voice, language, func(pcm []byte, rate int) error { + if len(pcm) == 0 { + return nil + } + if err := t.SendAudio(ctx, pcm, rate); err != nil { + return err + } + if webRTC { + return nil + } + if rate != 0 && rate != session.OutputSampleRate { + pcm = sound.Int16toBytesLE(sound.ResampleInt16(sound.BytesToInt16sLE(pcm), rate, session.OutputSampleRate)) + } + audio = append(audio, pcm...) + return t.SendEvent(types.ResponseOutputAudioDeltaEvent{ResponseID: responseID, ItemID: itemID, OutputIndex: 0, ContentIndex: 0, Delta: base64.StdEncoding.EncodeToString(pcm)}) + }) + return audio, err +} + +// emitBufferedSpeech synthesizes independent complete clauses in order. Each +// clause can stream its own backend audio; this is not a native live session. +func emitBufferedSpeech(ctx context.Context, input <-chan string, t Transport, session *Session, responseID, itemID string) ([]byte, error) { + var audio []byte + for { + if err := ctx.Err(); err != nil { + return audio, err + } + select { + case <-ctx.Done(): + return audio, ctx.Err() + case clause, ok := <-input: + if !ok { + return audio, nil + } + if err := ctx.Err(); err != nil { + return audio, err + } + chunk, err := emitSpeech(ctx, t, session, responseID, itemID, clause) + audio = append(audio, chunk...) + if err != nil { + return audio, err + } + } + } +} diff --git a/core/http/endpoints/openai/realtime_tts_live_test.go b/core/http/endpoints/openai/realtime_tts_live_test.go new file mode 100644 index 000000000..e74f2e8ac --- /dev/null +++ b/core/http/endpoints/openai/realtime_tts_live_test.go @@ -0,0 +1,140 @@ +package openai + +import ( + "context" + "errors" + "fmt" + . "github.com/onsi/ginkgo/v2" + "reflect" + "time" + + "github.com/mudler/LocalAI/core/backend" + "github.com/mudler/LocalAI/core/config" + "github.com/mudler/LocalAI/core/http/endpoints/openai/types" + "github.com/mudler/LocalAI/core/schema" +) + +type liveRouteModel struct { + fakeModel + early chan struct{} + received []string + calls int + failure error + failAfterAudio bool +} + +func (*liveRouteModel) SupportsTTSLive() bool { return true } +func (m *liveRouteModel) TTSLive(ctx context.Context, text <-chan string, _, _ string, audio func([]byte, int) error) error { + m.calls++ + if m.failure != nil && !m.failAfterAudio { + return m.failure + } + for { + select { + case <-ctx.Done(): + return ctx.Err() + case part, ok := <-text: + if !ok { + return nil + } + m.received = append(m.received, part) + if err := audio([]byte{byte(len(m.received)), 0}, 24000); err != nil { + return err + } + if len(m.received) == 1 { + close(m.early) + } + if m.failure != nil { + return m.failure + } + } + } +} +func (m *liveRouteModel) Predict(ctx context.Context, _ schema.Messages, _, _, _ []string, cb func(string, backend.TokenUsage) bool, _ []types.ToolUnion, _ *types.ToolChoiceUnion, _, _ *int, _ map[string]float64) (func() (backend.LLMResponse, error), error) { + return func() (backend.LLMResponse, error) { + if !cb("First ", backend.TokenUsage{}) { + return backend.LLMResponse{}, errors.New("callback stopped") + } + // No second text, input EOS, or completed LLM response exists at this point. + select { + case <-m.early: + case <-ctx.Done(): + return backend.LLMResponse{}, ctx.Err() + } + if !cb("late text.", backend.TokenUsage{}) { + return backend.LLMResponse{}, errors.New("callback stopped") + } + return backend.LLMResponse{Response: "First late text."}, nil + }, nil +} + +var _ = Describe("TTSLiveRealtimeRoute", func() { + for _, clauseFlag := range []*bool{nil, new(false), new(true)} { + It(fmt.Sprintf("preserves native synthesis with clause chunking %v", clauseFlag), func() { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + on := true + m := &liveRouteModel{early: make(chan struct{})} + session := &Session{OutputSampleRate: 24000, ModelInterface: m, ModelConfig: &config.ModelConfig{Pipeline: config.Pipeline{Streaming: config.PipelineStreaming{LLM: &on, TTS: &on, ClauseChunking: clauseFlag}}}} + transport := &fakeTransport{} + if !streamLLMResponse(ctx, session, &Conversation{}, transport, &liveResponse{id: "live"}, nil, nil, &config.ModelConfig{}, nil, nil, 0) { + Fail(fmt.Sprint("route unhandled")) + } + if ctx.Err() != nil { + Fail(fmt.Sprint("no audio before LLM finished", ctx.Err())) + } + if !reflect.DeepEqual(m.received, []string{"First ", "late text."}) || m.calls != 1 { + Fail(fmt.Sprintf("received %v in %d sessions", m.received, m.calls)) + } + if len(transport.audio) != 2 { + Fail(fmt.Sprintf("audio chunks %d", len(transport.audio))) + } + }) + } +}) +var _ = Describe("TTSLivePipelineCancellation", func() { + It("preserves the live synthesis contract", func() { + started := make(chan struct{}) + stopped := make(chan struct{}) + p := newLiveSpeechPipeline(context.Background(), func(ctx context.Context, _ <-chan string) ([]byte, error) { + close(started) + defer close(stopped) + <-ctx.Done() + return nil, ctx.Err() + }) + <-started + p.abort() + select { + case <-stopped: + case <-time.After(time.Second): + Fail(fmt.Sprint("worker leaked")) + } + if p.push("late") { + Fail(fmt.Sprint("accepted text after cancellation")) + } + }) +}) + +var _ = Describe("TTSLiveRealtimeFailureCancelsPrediction", func() { + It("preserves the live synthesis contract", func() { + for _, afterAudio := range []bool{false, true} { + By(map[bool]string{false: "before PCM", true: "after PCM"}[afterAudio]) + func() { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + on := true + m := &liveRouteModel{early: make(chan struct{}), failure: errors.New("native failed"), failAfterAudio: afterAudio} + session := &Session{OutputSampleRate: 24000, ModelInterface: m, ModelConfig: &config.ModelConfig{Pipeline: config.Pipeline{Streaming: config.PipelineStreaming{LLM: &on, TTS: &on}}}} + transport := &fakeTransport{} + r := &liveResponse{id: "failure"} + streamLLMResponse(ctx, session, &Conversation{}, transport, r, nil, nil, &config.ModelConfig{}, nil, nil, 0) + if ctx.Err() != nil { + Fail(fmt.Sprint("TTS failure did not cancel in-flight prediction promptly")) + } + if r.outcome != outcomeFailed || m.calls != 1 { + Fail(fmt.Sprintf("outcome=%v sessions=%d", r.outcome, m.calls)) + } + }() + } + }) +}) diff --git a/core/services/nodes/file_staging_client.go b/core/services/nodes/file_staging_client.go index e03671a15..a80de9fa1 100644 --- a/core/services/nodes/file_staging_client.go +++ b/core/services/nodes/file_staging_client.go @@ -726,3 +726,30 @@ func copyFile(src, dst string) error { } return closeErr } + +func (f *FileStagingClient) TTSLive(ctx context.Context, in *pb.TTSRequest, text <-chan string, fn func([]byte) error) error { + lifecycle := f.newStagedInputLifecycle() + defer lifecycle.release() + in = proto.Clone(in).(*pb.TTSRequest) + + // Translate model path from frontend to remote worker path (same as TTS above) + if in.Model != "" && isFilePath(in.Model) { + in.Model = f.translateModelPath(in.Model) + } + if in.Voice != "" && isFilePath(in.Voice) { + backendPath, err := f.stageInputFile(ctx, lifecycle, in.Voice, "inputs") + if err != nil { + return fmt.Errorf("staging streaming TTS voice reference: %w", err) + } + in.Voice = backendPath + } + if err := f.stageTTSReferences(ctx, lifecycle, in); err != nil { + return err + } + + live, ok := f.Backend.(grpc.TTSLiveBackend) + if !ok { + return fmt.Errorf("backend does not support incremental text TTS") + } + return live.TTSLive(ctx, in, text, fn) +} diff --git a/core/services/nodes/inflight.go b/core/services/nodes/inflight.go index 08df8411a..afe798f66 100644 --- a/core/services/nodes/inflight.go +++ b/core/services/nodes/inflight.go @@ -2,6 +2,7 @@ package nodes import ( "context" + "fmt" "sync" "time" @@ -294,3 +295,12 @@ func (c *InFlightTrackingClient) AudioTransform(ctx context.Context, in *pb.Audi // the done() func tied to stream close, which the Backend interface doesn't surface // here. If they ever need tracking, move them to grpc.InferenceBackend - the build // will then force an explicit wrapper here. + +func (c *InFlightTrackingClient) TTSLive(ctx context.Context, in *pb.TTSRequest, text <-chan string, fn func([]byte) error) error { + defer c.track(ctx)() + live, ok := c.inner.(grpc.TTSLiveBackend) + if !ok { + return fmt.Errorf("backend does not support incremental text TTS") + } + return c.reconcile(live.TTSLive(ctx, in, text, fn)) +} diff --git a/docs/content/features/openai-realtime.md b/docs/content/features/openai-realtime.md index 7f3f68b83..96f9c9e0b 100644 --- a/docs/content/features/openai-realtime.md +++ b/docs/content/features/openai-realtime.md @@ -33,6 +33,68 @@ Make sure all referenced models (`silero-vad-ggml`, `whisper-large-turbo`, `qwen A pipeline stage can name a [failover chain]({{%relref "features/model-failover" %}}); the stage then switches targets without closing the session. +### Install the Qwen streaming assistant preset + +The gallery entry `qwen3-streaming-assistant` configures an English voice +assistant with Silero VAD, Parakeet TDT 0.6B v3, Qwen3 0.6B, and Qwen3-TTS +0.6B CustomVoice using Ryan's voice. It streams LLM output text into one persistent TTS session while audio +is emitted. + +The preset installs only its pipeline configuration. Install all four component +models first, keeping their default names so the pipeline can resolve them: + +```bash +local-ai models install silero-vad-ggml +local-ai models install parakeet-cpp-tdt-0.6b-v3 +local-ai models install qwen3-0.6b +local-ai models install qwen3-tts-cpp-customvoice --variant qwen3-tts-cpp-customvoice +local-ai models install qwen3-streaming-assistant +``` + +The TTS command pins the Q8 CustomVoice entry. These are separate installs; +installing the preset does not install missing components. Use a LocalAI build +with incremental Qwen3-TTS input support, then select `qwen3-streaming-assistant` +on the Talk page or as the Realtime API model. WebRTC also requires the `opus` +backend, available from the Backends page. + +This preset targets English speech. It uses a named voice and does not require +reference audio or `ref_text`. CPU synthesis can be slower than playback; this +preset does not establish GPU or browser latency or throughput. + +### Install a clause-buffered assistant preset + +Two English presets use the same Silero VAD, Parakeet TDT 0.6B v3, and Qwen3 +0.6B components. `vits-streaming-assistant` uses VITS-LJS through Sherpa; +`piper-streaming-assistant` uses Piper's Amy voice. + +Install the shared components once, keeping their default names: + +```bash +local-ai models install silero-vad-ggml +local-ai models install parakeet-cpp-tdt-0.6b-v3 +local-ai models install qwen3-0.6b +``` + +Then install either TTS component and its matching preset, or install both: + +```bash +local-ai models install vits-ljs-sherpa +local-ai models install vits-streaming-assistant +local-ai models install voice-en-us-amy-low +local-ai models install piper-streaming-assistant +``` + +These presets install pipeline configurations only. They do not install missing +components. Both enable LLM streaming, TTS streaming, and `clause_chunking: true`. +Sherpa streams audio within each clause. Piper emits one audio result per clause. +Each clause starts a separate synthesis, which can introduce pauses between clauses. +The single-speaker models need no voice override or reference audio. + +Select the installed preset on the Talk page or use its name as the Realtime API +model. WebSocket and WebRTC share the pipeline; WebRTC also needs the `opus` +backend from the Backends page. Config installation and discovery checks do not +establish speech quality, browser latency, or hardware throughput. + ### Streaming the pipeline By default each stage runs to completion before the next begins: the whole utterance is transcribed, the full LLM reply is generated, then it is synthesized. Each stage can instead be streamed incrementally, which lowers the time-to-first-audio of a turn: @@ -53,10 +115,11 @@ pipeline: - **streaming.tts**: emit a `response.output_audio.delta` per audio chunk the TTS backend produces (requires a backend that supports streaming synthesis), instead of one delta for the whole utterance. Falls back to a single unary delta otherwise. - **streaming.transcription**: stream `conversation.item.input_audio_transcription.delta` events as the transcript is produced (requires a transcription backend that supports streaming). -- **streaming.llm**: stream the LLM reply token-by-token as `response.output_audio_transcript.delta` events. The full reply is buffered and synthesized once it is complete - streamed as audio chunks when `streaming.tts` is enabled (and the TTS backend supports it), otherwise as a single unary delta. Reasoning/thinking is always stripped from the spoken transcript. Tool calls are supported while streaming when the LLM uses its tokenizer template (`use_tokenizer_template: true`): the backend's autoparser then delivers content and tool calls separately, so the spoken transcript never leaks tool-call tokens. Grammar-based function calling keeps the buffered path. -- **streaming.clause_chunking**: instead of buffering the whole reply before TTS, split it into speakable clauses and synthesize each as soon as it completes, lowering the time-to-first-audio. The splitter is script-aware: it uses Unicode sentence segmentation (so it handles CJK `。！？` with no whitespace), CJK clause punctuation (`，、；：`), and Thai/Lao spaces - it does **not** rely on whitespace sentence boundaries, so it works for languages such as Chinese, Japanese and Thai where the old per-sentence approach degraded to whole-message buffering. Requires `streaming.llm`; scripts that genuinely need a dictionary (e.g. Khmer, Burmese) simply stay buffered until a space or end-of-message. Off by default. +- **streaming.llm**: stream the LLM reply token-by-token as `response.output_audio_transcript.delta` events. When TTS streaming is disabled, the full reply is synthesized once it is complete unless explicit clause chunking is enabled. Native incremental input or explicitly enabled clause buffering can produce audio before the reply finishes, as described below. Reasoning/thinking is always stripped from the spoken transcript. Tool calls are supported while streaming when the LLM uses its tokenizer template (`use_tokenizer_template: true`): the backend's autoparser then delivers content and tool calls separately, so the spoken transcript never leaks tool-call tokens. Grammar-based function calling keeps the buffered path. +- **Incremental Qwen3-TTS input**: with a `qwen3-tts-cpp` TTS model, enabling both `streaming.llm` and `streaming.tts` forwards cleaned LLM text fragments into one persistent synthesis session. Audio can arrive before the LLM finishes. WebSocket and WebRTC use this path. It takes precedence over `clause_chunking`; other backends can opt into clause buffering as described below. Pending text is bounded and slow audio playback applies backpressure. Cancellation stops the session. A failed live utterance is not replayed through a fallback model. The initial native implementation supports a single batch slot and reference audio without `ref_text`; transcript-based ICL cloning is unavailable for live input. Incomplete words are buffered until a stable boundary; CJK, Kana and Hangul characters can commit without whitespace. +- **streaming.clause_chunking**: instead of buffering the whole reply before TTS, split it into speakable clauses and synthesize each as soon as it completes, lowering the time-to-first-audio. The splitter is script-aware: it uses Unicode sentence segmentation (so it handles CJK `。！？` with no whitespace), CJK clause punctuation (`，、；：`), and Thai/Lao spaces - it does **not** rely on whitespace sentence boundaries, so it works for languages such as Chinese, Japanese and Thai where the old per-sentence approach degraded to whole-message buffering. Requires `streaming.llm`; scripts that genuinely need a dictionary (e.g. Khmer, Burmese) simply stay buffered until a space or end-of-message. Set `clause_chunking: true` to enable a bounded clause queue. Unset and false retain whole-reply synthesis for backends without native incremental text input. With `streaming.tts` enabled, Sherpa/VITS can stream audio within each clause; unary engines such as Piper produce one audio result per clause. With `streaming.tts` disabled, each clause uses unary synthesis. These engines start a separate synthesis for each clause, so they may pause at clause boundaries. Cancellation stops queued work; a streaming failure after audio has been emitted never replays that clause through unary synthesis. -All streaming flags are off by default, so existing pipelines are unaffected. +All streaming flags are off by default. Pipelines with streaming disabled remain unaffected. ### Model warm-up (cold start) diff --git a/gallery/index.yaml b/gallery/index.yaml index 8da50d158..236bf3c99 100644 --- a/gallery/index.yaml +++ b/gallery/index.yaml @@ -23949,6 +23949,104 @@ - filename: vibevoice-cpp-asr/tokenizer.gguf sha256: 37dc3b722d5677e37e29a57df55aa05c485116eeb5459e57ff8dde616b4986f6 uri: huggingface://mudler/vibevoice.cpp-models/tokenizer.gguf +- name: vits-streaming-assistant + urls: + - https://localai.io/features/openai-realtime/ + description: | + English voice assistant preset: Silero VAD, Parakeet TDT 0.6B v3, Qwen3 + 0.6B chat, and VITS-LJS English single-speaker TTS through sherpa-onnx. + Buffers streamed LLM output into clauses and emits audio progressively. + Each clause starts a separate synthesis; this is not native incremental + text input. This installs only the pipeline configuration. First install + silero-vad-ggml, parakeet-cpp-tdt-0.6b-v3, qwen3-0.6b, and vits-ljs-sherpa. + WebRTC also needs the opus backend. No browser or hardware performance + is established by this preset. + license: mit + tags: + - voice-assistant + - audio-pipeline + - realtime + - streaming + - english + - sherpa-onnx + config_file: + pipeline: + vad: silero-vad-ggml + transcription: parakeet-cpp-tdt-0.6b-v3 + llm: qwen3-0.6b + tts: vits-ljs-sherpa + disable_thinking: true + streaming: + llm: true + tts: true + clause_chunking: true +- name: piper-streaming-assistant + urls: + - https://localai.io/features/openai-realtime/ + description: | + English voice assistant preset: Silero VAD, Parakeet TDT 0.6B v3, Qwen3 + 0.6B chat, and Piper Amy US English TTS through piper. + Buffers streamed LLM output into clauses and emits audio progressively. + Each clause starts a separate synthesis; this is not native incremental + text input. This installs only the pipeline configuration. First install + silero-vad-ggml, parakeet-cpp-tdt-0.6b-v3, qwen3-0.6b, and voice-en-us-amy-low. + WebRTC also needs the opus backend. No browser or hardware performance + is established by this preset. + license: mit + tags: + - voice-assistant + - audio-pipeline + - realtime + - streaming + - english + - piper + config_file: + pipeline: + vad: silero-vad-ggml + transcription: parakeet-cpp-tdt-0.6b-v3 + llm: qwen3-0.6b + tts: voice-en-us-amy-low + disable_thinking: true + streaming: + llm: true + tts: true + clause_chunking: true +- name: qwen3-streaming-assistant + urls: + - https://github.com/localai-org/qwentts.cpp + - https://localai.io/features/openai-realtime/ + description: | + English voice assistant preset: Silero VAD, Parakeet TDT 0.6B v3, Qwen3 + 0.6B chat, and Qwen3-TTS 0.6B CustomVoice with Ryan's voice. Streams LLM + text into one persistent TTS session while audio is emitted. + This installs only the pipeline configuration. First install the separate + gallery models silero-vad-ggml, parakeet-cpp-tdt-0.6b-v3, qwen3-0.6b, and + qwen3-tts-cpp-customvoice. Pin its variant to qwen3-tts-cpp-customvoice + for Q8. WebRTC also needs the opus backend. + Requires a LocalAI build with incremental Qwen3-TTS input support. + CPU synthesis may be slower than playback; GPU and browser performance + are not established by this preset. + license: mit + tags: + - voice-assistant + - audio-pipeline + - realtime + - streaming + - english + - qwen3 + - qwen3-tts + config_file: + pipeline: + vad: silero-vad-ggml + transcription: parakeet-cpp-tdt-0.6b-v3 + llm: qwen3-0.6b + tts: qwen3-tts-cpp-customvoice + disable_thinking: true + streaming: + llm: true + tts: true + tts: + voice: ryan - &qwenttscpp_gallery name: qwen3-tts-cpp variants: diff --git a/pkg/grpc/tts_live.go b/pkg/grpc/tts_live.go new file mode 100644 index 000000000..5f859621e --- /dev/null +++ b/pkg/grpc/tts_live.go @@ -0,0 +1,167 @@ +package grpc + +import ( + "context" + "errors" + "io" + "unicode/utf8" + + pb "github.com/mudler/LocalAI/pkg/grpc/proto" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// TTSLiveModel optionally accepts text during a single persistent synthesis. +// recv must unblock when ctx is cancelled. io.EOF commits the final text. +// send applies backpressure and must return its first error to the caller. +type TTSLiveModel interface { + TTSLive(context.Context, func() (*pb.TTSRequest, error), func([]byte) error) error +} + +// TTSLiveBackend is optional: older backends retain their existing TTS APIs. +// Closing text signals end of input; audio may continue until this call returns. +type TTSLiveBackend interface { + TTSLive(context.Context, *pb.TTSRequest, <-chan string, func([]byte) error) error +} + +func (s *server) TTSLive(stream pb.Backend_TTSLiveServer) error { + model, ok := s.llm.(TTSLiveModel) + if !ok { + return status.Error(codes.Unimplemented, "backend does not support incremental text TTS") + } + first, err := stream.Recv() + if err != nil { + return err + } + if err := s.checkModelIdentity(first); err != nil { + return err + } + if s.llm.Locking() { + s.llm.Lock() + defer s.llm.Unlock() + } + pending := true + return model.TTSLive(stream.Context(), func() (*pb.TTSRequest, error) { + if pending { + pending = false + return first, nil + } + req, err := stream.Recv() + if err != nil { + return nil, err + } + if req.Voice != "" || req.Model != "" || req.Language != nil || req.Instructions != nil || len(req.Params) > 0 || req.Dst != "" || req.ModelIdentity != "" { + return nil, status.Error(codes.InvalidArgument, "only text is permitted after the first TTSLive frame") + } + return req, nil + }, func(pcm []byte) error { return stream.Send(&pb.Reply{Audio: pcm}) }) +} + +func (c *Client) TTSLive(ctx context.Context, config *pb.TTSRequest, text <-chan string, send func([]byte) error) (returnErr error) { + parentCtx := ctx + if !c.parallel { + if !c.opMutex.TryLock() { + return status.Error(codes.ResourceExhausted, "backend is busy") + } + defer c.opMutex.Unlock() + } + c.setBusy(true) + defer c.setBusy(false) + complete := c.wdMark() + defer complete() + conn, err := c.dial() + if err != nil { + return err + } + defer func() { _ = conn.Close() }() + ctx, cancel := context.WithCancel(ctx) + defer cancel() + stream, err := pb.NewBackendClient(conn).TTSLive(ctx) + if err != nil { + return err + } + if err := stream.Send(config); err != nil { + return err + } + sender := make(chan error, 1) + go func() { + for { + select { + case <-ctx.Done(): + sender <- ctx.Err() + return + case chunk, ok := <-text: + if !ok { + err := stream.CloseSend() + if err != nil { + cancel() + } + sender <- err + return + } + if !utf8.ValidString(chunk) { + sender <- status.Error(codes.InvalidArgument, "TTSLive text fragments must be valid UTF-8") + cancel() + return + } + if err := stream.Send(&pb.TTSRequest{Text: chunk}); err != nil { + sender <- err + cancel() + return + } + } + } + }() + defer func() { + cancel() + sendErr := <-sender + if sendErr != nil && !errors.Is(sendErr, context.Canceled) && (returnErr == nil || (status.Code(returnErr) == codes.Canceled && parentCtx.Err() == nil)) { + returnErr = sendErr + } + }() + for { + reply, err := stream.Recv() + if err == io.EOF { + return nil + } + if err != nil { + return err + } + if err := send(reply.Audio); err != nil { + return err + } + } +} + +func (e *embedBackend) TTSLive(ctx context.Context, config *pb.TTSRequest, text <-chan string, send func([]byte) error) error { + model, ok := e.s.llm.(TTSLiveModel) + if !ok { + return status.Error(codes.Unimplemented, "backend does not support incremental text TTS") + } + if err := e.s.checkModelIdentity(config); err != nil { + return err + } + if e.s.llm.Locking() { + e.s.llm.Lock() + defer e.s.llm.Unlock() + } + pending := true + return model.TTSLive(ctx, func() (*pb.TTSRequest, error) { + if pending { + pending = false + return config, nil + } + select { + case <-ctx.Done(): + return nil, ctx.Err() + case chunk, ok := <-text: + if !utf8.ValidString(chunk) { + return nil, status.Error(codes.InvalidArgument, "TTSLive text fragments must be valid UTF-8") + } + if !ok { + return nil, io.EOF + } + return &pb.TTSRequest{Text: chunk}, nil + } + }, send) +} diff --git a/pkg/grpc/tts_live_test.go b/pkg/grpc/tts_live_test.go new file mode 100644 index 000000000..8d9d43640 --- /dev/null +++ b/pkg/grpc/tts_live_test.go @@ -0,0 +1,150 @@ +package grpc + +import ( + "context" + "errors" + "fmt" + . "github.com/onsi/ginkgo/v2" + "io" + "net" + "time" + + "github.com/mudler/LocalAI/pkg/grpc/base" + pb "github.com/mudler/LocalAI/pkg/grpc/proto" + ggrpc "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +type liveTTSFixture struct{ base.SingleThread } + +func (*liveTTSFixture) TTSLive(ctx context.Context, recv func() (*pb.TTSRequest, error), send func([]byte) error) error { + for { + r, e := recv() + if e == io.EOF { + return send([]byte("tail")) + } + if e != nil { + return e + } + if r.Text == "fail" { + return errors.New("synthesis failed") + } + if r.Text != "" { + if e := send([]byte(r.Text)); e != nil { + return e + } + } + } +} + +var _ = Describe("TTSLiveEarlyAudioAndTail", func() { + It("preserves the live synthesis contract", func() { + listener, e := net.Listen("tcp", "127.0.0.1:0") + if e != nil { + Fail(fmt.Sprint(e)) + } + s := ggrpc.NewServer(serverOpts()...) + pb.RegisterBackendServer(s, &server{llm: &liveTTSFixture{}}) + go func() { _ = s.Serve(listener) }() + defer s.Stop() + c := NewClient(listener.Addr().String(), true, nil, false).(TTSLiveBackend) + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + text := make(chan string) + audio := make(chan string) + done := make(chan error, 1) + go func() { + done <- c.TTSLive(ctx, &pb.TTSRequest{}, text, func(b []byte) error { + select { + case audio <- string(b): + return nil + case <-ctx.Done(): + return ctx.Err() + } + }) + }() + text <- "early" + if got := <-audio; got != "early" { + Fail(fmt.Sprint(got)) + } + // Input deliberately remains open until audio has reached the receiver. + text <- "late" + if got := <-audio; got != "late" { + Fail(fmt.Sprint(got)) + } + close(text) + if got := <-audio; got != "tail" { + Fail(fmt.Sprint(got)) + } + if e := <-done; e != nil { + Fail(fmt.Sprint(e)) + } + }) +}) +var _ = Describe("TTSLiveCancellationAndSinkError", func() { + It("preserves the live synthesis contract", func() { + for _, sinkError := range []bool{false, true} { + By(map[bool]string{true: "sink error", false: "cancel waiting input"}[sinkError]) + func() { + listener, e := net.Listen("tcp", "127.0.0.1:0") + if e != nil { + Fail(fmt.Sprint(e)) + } + s := ggrpc.NewServer(serverOpts()...) + pb.RegisterBackendServer(s, &server{llm: &liveTTSFixture{}}) + go func() { _ = s.Serve(listener) }() + defer s.Stop() + c := NewClient(listener.Addr().String(), false, nil, false).(TTSLiveBackend) + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + text := make(chan string) + done := make(chan error, 1) + sinkErr := errors.New("sink failed") + go func() { + done <- c.TTSLive(ctx, &pb.TTSRequest{Text: "start"}, text, func([]byte) error { + if sinkError { + return sinkErr + } + cancel() + return nil + }) + }() + if e := <-done; e == nil { + Fail(fmt.Sprint("expected error")) + } else if sinkError && !errors.Is(e, sinkErr) { + Fail(fmt.Sprint(e)) + } + if c.(*Client).IsBusy() { + Fail(fmt.Sprint("backend remains busy")) + } + }() + } + }) +}) + +var _ = Describe("TTSLiveRejectsInvalidUnicode", func() { + It("preserves the live synthesis contract", func() { + listener, e := net.Listen("tcp", "127.0.0.1:0") + if e != nil { + Fail(fmt.Sprint(e)) + } + s := ggrpc.NewServer(serverOpts()...) + pb.RegisterBackendServer(s, &server{llm: &liveTTSFixture{}}) + go func() { _ = s.Serve(listener) }() + defer s.Stop() + c := NewClient(listener.Addr().String(), true, nil, false).(TTSLiveBackend) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + text := make(chan string, 1) + text <- string([]byte{0xe4, 0xb8}) + close(text) + e = c.TTSLive(ctx, &pb.TTSRequest{}, text, func([]byte) error { return nil }) + if status.Code(e) != codes.InvalidArgument { + Fail(fmt.Sprintf("invalid text: %v", e)) + } + if ctx.Err() != nil { + Fail(fmt.Sprint("invalid input left receiver blocked")) + } + }) +}) diff --git a/pkg/model/connection_evicting_client.go b/pkg/model/connection_evicting_client.go index 5cf920ede..92b4d6441 100644 --- a/pkg/model/connection_evicting_client.go +++ b/pkg/model/connection_evicting_client.go @@ -2,6 +2,7 @@ package model import ( "context" + "fmt" "sync" grpc "github.com/mudler/LocalAI/pkg/grpc" @@ -143,3 +144,13 @@ func (c *ConnectionEvictingClient) Score(ctx context.Context, in *pb.ScoreReques c.checkErr(err) return result, err } + +func (c *ConnectionEvictingClient) TTSLive(ctx context.Context, in *pb.TTSRequest, text <-chan string, send func([]byte) error) error { + live, ok := c.Backend.(grpc.TTSLiveBackend) + if !ok { + return fmt.Errorf("backend does not support incremental text TTS") + } + err := live.TTSLive(ctx, in, text, send) + c.checkErr(err) + return err +} diff --git a/pkg/model/tts_live_test.go b/pkg/model/tts_live_test.go new file mode 100644 index 000000000..a91178d7b --- /dev/null +++ b/pkg/model/tts_live_test.go @@ -0,0 +1,43 @@ +package model + +import ( + "context" + + grpcbackend "github.com/mudler/LocalAI/pkg/grpc" + pb "github.com/mudler/LocalAI/pkg/grpc/proto" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +type liveForwardFixture struct{ grpcbackend.Backend } + +func (*liveForwardFixture) TTSLive(ctx context.Context, _ *pb.TTSRequest, text <-chan string, send func([]byte) error) error { + for { + select { + case <-ctx.Done(): + return ctx.Err() + case chunk, ok := <-text: + if !ok { + return nil + } + if err := send([]byte(chunk)); err != nil { + return err + } + } + } +} + +var _ = Describe("TTSLive cache wrapper", func() { + It("preserves the optional incremental interface", func() { + wrapped := newConnectionEvictingClient(&liveForwardFixture{}, "fixture", func() { Fail("unexpected eviction") }) + live, ok := wrapped.(grpcbackend.TTSLiveBackend) + Expect(ok).To(BeTrue()) + text := make(chan string, 2) + text <- "early" + text <- "late" + close(text) + var got []string + Expect(live.TTSLive(context.Background(), &pb.TTSRequest{}, text, func(b []byte) error { got = append(got, string(b)); return nil })).To(Succeed()) + Expect(got).To(Equal([]string{"early", "late"})) + }) +})