Skip to content

fix(streaming): comply with OpenAI usage / stream_options spec #9815

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mudler merged 2 commits into from
May 14, 2026
Merged

fix(streaming): comply with OpenAI usage / stream_options spec #9815

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e794523
fix(streaming): comply with OpenAI usage / stream_options spec (#8546)
mudler May 13, 2026
122da9e
fix(streaming): silence errcheck on usage trailer Fprintf
mudler May 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 39 additions & 17 deletions core/http/endpoints/openai/chat.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,13 +131,19 @@ func ChatEndpoint(cl *config.ModelConfigLoader, ml *model.ModelLoader, evaluator
delta.Reasoning = &reasoningDelta
}

// Usage rides as a struct field for the consumer to track the
// running cumulative — it is stripped before JSON marshal so the
// wire chunk stays spec-compliant (no `usage` on intermediate
// chunks). The dedicated trailer chunk (when include_usage=true)
// carries the final totals.
usageForChunk := usage
resp := schema.OpenAIResponse{
ID: id,
Created: created,
Model: req.Model, // we have to return what the user sent here, due to OpenAI spec.
Choices: []schema.Choice{{Delta: delta, Index: 0, FinishReason: nil}},
Object: "chat.completion.chunk",
Usage: usage,
Usage: &usageForChunk,
}

responses <- resp
Expand All @@ -164,7 +170,7 @@ func ChatEndpoint(cl *config.ModelConfigLoader, ml *model.ModelLoader, evaluator
hasChatDeltaToolCalls := false
hasChatDeltaContent := false

_, tokenUsage, chatDeltas, err := ComputeChoices(req, prompt, config, cl, startupOptions, loader, func(s string, c *[]schema.Choice) {}, func(s string, usage backend.TokenUsage) bool {
_, _, chatDeltas, err := ComputeChoices(req, prompt, config, cl, startupOptions, loader, func(s string, c *[]schema.Choice) {}, func(s string, usage backend.TokenUsage) bool {
result += s

// Track whether ChatDeltas from the C++ autoparser contain
Expand Down Expand Up @@ -387,16 +393,11 @@ func ChatEndpoint(cl *config.ModelConfigLoader, ml *model.ModelLoader, evaluator

switch {
case noActionToRun:
usage := schema.OpenAIUsage{
PromptTokens: tokenUsage.Prompt,
CompletionTokens: tokenUsage.Completion,
TotalTokens: tokenUsage.Prompt + tokenUsage.Completion,
}
if extraUsage {
usage.TimingTokenGeneration = tokenUsage.TimingTokenGeneration
usage.TimingPromptProcessing = tokenUsage.TimingPromptProcessing
}

// Token-cumulative usage is communicated to the streaming
// consumer via the per-token callback's chunk struct (stripped
// before wire marshal). The final usage trailer — when the
// caller opted in with stream_options.include_usage — is built
// by the outer streaming loop, not here.
var result string
if !sentInitialRole {
var hqErr error
Expand All @@ -409,7 +410,7 @@ func ChatEndpoint(cl *config.ModelConfigLoader, ml *model.ModelLoader, evaluator
for _, chunk := range buildNoActionFinalChunks(
id, req.Model, created,
sentInitialRole, sentReasoning,
result, reasoning, usage,
result, reasoning,
) {
responses <- chunk
}
Expand Down Expand Up @@ -724,7 +725,13 @@ func ChatEndpoint(cl *config.ModelConfigLoader, ml *model.ModelLoader, evaluator
xlog.Debug("No choices in the response, skipping")
continue
}
usage = &ev.Usage // Copy a pointer to the latest usage chunk so that the stop message can reference it
// Capture the running cumulative usage from this chunk
// (when present) so the include_usage trailer can carry
// the final totals. Usage is stripped before marshal
// below so the wire chunk stays spec-compliant.
if ev.Usage != nil {
usage = ev.Usage
}
if len(ev.Choices[0].Delta.ToolCalls) > 0 {
toolsCalled = true
// Collect and merge tool call deltas for MCP execution
Expand All @@ -740,6 +747,11 @@ func ChatEndpoint(cl *config.ModelConfigLoader, ml *model.ModelLoader, evaluator
collectedContent += *sp
}
}
// OpenAI streaming spec: intermediate chunks must NOT
// carry a `usage` field. Strip the tracking copy
// before marshalling — usage is delivered via the
// dedicated trailer chunk when include_usage=true.
ev.Usage = nil
respData, err := json.Marshal(ev)
if err != nil {
xlog.Debug("Failed to marshal response", "error", err)
Expand Down Expand Up @@ -888,6 +900,9 @@ func ChatEndpoint(cl *config.ModelConfigLoader, ml *model.ModelLoader, evaluator
finishReason = FinishReasonFunctionCall
}

// Final delta chunk: empty delta with finish_reason set. Per
// OpenAI streaming spec this chunk does NOT carry usage —
// the optional trailer (below) does, gated on include_usage.
resp := &schema.OpenAIResponse{
ID: id,
Created: created,
Expand All @@ -899,11 +914,18 @@ func ChatEndpoint(cl *config.ModelConfigLoader, ml *model.ModelLoader, evaluator
Delta: &schema.Message{},
}},
Object: "chat.completion.chunk",
Usage: *usage,
}
respData, _ := json.Marshal(resp)

fmt.Fprintf(c.Response().Writer, "data: %s\n\n", respData)

// Trailing usage chunk per OpenAI spec: emit only when the
// caller opted in via stream_options.include_usage. Shape:
// {"choices":[],"usage":{...},"object":"chat.completion.chunk",...}
if input.StreamOptions != nil && input.StreamOptions.IncludeUsage && usage != nil {
trailer := streamUsageTrailerJSON(id, input.Model, created, *usage)
_, _ = fmt.Fprintf(c.Response().Writer, "data: %s\n\n", trailer)
}

fmt.Fprintf(c.Response().Writer, "data: [DONE]\n\n")
c.Response().Flush()
xlog.Debug("Stream ended")
Expand Down Expand Up @@ -1263,7 +1285,7 @@ func ChatEndpoint(cl *config.ModelConfigLoader, ml *model.ModelLoader, evaluator
Model: input.Model, // we have to return what the user sent here, due to OpenAI spec.
Choices: result,
Object: "chat.completion",
Usage: usage,
Usage: &usage,
}
respData, _ := json.Marshal(resp)
xlog.Debug("Response", "response", string(respData))
Expand Down
64 changes: 50 additions & 14 deletions core/http/endpoints/openai/chat_emit.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,12 +1,45 @@
package openai

import (
"encoding/json"
"fmt"

"github.com/mudler/LocalAI/core/schema"
"github.com/mudler/LocalAI/pkg/functions"
)

// streamUsageTrailerJSON returns the bytes of the OpenAI-spec trailing usage
// chunk emitted in streaming completions when the request opts in via
// `stream_options.include_usage: true`. The shape is:
//
// {"id":"...","object":"chat.completion.chunk","created":N,
// "model":"...","choices":[],"usage":{...}}
//
// `choices` is intentionally an empty array (not absent, not null) — that is
// what the OpenAI spec mandates, and what consumers like the official OpenAI
// SDK and Continue's openai-adapter look for to recognise this as the usage
// chunk rather than a content chunk. schema.OpenAIResponse has `omitempty`
// on Choices, so we cannot reuse it for the trailer.
func streamUsageTrailerJSON(id, model string, created int, usage schema.OpenAIUsage) []byte {
trailer := struct {
ID string `json:"id"`
Created int `json:"created"`
Model string `json:"model"`
Object string `json:"object"`
Choices []schema.Choice `json:"choices"`
Usage schema.OpenAIUsage `json:"usage"`
}{
ID: id,
Created: created,
Model: model,
Object: "chat.completion.chunk",
Choices: []schema.Choice{},
Usage: usage,
}
b, _ := json.Marshal(trailer)
return b
}

// hasRealCall reports whether functionResults contains at least one
// entry whose Name is something other than the noAction sentinel.
// Used by processTools to decide between the "answer the question"
Expand All @@ -25,44 +58,48 @@ func hasRealCall(functionResults []functions.FuncCallResults, noAction string) b
// pseudo-function or emitted no tool calls at all).
//
// When content was already streamed (contentAlreadyStreamed=true) the
// helper emits a single trailing usage chunk, optionally carrying
// reasoning that was produced but not streamed incrementally. When
// content was not streamed it emits a role chunk followed by a
// content+reasoning+usage chunk — the "send everything at once" fallback.
// helper emits a trailing reasoning chunk if any non-streamed reasoning
// remains, else nothing. When content was not streamed it emits a role
// chunk followed by a content (+reasoning) chunk — the "send everything
// at once" fallback.
//
// Reasoning re-emission is guarded by reasoningAlreadyStreamed, not by
// probing the extractor's Go-side state: the C++ autoparser delivers
// reasoning through ProcessChatDeltaReasoning which populates a
// separate accumulator that extractor.Reasoning() does not expose.
// Without this guard the callback would stream reasoning incrementally
// and the final chunk would duplicate it.
//
// The returned chunks intentionally do NOT carry a `usage` field. The
// usage trailer is emitted separately by the streaming handler when
// `stream_options.include_usage` is true, per OpenAI spec.
func buildNoActionFinalChunks(
id, model string,
created int,
contentAlreadyStreamed bool,
reasoningAlreadyStreamed bool,
content string,
reasoning string,
usage schema.OpenAIUsage,
) []schema.OpenAIResponse {
var out []schema.OpenAIResponse

if contentAlreadyStreamed {
delta := &schema.Message{}
if reasoning != "" && !reasoningAlreadyStreamed {
r := reasoning
delta.Reasoning = &r
if reasoning == "" || reasoningAlreadyStreamed {
return nil
}
r := reasoning
out = append(out, schema.OpenAIResponse{
ID: id, Created: created, Model: model,
Choices: []schema.Choice{{Delta: delta, Index: 0}},
Object: "chat.completion.chunk",
Usage: usage,
Choices: []schema.Choice{{
Delta: &schema.Message{Reasoning: &r},
Index: 0,
}},
Object: "chat.completion.chunk",
})
return out
}

// Content was not streamed — send role, then content (+reasoning) + usage.
// Content was not streamed — send role, then content (+reasoning).
out = append(out, schema.OpenAIResponse{
ID: id, Created: created, Model: model,
Choices: []schema.Choice{{
Expand All @@ -82,7 +119,6 @@ func buildNoActionFinalChunks(
ID: id, Created: created, Model: model,
Choices: []schema.Choice{{Delta: delta, Index: 0}},
Object: "chat.completion.chunk",
Usage: usage,
})
return out
}
Expand Down
46 changes: 22 additions & 24 deletions core/http/endpoints/openai/chat_emit_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -609,54 +609,52 @@ var _ = Describe("buildNoActionFinalChunks", func() {
testModel = "test-model"
testCreated = 1700000000
)
usage := schema.OpenAIUsage{PromptTokens: 5, CompletionTokens: 7, TotalTokens: 12}

Describe("Content streamed — trailing usage chunk", func() {
It("emits just one chunk with usage, no content, no reasoning when reasoning was streamed", func() {
Describe("Content streamed — trailing reasoning only", func() {
It("emits nothing when content and reasoning were already streamed", func() {
// Before the streaming-usage-spec fix this branch emitted a
// content-less chunk solely to carry `usage`. Per the OpenAI
// spec usage no longer rides on delta chunks; the dedicated
// trailer (when include_usage=true) carries it instead — so
// with nothing to deliver the helper returns no chunks.
chunks := buildNoActionFinalChunks(
testID, testModel, testCreated,
true, true,
"", "already-streamed-reasoning", usage,
"", "already-streamed-reasoning",
)

Expect(chunks).To(HaveLen(1))
Expect(chunks[0].Usage.TotalTokens).To(Equal(12))
Expect(contentOf(chunks[0])).To(BeEmpty())
Expect(reasoningOf(chunks[0])).To(BeEmpty(),
"reasoning must not be re-emitted once it was streamed via the callback")
Expect(chunks).To(BeEmpty())
})

It("emits a trailing reasoning delivery when reasoning came only at end", func() {
chunks := buildNoActionFinalChunks(
testID, testModel, testCreated,
true, false,
"", "autoparser final reasoning", usage,
"", "autoparser final reasoning",
)

Expect(chunks).To(HaveLen(1))
Expect(reasoningOf(chunks[0])).To(Equal("autoparser final reasoning"))
Expect(contentOf(chunks[0])).To(BeEmpty())
Expect(chunks[0].Usage.TotalTokens).To(Equal(12))
Expect(chunks[0].Usage).To(BeNil(),
"intermediate chunks must not carry usage per OpenAI spec")
})

It("omits reasoning when it's empty regardless of streamed flag", func() {
It("returns no chunks when reasoning is empty and content was streamed", func() {
chunks := buildNoActionFinalChunks(
testID, testModel, testCreated,
true, false,
"", "", usage,
"", "",
)

Expect(chunks).To(HaveLen(1))
Expect(reasoningOf(chunks[0])).To(BeEmpty())
Expect(chunks).To(BeEmpty())
})
})

Describe("Content not streamed — role, then content+usage", func() {
Describe("Content not streamed — role, then content", func() {
It("emits role chunk then content chunk without reasoning when reasoning was streamed", func() {
chunks := buildNoActionFinalChunks(
testID, testModel, testCreated,
false, true,
"the answer", "already-streamed-reasoning", usage,
"the answer", "already-streamed-reasoning",
)

Expect(chunks).To(HaveLen(2))
Expand All @@ -666,29 +664,29 @@ var _ = Describe("buildNoActionFinalChunks", func() {
Expect(contentOf(chunks[1])).To(Equal("the answer"))
Expect(reasoningOf(chunks[1])).To(BeEmpty(),
"reasoning must not be re-emitted if it was streamed earlier")
Expect(chunks[1].Usage.TotalTokens).To(Equal(12))
Expect(chunks[1].Usage).To(BeNil())
})

It("emits role, then content+reasoning when reasoning was not streamed", func() {
chunks := buildNoActionFinalChunks(
testID, testModel, testCreated,
false, false,
"the answer", "autoparser final reasoning", usage,
"the answer", "autoparser final reasoning",
)

Expect(chunks).To(HaveLen(2))
Expect(chunks[0].Choices[0].Delta.Role).To(Equal("assistant"))

Expect(contentOf(chunks[1])).To(Equal("the answer"))
Expect(reasoningOf(chunks[1])).To(Equal("autoparser final reasoning"))
Expect(chunks[1].Usage.TotalTokens).To(Equal(12))
Expect(chunks[1].Usage).To(BeNil())
})

It("still emits content even when reasoning is empty", func() {
chunks := buildNoActionFinalChunks(
testID, testModel, testCreated,
false, false,
"just an answer", "", usage,
"just an answer", "",
)

Expect(chunks).To(HaveLen(2))
Expand All @@ -702,7 +700,7 @@ var _ = Describe("buildNoActionFinalChunks", func() {
chunks := buildNoActionFinalChunks(
testID, testModel, testCreated,
false, false,
"hi", "reasoning", usage,
"hi", "reasoning",
)
for i, ch := range chunks {
Expect(ch.ID).To(Equal(testID), "chunk[%d] ID", i)
Expand Down
Loading
Loading