fix(relay):修复deepseek模型适配器处理chat协议的问题

This commit is contained in:
qian committed 2026-08-17 23:45:12 +08:00
1 parent 0a8ecbe4c7
commit e12cfb074c
6 files changed
+434 -16

No files matched your search

+46 -5
View File
@@ -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 顶层字段:
+141
View File
@@ -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)
}
}
+97 -6
View File
@@ -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())
}
}
+38
View File
@@ -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
}
+83
View File
@@ -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) {
+29 -5
View File
@@ -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)