opensource/cloudpods
3
0
Fork 0
mirror of https://github.com/yunionio/cloudpods.git synced 2026-08-29 03:51:54 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(aiproxy): emit done events before response.completed in stream converter (#25158)

Browse Source 
Emit output_item.done, reasoning_summary_part.done, function_call_arguments.done,
and related finalize events for text, reasoning, and tool calls before the
response.completed event to match the Responses API stream contract.
This commit is contained in:
Zexi Li
2026-07-16 17:02:18 +08:00
committed by GitHub
parent 49c5cfa421
commit 669fa8009c
2 changed files with 298 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/aiproxy/providers/openai/responses_compat_test.go
+76
View File
@@ -225,14 +225,90 @@ func TestResponsesStreamConverterText(t *testing.T) {
t.Fatal(err)
}
foundCompleted := false
foundOutputItemDone := false
for _, e := range end {
if e.Event == "response.completed" {
foundCompleted = true
}
if e.Event == "response.output_item.done" {
foundOutputItemDone = true
}
}
if !foundCompleted {
t.Fatalf("events = %+v", end)
}
if !foundOutputItemDone {
t.Fatalf("expected response.output_item.done in %v", responsesEventNames(end))
}
}
func TestResponsesStreamConverterReasoningAndToolDoneEvents(t *testing.T) {
conv := NewResponsesStreamConverter("deepseek-v4-flash", nil)
chunks := [][]byte{
[]byte(`{"id":"chatcmpl-1","model":"deepseek-v4-flash","choices":[{"delta":{"role":"assistant","reasoning_content":"think"}}]}`),
[]byte(`{"id":"chatcmpl-1","model":"deepseek-v4-flash","choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"exec_command","arguments":"{\"cmd\":\"ls\"}"}}]}}]}`),
[]byte(`{"id":"chatcmpl-1","model":"deepseek-v4-flash","choices":[{"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":10,"completion_tokens":5,"total_tokens":15}}`),
}
var events []ResponsesStreamEvent
for _, chunk := range chunks {
evs, err := conv.Feed(chunk, false)
if err != nil {
t.Fatal(err)
}
events = append(events, evs...)
}
end, err := conv.Feed(nil, true)
if err != nil {
t.Fatal(err)
}
events = append(events, end...)
counts := map[string]int{}
for _, e := range events {
counts[e.Event]++
}
if counts["response.reasoning_summary_part.done"] != 1 {
t.Fatalf("reasoning_summary_part.done = %d, events=%v", counts["response.reasoning_summary_part.done"], responsesEventNames(events))
}
if counts["response.function_call_arguments.done"] != 1 {
t.Fatalf("function_call_arguments.done = %d, events=%v", counts["response.function_call_arguments.done"], responsesEventNames(events))
}
if counts["response.output_item.done"] < 2 {
t.Fatalf("output_item.done = %d, want at least 2, events=%v", counts["response.output_item.done"], responsesEventNames(events))
}
if counts["response.completed"] != 1 {
t.Fatalf("completed = %d, events=%v", counts["response.completed"], responsesEventNames(events))
}
// output_item.done for tools must appear before response.completed
lastToolDone := -1
completedAt := -1
for i, e := range events {
if e.Event == "response.output_item.done" {
var payload map[string]interface{}
if err := json.Unmarshal(e.Data, &payload); err != nil {
t.Fatal(err)
}
item, _ := payload["item"].(map[string]interface{})
if item["type"] == "function_call" {
lastToolDone = i
}
}
if e.Event == "response.completed" {
completedAt = i
}
}
if lastToolDone < 0 || completedAt < 0 || lastToolDone >= completedAt {
t.Fatalf("tool output_item.done must precede completed: toolDone=%d completed=%d", lastToolDone, completedAt)
}
}
func responsesEventNames(events []ResponsesStreamEvent) []string {
out := make([]string, len(events))
for i, e := range events {
out[i] = e.Event
}
return out
}
func TestNonStreamChatCompletionToStreamPayloads(t *testing.T) {
pkg/aiproxy/providers/openai/responses_stream_compat.go
+222 -24
View File
@@ -17,6 +17,7 @@ package openai
import (
"encoding/json"
"fmt"
"sort"
"strings"
"github.com/google/uuid"
@@ -32,23 +33,27 @@ type ResponsesStreamEvent struct {
// ResponsesStreamConverter converts OpenAI chat.completion.chunk SSE to Responses SSE events.
type ResponsesStreamConverter struct {
requestModel string
toolMap CodexToolMap
created bool
completed bool
seq int64
responseID string
model string
outputIndex int
textItemID string
reasonItemID string
textStarted bool
reasonStarted bool
textBuf strings.Builder
reasonBuf strings.Builder
inputTokens int
outputTokens int
activeTools map[int]*responsesStreamToolState
requestModel string
toolMap CodexToolMap
created bool
completed bool
seq int64
responseID string
model string
outputIndex int
textItemID string
reasonItemID string
textOutputIndex int
reasonOutputIndex int
textStarted bool
reasonStarted bool
textDone bool
reasonDone bool
textBuf strings.Builder
reasonBuf strings.Builder
inputTokens int
outputTokens int
activeTools map[int]*responsesStreamToolState
}
type responsesStreamToolState struct {
@@ -59,6 +64,7 @@ type responsesStreamToolState struct {
namespace string
argsBuf strings.Builder
added bool
finalized bool
}
// NewResponsesStreamConverter creates stream state for one Responses response.
@@ -190,6 +196,7 @@ func (s *ResponsesStreamConverter) appendText(text string) ([]ResponsesStreamEve
var out []ResponsesStreamEvent
if !s.textStarted {
s.textStarted = true
s.textOutputIndex = s.outputIndex
s.textItemID = fmt.Sprintf("msg_%d", s.outputIndex)
added, err := s.outputItemAdded("message", s.textItemID, map[string]interface{}{
"type": "message",
@@ -222,6 +229,7 @@ func (s *ResponsesStreamConverter) appendReasoning(text string) ([]ResponsesStre
var out []ResponsesStreamEvent
if !s.reasonStarted {
s.reasonStarted = true
s.reasonOutputIndex = s.outputIndex
s.reasonItemID = fmt.Sprintf("rs_%d", s.outputIndex)
added, err := s.outputItemAdded("reasoning", s.reasonItemID, map[string]interface{}{
"type": "reasoning",
@@ -362,11 +370,182 @@ func (s *ResponsesStreamConverter) textDelta(text string) (ResponsesStreamEvent,
return ResponsesStreamEvent{Event: "response.output_text.delta", Data: b}, nil
}
func (s *ResponsesStreamConverter) emitCompleted() ([]ResponsesStreamEvent, error) {
if s.completed {
func (s *ResponsesStreamConverter) emitFinalizeEvents() ([]ResponsesStreamEvent, error) {
var out []ResponsesStreamEvent
if s.reasonStarted && !s.reasonDone {
events, err := s.emitReasoningDone()
if err != nil {
return nil, err
}
out = append(out, events...)
}
if s.textStarted && !s.textDone {
events, err := s.emitTextDone()
if err != nil {
return nil, err
}
out = append(out, events...)
}
toolIndexes := make([]int, 0, len(s.activeTools))
for idx := range s.activeTools {
toolIndexes = append(toolIndexes, idx)
}
sort.Ints(toolIndexes)
for _, idx := range toolIndexes {
st := s.activeTools[idx]
if st == nil || !st.added || st.finalized {
continue
}
events, err := s.emitToolDone(st)
if err != nil {
return nil, err
}
out = append(out, events...)
}
return out, nil
}
func (s *ResponsesStreamConverter) emitReasoningDone() ([]ResponsesStreamEvent, error) {
if !s.reasonStarted || s.reasonDone {
return nil, nil
}
s.completed = true
s.reasonDone = true
var out []ResponsesStreamEvent
partDone := map[string]interface{}{
"type": "response.reasoning_summary_part.done",
"sequence_number": s.nextSeq(),
"item_id": s.reasonItemID,
"output_index": s.reasonOutputIndex,
"summary_index": 0,
}
b, err := json.Marshal(partDone)
if err != nil {
return nil, err
}
out = append(out, ResponsesStreamEvent{Event: "response.reasoning_summary_part.done", Data: b})
item := map[string]interface{}{
"type": "reasoning",
"id": s.reasonItemID,
"status": "completed",
"summary": []map[string]interface{}{
{"type": "text", "text": s.reasonBuf.String()},
},
}
itemDone, err := s.outputItemDoneAt(s.reasonOutputIndex, item)
if err != nil {
return nil, err
}
out = append(out, itemDone)
return out, nil
}
func (s *ResponsesStreamConverter) emitTextDone() ([]ResponsesStreamEvent, error) {
if !s.textStarted || s.textDone {
return nil, nil
}
s.textDone = true
text := s.textBuf.String()
var out []ResponsesStreamEvent
textDone := map[string]interface{}{
"type": "response.output_text.done",
"sequence_number": s.nextSeq(),
"item_id": s.textItemID,
"output_index": s.textOutputIndex,
"content_index": 0,
"text": text,
}
b, err := json.Marshal(textDone)
if err != nil {
return nil, err
}
out = append(out, ResponsesStreamEvent{Event: "response.output_text.done", Data: b})
partDone := map[string]interface{}{
"type": "response.content_part.done",
"sequence_number": s.nextSeq(),
"item_id": s.textItemID,
"output_index": s.textOutputIndex,
"content_index": 0,
"part": map[string]interface{}{"type": "output_text", "text": text},
}
b, err = json.Marshal(partDone)
if err != nil {
return nil, err
}
out = append(out, ResponsesStreamEvent{Event: "response.content_part.done", Data: b})
item := map[string]interface{}{
"type": "message",
"id": s.textItemID,
"status": "completed",
"role": "assistant",
"content": []map[string]interface{}{
{"type": "output_text", "text": text},
},
}
itemDone, err := s.outputItemDoneAt(s.textOutputIndex, item)
if err != nil {
return nil, err
}
out = append(out, itemDone)
return out, nil
}
func (s *ResponsesStreamConverter) emitToolDone(st *responsesStreamToolState) ([]ResponsesStreamEvent, error) {
if st == nil || !st.added || st.finalized {
return nil, nil
}
st.finalized = true
args := st.argsBuf.String()
var out []ResponsesStreamEvent
argsDone := map[string]interface{}{
"type": "response.function_call_arguments.done",
"sequence_number": s.nextSeq(),
"item_id": st.itemID,
"output_index": st.outputIndex,
"arguments": args,
}
b, err := json.Marshal(argsDone)
if err != nil {
return nil, err
}
out = append(out, ResponsesStreamEvent{Event: "response.function_call_arguments.done", Data: b})
item := map[string]interface{}{
"type": "function_call",
"id": st.itemID,
"status": "completed",
"call_id": st.callID,
"name": st.name,
"arguments": args,
}
if st.namespace != "" {
item["namespace"] = st.namespace
}
itemDone, err := s.outputItemDoneAt(st.outputIndex, item)
if err != nil {
return nil, err
}
out = append(out, itemDone)
return out, nil
}
func (s *ResponsesStreamConverter) outputItemDoneAt(outputIndex int, item map[string]interface{}) (ResponsesStreamEvent, error) {
data := map[string]interface{}{
"type": "response.output_item.done",
"sequence_number": s.nextSeq(),
"output_index": outputIndex,
"item": item,
}
b, err := json.Marshal(data)
if err != nil {
return ResponsesStreamEvent{}, err
}
return ResponsesStreamEvent{Event: "response.output_item.done", Data: b}, nil
}
func (s *ResponsesStreamConverter) buildCompletedOutput() []interface{} {
output := make([]interface{}, 0)
if s.reasonBuf.Len() > 0 {
output = append(output, map[string]interface{}{
@@ -385,7 +564,13 @@ func (s *ResponsesStreamConverter) emitCompleted() ([]ResponsesStreamEvent, erro
"content": []map[string]interface{}{{"type": "text", "text": s.textBuf.String()}},
})
}
for _, st := range s.activeTools {
toolIndexes := make([]int, 0, len(s.activeTools))
for idx := range s.activeTools {
toolIndexes = append(toolIndexes, idx)
}
sort.Ints(toolIndexes)
for _, idx := range toolIndexes {
st := s.activeTools[idx]
if st == nil || !st.added {
continue
}
@@ -401,14 +586,26 @@ func (s *ResponsesStreamConverter) emitCompleted() ([]ResponsesStreamEvent, erro
}
output = append(output, item)
}
status := "completed"
return output
}
func (s *ResponsesStreamConverter) emitCompleted() ([]ResponsesStreamEvent, error) {
if s.completed {
return nil, nil
}
finalizeEvents, err := s.emitFinalizeEvents()
if err != nil {
return nil, err
}
s.completed = true
output := s.buildCompletedOutput()
data := map[string]interface{}{
"type": "response.completed",
"sequence_number": s.nextSeq(),
"response": map[string]interface{}{
"id": s.responseID,
"object": "response",
"status": status,
"status": "completed",
"model": s.model,
"output": output,
"output_text": s.textBuf.String(),
@@ -423,7 +620,8 @@ func (s *ResponsesStreamConverter) emitCompleted() ([]ResponsesStreamEvent, erro
if err != nil {
return nil, err
}
return []ResponsesStreamEvent{{Event: "response.completed", Data: b}}, nil
out := append(finalizeEvents, ResponsesStreamEvent{Event: "response.completed", Data: b})
return out, nil
}
// ToProviderChunks maps internal events to providerapi chunks.