From f12bcfac9a35107f0696bd4cdc64b6735ec47580 Mon Sep 17 00:00:00 2001 From: localai-org-maint-bot Date: Wed, 9 Sep 2026 10:38:23 +0200 Subject: [PATCH] fix(realtime): support voice profile switching (#11948) * fix(realtime): support session voice profile switching Keep the active resolved voice binding on the realtime session so updates can atomically replace model, voice, and profile parameters while releasing leases at the correct lifecycle boundaries. Assisted-by: Codex:gpt-5 * docs(realtime): explain voice profile switching Document the session.update payload for selecting a Voice Library URI and clarify precedence when changing the model in the same event.\n\nAssisted-by: Codex:gpt-5 --------- Co-authored-by: Ettore Di Giacinto --- core/http/endpoints/openai/realtime.go | 121 +++++++++--- core/http/endpoints/openai/realtime_model.go | 4 + .../openai/realtime_voice_profile_test.go | 174 ++++++++++++++++++ docs/content/features/text-to-audio.md | 17 +- 4 files changed, 285 insertions(+), 31 deletions(-) diff --git a/core/http/endpoints/openai/realtime.go b/core/http/endpoints/openai/realtime.go index b387805cf..811a5b80c 100644 --- a/core/http/endpoints/openai/realtime.go +++ b/core/http/endpoints/openai/realtime.go @@ -9,6 +9,7 @@ import ( "encoding/json" "errors" "fmt" + "maps" "math" "os" "strconv" @@ -32,6 +33,7 @@ import ( "github.com/mudler/LocalAI/core/http/endpoints/openai/types" "github.com/mudler/LocalAI/core/schema" "github.com/mudler/LocalAI/core/services/routing/router" + "github.com/mudler/LocalAI/core/services/voiceprofile" "github.com/mudler/LocalAI/core/templates" laudio "github.com/mudler/LocalAI/pkg/audio" "github.com/mudler/LocalAI/pkg/functions" @@ -136,6 +138,8 @@ type Session struct { Instructions string DefaultConversationID string ModelInterface Model + ttsParams map[string]string + voiceRelease func() // The pipeline model config or the config for an any-to-any model ModelConfig *config.ModelConfig InputSampleRate int @@ -199,6 +203,22 @@ type Session struct { respSink *responseSink } +func (s *Session) installVoiceBinding(voice string, params map[string]string, release func()) { + if release == nil { + release = func() {} + } + var once sync.Once + s.Voice = voice + s.ttsParams = maps.Clone(params) + s.voiceRelease = func() { once.Do(release) } +} + +func (s *Session) releaseVoiceBinding() { + if s.voiceRelease != nil { + s.voiceRelease() + } +} + func (s *Session) FromClient(session *types.SessionUnion) { } @@ -635,15 +655,15 @@ func runRealtimeSession(application *application.Application, t Transport, model return } if wrapped, ok := m.(*wrappedModel); ok { - resolvedVoice, params, release, resolveErr := resolveRealtimeVoice(context.Background(), session.Voice, wrapped.TTSConfig, application.VoiceProfileStore()) + resolvedVoice, params, release, resolveErr := resolveRealtimeVoice(context.Background(), wrapped.TTSConfig.TTSConfig.Voice, wrapped.TTSConfig, application.VoiceProfileStore()) if resolveErr != nil { xlog.Error("failed to resolve realtime voice", "error", resolveErr) sendError(t, "voice_profile_error", resolveErr.Error(), "", "") return } - defer release() - session.Voice = resolvedVoice - wrapped.ttsParams = params + session.installVoiceBinding(resolvedVoice, params, release) + defer session.releaseVoiceBinding() + wrapped.setTTSParams(params) } session.ModelInterface = m // A pipeline-seeded option list gets its scoring prompt prewarmed @@ -838,6 +858,7 @@ func runRealtimeSession(application *application.Application, t Transport, model application.ApplicationConfig(), evaluator, buildRealtimeRoutingContext(application, session.ID), + application.VoiceProfileStore(), ); err != nil { xlog.Error("failed to update session", "error", err) sendError(t, "session_update_error", fmt.Sprintf("Failed to update session: %v", err), "", "") @@ -1176,7 +1197,7 @@ func updateTransSession(session *Session, update *types.SessionUnion, cl *config return nil } -func updateSession(session *Session, update *types.SessionUnion, cl *config.ModelConfigLoader, ml *model.ModelLoader, appConfig *config.ApplicationConfig, evaluator *templates.Evaluator, routing *RealtimeRoutingContext) error { +func updateSession(session *Session, update *types.SessionUnion, cl *config.ModelConfigLoader, ml *model.ModelLoader, appConfig *config.ApplicationConfig, evaluator *templates.Evaluator, routing *RealtimeRoutingContext, profiles *voiceprofile.Store) error { sessionLock.Lock() defer sessionLock.Unlock() @@ -1184,8 +1205,13 @@ func updateSession(session *Session, update *types.SessionUnion, cl *config.Mode return nil } - session.TranscriptionOnly = false rt := update.Realtime + explicitVoice := rt.Audio != nil && rt.Audio.Output != nil && rt.Audio.Output.Voice != "" + rebuild := rt.Model != "" || explicitVoice || (rt.Audio != nil && rt.Audio.Input != nil && rt.Audio.Input.Transcription != nil) + + candidateModelName := session.Model + candidateConfig := session.ModelConfig + candidateTranscription := session.InputAudioTranscription if rt.Model != "" { cfg, err := cl.LoadModelConfigFileByNameDefaultOptions(rt.Model, appConfig) @@ -1196,40 +1222,78 @@ func updateSession(session *Session, update *types.SessionUnion, cl *config.Mode return fmt.Errorf("model is not a valid pipeline model: %s", rt.Model) } - if session.InputAudioTranscription == nil { - session.InputAudioTranscription = &types.AudioTranscription{} - } - session.InputAudioTranscription.Model = cfg.Pipeline.Transcription - session.Voice = cfg.TTSConfig.Voice - session.Model = rt.Model - session.ModelConfig = cfg - } - - if rt.Audio != nil && rt.Audio.Output != nil && rt.Audio.Output.Voice != "" { - session.Voice = string(rt.Audio.Output.Voice) + candidateModelName = rt.Model + candidateConfig = cfg + candidateTranscription = &types.AudioTranscription{Model: cfg.Pipeline.Transcription} } if rt.Audio != nil && rt.Audio.Input != nil && rt.Audio.Input.Transcription != nil { - trUpd := rt.Audio.Input.Transcription + trUpd := *rt.Audio.Input.Transcription // A language-only update (e.g. a client forcing the STT language) carries // an empty Model. Preserve the pipeline's configured transcription backend // instead of blanking it — otherwise the next utterance transcribes against // an empty model and the backend RPC fails with "unimplemented". - if trUpd.Model == "" && session.InputAudioTranscription != nil { - trUpd.Model = session.InputAudioTranscription.Model + if trUpd.Model == "" && candidateTranscription != nil { + trUpd.Model = candidateTranscription.Model } - session.InputAudioTranscription = trUpd + candidateTranscription = &trUpd if trUpd.Model != "" { - session.ModelConfig.Pipeline.Transcription = trUpd.Model + cfgCopy := *candidateConfig + candidateConfig = &cfgCopy + candidateConfig.Pipeline.Transcription = trUpd.Model } } - if rt.Model != "" || (rt.Audio != nil && rt.Audio.Output != nil && rt.Audio.Output.Voice != "") || (rt.Audio != nil && rt.Audio.Input != nil && rt.Audio.Input.Transcription != nil) { - m, err := newModel(&session.ModelConfig.Pipeline, cl, ml, appConfig, evaluator, routing) + candidateModel := session.ModelInterface + candidateVoice := session.Voice + candidateParams := maps.Clone(session.ttsParams) + var candidateRelease func() + selectVoice := rt.Model != "" || explicitVoice + if rebuild { + m, err := newModel(&candidateConfig.Pipeline, cl, ml, appConfig, evaluator, routing) if err != nil { return err } - session.ModelInterface = m + candidateModel = m + wrapped := m.(*wrappedModel) + if selectVoice { + configuredVoice := wrapped.TTSConfig.TTSConfig.Voice + if explicitVoice { + configuredVoice = string(rt.Audio.Output.Voice) + } + candidateVoice, candidateParams, candidateRelease, err = resolveRealtimeVoice( + context.Background(), configuredVoice, wrapped.TTSConfig, profiles, + ) + if err != nil { + return err + } + } + wrapped.setTTSParams(candidateParams) + } + + if rt.LocalAIClassifier != nil { + if err := validateClassifierActivation(candidateModel, rt.LocalAIClassifier); err != nil { + if candidateRelease != nil { + candidateRelease() + } + return err + } + } + + oldRelease := session.voiceRelease + session.TranscriptionOnly = false + session.Model = candidateModelName + session.ModelConfig = candidateConfig + session.ModelInterface = candidateModel + session.InputAudioTranscription = candidateTranscription + if selectVoice { + session.installVoiceBinding(candidateVoice, candidateParams, candidateRelease) + if oldRelease != nil { + oldRelease() + } + } + + if rebuild { // A session.update that swaps the model/voice rebuilds the pipeline, so // warm the new backends too (unless opted out) — otherwise the next turn // pays the cold-start load the original session warm-up already avoided. @@ -1238,9 +1302,9 @@ func updateSession(session *Session, update *types.SessionUnion, cl *config.Mode // stall every other session. Load errors are logged (and still surface on // first use); per-stage failures are already warned inside // backend.PreloadStages. - if !session.ModelConfig.Pipeline.DisableWarmup { + if !candidateConfig.Pipeline.DisableWarmup { go func() { - if err := m.Warmup(context.Background()); err != nil { + if err := candidateModel.Warmup(context.Background()); err != nil { xlog.Error("realtime warmup failed after session.update", "error", err) } }() @@ -1299,9 +1363,6 @@ func updateSession(session *Session, update *types.SessionUnion, cl *config.Mode // Replace-not-merge, like tools: the client owns the whole option // list. Invalid configs reject the update without touching the // session's current classifier. - if err := validateClassifierActivation(session.ModelInterface, rt.LocalAIClassifier); err != nil { - return err - } session.Classifier = rt.LocalAIClassifier prewarmClassifier(session) } diff --git a/core/http/endpoints/openai/realtime_model.go b/core/http/endpoints/openai/realtime_model.go index 33ca2c762..b525eee26 100644 --- a/core/http/endpoints/openai/realtime_model.go +++ b/core/http/endpoints/openai/realtime_model.go @@ -398,6 +398,10 @@ func (m *wrappedModel) TTS(ctx context.Context, text, voice, language string) (s return backend.ModelTTS(ctx, text, voice, language, "", maps.Clone(m.ttsParams), m.modelLoader, m.appConfig, *m.TTSConfig) } +func (m *wrappedModel) setTTSParams(params map[string]string) { + m.ttsParams = maps.Clone(params) +} + func (m *wrappedModel) TTSStream(ctx context.Context, text, voice, language string, onAudio func(pcm []byte, sampleRate int) error) error { return ttsStream(ctx, m.modelLoader, m.appConfig, *m.TTSConfig, text, voice, language, maps.Clone(m.ttsParams), onAudio) } diff --git a/core/http/endpoints/openai/realtime_voice_profile_test.go b/core/http/endpoints/openai/realtime_voice_profile_test.go index 801d37bbc..4281f7705 100644 --- a/core/http/endpoints/openai/realtime_voice_profile_test.go +++ b/core/http/endpoints/openai/realtime_voice_profile_test.go @@ -5,8 +5,12 @@ import ( "context" "encoding/binary" "errors" + "fmt" + "os" + "path/filepath" "time" + "github.com/mudler/LocalAI/core/http/endpoints/openai/types" grpcPkg "github.com/mudler/LocalAI/pkg/grpc" "github.com/mudler/LocalAI/pkg/grpc/proto" "github.com/mudler/LocalAI/pkg/model" @@ -177,4 +181,174 @@ var _ = Describe("wrappedModel voice profile parameters", func() { }) }) +var _ = Describe("realtime session voice switching", func() { + type fixture struct { + store *voiceprofile.Store + voiceDir string + loader *config.ModelConfigLoader + models *model.ModelLoader + appConfig *config.ApplicationConfig + profileA voiceprofile.Profile + profileB voiceprofile.Profile + } + + newFixture := func(ctx SpecContext) *fixture { + modelDir := GinkgoT().TempDir() + voiceDir := GinkgoT().TempDir() + store := voiceprofile.NewStore(voiceDir) + DeferCleanup(func() { Expect(store.Close()).To(Succeed()) }) + profileA, err := store.Create(ctx, voiceprofile.CreateInput{ + Name: "Alpha", Language: "en", Transcript: "Alpha transcript", ConsentConfirmed: true, + }, bytes.NewReader(realtimeProfileWAV(time.Second))) + Expect(err).NotTo(HaveOccurred()) + profileB, err := store.Create(ctx, voiceprofile.CreateInput{ + Name: "Beta", Language: "it", Transcript: "Beta transcript", ConsentConfirmed: true, + }, bytes.NewReader(realtimeProfileWAV(time.Second))) + Expect(err).NotTo(HaveOccurred()) + + configs := map[string]string{ + "vad": "name: vad\nbackend: test\nparameters:\n model: vad.bin\n", + "stt": "name: stt\nbackend: test\nparameters:\n model: stt.bin\n", + "llm": "name: llm\nbackend: test\nparameters:\n model: llm.bin\n", + "tts-a": fmt.Sprintf("name: tts-a\nbackend: qwen3-tts-cpp\nparameters:\n model: tts-a.bin\ntts:\n voice: %s\n voice_cloning: true\n", profileA.Voice), + "tts-b": fmt.Sprintf("name: tts-b\nbackend: qwen3-tts-cpp\nparameters:\n model: tts-b.bin\ntts:\n voice: %s\n voice_cloning: true\n", profileB.Voice), + "pipe-a": "name: pipe-a\npipeline:\n vad: vad\n transcription: stt\n llm: llm\n tts: tts-a\n disable_warmup: true\n", + "pipe-b": "name: pipe-b\npipeline:\n vad: vad\n transcription: stt\n llm: llm\n tts: tts-b\n disable_warmup: true\n", + } + for name, body := range configs { + Expect(os.WriteFile(filepath.Join(modelDir, name+".yaml"), []byte(body), 0o644)).To(Succeed()) + } + loader := config.NewModelConfigLoader(modelDir) + Expect(loader.LoadModelConfigsFromPath(modelDir)).To(Succeed()) + state, err := system.GetSystemState(system.WithModelPath(modelDir)) + Expect(err).NotTo(HaveOccurred()) + return &fixture{ + store: store, voiceDir: voiceDir, loader: loader, models: model.NewModelLoader(state), + appConfig: config.NewApplicationConfig(config.WithSystemState(state)), + profileA: profileA, profileB: profileB, + } + } + + newSession := func(f *fixture, voice string) *Session { + cfg, err := f.loader.LoadModelConfigFileByNameDefaultOptions("pipe-a", f.appConfig) + Expect(err).NotTo(HaveOccurred()) + m, err := newModel(&cfg.Pipeline, f.loader, f.models, f.appConfig, nil, nil) + Expect(err).NotTo(HaveOccurred()) + session := &Session{ + Model: "pipe-a", Voice: voice, ModelConfig: cfg, ModelInterface: m, + InputAudioTranscription: &types.AudioTranscription{Model: "stt"}, + } + resolved, params, release, err := resolveRealtimeVoice(context.Background(), voice, m.(*wrappedModel).TTSConfig, f.store) + Expect(err).NotTo(HaveOccurred()) + session.installVoiceBinding(resolved, params, release) + return session + } + + update := func(f *fixture, session *Session, rt *types.RealtimeSession) error { + return updateSession(session, &types.SessionUnion{Realtime: rt}, f.loader, f.models, f.appConfig, nil, nil, f.store) + } + + It("switches ordinary voices to profiles and clears the lease at final cleanup", func(ctx SpecContext) { + f := newFixture(ctx) + session := newSession(f, "speaker-1") + Expect(update(f, session, &types.RealtimeSession{Audio: &types.RealtimeSessionAudio{Output: &types.SessionAudioOutput{Voice: types.Voice(f.profileA.Voice)}}})).To(Succeed()) + + Expect(session.Voice).To(BeAnExistingFile()) + Expect(session.ttsParams).To(Equal(map[string]string{"ref_text": "Alpha transcript"})) + leased := session.Voice + session.releaseVoiceBinding() + session.releaseVoiceBinding() + Expect(leased).NotTo(BeAnExistingFile()) + }) + + It("replaces one profile lease with another and then an ordinary voice", func(ctx SpecContext) { + f := newFixture(ctx) + session := newSession(f, f.profileA.Voice) + firstLease := session.Voice + + Expect(update(f, session, &types.RealtimeSession{Audio: &types.RealtimeSessionAudio{Output: &types.SessionAudioOutput{Voice: types.Voice(f.profileB.Voice)}}})).To(Succeed()) + Expect(firstLease).NotTo(BeAnExistingFile()) + secondLease := session.Voice + Expect(secondLease).To(BeAnExistingFile()) + Expect(session.ttsParams).To(HaveKeyWithValue("ref_text", "Beta transcript")) + + Expect(update(f, session, &types.RealtimeSession{Audio: &types.RealtimeSessionAudio{Output: &types.SessionAudioOutput{Voice: "speaker-2"}}})).To(Succeed()) + Expect(secondLease).NotTo(BeAnExistingFile()) + Expect(session.Voice).To(Equal("speaker-2")) + Expect(session.ttsParams).To(BeNil()) + Expect(session.ModelInterface.(*wrappedModel).ttsParams).To(BeNil()) + }) + + It("uses a new model default profile unless an explicit voice takes precedence", func(ctx SpecContext) { + f := newFixture(ctx) + session := newSession(f, "speaker-1") + Expect(update(f, session, &types.RealtimeSession{Model: "pipe-b"})).To(Succeed()) + Expect(session.ttsParams).To(HaveKeyWithValue("ref_text", "Beta transcript")) + defaultLease := session.Voice + + Expect(update(f, session, &types.RealtimeSession{ + Model: "pipe-a", + Audio: &types.RealtimeSessionAudio{Output: &types.SessionAudioOutput{Voice: "speaker-explicit"}}, + })).To(Succeed()) + Expect(defaultLease).NotTo(BeAnExistingFile()) + Expect(session.Voice).To(Equal("speaker-explicit")) + Expect(session.ttsParams).To(BeNil()) + }) + + It("preserves a profile binding across a language-only rebuild", func(ctx SpecContext) { + f := newFixture(ctx) + session := newSession(f, f.profileA.Voice) + lease := session.Voice + Expect(update(f, session, &types.RealtimeSession{Audio: &types.RealtimeSessionAudio{Input: &types.SessionAudioInput{ + Transcription: &types.AudioTranscription{Language: "fr"}, + }}})).To(Succeed()) + + Expect(session.Voice).To(Equal(lease)) + Expect(session.InputAudioTranscription.Model).To(Equal("stt")) + Expect(session.InputAudioTranscription.Language).To(Equal("fr")) + wrapped := session.ModelInterface.(*wrappedModel) + Expect(wrapped.ttsParams).To(Equal(map[string]string{"ref_text": "Alpha transcript"})) + wrapped.ttsParams["ref_text"] = "wrapper mutation" + Expect(session.ttsParams).To(HaveKeyWithValue("ref_text", "Alpha transcript")) + }) + + It("rolls back the model, wrapper, voice, and lease when preparation fails", func(ctx SpecContext) { + f := newFixture(ctx) + session := newSession(f, f.profileA.Voice) + oldModel, oldConfig, oldVoice := session.ModelInterface, session.ModelConfig, session.Voice + err := update(f, session, &types.RealtimeSession{ + Model: "pipe-b", + Audio: &types.RealtimeSessionAudio{Output: &types.SessionAudioOutput{Voice: "localai://voice-profiles/00000000-0000-0000-0000-000000000001"}}, + }) + + Expect(err).To(HaveOccurred()) + Expect(session.Model).To(Equal("pipe-a")) + Expect(session.ModelConfig).To(BeIdenticalTo(oldConfig)) + Expect(session.ModelInterface).To(BeIdenticalTo(oldModel)) + Expect(session.Voice).To(Equal(oldVoice)) + Expect(oldVoice).To(BeAnExistingFile()) + Expect(session.ttsParams).To(HaveKeyWithValue("ref_text", "Alpha transcript")) + }) + + It("releases a candidate profile lease when later validation fails", func(ctx SpecContext) { + f := newFixture(ctx) + session := newSession(f, "speaker-1") + leases := func() []string { + matches, err := filepath.Glob(filepath.Join(f.voiceDir, voiceprofile.DirectoryName, ".leases", "*", "*.wav")) + Expect(err).NotTo(HaveOccurred()) + return matches + } + Expect(leases()).To(BeEmpty()) + + err := update(f, session, &types.RealtimeSession{ + Audio: &types.RealtimeSessionAudio{Output: &types.SessionAudioOutput{Voice: types.Voice(f.profileA.Voice)}}, + LocalAIClassifier: classifierTestConfig(0, nil), + }) + + Expect(err).To(HaveOccurred()) + Expect(session.Voice).To(Equal("speaker-1")) + Expect(leases()).To(BeEmpty()) + }) +}) + func ptrTo[T any](value T) *T { return &value } diff --git a/docs/content/features/text-to-audio.md b/docs/content/features/text-to-audio.md index 877bd4796..a30e5f877 100644 --- a/docs/content/features/text-to-audio.md +++ b/docs/content/features/text-to-audio.md @@ -141,7 +141,22 @@ pipeline: LocalAI resolves this profile when the realtime session starts. The selected TTS model must support Voice Library cloning. -This feature does not resolve Voice Library URIs sent later through realtime `session.update`. You can still use `session.update` with ordinary backend voice names or IDs. +You can also change the profile during a realtime session. Set `audio.output.voice` to a Voice Library URI in a `session.update` event: + +```json +{ + "type": "session.update", + "session": { + "audio": { + "output": { + "voice": "localai://voice-profiles/550e8400-e29b-41d4-a716-446655440000" + } + } + } +} +``` + +If the same update changes `model`, the explicit `audio.output.voice` value takes precedence over the new model's `tts.voice` default. The selected model must support Voice Library cloning. #### Supported backend and model variants