Skip to content

Commit d05d83f

Browse files
mudlermudler
committed
feat(realtime): stream tool-call turns via tokenizer-template autoparser
Per review (richiejp): tool-call deltas exist, so streaming should work with tools too. It does — for models that use their tokenizer template. The C++ autoparser then clears reply.Message and delivers content + tool calls via ChatDeltas, so the streamed transcript carries only spoken content (no tool-call JSON leak) and the tool calls are parsed from the final response. - Drop the len(tools)==0 gate; stream when no tools OR use_tokenizer_template (grammar-based function calling still buffers, since its call is emitted as JSON in the token stream and would leak into the transcript). - streamLLMResponse takes tools/toolChoice/toolTurn, reads ChatDelta content in the token callback, parses tool calls from the final ChatDeltas, and creates the assistant content item lazily so a content-less tool turn emits only the tool calls. - Extract emitToolCallItems from the buffered path so both paths finalize tool calls, response.done, and server-side assistant-tool follow-ups identically. Assisted-by: Claude:claude-opus-4-8 go test, golangci-lint Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
1 parent 076dcdb commit d05d83f

4 files changed

Lines changed: 233 additions & 109 deletions

File tree

‎core/http/endpoints/openai/realtime.go‎

Lines changed: 32 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -1496,18 +1496,21 @@ func triggerResponseAtTurn(ctx context.Context, session *Session, conv *Conversa
14961496
},
14971497
})
14981498

