diff --git a/backend/go/vibevoice-cpp/Makefile b/backend/go/vibevoice-cpp/Makefile index 77f597738..3d5cd43a1 100644 --- a/backend/go/vibevoice-cpp/Makefile +++ b/backend/go/vibevoice-cpp/Makefile @@ -11,7 +11,7 @@ JOBS?=$(shell nproc --ignore=1) # already do for ik_llama.cpp / llama.cpp / whisper.cpp). Floating on # `master` led to silent ABI breaks reaching CI — pin it. VIBEVOICE_REPO?=https://github.com/mudler/vibevoice.cpp -VIBEVOICE_CPP_VERSION?=ad856bda6b1311b7f3d7c4a667be43eeb8a8249a +VIBEVOICE_CPP_VERSION?=000e37282bc5bb09edc20f7047a47924122ba3a0 SO_TARGET?=libgovibevoicecpp.so CMAKE_ARGS+=-DBUILD_SHARED_LIBS=OFF diff --git a/backend/go/vibevoice-cpp/govibevoicecpp.go b/backend/go/vibevoice-cpp/govibevoicecpp.go index cf4945416..751aaff35 100644 --- a/backend/go/vibevoice-cpp/govibevoicecpp.go +++ b/backend/go/vibevoice-cpp/govibevoicecpp.go @@ -2,6 +2,7 @@ package main import ( "context" + "encoding/binary" "encoding/json" "fmt" "io" @@ -10,7 +11,9 @@ import ( "path/filepath" "runtime" "strings" + "unsafe" + "github.com/ebitengine/purego" laudio "github.com/mudler/LocalAI/pkg/audio" "github.com/mudler/LocalAI/pkg/grpc/base" pb "github.com/mudler/LocalAI/pkg/grpc/proto" @@ -113,12 +116,76 @@ var ( refAudioPaths []*byte, nRefAudioPaths int32, dstWav string, nSteps int32, cfgScale float32, maxSpeechFrames int32, seed uint32) int32 + // CppTTSStream drives vv_capi_tts_stream: it synthesizes `text` and + // invokes the C callback `cb` once per decoded PCM window instead of + // writing a file. `cb` is the address of a purego callback (see + // streamCB); `user` is an opaque pointer handed back to every + // callback invocation - we route via the package-level activeStream + // instead, so it is always nil here. + CppTTSStream func(text, voicePath string, + nSteps int32, cfgScale float32, maxFrames int32, seed uint32, + cb uintptr, user unsafe.Pointer) int32 CppASR func(srcWav string, outJSON []byte, capacity uint64, maxNewTokens int32) int32 CppUnload func() CppVersion func() string ) +// streamState carries the destination channel for one in-flight +// TTSStream call. vibevoice's engine is a single process-global, and +// backend calls are serialized through base.SingleThread, so a single +// package-level pointer is safe: only one TTSStream runs at a time. +type streamState struct { + results chan []byte +} + +// activeStream points at the streamState for the currently-running +// TTSStream. The C callback (streamCB) and the deliverPCMForTest hook +// read it to find the channel. Guarded by base.SingleThread +// serialization; TTSStream sets it and clears it in a defer. +var activeStream *streamState + +// pushPCM copies a transient int16 PCM window into a fresh little-endian +// []byte and pushes it onto the active stream. The C buffer handed to +// the callback is only valid for the duration of the call, so we must +// copy before returning. A nil/empty input or a missing active stream +// is a no-op. +func pushPCM(pcm []int16) { + s := activeStream + if s == nil || len(pcm) == 0 { + return + } + buf := make([]byte, len(pcm)*2) + for i, v := range pcm { + binary.LittleEndian.PutUint16(buf[i*2:], uint16(v)) + } + s.results <- buf +} + +// streamCB is the ONE reusable purego callback bound to the C ABI's +// vv_pcm_cb. purego cannot free callbacks and enforces a process-global +// limit, so we create exactly one at package init and reuse it for every +// TTSStream call - the per-call state lives in activeStream, not here. +// purego marshals the C `const int16_t*` first argument straight into a +// Go *int16, so we can unsafe.Slice it without a uintptr round-trip +// (keeps go vet clean); pushPCM copies the transient buffer out and +// returns 0 to keep synthesizing. +var streamCB = purego.NewCallback(func(samples *int16, n int32, _ uintptr) uintptr { + if activeStream == nil || samples == nil || n <= 0 { + return 0 + } + pcm := unsafe.Slice(samples, int(n)) + pushPCM(pcm) + return 0 +}) + +// deliverPCMForTest exercises the exact copy-and-push path streamCB runs +// against activeStream, but from a Go []int16 - so unit tests can +// validate the callback -> channel framing without the C library. +func deliverPCMForTest(samples []int16) { + pushPCM(samples) +} + // VibevoiceCpp speaks gRPC against vibevoice.cpp's flat C ABI. The // engine is a single global, so we serialize calls through SingleThread. type VibevoiceCpp struct { @@ -404,14 +471,16 @@ func (v *VibevoiceCpp) callASR(srcWav string, maxNewTokens int32) (string, error return string(buf[:rc]), nil } -// TTSStream is the streaming counterpart to TTS. vibevoice's C ABI is -// file-only (vv_capi_tts writes a complete WAV), so we synthesize to -// a tempfile, then emit a streaming-WAV header followed by the PCM -// body in chunks. The main reason this exists at all is the gRPC -// server wrapper (pkg/grpc/server.go:TTSStream) blocks on a channel -// that only this method can close - if we leave the default Base -// stub in place, every TTSStream call hangs until the client -// deadline. +// TTSStream is the streaming counterpart to TTS. It drives +// vv_capi_tts_stream, which synthesizes `text` and invokes our C +// callback (streamCB) once per decoded PCM window instead of writing a +// file - so the client starts receiving audio while the model is still +// generating. We first emit a streaming-WAV header, install the results +// channel as the active stream, then let the callback push each PCM +// window (copied to little-endian bytes) onto that channel. The gRPC +// server wrapper (pkg/grpc/server.go:TTSStream) blocks on the channel +// until this method closes it, so `defer close(results)` is mandatory +// even on the error paths. func (v *VibevoiceCpp) TTSStream(req *pb.TTSRequest, results chan []byte) error { defer close(results) if v.ttsModel == "" { @@ -421,28 +490,6 @@ func (v *VibevoiceCpp) TTSStream(req *pb.TTSRequest, results chan []byte) error return fmt.Errorf("vibevoice-cpp: TTSStream requires text") } - tmp, err := os.CreateTemp("", "vibevoice-cpp-stream-*.wav") - if err != nil { - return fmt.Errorf("vibevoice-cpp: tempfile: %w", err) - } - dst := tmp.Name() - _ = tmp.Close() - defer func() { _ = os.Remove(dst) }() - - if err := v.TTS(&pb.TTSRequest{ - Text: req.Text, - Voice: req.Voice, - Dst: dst, - Language: req.Language, - }); err != nil { - return err - } - - wav, err := os.ReadFile(dst) - if err != nil { - return fmt.Errorf("vibevoice-cpp: read tempfile: %w", err) - } - // Streaming WAV header: declare 0xFFFFFFFF for chunk sizes so HTTP // clients can start playback before they see the full PCM. const streamingSize = 0xFFFFFFFF @@ -455,18 +502,38 @@ func (v *VibevoiceCpp) TTSStream(req *pb.TTSRequest, results chan []byte) error } results <- hdrBuf - // PCM body: send in ~64 KB slices so the client gets multiple - // reply chunks (e2e harness asserts >=2 frames). - pcm := laudio.StripWAVHeader(wav) - const chunkBytes = 64 * 1024 - for off := 0; off < len(pcm); off += chunkBytes { - end := off + chunkBytes - if end > len(pcm) { - end = len(pcm) - } - chunk := make([]byte, end-off) - copy(chunk, pcm[off:end]) - results <- chunk + // vv_capi_tts_stream takes a single voice_path (realtime-0.5B path); + // unlike vv_capi_tts it has no ref_audio array. Resolve the per-call + // override when it names a voice gguf, otherwise fall back to the + // load-time default that already went to vv_capi_load. + voice := v.voice + if reqVoice := strings.TrimSpace(req.Voice); reqVoice != "" && !isRefAudioOverride(reqVoice) { + voice = resolvePath(reqVoice, v.modelRoot) + } + + if req.Language != nil && *req.Language != "" { + fmt.Fprintf(os.Stderr, + "[vibevoice-cpp] note: TTSRequest.language=%q ignored - vibevoice picks language from the voice prompt\n", + *req.Language) + } + + const ( + defaultSteps = 20 + defaultMaxFrames = 200 + ) + defaultCfg := float32(1.3) + + // Serialized by base.SingleThread, so a single package-level + // activeStream is race-free: exactly one TTSStream runs at a time. + // The callback reads it to find `results`; clear it on the way out. + activeStream = &streamState{results: results} + defer func() { activeStream = nil }() + + rc := CppTTSStream(req.Text, voice, + int32(defaultSteps), defaultCfg, int32(defaultMaxFrames), 0, + streamCB, nil) + if rc != 0 { + return fmt.Errorf("vibevoice-cpp: vv_capi_tts_stream failed (rc=%d)", rc) } return nil } diff --git a/backend/go/vibevoice-cpp/main.go b/backend/go/vibevoice-cpp/main.go index b9a696d82..8add177ed 100644 --- a/backend/go/vibevoice-cpp/main.go +++ b/backend/go/vibevoice-cpp/main.go @@ -37,6 +37,7 @@ func main() { libFuncs := []LibFuncs{ {&CppLoad, "vv_capi_load"}, {&CppTTS, "vv_capi_tts"}, + {&CppTTSStream, "vv_capi_tts_stream"}, {&CppASR, "vv_capi_asr"}, {&CppUnload, "vv_capi_unload"}, {&CppVersion, "vv_capi_version"}, diff --git a/backend/go/vibevoice-cpp/vibevoicecpp_integration_test.go b/backend/go/vibevoice-cpp/vibevoicecpp_integration_test.go new file mode 100644 index 000000000..cc0a2b218 --- /dev/null +++ b/backend/go/vibevoice-cpp/vibevoicecpp_integration_test.go @@ -0,0 +1,262 @@ +package main + +// Real end-to-end streaming integration test for the vibevoice-cpp +// backend. It drives the actual Go -> purego -> C path against the real +// model, proving that TTSStream delivers audio incrementally (measurable +// time-to-first-audio) while TTS only yields anything after full +// synthesis. Gated behind VIBEVOICE_IT=1 so normal `go test` and CI stay +// unaffected - it needs the built engine .so and ~1.7 GB of model files. +// +// Run: +// +// VIBEVOICE_IT=1 \ +// VIBEVOICECPP_LIBRARY= \ +// go test ./... -run TestVibevoiceCpp -v -timeout 600s +// +// Optional overrides (default to the staged bundle under +// ~/_git/vibevoice-models): +// +// VIBEVOICE_IT_MODEL vibevoice-realtime-0.5B-q8_0.gguf (abs path) +// VIBEVOICE_IT_TOKENIZER tokenizer.gguf (abs path) +// VIBEVOICE_IT_VOICE voice-en-Carter_man.gguf (abs path) + +import ( + "fmt" + "io" + "os" + "path/filepath" + "sync" + "time" + + "github.com/ebitengine/purego" + pb "github.com/mudler/LocalAI/pkg/grpc/proto" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// itDefaultModelDir is where Task 6 staged the model bundle. Individual +// files can be overridden via the VIBEVOICE_IT_* env vars. +const itDefaultModelDir = "/home/mudler/_git/vibevoice-models" + +// itLibLoaded guards the one-time purego Dlopen + RegisterLibFunc into +// the package-level Cpp* vars. purego cannot free a loaded library and +// the Cpp* symbols are process-global, so we bind exactly once. +var itLibLoaded sync.Once + +// integrationOrSkip enforces the VIBEVOICE_IT=1 gate and binds the C ABI +// symbols into the package vars the backend calls (mirrors main.go's +// libFuncs list). It Skip()s the spec when the gate is off or the .so / +// model files are missing, so the suite stays green in every other run. +func integrationOrSkip() (model, tokenizer, voice string) { + if os.Getenv("VIBEVOICE_IT") != "1" { + Skip("VIBEVOICE_IT!=1, skipping real-model streaming integration test") + } + + lib := os.Getenv("VIBEVOICECPP_LIBRARY") + if lib == "" { + Skip("VIBEVOICECPP_LIBRARY not set, cannot dlopen the engine .so") + } + if _, err := os.Stat(lib); err != nil { + Skip("engine .so not found at " + lib) + } + + model = itEnvOr("VIBEVOICE_IT_MODEL", filepath.Join(itDefaultModelDir, "vibevoice-realtime-0.5B-q8_0.gguf")) + tokenizer = itEnvOr("VIBEVOICE_IT_TOKENIZER", filepath.Join(itDefaultModelDir, "tokenizer.gguf")) + voice = itEnvOr("VIBEVOICE_IT_VOICE", filepath.Join(itDefaultModelDir, "voice-en-Carter_man.gguf")) + for _, p := range []string{model, tokenizer, voice} { + if _, err := os.Stat(p); err != nil { + Skip("model file missing: " + p) + } + } + + itLibLoaded.Do(func() { + handle, err := purego.Dlopen(lib, purego.RTLD_NOW|purego.RTLD_GLOBAL) + Expect(err).ToNot(HaveOccurred(), "dlopen %s", lib) + // Mirror the libFuncs list in main.go verbatim so the test binds + // the exact same symbols the production backend binary does. + purego.RegisterLibFunc(&CppLoad, handle, "vv_capi_load") + purego.RegisterLibFunc(&CppTTS, handle, "vv_capi_tts") + purego.RegisterLibFunc(&CppTTSStream, handle, "vv_capi_tts_stream") + purego.RegisterLibFunc(&CppASR, handle, "vv_capi_asr") + purego.RegisterLibFunc(&CppUnload, handle, "vv_capi_unload") + purego.RegisterLibFunc(&CppVersion, handle, "vv_capi_version") + }) + return model, tokenizer, voice +} + +func itEnvOr(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + return def +} + +// itParseWavPCM returns the int16 PCM samples from a standard WAV file by +// locating the "data" sub-chunk, so it works regardless of how many +// bytes of header/metadata the engine wrote. +func itParseWavPCM(b []byte) []int16 { + // Find the "data" sub-chunk id, then its 4-byte little-endian size. + idx := -1 + for i := 12; i+8 <= len(b); i += 2 { + if string(b[i:i+4]) == "data" { + idx = i + break + } + } + Expect(idx).To(BeNumerically(">=", 0), "no data chunk in WAV") + pcmStart := idx + 8 + size := int(b[idx+4]) | int(b[idx+5])<<8 | int(b[idx+6])<<16 | int(b[idx+7])<<24 + if size <= 0 || pcmStart+size > len(b) { + size = len(b) - pcmStart + } + n := size / 2 + out := make([]int16, n) + for i := 0; i < n; i++ { + out[i] = int16(uint16(b[pcmStart+i*2]) | uint16(b[pcmStart+i*2+1])<<8) + } + return out +} + +// itRMS computes the root-mean-square of int16 PCM as a sanity signal: +// > 0 means non-silent, and NaN/Inf-free means well-formed samples. +func itRMS(pcm []int16) float64 { + if len(pcm) == 0 { + return 0 + } + var sumSq float64 + for _, s := range pcm { + v := float64(s) + sumSq += v * v + } + return sqrt(sumSq / float64(len(pcm))) +} + +// sqrt via Newton's method to avoid dragging math into the file's tiny +// need (keeps the dependency surface of the test minimal). +func sqrt(x float64) float64 { + if x <= 0 { + return 0 + } + z := x + for i := 0; i < 40; i++ { + z = 0.5 * (z + x/z) + } + return z +} + +var _ = Describe("VibeVoice-cpp real-model streaming (VIBEVOICE_IT=1)", Ordered, func() { + // The paragraph is deliberately long (multiple sentences) so the + // engine decodes several streaming windows and TTFA is meaningfully + // earlier than full synthesis. + const paragraph = "The quick brown fox jumps over the lazy dog near the riverbank. " + + "A gentle breeze carried the sound of distant bells across the quiet valley. " + + "Streaming synthesis lets you hear the very first words while the rest is still being generated." + + var v *VibevoiceCpp + + BeforeAll(func() { + model, tokenizer, voice := integrationOrSkip() + + v = &VibevoiceCpp{} + err := v.Load(&pb.ModelOptions{ + ModelFile: model, + ModelPath: filepath.Dir(model), + Options: []string{ + "tokenizer=" + tokenizer, + "voice=" + voice, + }, + Threads: 4, + }) + Expect(err).ToNot(HaveOccurred(), "Load must succeed with the real model") + }) + + It("streams audio incrementally and beats the batch path to first audio", func() { + // ---- Streaming run ------------------------------------------- + results := make(chan []byte, 256) + streamErr := make(chan error, 1) + + start := time.Now() + go func() { + streamErr <- v.TTSStream(&pb.TTSRequest{Text: paragraph}, results) + }() + + var ( + header []byte + ttfa time.Duration + totalStream time.Duration + chunkCount int + pcmBytes int + firstPCM bool + ) + for buf := range results { + if header == nil { + header = buf + continue + } + if len(buf) == 0 { + continue + } + if !firstPCM { + ttfa = time.Since(start) + firstPCM = true + } + chunkCount++ + pcmBytes += len(buf) + } + totalStream = time.Since(start) + Expect(<-streamErr).ToNot(HaveOccurred(), "TTSStream returned an error") + + // First message must be a 44-byte streaming WAV header. + Expect(header).To(HaveLen(44), "first stream message must be the WAV header") + Expect(string(header[0:4])).To(Equal("RIFF")) + Expect(string(header[8:12])).To(Equal("WAVE")) + + // Streaming invariants: multiple windows, real audio, early + // delivery (first audio strictly before the stream completes). + Expect(chunkCount).To(BeNumerically(">=", 2), "expected multiple streamed PCM windows") + Expect(pcmBytes).To(BeNumerically(">", 0), "no PCM bytes streamed") + Expect(firstPCM).To(BeTrue(), "never received a PCM chunk") + Expect(ttfa).To(BeNumerically("<", totalStream), "TTFA must precede stream completion") + + // ---- Batch baseline ------------------------------------------ + tmp, err := os.MkdirTemp("", "vv-it-batch-*") + Expect(err).ToNot(HaveOccurred()) + DeferCleanup(func() { _ = os.RemoveAll(tmp) }) + dst := filepath.Join(tmp, "batch.wav") + + batchStart := time.Now() + Expect(v.TTS(&pb.TTSRequest{Text: paragraph, Dst: dst})).To(Succeed()) + totalBatch := time.Since(batchStart) + + wav, err := os.ReadFile(dst) + Expect(err).ToNot(HaveOccurred()) + Expect(len(wav)).To(BeNumerically(">", 44), "batch wav has no PCM payload") + Expect(string(wav[0:4])).To(Equal("RIFF")) + Expect(string(wav[8:12])).To(Equal("WAVE")) + batchPCM := itParseWavPCM(wav) + Expect(len(batchPCM)).To(BeNumerically(">", 0), "batch produced no samples") + rms := itRMS(batchPCM) + Expect(rms).To(BeNumerically(">", 0), "batch audio is silent") + Expect(rms).To(BeNumerically("<", 40000), "batch rms out of int16 range (corrupt samples)") + + // ---- Headline numbers ---------------------------------------- + // Emit to both GinkgoWriter and stderr: GinkgoWriter needs + // -ginkgo.v to surface, stderr is always captured so the headline + // TTFA vs batch numbers are never lost in an unattended run. + ratio := float64(ttfa) / float64(totalBatch) + report := fmt.Sprintf("\n================ vibevoice-cpp streaming TTFA ================\n"+ + "input words : ~%d\n"+ + "TTFA (first audio) : %v\n"+ + "total_stream : %v\n"+ + "total_batch : %v\n"+ + "stream chunks : %d\n"+ + "stream PCM bytes : %d\n"+ + "batch samples/rms : %d / %.1f\n"+ + "TTFA / total_batch : %.3f (first audio in this fraction of batch's deliver time)\n"+ + "==============================================================\n\n", + len(paragraph)/6, ttfa, totalStream, totalBatch, chunkCount, pcmBytes, len(batchPCM), rms, ratio) + for _, out := range []io.Writer{GinkgoWriter, os.Stderr} { + _, _ = io.WriteString(out, report) + } + }) +}) diff --git a/backend/go/vibevoice-cpp/vibevoicecpp_stream_test.go b/backend/go/vibevoice-cpp/vibevoicecpp_stream_test.go new file mode 100644 index 000000000..139980f81 --- /dev/null +++ b/backend/go/vibevoice-cpp/vibevoicecpp_stream_test.go @@ -0,0 +1,46 @@ +package main + +import ( + "encoding/binary" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("TTSStream callback framing", func() { + // The real C callback copies a transient int16 PCM buffer out of the + // engine into a fresh []byte and pushes it onto the active stream's + // results channel as little-endian bytes. deliverPCMForTest runs that + // exact copy-and-push path against a []int16 so we can validate the + // framing without the C library (full audio e2e is a later task). + It("copies int16 PCM from the C callback into the results channel as LE bytes", func() { + samples := []int16{0, 1, -1, 32767, -32768, 1234, -4321} + + prev := activeStream + DeferCleanup(func() { activeStream = prev }) + + s := &streamState{results: make(chan []byte, 1)} + activeStream = s + + deliverPCMForTest(samples) + + var got []byte + Eventually(s.results).Should(Receive(&got)) + Expect(got).To(HaveLen(len(samples) * 2)) + + want := make([]byte, len(samples)*2) + for i, v := range samples { + binary.LittleEndian.PutUint16(want[i*2:], uint16(v)) + } + Expect(got).To(Equal(want)) + }) + + It("is a no-op when there is no active stream", func() { + prev := activeStream + DeferCleanup(func() { activeStream = prev }) + activeStream = nil + + // Must not panic when no stream is installed. + Expect(func() { deliverPCMForTest([]int16{1, 2, 3}) }).ToNot(Panic()) + }) +})