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: