diff --git a/core/http/endpoints/openai/realtime.go b/core/http/endpoints/openai/realtime.go index ebe8b5241..3aed85c0a 100644 --- a/core/http/endpoints/openai/realtime.go +++ b/core/http/endpoints/openai/realtime.go @@ -285,7 +285,9 @@ func (session *Session) issueCommit(parent context.Context, source respcoord.Sou session.commitOrderMu.Lock() defer session.commitOrderMu.Unlock() slot := session.nextCommitSlotLocked() - session.respSink.issue(parent, source, func(ctx context.Context) { + // A non-interrupting VAD commit must not supersede a response at speech end. + ifIdle := source == respcoord.SourceVAD && !session.interruptResponseEnabled() + session.respSink.issueWithPolicy(parent, source, ifIdle, func(ctx context.Context) { run(ctx, slot) }) } diff --git a/core/http/endpoints/openai/realtime_interrupt_test.go b/core/http/endpoints/openai/realtime_interrupt_test.go new file mode 100644 index 000000000..10e96a381 --- /dev/null +++ b/core/http/endpoints/openai/realtime_interrupt_test.go @@ -0,0 +1,135 @@ +// SPDX-License-Identifier: MIT +package openai + +import ( + "context" + "encoding/json" + "time" + + "github.com/mudler/LocalAI/core/config" + "github.com/mudler/LocalAI/core/http/endpoints/openai/respcoord" + "github.com/mudler/LocalAI/core/http/endpoints/openai/types" + "github.com/mudler/LocalAI/core/schema" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("realtime interrupt_response", func() { + decode := func(mode, setting string) *types.TurnDetectionUnion { + var td types.TurnDetectionUnion + Expect(json.Unmarshal([]byte(`{"type":"`+mode+`"`+setting+`}`), &td)).To(Succeed()) + return &td + } + holdResponse := func(s *Session) context.Context { + started := make(chan context.Context, 1) + s.respSink.issue(context.Background(), respcoord.SourceClient, func(ctx context.Context) { + started <- ctx + <-ctx.Done() + }) + DeferCleanup(s.respSink.shutdown) + var ctx context.Context + Eventually(started).Should(Receive(&ctx)) + return ctx + } + + DescribeTable("controls cancellation at speech onset", + func(mode, setting string, interrupts bool) { + s := &Session{TurnDetection: decode(mode, setting), InputSampleRate: 16000, + InputAudioBuffer: make([]byte, 32000), ModelConfig: &config.ModelConfig{}, + ModelInterface: &fakeModel{vadSegments: []schema.VADSegment{{Start: 0.1}}}, respSink: newResponseSink()} + ctx := holdResponse(s) + tr := &fakeTransport{} + sink := newTurnSink(s, &Conversation{}, tr, newLiveTurnState(s, tr), context.Background(), time.Now()) + vadTick(sink, 0.5) + Expect(tr.countEvents(types.ServerEventTypeInputAudioBufferSpeechStarted)).To(Equal(1)) + Expect(ctx.Err() != nil).To(Equal(interrupts)) + }, + Entry("server false", "server_vad", `,"interrupt_response":false`, false), + Entry("semantic false", "semantic_vad", `,"interrupt_response":false`, false), + Entry("server true", "server_vad", `,"interrupt_response":true`, true), + Entry("semantic true", "semantic_vad", `,"interrupt_response":true`, true), + Entry("server omitted", "server_vad", "", true), + Entry("semantic omitted", "semantic_vad", "", true), + ) + + DescribeTable("preserves the active response and commits overlapping speech", + func(mode string) { + s := &Session{TurnDetection: decode(mode, `,"interrupt_response":false`), + InputAudioTranscription: &types.AudioTranscription{}, + ModelInterface: &fakeModel{cfg: &config.ModelConfig{}, transcribeFinal: &schema.TranscriptionResult{Text: "next turn"}}, + ModelConfig: &config.ModelConfig{}, respSink: newResponseSink()} + ctx := holdResponse(s) + conv, tr := &Conversation{}, &fakeTransport{} + committed := make(chan struct{}) + s.issueCommit(context.Background(), respcoord.SourceVAD, func(turnCtx context.Context, slot *commitSlot) { + commitUtteranceWithTranscript(turnCtx, []byte{1, 2, 3, 4}, nil, nil, "", s, conv, tr, slot) + close(committed) + }) + Eventually(committed).Should(BeClosed()) + Expect(ctx.Err()).NotTo(HaveOccurred()) + Expect(tr.countEvents(types.ServerEventTypeResponseCreated)).To(BeZero()) + Expect(tr.countEvents(types.ServerEventTypeConversationItemInputAudioTranscriptionCompleted)).To(Equal(1)) + conv.Lock.Lock() + defer conv.Lock.Unlock() + Expect(conv.Items).To(HaveLen(1)) + Expect(conv.Items[0].User.Content[0].Transcript).To(Equal("next turn")) + }, + Entry("server", "server_vad"), Entry("semantic", "semantic_vad"), + ) + + It("applies session updates to interruption and echoes false", func() { + s := &Session{TurnDetection: defaultTurnDetection(nil), ModelConfig: &config.ModelConfig{}} + var update types.SessionUpdateEvent + Expect(json.Unmarshal([]byte(`{"type":"session.update","session":{"type":"realtime","audio":{"input":{"turn_detection":{"type":"semantic_vad","interrupt_response":false}}}}}`), &update)).To(Succeed()) + Expect(updateSession(s, &update.Session, nil, nil, nil, nil, nil, nil)).To(Succeed()) + Expect(s.interruptResponseEnabled()).To(BeFalse()) + raw, err := json.Marshal(s.ToServer()) + Expect(err).NotTo(HaveOccurred()) + Expect(string(raw)).To(ContainSubstring(`"interrupt_response":false`)) + Expect(json.Unmarshal([]byte(`{"type":"session.update","session":{"type":"realtime","instructions":"Keep talking"}}`), &update)).To(Succeed()) + Expect(updateSession(s, &update.Session, nil, nil, nil, nil, nil, nil)).To(Succeed()) + Expect(s.interruptResponseEnabled()).To(BeFalse()) + s.TurnDetection = decode("server_vad", `,"interrupt_response":true`) + Expect(s.interruptResponseEnabled()).To(BeTrue()) + }) + + DescribeTable("starts commits when idle and retains explicit client cancellation", + func(source respcoord.Source) { + s := &Session{TurnDetection: decode("server_vad", `,"interrupt_response":false`), respSink: newResponseSink()} + DeferCleanup(s.respSink.shutdown) + started := make(chan context.Context, 1) + s.issueCommit(context.Background(), source, func(ctx context.Context, slot *commitSlot) { + defer close(slot.done) + started <- ctx + <-ctx.Done() + }) + var ctx context.Context + Eventually(started).Should(Receive(&ctx)) + Expect(ctx.Err()).NotTo(HaveOccurred()) + s.respSink.cancel(respcoord.SourceClient) + Expect(ctx.Err()).To(MatchError(context.Canceled)) + }, Entry("VAD", respcoord.SourceVAD), Entry("client", respcoord.SourceClient), + ) + + It("keeps explicit client supersession when interruption is disabled", func() { + s := &Session{TurnDetection: decode("server_vad", `,"interrupt_response":false`), respSink: newResponseSink()} + ctx := holdResponse(s) + done := make(chan struct{}) + s.issueCommit(context.Background(), respcoord.SourceClient, func(_ context.Context, slot *commitSlot) { + close(slot.done) + close(done) + }) + Eventually(done).Should(BeClosed()) + Expect(ctx.Err()).To(MatchError(context.Canceled)) + }) + + DescribeTable("echoes an explicit false through JSON", + func(mode string) { + raw, err := json.Marshal(decode(mode, `,"interrupt_response":false`)) + Expect(err).NotTo(HaveOccurred()) + var fields map[string]any + Expect(json.Unmarshal(raw, &fields)).To(Succeed()) + Expect(fields).To(HaveKeyWithValue("interrupt_response", false)) + }, Entry("server", "server_vad"), Entry("semantic", "semantic_vad"), + ) +}) diff --git a/core/http/endpoints/openai/realtime_respcoord.go b/core/http/endpoints/openai/realtime_respcoord.go index c34ef8bd4..9a3f28e86 100644 --- a/core/http/endpoints/openai/realtime_respcoord.go +++ b/core/http/endpoints/openai/realtime_respcoord.go @@ -58,11 +58,17 @@ func newResponseSink() *responseSink { // race. Non-blocking: the superseded response drains concurrently and its later // Finished is ignored as stale. func (s *responseSink) issue(parent context.Context, source respcoord.Source, run func(ctx context.Context)) { + s.issueWithPolicy(parent, source, false, run) +} + +// issueWithPolicy checks IfIdle inside the coordinator's serialized transition, +// avoiding a race between checking for an active response and starting one. +func (s *responseSink) issueWithPolicy(parent context.Context, source respcoord.Source, ifIdle bool, run func(ctx context.Context)) { id := respcoord.ResponseID(s.seq.Add(1)) s.mu.Lock() s.bodies[id] = responseBody{parent: parent, run: run} s.mu.Unlock() - if err := s.coord.Apply(respcoord.Start{ID: id, Source: source}); err != nil { + if err := s.coord.Apply(respcoord.Start{ID: id, Source: source, IfIdle: ifIdle}); err != nil { xlog.Error("respcoord: start failed", "error", err) } } @@ -102,33 +108,9 @@ func (s *responseSink) shutdown() { func (s *responseSink) Perform(e respcoord.Effect) { switch eff := e.(type) { case respcoord.StartResponse: - s.mu.Lock() - body := s.bodies[eff.ID] - delete(s.bodies, eff.ID) - parent := body.parent - if parent == nil { - parent = context.Background() - } - ctx, cancel := context.WithCancel(parent) - s.cancels[eff.ID] = cancel - s.mu.Unlock() - - s.wg.Go(func() { - defer func() { - s.mu.Lock() - delete(s.cancels, eff.ID) - s.mu.Unlock() - // Report completion. If this response was superseded/cancelled - // the id is stale and the coordinator ignores it (so the - // terminal is never emitted twice). - if err := s.coord.Apply(respcoord.Finished{ID: eff.ID}); err != nil { - xlog.Error("respcoord: finished apply failed", "error", err) - } - }() - if body.run != nil { - body.run(ctx) - } - }) + s.runBody(eff.ID, false) + case respcoord.SkipResponse: + s.runBody(eff.ID, true) case respcoord.CancelResponse: s.mu.Lock() cancel := s.cancels[eff.ID] @@ -141,3 +123,43 @@ func (s *responseSink) Perform(e respcoord.Effect) { // Wiring the authoritative single terminal here is the next step. } } + +// runBody runs under the coordinator lock, just like Perform. +func (s *responseSink) runBody(id respcoord.ResponseID, skipped bool) { + s.mu.Lock() + body := s.bodies[id] + delete(s.bodies, id) + parent := body.parent + if parent == nil { + parent = context.Background() + } + ctx, cancel := context.WithCancel(parent) + if skipped { + // Commit bodies use the session context for transcription and this cancelled + // response context to append the user item without generating a response. + cancel() + } else { + s.cancels[id] = cancel + } + s.mu.Unlock() + + s.wg.Go(func() { + defer cancel() + defer func() { + s.mu.Lock() + delete(s.cancels, id) + s.mu.Unlock() + // Report completion. If this response was superseded/cancelled + // the id is stale and the coordinator ignores it (so the + // terminal is never emitted twice). + if !skipped { + if err := s.coord.Apply(respcoord.Finished{ID: id}); err != nil { + xlog.Error("respcoord: finished apply failed", "error", err) + } + } + }() + if body.run != nil { + body.run(ctx) + } + }) +} diff --git a/core/http/endpoints/openai/realtime_semantic_vad.go b/core/http/endpoints/openai/realtime_semantic_vad.go index fa1ed0713..ac5e6b4d8 100644 --- a/core/http/endpoints/openai/realtime_semantic_vad.go +++ b/core/http/endpoints/openai/realtime_semantic_vad.go @@ -389,3 +389,19 @@ func defaultTurnDetection(cfg *config.ModelConfig) *types.TurnDetectionUnion { }, } } + +// interruptResponseEnabled treats omission as true while preserving an explicit +// false. The session lock also makes updates visible to the VAD goroutine. +func (s *Session) interruptResponseEnabled() bool { + sessionLock.Lock() + defer sessionLock.Unlock() + var enabled *bool + if td := s.TurnDetection; td != nil { + if td.ServerVad != nil { + enabled = td.ServerVad.InterruptResponse + } else if td.SemanticVad != nil { + enabled = td.SemanticVad.InterruptResponse + } + } + return enabled == nil || *enabled +} diff --git a/core/http/endpoints/openai/realtime_turncoord.go b/core/http/endpoints/openai/realtime_turncoord.go index 8faf12479..066614b5a 100644 --- a/core/http/endpoints/openai/realtime_turncoord.go +++ b/core/http/endpoints/openai/realtime_turncoord.go @@ -92,7 +92,9 @@ func (s *turnSink) Perform(e turncoord.Effect) { s.lts.feedNewAudio(s.onsetAudio) } case turncoord.BargeIn: - s.session.respSink.cancel(respcoord.SourceVAD) + if s.session.interruptResponseEnabled() { + s.session.respSink.cancel(respcoord.SourceVAD) + } case turncoord.EmitSpeechStarted: sendEvent(s.transport, types.InputAudioBufferSpeechStartedEvent{ ServerEventBase: types.ServerEventBase{EventID: "event_TODO"}, diff --git a/core/http/endpoints/openai/respcoord/respcoord.go b/core/http/endpoints/openai/respcoord/respcoord.go index 6c8c6d80f..9cfc4803f 100644 --- a/core/http/endpoints/openai/respcoord/respcoord.go +++ b/core/http/endpoints/openai/respcoord/respcoord.go @@ -122,6 +122,8 @@ type Event interface { type Start struct { ID ResponseID Source Source + // IfIdle skips this attempt when a response is already active. + IfIdle bool } // Finished reports that the response goroutine for ID reached its own terminal. @@ -152,7 +154,7 @@ func (Shutdown) String() string { return "Shutdown" } // Returning effects as data (rather than firing callbacks inside the // transition) keeps Next pure and exhaustively testable, and lets the // Coordinator decide how/when to perform them. Exhaustively: -// CancelResponse | StartResponse | EmitTerminal. +// CancelResponse | StartResponse | SkipResponse | EmitTerminal. type Effect interface { isEffect() String() string @@ -164,6 +166,10 @@ type CancelResponse struct{ ID ResponseID } // StartResponse: spawn the response goroutine for ID. type StartResponse struct{ ID ResponseID } +// SkipResponse keeps the active response. The sink must release the skipped +// attempt's resources; a committed audio turn can still retain its transcript. +type SkipResponse struct{ ID ResponseID } + // EmitTerminal: send response.done for ID with Status. type EmitTerminal struct { ID ResponseID @@ -172,10 +178,12 @@ type EmitTerminal struct { func (CancelResponse) isEffect() {} func (StartResponse) isEffect() {} +func (SkipResponse) isEffect() {} func (EmitTerminal) isEffect() {} func (e CancelResponse) String() string { return fmt.Sprintf("CancelResponse(%d)", e.ID) } func (e StartResponse) String() string { return fmt.Sprintf("StartResponse(%d)", e.ID) } +func (e SkipResponse) String() string { return fmt.Sprintf("SkipResponse(%d)", e.ID) } func (e EmitTerminal) String() string { return fmt.Sprintf("EmitTerminal(%d,%s)", e.ID, e.Status) } @@ -213,6 +221,9 @@ func Next(s State, e Event) (State, []Effect, error) { case Active: switch ev := e.(type) { case Start: + if ev.IfIdle { + return st, []Effect{SkipResponse{ID: ev.ID}}, nil + } return Active{ID: ev.ID}, []Effect{ CancelResponse{ID: st.ID}, EmitTerminal{ID: st.ID, Status: StatusCancelled}, diff --git a/core/http/endpoints/openai/respcoord/respcoord_test.go b/core/http/endpoints/openai/respcoord/respcoord_test.go index 6a3c7c297..2af5dcc61 100644 --- a/core/http/endpoints/openai/respcoord/respcoord_test.go +++ b/core/http/endpoints/openai/respcoord/respcoord_test.go @@ -85,6 +85,12 @@ var _ = Describe("respcoord.Next", func() { Entry("idle+start -> active, spawns response", Idle{}, Start{ID: 1, Source: SourceClient}, Active{ID: 1}, []Effect{StartResponse{ID: 1}}), + Entry("idle+conditional start -> active", + Idle{}, Start{ID: 2, Source: SourceVAD, IfIdle: true}, + Active{ID: 2}, []Effect{StartResponse{ID: 2}}), + Entry("active+conditional start -> retain active and skip new response", + Active{ID: 1}, Start{ID: 2, Source: SourceVAD, IfIdle: true}, + Active{ID: 1}, []Effect{SkipResponse{ID: 2}}), Entry("idle+cancel -> idle, no-op", Idle{}, Cancel{Source: SourceVAD}, Idle{}, []Effect(nil)), diff --git a/core/http/endpoints/openai/types/types.go b/core/http/endpoints/openai/types/types.go index b6f6d9525..15b58e147 100644 --- a/core/http/endpoints/openai/types/types.go +++ b/core/http/endpoints/openai/types/types.go @@ -602,7 +602,7 @@ type ServerVad struct { CreateResponse bool `json:"create_response,omitempty"` // Whether or not to automatically interrupt any ongoing response with output to the default conversation (i.e. conversation of auto) when a VAD start event occurs. - InterruptResponse bool `json:"interrupt_response,omitempty"` + InterruptResponse *bool `json:"interrupt_response,omitempty"` // Used only for server_vad mode. Amount of audio to include before the VAD detected speech (in milliseconds). Defaults to 300ms. PrefixPaddingMs int64 `json:"prefix_padding_ms,omitempty"` @@ -636,7 +636,7 @@ type RealtimeSessionSemanticVad struct { CreateResponse bool `json:"create_response,omitempty"` // Whether or not to automatically interrupt any ongoing response with output to the default conversation (i.e. conversation of auto) when a VAD start event occurs. - InterruptResponse bool `json:"interrupt_response,omitempty"` + InterruptResponse *bool `json:"interrupt_response,omitempty"` // Used only for semantic_vad mode. The eagerness of the model to respond. low will wait longer for the user to continue speaking, high will respond more quickly. auto is the default and is equivalent to medium. low, medium, and high have max timeouts of 8s, 4s, and 2s respectively. Eagerness string `json:"eagerness,omitempty"` diff --git a/docs/content/features/openai-realtime.md b/docs/content/features/openai-realtime.md index 5799ed921..ce44ae92f 100644 --- a/docs/content/features/openai-realtime.md +++ b/docs/content/features/openai-realtime.md @@ -91,6 +91,40 @@ curl -X POST http://localhost:8080/backend/load \ The endpoint is not realtime-specific - it pre-loads any model. See [Backend Monitor]({{%relref "operations/backend-monitor" %}}) for the full request/response reference (it is the inverse of `/backend/shutdown`). +### Controlling response interruption + +By default, speech detected by either VAD mode interrupts the active response. +Set `interrupt_response: false` to let the response finish while the user speaks: + +```json +{ + "type": "session.update", + "session": { + "type": "realtime", + "audio": { + "input": { + "turn_detection": { + "type": "server_vad", + "silence_duration_ms": 500, + "interrupt_response": false + } + } + } + } +} +``` + +This option also applies to `semantic_vad`. Speech events and transcription +continue. If a response is active when another turn commits, LocalAI retains +that turn in conversation history but skips its automatic response. It does +not queue a response for that turn. A later turn can generate a response once +the active one finishes. To request a response manually, wait for both +`response.done` and the overlapping turn's `conversation.item.added` event, then +send `response.create`. Transcription can finish after the active response. +Explicit `response.cancel` and client response creation keep their existing +behavior. When replacing the turn-detection settings, omit `interrupt_response` +or set it to `true` to enable interruption again. + ### Turn detection Turn detection decides when the user has finished speaking and the pipeline should respond. Two modes are supported, matching the OpenAI session schema: