fix(deepseek): keep parallel Responses tool outputs contiguous

This commit is contained in:
laoxiechuzheng
2026-09-17 15:45:56 +08:00
parent 881ab1b0c6
commit acc05620cb
2 changed files with 171 additions and 66 deletions
@@ -23,93 +23,139 @@ func LiftResponsesToolOutputMedia(input any) (any, bool) {
}
rewritten := make([]any, 0, len(items)+1)
pending := make([]responsesToolOutputMedia, 0)
changed := false
flushMedia := func() {
if len(pending) == 0 {
return
}
content := make([]map[string]any, 0, len(pending)*2)
lastCallID := ""
for _, media := range pending {
if media.callID != lastCallID {
text := "Tool output media"
if media.callID != "" {
text = fmt.Sprintf(toolOutputMediaAttribution, media.callID)
}
content = append(content, map[string]any{
"type": "input_text",
"text": text,
})
lastCallID = media.callID
}
content = append(content, map[string]any{
"type": "input_image",
"image_url": media.imageURL,
})
}
rewritten = append(rewritten, map[string]any{
"type": "message",
"role": "user",
"content": content,
})
pending = pending[:0]
}
for _, rawItem := range items {
item, ok := rawItem.(map[string]any)
// DeepSeek validates parallel tool outputs as one contiguous run. Codex can
// inject developer notices between image outputs, so keep those notices
// pending until the full batch and its lifted media have been emitted.
for index := 0; index < len(items); {
item, ok := items[index].(map[string]any)
if !ok || !isResponsesToolOutputItem(item) {
flushMedia()
rewritten = append(rewritten, rawItem)
rewritten = append(rewritten, items[index])
index++
continue
}
output, exists := item["output"]
if !exists {
rewritten = append(rewritten, item)
continue
}
outputRaw, err := json.Marshal(output)
if err != nil {
rewritten = append(rewritten, item)
continue
}
batchStart := index
outputs := make([]any, 0)
trailing := make([]any, 0)
pending := make([]responsesToolOutputMedia, 0)
batchChanged := false
outputText, media, didRewrite := extractToolOutputMedia(outputRaw)
if !didRewrite {
rewritten = append(rewritten, item)
continue
}
item["output"] = outputText
callID := strings.TrimSpace(stringValue(item["call_id"]))
for _, part := range media {
if part.ImageURL == nil {
for index < len(items) {
rawItem := items[index]
item, ok := rawItem.(map[string]any)
if !ok {
break
}
if isResponsesToolOutputItem(item) {
rewrittenItem, media, didRewrite := liftResponsesToolOutputMediaItem(item)
if didRewrite {
batchChanged = true
pending = append(pending, media...)
}
outputs = append(outputs, rewrittenItem)
index++
continue
}
imageURL := strings.TrimSpace(part.ImageURL.URL)
if imageURL == "" {
if isResponsesToolBatchInstruction(item) {
trailing = append(trailing, rawItem)
index++
continue
}
pending = append(pending, responsesToolOutputMedia{
callID: callID,
imageURL: imageURL,
})
break
}
if !batchChanged {
rewritten = append(rewritten, items[batchStart:index]...)
continue
}
rewritten = append(rewritten, outputs...)
if len(pending) > 0 {
rewritten = append(rewritten, buildResponsesToolOutputMediaMessage(pending))
}
rewritten = append(rewritten, trailing...)
changed = true
rewritten = append(rewritten, item)
}
flushMedia()
if !changed {
return input, false
}
return rewritten, true
}
func liftResponsesToolOutputMediaItem(item map[string]any) (any, []responsesToolOutputMedia, bool) {
output, exists := item["output"]
if !exists {
return item, nil, false
}
outputRaw, err := json.Marshal(output)
if err != nil {
return item, nil, false
}
outputText, media, didRewrite := extractToolOutputMedia(outputRaw)
if !didRewrite {
return item, nil, false
}
item["output"] = outputText
callID := strings.TrimSpace(stringValue(item["call_id"]))
lifted := make([]responsesToolOutputMedia, 0, len(media))
for _, part := range media {
if part.ImageURL == nil {
continue
}
imageURL := strings.TrimSpace(part.ImageURL.URL)
if imageURL == "" {
continue
}
lifted = append(lifted, responsesToolOutputMedia{
callID: callID,
imageURL: imageURL,
})
}
return item, lifted, true
}
func buildResponsesToolOutputMediaMessage(pending []responsesToolOutputMedia) map[string]any {
content := make([]map[string]any, 0, len(pending)*2)
lastCallID := ""
for _, media := range pending {
if media.callID != lastCallID {
text := "Tool output media"
if media.callID != "" {
text = fmt.Sprintf(toolOutputMediaAttribution, media.callID)
}
content = append(content, map[string]any{
"type": "input_text",
"text": text,
})
lastCallID = media.callID
}
content = append(content, map[string]any{
"type": "input_image",
"image_url": media.imageURL,
})
}
return map[string]any{
"type": "message",
"role": "user",
"content": content,
}
}
func isResponsesToolBatchInstruction(item map[string]any) bool {
itemType := strings.TrimSpace(stringValue(item["type"]))
if itemType != "" && itemType != "message" {
return false
}
role := strings.TrimSpace(stringValue(item["role"]))
return role == "developer" || role == "system"
}
func isResponsesToolOutputItem(item map[string]any) bool {
switch strings.TrimSpace(stringValue(item["type"])) {
case "function_call_output", "custom_tool_call_output",
@@ -55,3 +55,62 @@ func TestLiftResponsesToolOutputMediaLeavesPlainOutputUntouched(t *testing.T) {
require.False(t, changed)
require.Equal(t, input, lifted)
}
func TestLiftResponsesToolOutputMediaKeepsParallelBatchContiguous(t *testing.T) {
var input any
require.NoError(t, json.Unmarshal([]byte(`[
{"type":"function_call","call_id":"call_A","name":"view_image","arguments":"{}"},
{"type":"function_call","call_id":"call_B","name":"view_image","arguments":"{}"},
{"type":"function_call","call_id":"call_C","name":"view_image","arguments":"{}"},
{"type":"function_call_output","call_id":"call_A","output":[{"type":"input_image","image_url":"data:image/png;base64,QQ=="}]},
{"type":"message","role":"developer","content":[{"type":"input_text","text":"resized A"}]},
{"type":"function_call_output","call_id":"call_B","output":[{"type":"input_image","image_url":"data:image/png;base64,Qg=="}]},
{"type":"message","role":"developer","content":[{"type":"input_text","text":"resized B"}]},
{"type":"function_call_output","call_id":"call_C","output":[{"type":"input_image","image_url":"data:image/png;base64,Qw=="}]},
{"type":"message","role":"developer","content":[{"type":"input_text","text":"resized C"}]}
]`), &input))
lifted, changed := LiftResponsesToolOutputMedia(input)
require.True(t, changed)
items, ok := lifted.([]any)
require.True(t, ok)
require.Len(t, items, 10)
for index, callID := range []string{"call_A", "call_B", "call_C"} {
item, ok := items[index+3].(map[string]any)
require.True(t, ok)
require.Equal(t, "function_call_output", item["type"])
require.Equal(t, callID, item["call_id"])
outputText, ok := item["output"].(string)
require.True(t, ok)
require.Contains(t, outputText, toolOutputMediaMarker)
require.NotContains(t, outputText, "data:image/png")
}
mediaMessage, ok := items[6].(map[string]any)
require.True(t, ok)
require.Equal(t, "message", mediaMessage["type"])
require.Equal(t, "user", mediaMessage["role"])
parts, ok := mediaMessage["content"].([]map[string]any)
require.True(t, ok)
require.Len(t, parts, 6)
require.Equal(t, "[Tool output media for call call_A]", parts[0]["text"])
require.Equal(t, "data:image/png;base64,QQ==", parts[1]["image_url"])
require.Equal(t, "[Tool output media for call call_B]", parts[2]["text"])
require.Equal(t, "data:image/png;base64,Qg==", parts[3]["image_url"])
require.Equal(t, "[Tool output media for call call_C]", parts[4]["text"])
require.Equal(t, "data:image/png;base64,Qw==", parts[5]["image_url"])
for index, text := range []string{"resized A", "resized B", "resized C"} {
notice, ok := items[index+7].(map[string]any)
require.True(t, ok)
require.Equal(t, "developer", notice["role"])
content, ok := notice["content"].([]any)
require.True(t, ok)
require.Len(t, content, 1)
contentPart, ok := content[0].(map[string]any)
require.True(t, ok)
require.Equal(t, text, contentPart["text"])
}
}