1499-
// Streamed LLM path: when the pipeline opts into LLM streaming and the turn
1500-
// cannot produce a tool call (no tools), stream tokens straight to the client
1501-
// as transcript deltas and sentence-pipe them into TTS. Tool turns fall
1502-
// through to the buffered path below, since partial tool-call output can't be
1503-
// safely spoken mid-stream.
1504-
if config != nil && session.ModelConfig != nil && session.ModelConfig.Pipeline.StreamLLM() && len(tools) == 0 {
1499+
// Streamed LLM path: when the pipeline opts into LLM streaming, stream the
1500+
// transcript to the client as it is generated and synthesize the buffered
1501+
// message once. Tool turns are supported only when the model uses its
1502+
// tokenizer template: the C++ autoparser then delivers content and tool
1503+
// calls via ChatDeltas (clearing the text stream), so the spoken transcript
1504+
// never leaks tool-call tokens. Grammar-based function calling emits the
1505+
// call as JSON in the token stream, so those turns keep the buffered path.
1506+
if config != nil && session.ModelConfig != nil && session.ModelConfig.Pipeline.StreamLLM() {
1507+
canStream := len(tools) == 0 || config.TemplateConfig.UseTokenizerTemplate
15051508
var respMods []types.Modality
15061509
if overrides != nil {
15071510
respMods = overrides.OutputModalities
15081511
}
1509-
if modalitiesContainAudio(resolveOutputModalities(session.OutputModalities, respMods)) {
1510-
if streamLLMResponse(ctx, session, conv, t, responseID, conversationHistory, images, config) {
1512+
if canStream && modalitiesContainAudio(resolveOutputModalities(session.OutputModalities, respMods)) {
1513+
if streamLLMResponse(ctx, session, conv, t, responseID, conversationHistory, images, config, tools, toolChoice, toolTurn) {
15111514
return
15121515
}
15131516
}
@@ -1814,17 +1817,27 @@ func triggerResponseAtTurn(ctx context.Context, session *Session, conv *Conversa
18141817
})
18151818
}
18161819

1817-
// Handle Tool Calls. Two paths:
1818-
// - LocalAI Assistant tools (session.AssistantExecutor.IsTool) run
1819-
// server-side; we append both the call and its output to conv.Items
1820-
// and re-trigger a follow-up response so the model can speak the
1821-
// result. The client only sees observability events.
1822-
// - All other tools follow the standard OpenAI flow: emit
1823-
// function_call_arguments.done and wait for the client to send
1824-
// conversation.item.create back.
1825-
xlog.Debug("About to handle tool calls", "finalToolCallsCount", len(finalToolCalls))
1820+
// Emit the parsed tool calls, the terminal response.done, and (for
1821+
// server-side assistant tools) the follow-up response. Shared with the
1822+
// streamed path so both finalize tool calls identically.
1823+
emitToolCallItems(ctx, session, conv, t, responseID, finalToolCalls, finalSpeech != "", toolTurn)
1824+
}
1825+
1826+
// emitToolCallItems emits the realtime function_call items for the parsed tool
1827+
// calls, the terminal response.done, and — for server-side LocalAI Assistant
1828+
// tools — re-triggers a follow-up response so the model can speak the result.
1829+
// hasContent shifts the tool-call output index past the assistant content item
1830+
// when the same turn also produced spoken/text content. Two tool paths:
1831+
// - LocalAI Assistant tools (session.AssistantExecutor.IsTool) run server-side;
1832+
// we append both the call and its output to conv.Items and re-trigger. The
1833+
// client only sees observability events.
1834+
// - All other tools follow the standard OpenAI flow: emit
1835+
// function_call_arguments.done and wait for the client to send
1836+
// conversation.item.create back.
1837+
func emitToolCallItems(ctx context.Context, session *Session, conv *Conversation, t Transport, responseID string, toolCalls []functions.FuncCallResults, hasContent bool, toolTurn int) {
1838+
xlog.Debug("About to handle tool calls", "finalToolCallsCount", len(toolCalls))
18261839
executedAssistantTool := false
1827-
for i, tc := range finalToolCalls {
1840+
for i, tc := range toolCalls {
18281841
toolCallID := generateItemID()
18291842
callID := "call_" + generateUniqueID() // OpenAI uses call_xyz
18301843

@@ -1844,7 +1857,7 @@ func triggerResponseAtTurn(ctx context.Context, session *Session, conv *Conversa
18441857
conv.Lock.Unlock()
18451858

18461859
outputIndex := i
1847-
if finalSpeech != "" {
1860+
if hasContent {
18481861
outputIndex++
18491862
}
18501863

‎core/http/endpoints/openai/realtime_stream.go‎

Lines changed: 130 additions & 87 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import (
99
"github.com/mudler/LocalAI/core/config"
1010
"github.com/mudler/LocalAI/core/http/endpoints/openai/types"
1111
"github.com/mudler/LocalAI/core/schema"
12+
"github.com/mudler/LocalAI/pkg/functions"
1213
"github.com/mudler/LocalAI/pkg/reasoning"
1314
)
1415

@@ -25,6 +26,12 @@ type transcriptStreamer struct {
2526
responseID string
2627
itemID string
2728
extractor *reasoning.ReasoningExtractor
29+
30+
// announce, if set, is invoked once just before the first transcript delta.
31+
// It lets the caller create the assistant item lazily, so a content-less
32+
// tool-call turn never emits a spurious empty assistant item.
33+
announce func()
34+
announced bool
2835
}
2936

3037
func newTranscriptStreamer(ctx context.Context, t Transport, responseID, itemID, thinkingStartToken string, reasoningCfg reasoning.Config) *transcriptStreamer {
@@ -46,6 +53,12 @@ func (s *transcriptStreamer) onToken(token string) {
4653
if content == "" {
4754
return
4855
}
56+
if !s.announced {
57+
s.announced = true
58+
if s.announce != nil {
59+
s.announce()
60+
}
61+
}
4962
_ = s.t.SendEvent(types.ResponseOutputAudioTranscriptDeltaEvent{
5063
ServerEventBase: types.ServerEventBase{},
5164
ResponseID: s.responseID,
@@ -61,50 +74,61 @@ func (s *transcriptStreamer) content() string {
6174
return s.extractor.CleanedContent()
6275
}
6376

64-
// streamLLMResponse drives a streamed, plain-content (no tools) realtime reply.
65-
// It announces the assistant item before tokens arrive, streams transcript
66-
// deltas as the LLM generates, then synthesizes the whole buffered message once
67-
// (streaming the audio chunks when the TTS backend supports it, otherwise a
68-
// single unary delta) and emits the terminal events. It returns true when it has
69-
// fully handled the response so the caller can return; callers must only invoke
70-
// it for turns with no tools and an audio modality (see triggerResponseAtTurn).
71-
func streamLLMResponse(ctx context.Context, session *Session, conv *Conversation, t Transport, responseID string, history schema.Messages, images []string, llmCfg *config.ModelConfig) bool {
72-
// Announce the assistant item up front so streamed deltas target a known item.
77+
// streamLLMResponse drives a streamed realtime reply. It streams the assistant
78+
// transcript as the LLM generates, then synthesizes the whole buffered message
79+
// once (streaming the audio chunks when the TTS backend supports it, otherwise a
80+
// single unary delta). Tool calls parsed from the autoparser ChatDeltas are
81+
// emitted after the spoken content. The assistant content item is created lazily
82+
// on the first content delta, so a content-less tool-call turn emits only the
83+
// tool calls. It returns true when it has fully handled the response so the
84+
// caller can return; callers must only invoke it for an audio modality, and with
85+
// tools only when the model uses its tokenizer template (see triggerResponseAtTurn).
86+
func streamLLMResponse(ctx context.Context, session *Session, conv *Conversation, t Transport, responseID string, history schema.Messages, images []string, llmCfg *config.ModelConfig, tools []types.ToolUnion, toolChoice *types.ToolChoiceUnion, toolTurn int) bool {
87+
itemID := generateItemID()
7388
item := types.MessageItemUnion{
7489
Assistant: &types.MessageItemAssistant{
75-
ID: generateItemID(),
90+
ID: itemID,
7691
Status: types.ItemStatusInProgress,
7792
Content: []types.MessageContentOutput{{Type: types.MessageContentTypeOutputAudio}},
7893
},
7994
}
80-
conv.Lock.Lock()
81-
conv.Items = append(conv.Items, &item)
82-
conv.Lock.Unlock()
8395

84-
sendEvent(t, types.ResponseOutputItemAddedEvent{
85-
ServerEventBase: types.ServerEventBase{},
86-
ResponseID: responseID,
87-
OutputIndex: 0,
88-
Item: item,
89-
})
90-
sendEvent(t, types.ResponseContentPartAddedEvent{
91-
ServerEventBase: types.ServerEventBase{},
92-
ResponseID: responseID,
93-
ItemID: item.Assistant.ID,
94-
OutputIndex: 0,
95-
ContentIndex: 0,
96-
Part: item.Assistant.Content[0],
97-
})
96+
// announce creates the assistant content item lazily, just before the first
97+
// transcript delta — a tool-only turn never produces content, so it stays out
98+
// of the conversation and the client sees only the tool calls.
99+
announced := false
100+
announce := func() {
101+
announced = true
102+
conv.Lock.Lock()
103+
conv.Items = append(conv.Items, &item)
104+
conv.Lock.Unlock()
105+
sendEvent(t, types.ResponseOutputItemAddedEvent{
106+
ServerEventBase: types.ServerEventBase{},
107+
ResponseID: responseID,
108+
OutputIndex: 0,
109+
Item: item,
110+
})
111+
sendEvent(t, types.ResponseContentPartAddedEvent{
112+
ServerEventBase: types.ServerEventBase{},
113+
ResponseID: responseID,
114+
ItemID: itemID,
115+
OutputIndex: 0,
116+
ContentIndex: 0,
117+
Part: item.Assistant.Content[0],
118+
})
119+
}
98120

99121
cancel := func() {
100-
conv.Lock.Lock()
101-
for i := len(conv.Items) - 1; i >= 0; i-- {
102-
if conv.Items[i].Assistant != nil && conv.Items[i].Assistant.ID == item.Assistant.ID {
103-
conv.Items = append(conv.Items[:i], conv.Items[i+1:]...)
104-
break
122+
if announced {
123+
conv.Lock.Lock()
124+
for i := len(conv.Items) - 1; i >= 0; i-- {
125+
if conv.Items[i].Assistant != nil && conv.Items[i].Assistant.ID == itemID {
126+
conv.Items = append(conv.Items[:i], conv.Items[i+1:]...)
127+
break
128+
}
105129
}
130+
conv.Lock.Unlock()
106131
}
107-
conv.Lock.Unlock()
108132
sendEvent(t, types.ResponseDoneEvent{
109133
ServerEventBase: types.ServerEventBase{},
110134
Response: types.Response{ID: responseID, Object: "realtime.response", Status: types.ResponseStatusCancelled},
@@ -119,92 +143,111 @@ func streamLLMResponse(ctx context.Context, session *Session, conv *Conversation
119143
}
120144
thinkingStartToken := reasoning.DetectThinkingStartToken(template, &llmCfg.ReasoningConfig)
121145

122-
streamer := newTranscriptStreamer(ctx, t, responseID, item.Assistant.ID, thinkingStartToken, llmCfg.ReasoningConfig)
123-
cb := func(token string, _ backend.TokenUsage) bool {
146+
streamer := newTranscriptStreamer(ctx, t, responseID, itemID, thinkingStartToken, llmCfg.ReasoningConfig)
147+
streamer.announce = announce
148+
cb := func(token string, usage backend.TokenUsage) bool {
124149
if ctx.Err() != nil {
125150
return false
126151
}
127-
streamer.onToken(token)
152+
// Plain-content models stream text via the token; autoparser tool turns
153+
// clear the text and deliver content via ChatDeltas, so prefer the latter
154+
// when present. Either way only content reaches the transcript — tool-call
155+
// deltas are parsed from the final response below.
156+
text := token
157+
if len(usage.ChatDeltas) > 0 {
158+
text = functions.ContentFromChatDeltas(usage.ChatDeltas)
159+
}
160+
streamer.onToken(text)
128161
return true
129162
}
130163

131-
predFunc, err := session.ModelInterface.Predict(ctx, history, images, nil, nil, cb, nil, nil, nil, nil, nil)
164+
predFunc, err := session.ModelInterface.Predict(ctx, history, images, nil, nil, cb, tools, toolChoice, nil, nil, nil)
132165
if err != nil {
133-
sendError(t, "inference_failed", fmt.Sprintf("backend error: %v", err), "", item.Assistant.ID)
166+
sendError(t, "inference_failed", fmt.Sprintf("backend error: %v", err), "", itemID)
134167
return true
135168
}
136-
if _, err := predFunc(); err != nil {
169+
pred, err := predFunc()
170+
if err != nil {
137171
if ctx.Err() != nil {
138172
cancel()
139173
return true
140174
}
141-
sendError(t, "prediction_failed", fmt.Sprintf("backend error: %v", err), "", item.Assistant.ID)
175+
sendError(t, "prediction_failed", fmt.Sprintf("backend error: %v", err), "", itemID)
142176
return true
143177
}
144178
if ctx.Err() != nil {
145179
cancel()
146180
return true
147181
}
148182

149-
// Buffer the whole message, then synthesize it once. emitSpeech streams the
150-
// audio chunks when the TTS backend supports TTSStream, otherwise it sends a
151-
// single unary delta — no per-sentence segmentation either way.
152183
content := streamer.content()
153-
audio, err := emitSpeech(ctx, t, session, responseID, item.Assistant.ID, content)
154-
if err != nil {
155-
if ctx.Err() != nil {
156-
cancel()
184+
toolCalls := functions.ToolCallsFromChatDeltas(pred.ChatDeltas)
185+
186+
// Finalize the spoken content item only when the turn produced content. A
187+
// tool-only turn skips this entirely (no empty assistant item).
188+
if content != "" {
189+
if !announced {
190+
announce()
191+
}
192+
// Buffer the whole message, then synthesize it once. emitSpeech streams
193+
// the audio chunks when the TTS backend supports TTSStream, otherwise it
194+
// sends a single unary delta — no per-sentence segmentation either way.
195+
audio, err := emitSpeech(ctx, t, session, responseID, itemID, content)
196+
if err != nil {
197+
if ctx.Err() != nil {
198+
cancel()
199+
return true
200+
}
201+
sendError(t, "tts_error", fmt.Sprintf("TTS generation failed: %v", err), "", itemID)
157202
return true
158203
}
159-
sendError(t, "tts_error", fmt.Sprintf("TTS generation failed: %v", err), "", item.Assistant.ID)
160-
return true
161-
}
162204

163-
_, isWebRTC := t.(*WebRTCTransport)
205+
_, isWebRTC := t.(*WebRTCTransport)
164206

165-
sendEvent(t, types.ResponseOutputAudioTranscriptDoneEvent{
166-
ServerEventBase: types.ServerEventBase{},
167-
ResponseID: responseID,
168-
ItemID: item.Assistant.ID,
169-
OutputIndex: 0,
170-
ContentIndex: 0,
171-
Transcript: content,
172-
})
173-
if !isWebRTC {
174-
sendEvent(t, types.ResponseOutputAudioDoneEvent{
207+
sendEvent(t, types.ResponseOutputAudioTranscriptDoneEvent{
175208
ServerEventBase: types.ServerEventBase{},
176209
ResponseID: responseID,
177-
ItemID: item.Assistant.ID,
210+
ItemID: itemID,
178211
OutputIndex: 0,
179212
ContentIndex: 0,
213+
Transcript: content,
180214
})
181-
}
215+
if !isWebRTC {
216+
sendEvent(t, types.ResponseOutputAudioDoneEvent{
217+
ServerEventBase: types.ServerEventBase{},
218+
ResponseID: responseID,
219+
ItemID: itemID,
220+
OutputIndex: 0,
221+
ContentIndex: 0,
222+
})
223+
}
224+
225+
conv.Lock.Lock()
226+
item.Assistant.Status = types.ItemStatusCompleted
227+
item.Assistant.Content[0].Transcript = content
228+
if !isWebRTC {
229+
item.Assistant.Content[0].Audio = base64.StdEncoding.EncodeToString(audio)
230+
}
231+
conv.Lock.Unlock()
182232

183-
conv.Lock.Lock()
184-
item.Assistant.Status = types.ItemStatusCompleted
185-
item.Assistant.Content[0].Transcript = content
186-
if !isWebRTC {
187-
item.Assistant.Content[0].Audio = base64.StdEncoding.EncodeToString(audio)
233+
sendEvent(t, types.ResponseContentPartDoneEvent{
234+
ServerEventBase: types.ServerEventBase{},
235+
ResponseID: responseID,
236+
ItemID: itemID,
237+
OutputIndex: 0,
238+
ContentIndex: 0,
239+
Part: item.Assistant.Content[0],
240+
})
241+
sendEvent(t, types.ResponseOutputItemDoneEvent{
242+
ServerEventBase: types.ServerEventBase{},
243+
ResponseID: responseID,
244+
OutputIndex: 0,
245+
Item: item,
246+
})
188247
}
189-
conv.Lock.Unlock()
190248

191-
sendEvent(t, types.ResponseContentPartDoneEvent{
192-
ServerEventBase: types.ServerEventBase{},
193-
ResponseID: responseID,
194-
ItemID: item.Assistant.ID,
195-
OutputIndex: 0,
196-
ContentIndex: 0,
197-
Part: item.Assistant.Content[0],
198-
})
199-
sendEvent(t, types.ResponseOutputItemDoneEvent{
200-
ServerEventBase: types.ServerEventBase{},
201-
ResponseID: responseID,
202-
OutputIndex: 0,
203-
Item: item,
204-
})
205-
sendEvent(t, types.ResponseDoneEvent{
206-
ServerEventBase: types.ServerEventBase{},
207-
Response: types.Response{ID: responseID, Object: "realtime.response", Status: types.ResponseStatusCompleted},
208-
})
249+
// Emit any tool calls, the terminal response.done, and (for server-side
250+
// assistant tools) the follow-up turn — shared with the buffered path.
251+
emitToolCallItems(ctx, session, conv, t, responseID, toolCalls, content != "", toolTurn)
209252
return true
210253
}

0 commit comments

Comments
 (0)