From 70938ea3c35c286a0626dede552b24caa4e88d26 Mon Sep 17 00:00:00 2001 From: localai-org-maint-bot <306269227+localai-org-maint-bot@users.noreply.github.com> Date: Mon, 14 Sep 2026 16:07:21 +0000 Subject: [PATCH] fix(responses): preserve streamed output items Keep each message and reasoning item at its announced output index. Include the answer in completed responses with reasoning or fallback function calls, and retain reasoning supplied through backend deltas. Add regression coverage for stream indices, final output, plain text, and automatic tool parsing. Assisted-by: Codex:GPT-6 --- .../http/endpoints/openresponses/responses.go | 60 +++---- .../openresponses/responses_stream_test.go | 148 ++++++++++++++++++ docs/content/features/text-generation.md | 9 ++ 3 files changed, 176 insertions(+), 41 deletions(-) create mode 100644 core/http/endpoints/openresponses/responses_stream_test.go diff --git a/core/http/endpoints/openresponses/responses.go b/core/http/endpoints/openresponses/responses.go index 553c01558..2b7daa471 100644 --- a/core/http/endpoints/openresponses/responses.go +++ b/core/http/endpoints/openresponses/responses.go @@ -2424,6 +2424,8 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6 } // Non-tool-call streaming path + messageOutputIndex := outputIndex + var reasoningOutputIndex int // Emit output_item.added for message currentMessageID = fmt.Sprintf("msg_%s", uuid.New().String()) messageItem := &schema.ORItemField{ @@ -2436,7 +2438,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6 sendSSEEvent(c, &schema.ORStreamEvent{ Type: "response.output_item.added", SequenceNumber: sequenceNumber, - OutputIndex: &outputIndex, + OutputIndex: &messageOutputIndex, Item: messageItem, }) sequenceNumber++ @@ -2448,7 +2450,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6 Type: "response.content_part.added", SequenceNumber: sequenceNumber, ItemID: currentMessageID, - OutputIndex: &outputIndex, + OutputIndex: &messageOutputIndex, ContentIndex: ¤tContentIndex, Part: &emptyTextPart, }) @@ -2471,10 +2473,11 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6 } // Handle reasoning item - if extractor.Reasoning() != "" { + if extractor.Reasoning() != "" || reasoningDelta != "" { // Check if we need to create reasoning item if currentReasoningID == "" { outputIndex++ + reasoningOutputIndex = outputIndex currentReasoningID = fmt.Sprintf("reasoning_%s", uuid.New().String()) reasoningItem := &schema.ORItemField{ Type: "reasoning", @@ -2484,7 +2487,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6 sendSSEEvent(c, &schema.ORStreamEvent{ Type: "response.output_item.added", SequenceNumber: sequenceNumber, - OutputIndex: &outputIndex, + OutputIndex: &reasoningOutputIndex, Item: reasoningItem, }) sequenceNumber++ @@ -2496,7 +2499,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6 Type: "response.content_part.added", SequenceNumber: sequenceNumber, ItemID: currentReasoningID, - OutputIndex: &outputIndex, + OutputIndex: &reasoningOutputIndex, ContentIndex: ¤tReasoningContentIndex, Part: &emptyPart, }) @@ -2509,7 +2512,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6 Type: "response.output_text.delta", SequenceNumber: sequenceNumber, ItemID: currentReasoningID, - OutputIndex: &outputIndex, + OutputIndex: &reasoningOutputIndex, ContentIndex: ¤tReasoningContentIndex, Delta: strPtr(reasoningDelta), Logprobs: emptyLogprobs(), @@ -2526,7 +2529,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6 Type: "response.output_text.delta", SequenceNumber: sequenceNumber, ItemID: currentMessageID, - OutputIndex: &outputIndex, + OutputIndex: &messageOutputIndex, ContentIndex: ¤tContentIndex, Delta: strPtr(contentDelta), Logprobs: emptyLogprobs(), @@ -2595,7 +2598,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6 Type: "response.output_text.done", SequenceNumber: sequenceNumber, ItemID: currentReasoningID, - OutputIndex: &outputIndex, + OutputIndex: &reasoningOutputIndex, ContentIndex: ¤tReasoningContentIndex, Text: strPtr(finalReasoning), Logprobs: emptyLogprobs(), @@ -2608,7 +2611,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6 Type: "response.content_part.done", SequenceNumber: sequenceNumber, ItemID: currentReasoningID, - OutputIndex: &outputIndex, + OutputIndex: &reasoningOutputIndex, ContentIndex: ¤tReasoningContentIndex, Part: &reasoningPart, }) @@ -2624,7 +2627,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6 sendSSEEvent(c, &schema.ORStreamEvent{ Type: "response.output_item.done", SequenceNumber: sequenceNumber, - OutputIndex: &outputIndex, + OutputIndex: &reasoningOutputIndex, Item: reasoningItem, }) sequenceNumber++ @@ -2658,7 +2661,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6 Type: "response.output_text.done", SequenceNumber: sequenceNumber, ItemID: currentMessageID, - OutputIndex: &outputIndex, + OutputIndex: &messageOutputIndex, ContentIndex: ¤tContentIndex, Text: strPtr(result), Logprobs: logprobsPtr(mcpStreamLogprobs), @@ -2671,7 +2674,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6 Type: "response.content_part.done", SequenceNumber: sequenceNumber, ItemID: currentMessageID, - OutputIndex: &outputIndex, + OutputIndex: &messageOutputIndex, ContentIndex: ¤tContentIndex, Part: &resultPart, }) @@ -2683,7 +2686,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6 sendSSEEvent(c, &schema.ORStreamEvent{ Type: "response.output_item.done", SequenceNumber: sequenceNumber, - OutputIndex: &outputIndex, + OutputIndex: &messageOutputIndex, Item: messageItem, }) sequenceNumber++ @@ -2723,34 +2726,9 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6 // Emit response.completed now := time.Now().Unix() - // Collect final output items (reasoning first, then messages, then tool calls) - var finalOutputItems []schema.ORItemField - // Add reasoning item if it exists - if currentReasoningID != "" && finalReasoning != "" { - finalOutputItems = append(finalOutputItems, schema.ORItemField{ - Type: "reasoning", - ID: currentReasoningID, - Status: "completed", - Content: []schema.ORContentPart{makeOutputTextPart(finalReasoning)}, - }) - } - // Add message item - if len(collectedOutputItems) > 0 { - // Use collected items (may include reasoning already) - for _, item := range collectedOutputItems { - if item.Type == "message" { - finalOutputItems = append(finalOutputItems, item) - } - } - } else { - finalOutputItems = append(finalOutputItems, *messageItem) - } - // Add function_call items from fallback - for _, item := range collectedOutputItems { - if item.Type == "function_call" { - finalOutputItems = append(finalOutputItems, item) - } - } + // The final output array must use the indices announced in the stream. + // The message is opened first, followed by reasoning and fallback calls. + finalOutputItems := append([]schema.ORItemField{*messageItem}, collectedOutputItems...) responseCompleted := buildORResponse(responseID, createdAt, &now, "completed", input, finalOutputItems, &schema.ORUsage{ InputTokens: noToolTokenUsage.Prompt, OutputTokens: noToolTokenUsage.Completion, diff --git a/core/http/endpoints/openresponses/responses_stream_test.go b/core/http/endpoints/openresponses/responses_stream_test.go new file mode 100644 index 000000000..13ddb10e1 --- /dev/null +++ b/core/http/endpoints/openresponses/responses_stream_test.go @@ -0,0 +1,148 @@ +// SPDX-License-Identifier: MIT +package openresponses + +import ( + "context" + "encoding/json" + "net/http/httptest" + "strings" + + "github.com/labstack/echo/v4" + "github.com/mudler/LocalAI/core/backend" + "github.com/mudler/LocalAI/core/config" + "github.com/mudler/LocalAI/core/schema" + pb "github.com/mudler/LocalAI/pkg/grpc/proto" + "github.com/mudler/LocalAI/pkg/model" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("Responses stream item consistency", func() { + DescribeTable("preserves every item and its announced output index", func(tokens []string, chatDeltas []*pb.ChatDelta, wantReasoning, wantAnswer string, fallback bool) { + originalInference := backend.ModelInferenceFunc + DeferCleanup(func() { backend.ModelInferenceFunc = originalInference }) + backend.ModelInferenceFunc = func( + ctx context.Context, prompt string, messages schema.Messages, + images, videos, audios []string, loader *model.ModelLoader, + cfg *config.ModelConfig, cl *config.ModelConfigLoader, app *config.ApplicationConfig, + tokenCallback func(string, backend.TokenUsage) bool, tools, toolChoice string, + logprobs, topLogprobs *int, logitBias map[string]float64, metadata map[string]string, + ) (func() (backend.LLMResponse, error), error) { + return func() (backend.LLMResponse, error) { + for i, token := range tokens { + usage := backend.TokenUsage{} + if len(chatDeltas) > 0 { + usage.ChatDeltas = []*pb.ChatDelta{chatDeltas[i]} + } + if !tokenCallback(token, usage) { + break + } + } + return backend.LLMResponse{Response: strings.Join(tokens, ""), ChatDeltas: chatDeltas, Usage: backend.TokenUsage{Prompt: 3, Completion: 8}}, nil + }, nil + } + cfg := &config.ModelConfig{} + cfg.FunctionsConfig.AutomaticToolParsingFallback = fallback + cfg.FunctionsConfig.JSONRegexMatch = []string{`(?s)(.*?)`} + recorder := httptest.NewRecorder() + request := httptest.NewRequest("POST", "/v1/responses", nil) + c := echo.New().NewContext(request, recorder) + input := &schema.OpenResponsesRequest{Model: "test-model", Input: "hello", Stream: true} + err := handleOpenResponsesStream(c, "resp_test", 1, input, cfg, nil, nil, config.NewApplicationConfig(), "hello", &schema.OpenAIRequest{Context: request.Context()}, nil, false, false, nil, nil) + Expect(err).NotTo(HaveOccurred()) + Expect(recorder.Body.String()).To(HaveSuffix("data: [DONE]\n\n")) + + var events []schema.ORStreamEvent + var completed *schema.ORResponseResource + for _, line := range strings.Split(recorder.Body.String(), "\n") { + if !strings.HasPrefix(line, "data: ") || line == "data: [DONE]" { + continue + } + var event schema.ORStreamEvent + Expect(json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &event)).To(Succeed()) + Expect(event.Type).NotTo(Equal("error")) + events = append(events, event) + if event.Type == "response.completed" { + completed = event.Response + } + } + Expect(completed).NotTo(BeNil()) + wantCount := 1 + if wantReasoning != "" { + wantCount++ + } + if fallback { + wantCount++ + } + Expect(completed.Output).To(HaveLen(wantCount), "final output must retain the answer alongside reasoning and fallback calls") + + indices := map[string]int{} + done := map[string]int{} + deltas := map[string]string{} + for i, event := range events { + Expect(event.SequenceNumber).To(Equal(i)) + if event.Type == "response.output_item.added" { + Expect(event.Item).NotTo(BeNil()) + Expect(event.OutputIndex).NotTo(BeNil()) + Expect(indices).NotTo(HaveKey(event.Item.ID)) + Expect(*event.OutputIndex).To(Equal(len(indices))) + indices[event.Item.ID] = *event.OutputIndex + } + id := event.ItemID + if event.Item != nil { + id = event.Item.ID + } + if id == "" { + continue + } + Expect(indices).To(HaveKey(id)) + Expect(event.OutputIndex).NotTo(BeNil()) + Expect(*event.OutputIndex).To(Equal(indices[id]), "event %s changes the index for %s", event.Type, id) + Expect(completed.Output[indices[id]].ID).To(Equal(id)) + if event.Type == "response.output_item.done" { + done[id]++ + Expect(event.Item.Status).To(Equal("completed")) + Expect(event.Item.Type).To(Equal(completed.Output[indices[id]].Type)) + if event.Item.Type == "function_call" { + Expect(event.Item.Name).To(Equal(completed.Output[indices[id]].Name)) + Expect(event.Item.Arguments).To(Equal(completed.Output[indices[id]].Arguments)) + } else { + Expect(event.Item.Content).To(Equal(completed.Output[indices[id]].Content)) + } + } + if event.Type == "response.output_text.delta" { + deltas[id] += *event.Delta + } + } + Expect(indices).To(HaveLen(wantCount)) + for _, item := range completed.Output { + Expect(done[item.ID]).To(Equal(1)) + switch item.Type { + case "message", "reasoning": + want := wantAnswer + if item.Type == "reasoning" { + want = wantReasoning + } + parts, ok := item.Content.([]any) + Expect(ok).To(BeTrue()) + Expect(parts).To(HaveLen(1)) + Expect(parts[0].(map[string]any)["text"]).To(Equal(want)) + if !fallback { + Expect(deltas[item.ID]).To(Equal(want)) + } + case "function_call": + Expect(item.Name).To(Equal("get_weather")) + Expect(item.Arguments).To(MatchJSON(`{"city":"Rome"}`)) + Expect(item.CallID).NotTo(BeEmpty()) + default: + Fail("unexpected output item type: " + item.Type) + } + } + }, + Entry("tagged reasoning and answer", []string{"", "Let me think.", "", "The answer is 42."}, nil, "Let me think.", "The answer is 42.", false), + Entry("backend reasoning and answer deltas", []string{"", ""}, []*pb.ChatDelta{{ReasoningContent: "Let me think."}, {Content: "The answer is 42."}}, "Let me think.", "The answer is 42.", false), + Entry("plain text", []string{"Hello", " world."}, nil, "", "Hello world.", false), + Entry("automatic fallback tool call", []string{`{"name":"get_weather","arguments":{"city":"Rome"}}`}, nil, "", "", true), + Entry("reasoning and automatic fallback tool call", []string{"", "Let me think.", "", `{"name":"get_weather","arguments":{"city":"Rome"}}`}, nil, "Let me think.", "", true), + ) +}) diff --git a/docs/content/features/text-generation.md b/docs/content/features/text-generation.md index 490877e21..920f5d2b7 100644 --- a/docs/content/features/text-generation.md +++ b/docs/content/features/text-generation.md @@ -340,6 +340,15 @@ curl http://localhost:8080/v1/responses \ }' ``` +#### Streaming responses + +Set `"stream": true` to receive Server-Sent Events. Each `response.output_item.added` event assigns an `output_index` to an item. +Use that index and the item ID to associate later deltas and completion events with the same item. + +If a request without explicit tools produces reasoning, the stream uses separate items for reasoning and answer text. +Each item keeps its original index throughout the stream. +The `response.completed` event includes both items in the same index order, followed by any automatically parsed tool calls. + #### Background Processing Run requests in the background for long-running tasks: