diff --git a/core/http/endpoints/openresponses/responses.go b/core/http/endpoints/openresponses/responses.go index 553c01558..6da7f2adc 100644 --- a/core/http/endpoints/openresponses/responses.go +++ b/core/http/endpoints/openresponses/responses.go @@ -1873,49 +1873,37 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6 return true } - // Try JSON parsing as fallback - jsonResults, jsonErr := functions.ParseJSONIterative(cleanedResult, true) - if jsonErr == nil && len(jsonResults) > lastEmittedToolCallCount { + // Only completed JSON calls can be emitted as completed SSE items. + jsonResults := parseStreamingJSONToolCalls(cleanedResult) + if len(jsonResults) > lastEmittedToolCallCount { for i := lastEmittedToolCallCount; i < len(jsonResults); i++ { - jsonObj := jsonResults[i] - if name, ok := jsonObj["name"].(string); ok && name != "" { - args := "{}" - if argsVal, ok := jsonObj["arguments"]; ok { - if argsStr, ok := argsVal.(string); ok { - args = argsStr - } else { - argsBytes, _ := json.Marshal(argsVal) - args = string(argsBytes) - } - } + tc := jsonResults[i] + toolCallID := fmt.Sprintf("fc_%s", uuid.New().String()) + outputIndex++ - toolCallID := fmt.Sprintf("fc_%s", uuid.New().String()) - outputIndex++ - - functionCallItem := &schema.ORItemField{ - Type: "function_call", - ID: toolCallID, - Status: "completed", - CallID: toolCallID, - Name: name, - Arguments: args, - } - sendSSEEvent(c, &schema.ORStreamEvent{ - Type: "response.output_item.added", - SequenceNumber: sequenceNumber, - OutputIndex: &outputIndex, - Item: functionCallItem, - }) - sequenceNumber++ - - sendSSEEvent(c, &schema.ORStreamEvent{ - Type: "response.output_item.done", - SequenceNumber: sequenceNumber, - OutputIndex: &outputIndex, - Item: functionCallItem, - }) - sequenceNumber++ + functionCallItem := &schema.ORItemField{ + Type: "function_call", + ID: toolCallID, + Status: "completed", + CallID: toolCallID, + Name: tc.Name, + Arguments: tc.Arguments, } + sendSSEEvent(c, &schema.ORStreamEvent{ + Type: "response.output_item.added", + SequenceNumber: sequenceNumber, + OutputIndex: &outputIndex, + Item: functionCallItem, + }) + sequenceNumber++ + + sendSSEEvent(c, &schema.ORStreamEvent{ + Type: "response.output_item.done", + SequenceNumber: sequenceNumber, + OutputIndex: &outputIndex, + Item: functionCallItem, + }) + sequenceNumber++ } lastEmittedToolCallCount = len(jsonResults) c.Response().Flush() diff --git a/core/http/endpoints/openresponses/stream_tool_calls.go b/core/http/endpoints/openresponses/stream_tool_calls.go new file mode 100644 index 000000000..b8f0f185d --- /dev/null +++ b/core/http/endpoints/openresponses/stream_tool_calls.go @@ -0,0 +1,35 @@ +package openresponses + +import ( + "encoding/json" + + "github.com/mudler/LocalAI/pkg/functions" +) + +func parseStreamingJSONToolCalls(text string) []functions.FuncCallResults { + // Partial parsing heals unfinished arguments. The caller emits terminal + // events and never revisits emitted calls, so only accept complete JSON. + // Keep completed objects returned before an unfinished trailing object. + objects, _ := functions.ParseJSONIterative(text, false) + var calls []functions.FuncCallResults + for _, object := range objects { + name, ok := object["name"].(string) + if !ok || name == "" { + continue + } + arguments := "{}" + if value, ok := object["arguments"]; ok { + if s, ok := value.(string); ok { + arguments = s + } else { + data, err := json.Marshal(value) + if err != nil { + continue + } + arguments = string(data) + } + } + calls = append(calls, functions.FuncCallResults{Name: name, Arguments: arguments}) + } + return calls +} diff --git a/core/http/endpoints/openresponses/stream_tool_calls_test.go b/core/http/endpoints/openresponses/stream_tool_calls_test.go new file mode 100644 index 000000000..1fa6bca2a --- /dev/null +++ b/core/http/endpoints/openresponses/stream_tool_calls_test.go @@ -0,0 +1,44 @@ +package openresponses + +import ( + "github.com/mudler/LocalAI/pkg/functions" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("Streaming JSON tool calls", func() { + It("waits for the arguments before completing a split call", func() { + Expect(parseStreamingJSONToolCalls(`{"name":"Bash",`)).To(BeEmpty()) + Expect(parseStreamingJSONToolCalls(`{"name":"Bash","arguments":{"command":"ls`)).To(BeEmpty()) + Expect(parseStreamingJSONToolCalls(`{"name":"Bash","arguments":{"command":"ls -la"}}`)).To(Equal([]functions.FuncCallResults{ + {Name: "Bash", Arguments: `{"command":"ls -la"}`}, + })) + }) + + It("does not complete a call at any intermediate token boundary", func() { + text := `{"name":"Bash","arguments":{"command":"printf \"hello\"","options":[1,2]}}` + for end := 1; end < len(text); end++ { + Expect(parseStreamingJSONToolCalls(text[:end])).To(BeEmpty(), "prefix: %s", text[:end]) + } + Expect(parseStreamingJSONToolCalls(text)).To(HaveLen(1)) + }) + + It("keeps completed calls while the next call is incomplete", func() { + Expect(parseStreamingJSONToolCalls(`{"name":"Bash","arguments":{"command":"ls -la"}} {"name":"Read",`)).To(Equal([]functions.FuncCallResults{ + {Name: "Bash", Arguments: `{"command":"ls -la"}`}, + })) + }) + + It("preserves string arguments and calls that take no arguments", func() { + Expect(parseStreamingJSONToolCalls(`[{"name":"Bash","arguments":"{\"command\":\"ls -la\"}"},{"name":"status"}]`)).To(Equal([]functions.FuncCallResults{ + {Name: "Bash", Arguments: `{"command":"ls -la"}`}, + {Name: "status", Arguments: `{}`}, + })) + }) + + It("does not count unrelated JSON objects as emitted calls", func() { + Expect(parseStreamingJSONToolCalls(`{"message":"checking"} {"name":"status","arguments":{}}`)).To(Equal([]functions.FuncCallResults{ + {Name: "status", Arguments: `{}`}, + })) + }) +}) diff --git a/docs/content/features/text-generation.md b/docs/content/features/text-generation.md index 490877e21..286359339 100644 --- a/docs/content/features/text-generation.md +++ b/docs/content/features/text-generation.md @@ -434,6 +434,11 @@ curl http://localhost:8080/v1/responses \ }' ``` +For streaming requests with JSON tool output, LocalAI waits for the complete JSON +object before emitting a completed `function_call` item. Arguments can span +multiple tokens. Read the arguments from the `response.output_item.done` event +before executing the tool. + #### Reasoning Configuration Configure reasoning effort and summary style: