diff --git a/relay/channel/deepseek/adaptor.go b/relay/channel/deepseek/adaptor.go index 9b89957..ed828b7 100644 --- a/relay/channel/deepseek/adaptor.go +++ b/relay/channel/deepseek/adaptor.go @@ -26,7 +26,7 @@ func (a *Adaptor) Init(info *common.RelayInfo) { } // GetRequestURL 构建上游请求 URL。 -// Completions(FIM) 模式使用 /beta/completions 端点,Responses 使用 /v1/responses,其余走标准 OpenAI 路径。 +// Completions(FIM) 模式使用 /beta/completions 端点,Responses 视渠道开关直连 /v1/responses 或转 chat,其余走标准 OpenAI 路径。 func (a *Adaptor) GetRequestURL(info *common.RelayInfo) (string, error) { baseURL := strings.TrimSuffix(info.ChannelMeta.BaseURL, "/") @@ -41,8 +41,16 @@ func (a *Adaptor) GetRequestURL(info *common.RelayInfo) (string, error) { return baseURL + "/v1/chat/completions", nil case constant.RelayModeClaudeMessages: return baseURL + "/anthropic/v1/messages", nil - case constant.RelayModeResponses: - return baseURL + "/v1/responses", nil + case constant.RelayModeResponses, constant.RelayModeResponsesCompact: + // 上游原生支持 Responses 协议(渠道开启 supports_responses):直连 Responses 端点 + if info.ChannelMeta.UpstreamSpeaksResponses() { + if constant.RelayMode(info.RelayMode) == constant.RelayModeResponsesCompact { + return baseURL + "/v1/responses/compact", nil + } + return baseURL + "/v1/responses", nil + } + // 兜底(chat-only 上游):请求体先转 Chat Completions 格式再发送(与 openai.Adaptor 行为对齐) + return baseURL + "/v1/chat/completions", nil case constant.RelayModeEmbeddings: return baseURL + "/v1/embeddings", nil default: @@ -65,10 +73,14 @@ func (a *Adaptor) ConvertRequest(ctx context.Context, info *common.RelayInfo, re return convertClaudeRequestForDeepSeek(requestBody, info) } - // Responses 模式:处理 DeepSeek V4 的 reasoning_effort 映射 + // Responses 模式:上游原生支持时保持 Responses 格式(模型映射 + reasoning.effort 注入); + // chat-only 上游(渠道未开 supports_responses)先转 Chat 格式,再叠加 DeepSeek 特有适配 if constant.RelayMode(info.RelayMode) == constant.RelayModeResponses || constant.RelayMode(info.RelayMode) == constant.RelayModeResponsesCompact { - return convertResponsesRequestForDeepSeek(requestBody, info) + if info.ChannelMeta.UpstreamSpeaksResponses() { + return convertResponsesRequestForDeepSeek(requestBody, info) + } + return convertResponsesToChatForDeepSeek(requestBody, info) } // 非 OpenAI 格式先转换为 OpenAI @@ -295,6 +307,35 @@ func convertResponsesRequestForDeepSeek(requestBody []byte, info *common.RelayIn return bytes.NewReader(result), nil } +// convertResponsesToChatForDeepSeek chat-only 上游兜底:将 Responses 请求转换为 Chat Completions 格式, +// 再叠加 DeepSeek 特有适配(stream_options / 思考参数注入)。 +// 模型映射与 ResponsesRequest 快照(响应侧合成 Responses 格式时 echo 请求参数)由 +// ConvertResponsesToOpenAI 完成;思考后缀经 chat 侧 reasoning_effort 映射注入。 +func convertResponsesToChatForDeepSeek(requestBody []byte, info *common.RelayInfo) (io.Reader, error) { + converted, err := openai.ConvertResponsesToOpenAI(requestBody, info) + if err != nil { + return nil, err + } + chatBody, err := io.ReadAll(converted) + if err != nil { + return nil, fmt.Errorf("read converted chat body failed: %w", err) + } + + var rawMap map[string]json.RawMessage + if err := json.Unmarshal(chatBody, &rawMap); err != nil { + return bytes.NewReader(chatBody), nil + } + + rawMap = injectStreamOptions(rawMap, info) + rawMap = injectThinkingParams(rawMap, info) + + result, err := json.Marshal(rawMap) + if err != nil { + return nil, fmt.Errorf("marshal converted request failed: %w", err) + } + return bytes.NewReader(result), nil +} + // injectResponsesReasoningEffort 为 Responses API 请求注入 DeepSeek reasoning.effort 参数。 // // Responses API 用 reasoning.effort 而非 Chat API 的 reasoning_effort 顶层字段: diff --git a/relay/channel/deepseek/adaptor_test.go b/relay/channel/deepseek/adaptor_test.go new file mode 100644 index 0000000..6b3a24f --- /dev/null +++ b/relay/channel/deepseek/adaptor_test.go @@ -0,0 +1,141 @@ +package deepseek + +import ( + "context" + "encoding/json" + "io" + "strings" + "testing" + + "github.com/qianfree/team-api/relay/common" + "github.com/qianfree/team-api/relay/constant" +) + +// deepseekResponsesInfo 构造 responses 入站的 DeepSeek 渠道 RelayInfo, +// supportsResponses 控制渠道是否声明上游原生支持 Responses 协议。 +func deepseekResponsesInfo(supportsResponses bool) *common.RelayInfo { + return &common.RelayInfo{ + RelayMode: int(constant.RelayModeResponses), + IsStream: true, + InboundFormat: constant.RelayFormatResponses, + ClientFormat: constant.RelayFormatResponses, + OriginModelName: "deepseek-v4-flash", + StreamStatus: common.NewStreamStatus(), + ChannelMeta: &common.ChannelMeta{ + ChannelType: int(constant.ProviderDeepSeek), + BaseURL: "https://api.deepseek.example.com", + UpstreamModelName: "deepseek-v4-flash", + IsModelMapped: false, + SupportsResponses: supportsResponses, + }, + } +} + +// TestGetRequestURL_ResponsesFlagRouting responses 模式按渠道开关路由: +// 开 → /v1/responses(compact → /v1/responses/compact),关 → /v1/chat/completions 兜底。 +func TestGetRequestURL_ResponsesFlagRouting(t *testing.T) { + cases := []struct { + name string + mode constant.RelayMode + supportsResponse bool + want string + }{ + {"responses_on", constant.RelayModeResponses, true, "https://api.deepseek.example.com/v1/responses"}, + {"responses_compact_on", constant.RelayModeResponsesCompact, true, "https://api.deepseek.example.com/v1/responses/compact"}, + {"responses_off", constant.RelayModeResponses, false, "https://api.deepseek.example.com/v1/chat/completions"}, + {"responses_compact_off", constant.RelayModeResponsesCompact, false, "https://api.deepseek.example.com/v1/chat/completions"}, + } + for _, c := range cases { + a := &Adaptor{} + info := deepseekResponsesInfo(c.supportsResponse) + info.RelayMode = int(c.mode) + got, err := a.GetRequestURL(info) + if err != nil { + t.Fatalf("GetRequestURL(%s) error: %v", c.name, err) + } + if got != c.want { + t.Errorf("GetRequestURL(%s) = %q, want %q", c.name, got, c.want) + } + } +} + +const deepseekResponsesRequestBody = `{"model":"deepseek-v4-flash","instructions":"You are helpful.","input":[{"type":"message","role":"user","content":[{"type":"input_text","text":"你好"}]}],"stream":true}` + +// TestConvertRequest_ResponsesChatFallback 渠道未开 supports_responses 时, +// Responses 请求必须转换为 Chat 格式(instructions→system、input→messages、stream 保留), +// 不能再以 Responses 体直发 chat-only 上游。 +func TestConvertRequest_ResponsesChatFallback(t *testing.T) { + a := &Adaptor{} + info := deepseekResponsesInfo(false) + out, err := a.ConvertRequest(context.Background(), info, []byte(deepseekResponsesRequestBody)) + if err != nil { + t.Fatalf("ConvertRequest error: %v", err) + } + raw, _ := io.ReadAll(out) + var m map[string]json.RawMessage + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatalf("converted body is not json: %v, body: %s", err, raw) + } + + // Responses 专属字段必须消失,chat 专属字段必须存在 + if _, ok := m["input"]; ok { + t.Error("input field should be converted to messages") + } + msgsRaw, ok := m["messages"] + if !ok { + t.Fatalf("messages field missing: %s", raw) + } + + // instructions → system 消息,input → user 消息 + var msgs []struct { + Role string `json:"role"` + Content json.RawMessage `json:"content"` + } + if err := json.Unmarshal(msgsRaw, &msgs); err != nil { + t.Fatalf("unmarshal messages: %v", err) + } + if len(msgs) != 2 { + t.Fatalf("messages len = %d, want 2 (system + user): %s", len(msgs), raw) + } + if msgs[0].Role != "system" || strings.TrimSpace(string(msgs[0].Content)) != `"You are helpful."` { + t.Errorf("first message should be system with instructions, got role=%s content=%s", msgs[0].Role, msgs[0].Content) + } + if msgs[1].Role != "user" || !strings.Contains(string(msgs[1].Content), "你好") { + t.Errorf("second message should be user with input text, got role=%s content=%s", msgs[1].Role, msgs[1].Content) + } + + // stream 保留 + 注入 stream_options(usage 计费需要) + if string(m["stream"]) != "true" { + t.Errorf("stream should be preserved as true, got %s", m["stream"]) + } + if _, ok := m["stream_options"]; !ok { + t.Errorf("stream_options should be injected for stream request: %s", raw) + } + + // ResponsesRequest 快照需落 RelayInfo,供响应侧合成 Responses 格式时 echo + if info.ResponsesRequest == nil { + t.Error("info.ResponsesRequest snapshot should be stashed by conversion") + } +} + +// TestConvertRequest_ResponsesUpstreamPassthrough 渠道开启 supports_responses 时, +// 保持 Responses 格式直连(仅模型映射 + reasoning 注入),不得转成 chat 体。 +func TestConvertRequest_ResponsesUpstreamPassthrough(t *testing.T) { + a := &Adaptor{} + info := deepseekResponsesInfo(true) + out, err := a.ConvertRequest(context.Background(), info, []byte(deepseekResponsesRequestBody)) + if err != nil { + t.Fatalf("ConvertRequest error: %v", err) + } + raw, _ := io.ReadAll(out) + var m map[string]json.RawMessage + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatalf("converted body is not json: %v, body: %s", err, raw) + } + if _, ok := m["input"]; !ok { + t.Errorf("input field should be preserved for responses-native upstream: %s", raw) + } + if _, ok := m["messages"]; ok { + t.Errorf("messages field should not appear for responses-native upstream: %s", raw) + } +} diff --git a/relay/channel/openai/adaptor_responses_test.go b/relay/channel/openai/adaptor_responses_test.go index fd797a2..87d9b90 100644 --- a/relay/channel/openai/adaptor_responses_test.go +++ b/relay/channel/openai/adaptor_responses_test.go @@ -698,9 +698,90 @@ func TestAdaptor_DoResponse_ResponsesInboundStream_EmbeddedErrorMidStream(t *tes } } +// TestAdaptor_DoResponse_ResponsesInboundStream_UpstreamResponsesSSE +// 上游返回 Responses 格式 SSE(渠道误配为 chat-only 但上游实际说 Responses 的典型形态): +// 每个 data 行都能解析为"空 chat chunk",绝不能静默合成空 response.completed 假成功, +// 必须返回 502 upstream error(此时已发过前置事件,无法回写错误体)。 +func TestAdaptor_DoResponse_ResponsesInboundStream_UpstreamResponsesSSE(t *testing.T) { + ss := strings.Join([]string{ + `data: {"type":"response.created","response":{"id":"resp_1"}}`, + "", + `data: {"type":"response.output_text.delta","item_id":"msg_1","delta":"Hi"}`, + "", + `data: {"type":"response.completed","response":{"id":"resp_1","usage":{"input_tokens":10,"output_tokens":5,"total_tokens":15}}}`, + "", + }, "\n") + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": {"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(ss)), + } + + info := responsesInboundInfo(true) + rec := httptest.NewRecorder() + a := &Adaptor{} + usage, err := a.DoResponse(context.Background(), resp, info, rec) + if err == nil { + t.Fatal("non-chat upstream stream should return an upstream error, got nil") + } + if usage == nil { + t.Fatal("usage should be non-nil for error path") + } + var relayErr *constant.RelayError + if !errors.As(err, &relayErr) { + t.Fatalf("error should be *constant.RelayError, got %T", err) + } + if relayErr.StatusCode != http.StatusBadGateway { + t.Errorf("status code = %d, want %d", relayErr.StatusCode, http.StatusBadGateway) + } + out := rec.Body.String() + // 首个"空 chunk"触发的 created 系列事件可以保留,但不得合成假成功、不得转发 Responses delta + if strings.Contains(out, "response.completed") { + t.Errorf("should not synthesize empty response.completed:\n%s", out) + } + if strings.Contains(out, `"delta":"Hi"`) { + t.Errorf("upstream responses delta must not be forwarded as chat delta:\n%s", out) + } +} + +// TestAdaptor_DoResponse_ResponsesInboundStream_NonSSEJSONBody +// 上游 200 但返回非流式 JSON 体(无 data: 行):应向客户端写入 502 错误并返回 upstream error, +// 而不是合成空 response.completed。 +func TestAdaptor_DoResponse_ResponsesInboundStream_NonSSEJSONBody(t *testing.T) { + body := `{"id":"chatcmpl-1","object":"chat.completion","choices":[{"message":{"role":"assistant","content":"Hi"}}],"usage":{"prompt_tokens":5,"completion_tokens":2,"total_tokens":7}}` + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": {"application/json"}}, + Body: io.NopCloser(strings.NewReader(body)), + } + + info := responsesInboundInfo(true) + rec := httptest.NewRecorder() + a := &Adaptor{} + _, err := a.DoResponse(context.Background(), resp, info, rec) + if err == nil { + t.Fatal("non-SSE upstream body should return an upstream error, got nil") + } + var relayErr *constant.RelayError + if !errors.As(err, &relayErr) { + t.Fatalf("error should be *constant.RelayError, got %T", err) + } + if !relayErr.ResponseWritten { + t.Error("ResponseWritten should be true after writing error body to client") + } + // SetEventStreamHeaders 已提前提交 HTTP 200,状态码无法改写为 502; + // 但错误体必须写入(客户端可见明确错误而非空 SSE 流) + if !strings.Contains(rec.Body.String(), "upstream_protocol_mismatch") { + t.Errorf("error body should contain protocol mismatch code: %s", rec.Body.String()) + } + if strings.Contains(rec.Body.String(), "response.completed") { + t.Errorf("should not synthesize empty response.completed: %s", rec.Body.String()) + } +} + // TestAdaptor_DoResponse_ResponsesInboundStream_UnparseableChunks // 上游 data 行全部无法解析为 chat chunk(如返回了非 chat 格式): -// 保持既有兜底行为(合成空 response.completed、不报错),但不再静默无痕。 +// 不再合成空 response.completed 假成功,必须返回 upstream error 并向客户端写入错误体。 func TestAdaptor_DoResponse_ResponsesInboundStream_UnparseableChunks(t *testing.T) { ss := strings.Join([]string{ "data: {invalid json", @@ -718,13 +799,23 @@ func TestAdaptor_DoResponse_ResponsesInboundStream_UnparseableChunks(t *testing. rec := httptest.NewRecorder() a := &Adaptor{} usage, err := a.DoResponse(context.Background(), resp, info, rec) - if err != nil { - t.Fatalf("unparseable chunks keep fallback behavior, unexpected error: %v", err) - } - if !strings.Contains(rec.Body.String(), "response.completed") { - t.Errorf("fallback response.completed missing:\n%s", rec.Body.String()) + if err == nil { + t.Fatal("unparseable non-chat stream should return an upstream error, got nil") } if usage == nil { t.Fatal("usage should be non-nil") } + var relayErr *constant.RelayError + if !errors.As(err, &relayErr) { + t.Fatalf("error should be *constant.RelayError, got %T", err) + } + if !relayErr.ResponseWritten { + t.Error("ResponseWritten should be true after writing error body to client") + } + if strings.Contains(rec.Body.String(), "response.completed") { + t.Errorf("should not synthesize empty response.completed:\n%s", rec.Body.String()) + } + if !strings.Contains(rec.Body.String(), "upstream_protocol_mismatch") { + t.Errorf("error body should contain protocol mismatch code: %s", rec.Body.String()) + } } diff --git a/relay/channel/openai/converter.go b/relay/channel/openai/converter.go index 36a1d78..705ae24 100644 --- a/relay/channel/openai/converter.go +++ b/relay/channel/openai/converter.go @@ -714,6 +714,9 @@ type r2cInputItem struct { CallID string `json:"call_id,omitempty"` Output string `json:"output,omitempty"` Text string `json:"text,omitempty"` + // function_call 项字段(Responses 历史中的助手工具调用) + Name string `json:"name,omitempty"` + Arguments string `json:"arguments,omitempty"` } type r2cContentPart struct { @@ -749,6 +752,16 @@ func r2cConvertInputToMessages(input json.RawMessage) ([]map[string]any, error) return nil, fmt.Errorf("input must be string or array: %w", err) } messages := make([]map[string]any, 0, len(items)) + // 连续的 function_call 项聚合为一条 assistant 消息(chat 协议的 tool_calls 数组语义), + // 其后的 function_call_output 转为引用对应 tool_call_id 的 tool 消息 + var pendingToolCalls []map[string]any + flushToolCalls := func() { + if len(pendingToolCalls) == 0 { + return + } + messages = append(messages, map[string]any{"role": "assistant", "content": nil, "tool_calls": pendingToolCalls}) + pendingToolCalls = nil + } for _, raw := range items { var item r2cInputItem if err := json.Unmarshal(raw, &item); err != nil { @@ -756,12 +769,31 @@ func r2cConvertInputToMessages(input json.RawMessage) ([]map[string]any, error) } switch item.Type { case "message": + flushToolCalls() if msg := r2cConvertMessage(item); msg != nil { messages = append(messages, msg) } + case "function_call": + // 历史中的助手工具调用:转为 assistant.tool_calls 条目,id 用 call_id(与 tool 消息的 tool_call_id 对应) + if item.CallID == "" && item.Name == "" { + continue + } + pendingToolCalls = append(pendingToolCalls, map[string]any{ + "id": item.CallID, + "type": "function", + "function": map[string]any{ + "name": item.Name, + "arguments": item.Arguments, + }, + }) case "function_call_output": + flushToolCalls() messages = append(messages, map[string]any{"role": "tool", "tool_call_id": item.CallID, "content": item.Output}) + case "reasoning": + // reasoning 项(含加密思考内容)无 chat 协议对应物,跳过 + continue default: + flushToolCalls() if item.Role != "" { if msg := r2cConvertMessage(item); msg != nil { messages = append(messages, msg) @@ -769,6 +801,7 @@ func r2cConvertInputToMessages(input json.RawMessage) ([]map[string]any, error) } } } + flushToolCalls() return messages, nil } @@ -777,6 +810,11 @@ func r2cConvertMessage(item r2cInputItem) map[string]any { if role == "" { role = "user" } + // Responses 的 developer 角色(OpenAI 新式系统提示,codex 等客户端常用) + // 多数第三方 chat 上游不识别(serde 严格校验直接拒绝),统一映射为 system + if role == "developer" { + role = "system" + } if len(item.Content) == 0 { return nil } diff --git a/relay/channel/openai/converter_test.go b/relay/channel/openai/converter_test.go index cab7c47..699976f 100644 --- a/relay/channel/openai/converter_test.go +++ b/relay/channel/openai/converter_test.go @@ -436,6 +436,89 @@ func TestConvertResponsesToOpenAI_InputAudioAndFile(t *testing.T) { } } +// TestConvertResponsesToOpenAI_DeveloperRoleMappedToSystem codex 等客户端会在 input 中 +// 发送 developer 角色消息(新式系统提示);第三方 chat 上游(serde 严格校验)不识别该角色, +// 必须映射为 system,否则上游直接拒绝整个请求体。 +func TestConvertResponsesToOpenAI_DeveloperRoleMappedToSystem(t *testing.T) { + info := &common.RelayInfo{ChannelMeta: &common.ChannelMeta{}} + body := []byte(`{"model":"deepseek-v4-flash","instructions":"You are helpful.",` + + `"input":[{"type":"message","role":"developer","content":[{"type":"input_text","text":"You are a coding agent."}]},` + + `{"type":"message","role":"user","content":[{"type":"input_text","text":"hi"}]}],"stream":true}`) + out, err := ConvertResponsesToOpenAI(body, info) + if err != nil { + t.Fatalf("ConvertResponsesToOpenAI: %v", err) + } + raw, _ := io.ReadAll(out) + var m map[string]any + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatalf("bad json: %v\n%s", err, raw) + } + msgs, _ := m["messages"].([]any) + if len(msgs) != 3 { + t.Fatalf("messages = %v, want 3 (instructions→system + developer→system + user)", msgs) + } + for i, want := range []string{"system", "system", "user"} { + msg := msgs[i].(map[string]any) + if got := msg["role"]; got != want { + t.Errorf("messages[%d].role = %v, want %q", i, got, want) + } + } +} + +// TestConvertResponsesToOpenAI_FunctionCallHistory codex 多轮会把历史 function_call / +// function_call_output / reasoning 项放进 input:function_call 需聚合为 assistant.tool_calls +// (其后的 tool 消息才有对应的 tool_call_id),reasoning 项直接跳过。 +func TestConvertResponsesToOpenAI_FunctionCallHistory(t *testing.T) { + info := &common.RelayInfo{ChannelMeta: &common.ChannelMeta{}} + body := []byte(`{"model":"deepseek-v4-flash","input":[` + + `{"type":"message","role":"user","content":[{"type":"input_text","text":"list files"}]},` + + `{"type":"reasoning","summary":[]},` + + `{"type":"function_call","call_id":"call_a","name":"shell","arguments":"{\"cmd\":\"ls\"}"},` + + `{"type":"function_call","call_id":"call_b","name":"read","arguments":"{\"path\":\"a.go\"}"},` + + `{"type":"function_call_output","call_id":"call_a","output":"a.go b.go"},` + + `{"type":"function_call_output","call_id":"call_b","output":"package main"}` + + `]}`) + out, err := ConvertResponsesToOpenAI(body, info) + if err != nil { + t.Fatalf("ConvertResponsesToOpenAI: %v", err) + } + raw, _ := io.ReadAll(out) + var m map[string]any + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatalf("bad json: %v\n%s", err, raw) + } + msgs, _ := m["messages"].([]any) + if len(msgs) != 4 { + t.Fatalf("messages = %v, want 4 (user + assistant[2 calls] + tool + tool)", msgs) + } + + // 连续两条 function_call 聚合为一条 assistant 消息 + assistant := msgs[1].(map[string]any) + if assistant["role"] != "assistant" { + t.Fatalf("messages[1].role = %v, want assistant", assistant["role"]) + } + calls, _ := assistant["tool_calls"].([]any) + if len(calls) != 2 { + t.Fatalf("tool_calls = %v, want 2 aggregated calls", calls) + } + first, _ := calls[0].(map[string]any) + if first["id"] != "call_a" || first["type"] != "function" { + t.Errorf("tool_calls[0] = %v", first) + } + fn, _ := first["function"].(map[string]any) + if fn["name"] != "shell" || fn["arguments"] != `{"cmd":"ls"}` { + t.Errorf("tool_calls[0].function = %v", fn) + } + + // function_call_output → tool 消息,tool_call_id 与聚合条目对应 + for i, wantID := range []string{"call_a", "call_b"} { + tool := msgs[2+i].(map[string]any) + if tool["role"] != "tool" || tool["tool_call_id"] != wantID { + t.Errorf("messages[%d] = %v, want tool with tool_call_id=%s", 2+i, tool, wantID) + } + } +} + // TestConvertOpenAIToResponses_TextFormatUnpack chat response_format 转 Responses // text.format:json_schema 需解包为扁平结构(不能原样塞入嵌套形状)。 func TestConvertOpenAIToResponses_TextFormatUnpack(t *testing.T) { diff --git a/relay/channel/openai/responses.go b/relay/channel/openai/responses.go index 625ed09..84ee7e8 100644 --- a/relay/channel/openai/responses.go +++ b/relay/channel/openai/responses.go @@ -233,6 +233,7 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R sentCreated := false sentTextDone := false parsedChunks := 0 + sawChoices := false outputIndex := 0 contentIndex := 0 toolCallIndexByID := make(map[string]int) @@ -297,6 +298,9 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R continue } parsedChunks++ + if len(chunk.Choices) > 0 { + sawChoices = true + } // 第一个 chunk:发送 response.created + output_item.added + content_part.added if !sentCreated { @@ -549,12 +553,32 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R } } - // 上游零有效 chunk 时不再静默合成空响应:记录告警便于定位 - // (上游返回非 SSE 体 / 非 chat 格式 / 空 body / 纯 [DONE] 等异常都会走到这里) - if !sentCreated { + // 上游流始终不是 chat 格式(全程无 choices 且无内容/工具调用/usage)时不再静默合成空响应: + // 典型形态包括 Responses/Claude 格式 SSE(JSON 能解析为空 chat chunk)、非流式 JSON 体、空 body 等。 + // 假成功的空 response.completed 会让客户端"成功但无内容",必须报错驱动重试换渠道/健康上报。 + if !sawChoices && contentBuilder.Len() == 0 && len(toolCallIndexByID) == 0 && usage.TotalTokens == 0 { + errUpstream := fmt.Errorf("upstream stream is not chat completions format: %d chunks parsed, none contained choices", parsedChunks) g.Log().Warningf(ctx, - "[OpenAI.handleResponsesInboundStream] upstream stream yielded no parseable chat chunks (parsed=%d), synthesizing empty response.completed: channel=%d(%s) model=%s request_id=%s", - parsedChunks, info.ChannelMeta.ChannelID, info.ChannelMeta.ChannelName, info.OriginModelName, info.RequestID) + "[OpenAI.handleResponsesInboundStream] %v: channel=%d(%s) model=%s request_id=%s", + errUpstream, info.ChannelMeta.ChannelID, info.ChannelMeta.ChannelName, info.OriginModelName, info.RequestID) + info.StreamStatus.SetEndReason(common.StreamEndReasonError, errUpstream) + if !sentCreated { + // 尚未向客户端发送任何事件:写入错误响应(与 SSE 开始前的非 200 分支行为一致) + errBody, _ := json.Marshal(map[string]any{ + "error": map[string]any{ + "message": "upstream returned a non-chat-completions stream", + "type": "upstream_error", + "param": nil, + "code": "upstream_protocol_mismatch", + }, + }) + writeUpstreamErrorResponse(writer, http.StatusBadGateway, errBody) + upstreamErr := constant.NewUpstreamError(http.StatusBadGateway, string(errBody), nil) + upstreamErr.ResponseWritten = true + return &usage, upstreamErr + } + // 已发送部分事件:流已污染无法回写错误体,直接返回错误终止处理 + return &usage, constant.NewUpstreamError(http.StatusBadGateway, errUpstream.Error(), nil) } // 估算 usage(正常结束 4 字符/token;scanner 异常等部分传输场景 2 字符/token)