Skip to content

Commit cdbadc9

Browse files
mudlermudler
committed
fix(streaming): deduplicate tool call emissions during streaming
The Go-side incremental JSON parser was emitting the same tool call on every streaming token because it lacked the len > lastEmittedCount guard that the XML parser had. On top of that, the post-streaming default: case re-emitted all tool calls from index 0, duplicating everything. This produced duplicate delta.tool_calls events causing clients to accumulate arguments as "{args}{args}" — invalid JSON. Fixes: - JSON incremental parser: add len(jsonResults) > lastEmittedCount guard and loop from lastEmittedCount (matching the XML parser pattern) - Post-streaming default: case: skip i < lastEmittedCount entries that were already emitted during streaming - JSON parser: use blocking channel send (matching XML parser behavior)
1 parent 706cf5d commit cdbadc9

3 files changed

Lines changed: 223 additions & 44 deletions

File tree

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

Lines changed: 48 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -265,55 +265,52 @@ func ChatEndpoint(cl *config.ModelConfigLoader, ml *model.ModelLoader, evaluator
265265
lastEmittedCount = len(partialResults)
266266
}
267267
} else {
268-
// Try JSON tool call parsing for streaming
269-
// Check if the result looks like JSON tool calls
268+
// Try JSON tool call parsing for streaming.
269+
// Only emit NEW tool calls (same guard as XML parser above).
270270
jsonResults, jsonErr := functions.ParseJSONIterative(cleanedResult, true)
271-
if jsonErr == nil && len(jsonResults) > 0 {
272-
// Check if these are tool calls (have "name" and optionally "arguments")
273-
for _, jsonObj := range jsonResults {
274-
if name, ok := jsonObj["name"].(string); ok && name != "" {
275-
// This looks like a tool call
276-
args := "{}"
277-
if argsVal, ok := jsonObj["arguments"]; ok {
278-
if argsStr, ok := argsVal.(string); ok {
279-
args = argsStr
280-
} else {
281-
argsBytes, _ := json.Marshal(argsVal)
282-
args = string(argsBytes)
283-
}
271+
if jsonErr == nil && len(jsonResults) > lastEmittedCount {
272+
for i := lastEmittedCount; i < len(jsonResults); i++ {
273+
jsonObj := jsonResults[i]
274+
name, ok := jsonObj["name"].(string)
275+
if !ok || name == "" {
276+
continue
277+
}
278+
args := "{}"
279+
if argsVal, ok := jsonObj["arguments"]; ok {
280+
if argsStr, ok := argsVal.(string); ok {
281+
args = argsStr
282+
} else {
283+
argsBytes, _ := json.Marshal(argsVal)
284+
args = string(argsBytes)
284285
}
285-
// Emit tool call
286-
initialMessage := schema.OpenAIResponse{
287-
ID: id,
288-
Created: created,
289-
Model: req.Model,
290-
Choices: []schema.Choice{{
291-
Delta: &schema.Message{
292-
Role: "assistant",
293-
ToolCalls: []schema.ToolCall{
294-
{
295-
Index: lastEmittedCount,
296-
ID: id,
297-
Type: "function",
298-
FunctionCall: schema.FunctionCall{
299-
Name: name,
300-
Arguments: args,
301-
},
286+
}
287+
initialMessage := schema.OpenAIResponse{
288+
ID: id,
289+
Created: created,
290+
Model: req.Model,
291+
Choices: []schema.Choice{{
292+
Delta: &schema.Message{
293+
Role: "assistant",
294+
ToolCalls: []schema.ToolCall{
295+
{
296+
Index: i,
297+
ID: id,
298+
Type: "function",
299+
FunctionCall: schema.FunctionCall{
300+
Name: name,
301+
Arguments: args,
302302
},
303303
},
304304
},
305-
Index: 0,
306-
FinishReason: nil,
307-
}},
308-
Object: "chat.completion.chunk",
309-
}
310-
select {
311-
case responses <- initialMessage:
312-
default:
313-
}
314-
lastEmittedCount++
305+
},
306+
Index: 0,
307+
FinishReason: nil,
308+
}},
309+
Object: "chat.completion.chunk",
315310
}
311+
responses <- initialMessage
316312
}
313+
lastEmittedCount = len(jsonResults)
317314
}
318315
}
319316
return true
@@ -426,10 +423,17 @@ func ChatEndpoint(cl *config.ModelConfigLoader, ml *model.ModelLoader, evaluator
426423
toolCallID = id
427424
}
428425

426+
if i < lastEmittedCount {
427+
// Already emitted during streaming by the incremental
428+
// JSON/XML parser — skip to avoid duplicate tool calls.
429+
continue
430+
}
431+
432+
// Tool call not yet emitted — send name + args (two chunks).
429433
initialMessage := schema.OpenAIResponse{
430434
ID: id,
431435
Created: created,
432-
Model: req.Model, // we have to return what the user sent here, due to OpenAI spec.
436+
Model: req.Model,
433437
Choices: []schema.Choice{{
434438
Delta: &schema.Message{
435439
Role: "assistant",
@@ -454,7 +458,7 @@ func ChatEndpoint(cl *config.ModelConfigLoader, ml *model.ModelLoader, evaluator
454458
responses <- schema.OpenAIResponse{
455459
ID: id,
456460
Created: created,
457-
Model: req.Model, // we have to return what the user sent here, due to OpenAI spec.
461+
Model: req.Model,
458462
Choices: []schema.Choice{{
459463
Delta: &schema.Message{
460464
Role: "assistant",

‎tests/e2e/mock-backend/main.go‎

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -115,6 +115,15 @@ func (m *MockBackend) Predict(ctx context.Context, in *pb.PredictOptions) (*pb.R
115115
}, nil
116116
}
117117

118+
// Simulate multiple tool calls in a single response (Go-side JSON parser path).
119+
if strings.Contains(in.Prompt, "MULTI_TOOL_CALL") {
120+
return &pb.Reply{
121+
Message: []byte(`{"name": "get_weather", "arguments": {"location": "Rome"}}
122+
{"name": "get_weather", "arguments": {"location": "Paris"}}`),
123+
Tokens: 30,
124+
PromptTokens: 10,
125+
}, nil
126+
}
118127
var response string
119128
toolName := mockToolNameFromRequest(in)
120129
if toolName != "" && !promptHasToolResults(in.Prompt) {
@@ -219,6 +228,38 @@ func (m *MockBackend) PredictStream(in *pb.PredictOptions, stream pb.Backend_Pre
219228
return nil
220229
}
221230

231+
// Simulate tool calls streamed as whole JSON objects (Go-side parser path).
232+
// Each object is sent as a complete chunk so the incremental parser can
233+
// detect tool calls mid-stream (unlike char-by-char which only parses after
234+
// streaming completes).
235+
if strings.Contains(in.Prompt, "MULTI_TOOL_CALL") {
236+
chunks := []string{
237+
`{"name": "get_weather", "arguments": {"location": "Rome"}}`,
238+
"\n",
239+
`{"name": "get_weather", "arguments": {"location": "Paris"}}`,
240+
}
241+
for i, chunk := range chunks {
242+
if err := stream.Send(&pb.Reply{
243+
Message: []byte(chunk),
244+
Tokens: int32(i + 1),
245+
}); err != nil {
246+
return err
247+
}
248+
}
249+
return nil
250+
}
251+
252+
// Simulate single tool call streamed as whole JSON (Go-side parser path).
253+
if strings.Contains(in.Prompt, "SINGLE_TOOL_CALL") {
254+
if err := stream.Send(&pb.Reply{
255+
Message: []byte(`{"name": "get_weather", "arguments": {"location": "San Francisco"}}`),
256+
Tokens: 1,
257+
}); err != nil {
258+
return err
259+
}
260+
return nil
261+
}
262+
222263
var toStream string
223264
toolName := mockToolNameFromRequest(in)
224265
if toolName != "" && !promptHasToolResults(in.Prompt) {

‎tests/e2e/mock_backend_test.go‎

Lines changed: 134 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -509,4 +509,138 @@ var _ = Describe("Mock Backend E2E Tests", Label("MockBackend"), func() {
509509
})
510510
})
511511
})
512+
513+
// Tests for duplicate tool call emissions during streaming.
514+
// The Go-side incremental JSON parser was emitting the same tool call on
515+
// every streaming token, and the post-streaming default: case re-emitted
516+
// all tool calls again, producing massive duplication.
517+
Describe("Streaming Tool Call Deduplication", Label("ToolDedup"), func() {
518+
// Helper: parse SSE lines and count tool call name/arguments chunks
519+
parseToolCallChunks := func(data []byte) (nameChunks int, argChunks int) {
520+
for _, line := range strings.Split(string(data), "\n") {
521+
line = strings.TrimSpace(line)
522+
if !strings.HasPrefix(line, "data: ") || line == "data: [DONE]" {
523+
continue
524+
}
525+
var chunk map[string]any
526+
if err := json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &chunk); err != nil {
527+
continue
528+
}
529+
choices, _ := chunk["choices"].([]any)
530+
if len(choices) == 0 {
531+
continue
532+
}
533+
delta, _ := choices[0].(map[string]any)["delta"].(map[string]any)
534+
if delta == nil {
535+
continue
536+
}
537+
toolCalls, _ := delta["tool_calls"].([]any)
538+
for _, tc := range toolCalls {
539+
tcMap, _ := tc.(map[string]any)
540+
fn, _ := tcMap["function"].(map[string]any)
541+
if fn == nil {
542+
continue
543+
}
544+
if name, _ := fn["name"].(string); name != "" {
545+
nameChunks++
546+
}
547+
if args, _ := fn["arguments"].(string); args != "" {
548+
argChunks++
549+
}
550+
}
551+
}
552+
return
553+
}
554+
555+
Context("Single tool call via Go-side JSON parser", func() {
556+
It("should emit exactly one tool call name without duplicates", func() {
557+
body := `{
558+
"model": "mock-model-autoparser",
559+
"messages": [{"role": "user", "content": "SINGLE_TOOL_CALL"}],
560+
"tools": [{"type": "function", "function": {"name": "get_weather", "description": "Get weather", "parameters": {"type": "object", "properties": {"location": {"type": "string"}}, "required": ["location"]}}}],
561+
"stream": true
562+
}`
563+
req, err := http.NewRequest("POST", apiURL+"/chat/completions", strings.NewReader(body))
564+
Expect(err).ToNot(HaveOccurred())
565+
req.Header.Set("Content-Type", "application/json")
566+
567+
httpClient := &http.Client{Timeout: 60 * time.Second}
568+
resp, err := httpClient.Do(req)
569+
Expect(err).ToNot(HaveOccurred())
570+
defer resp.Body.Close()
571+
Expect(resp.StatusCode).To(Equal(200))
572+
573+
data, err := io.ReadAll(resp.Body)
574+
Expect(err).ToNot(HaveOccurred())
575+
576+
nameChunks, argChunks := parseToolCallChunks(data)
577+
578+
Expect(nameChunks).To(Equal(1),
579+
"Expected exactly 1 tool call name chunk, got %d. Full SSE:\n%s",
580+
nameChunks, string(data))
581+
Expect(argChunks).To(BeNumerically(">=", 1),
582+
"Expected at least 1 arguments chunk. Full SSE:\n%s", string(data))
583+
})
584+
})
585+
586+
Context("ChatDelta tool calls (regression guard)", func() {
587+
It("should emit exactly one tool call name per tool", func() {
588+
body := `{
589+
"model": "mock-model-autoparser",
590+
"messages": [{"role": "user", "content": "AUTOPARSER_TOOL_CALL"}],
591+
"tools": [{"type": "function", "function": {"name": "search_collections", "description": "Search documents", "parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}}}],
592+
"stream": true
593+
}`
594+
req, err := http.NewRequest("POST", apiURL+"/chat/completions", strings.NewReader(body))
595+
Expect(err).ToNot(HaveOccurred())
596+
req.Header.Set("Content-Type", "application/json")
597+
598+
httpClient := &http.Client{Timeout: 60 * time.Second}
599+
resp, err := httpClient.Do(req)
600+
Expect(err).ToNot(HaveOccurred())
601+
defer resp.Body.Close()
602+
Expect(resp.StatusCode).To(Equal(200))
603+
604+
data, err := io.ReadAll(resp.Body)
605+
Expect(err).ToNot(HaveOccurred())
606+
607+
nameChunks, _ := parseToolCallChunks(data)
608+
609+
Expect(nameChunks).To(Equal(1),
610+
"Expected exactly 1 tool call name chunk from ChatDeltas, got %d. Full SSE:\n%s",
611+
nameChunks, string(data))
612+
})
613+
})
614+
615+
Context("Multiple tool calls via Go-side JSON parser", func() {
616+
It("should emit exactly two tool call names without duplicates", func() {
617+
body := `{
618+
"model": "mock-model-autoparser",
619+
"messages": [{"role": "user", "content": "MULTI_TOOL_CALL"}],
620+
"tools": [{"type": "function", "function": {"name": "get_weather", "description": "Get weather", "parameters": {"type": "object", "properties": {"location": {"type": "string"}}, "required": ["location"]}}}],
621+
"stream": true
622+
}`
623+
req, err := http.NewRequest("POST", apiURL+"/chat/completions", strings.NewReader(body))
624+
Expect(err).ToNot(HaveOccurred())
625+
req.Header.Set("Content-Type", "application/json")
626+
627+
httpClient := &http.Client{Timeout: 60 * time.Second}
628+
resp, err := httpClient.Do(req)
629+
Expect(err).ToNot(HaveOccurred())
630+
defer resp.Body.Close()
631+
Expect(resp.StatusCode).To(Equal(200))
632+
633+
data, err := io.ReadAll(resp.Body)
634+
Expect(err).ToNot(HaveOccurred())
635+
636+
nameChunks, argChunks := parseToolCallChunks(data)
637+
638+
Expect(nameChunks).To(Equal(2),
639+
"Expected exactly 2 tool call name chunks (one per tool), got %d. Full SSE:\n%s",
640+
nameChunks, string(data))
641+
Expect(argChunks).To(BeNumerically(">=", 2),
642+
"Expected at least 2 arguments chunks. Full SSE:\n%s", string(data))
643+
})
644+
})
645+
})
512646
})

0 commit comments

Comments
 (0)