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