From 8a5ef916257e005fbde150e7f67be136c911597c Mon Sep 17 00:00:00 2001 From: qian Date: Sun, 2 Aug 2026 11:52:28 +0800 Subject: [PATCH] =?UTF-8?q?feat(relay):deepseek=E6=96=B0=E5=A2=9Eresponse?= =?UTF-8?q?=E7=AB=AF=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- relay/channel/claude/response.go | 33 +++++++++--- relay/channel/deepseek/adaptor.go | 86 ++++++++++++++++++++++++++++++- 2 files changed, 111 insertions(+), 8 deletions(-) diff --git a/relay/channel/claude/response.go b/relay/channel/claude/response.go index a80f099..f37e6ec 100644 --- a/relay/channel/claude/response.go +++ b/relay/channel/claude/response.go @@ -415,16 +415,28 @@ func (a *Adaptor) handleClaudeNativeNonStream(ctx context.Context, resp *http.Re // handleClaudeNativeStream 直通 Claude 流式响应 func (a *Adaptor) handleClaudeNativeStream(ctx context.Context, resp *http.Response, info *common.RelayInfo, writer http.ResponseWriter) (*common.Usage, error) { - defer resp.Body.Close() - if resp.StatusCode != http.StatusOK { body, _ := io.ReadAll(resp.Body) + resp.Body.Close() return nil, constant.NewUpstreamError(resp.StatusCode, string(body), nil).WithRetryAfter(constant.RetryAfterFromHeader(resp.Header)) } + // 创建可取消的上下文，用于在客户端断开时立即中止上游读取 + streamCtx, cancelStream := context.WithCancel(ctx) + defer cancelStream() + + // cleanup 函数：关闭上游连接，停止 token 生成 + cleanup := func() { + cancelStream() + if resp.Body != nil { + resp.Body.Close() + } + } + helper.SetEventStreamHeaders(writer) writer = helper.NewSafeWriter(writer) - defer helper.PingTicker(writer, 15*time.Second)() + stopPing := helper.PingTicker(writer, 15*time.Second) + defer stopPing() reader := bufio.NewReaderSize(resp.Body, 64*1024) var usage dto.ClaudeUsage @@ -432,8 +444,10 @@ func (a *Adaptor) handleClaudeNativeStream(ctx context.Context, resp *http.Respo for { select { - case <-ctx.Done(): - info.StreamStatus.SetEndReason(common.StreamEndReasonClientGone, ctx.Err()) + case <-streamCtx.Done(): + // 客户端断开：立即关闭上游连接，停止生成 + cleanup() + info.StreamStatus.SetEndReason(common.StreamEndReasonClientGone, streamCtx.Err()) interruptedUsage := buildUsageFromClaude(&usage) helper.ApplyInterruptedUsageFallback(info, interruptedUsage, transferredTextLen) return interruptedUsage, common.ErrStreamInterrupted @@ -446,7 +460,12 @@ func (a *Adaptor) handleClaudeNativeStream(ctx context.Context, resp *http.Respo break } info.StreamStatus.SetEndReason(common.StreamEndReasonError, err) - return buildUsageFromClaude(&usage), fmt.Errorf("stream read error: %w", err) + // 已有部分输出时按部分成功处理（避免标记为完全失败） + interruptedUsage := buildUsageFromClaude(&usage) + if transferredTextLen > 0 { + helper.ApplyInterruptedUsageFallback(info, interruptedUsage, transferredTextLen) + } + return interruptedUsage, fmt.Errorf("upstream stream interrupted: %w", err) } if strings.HasPrefix(line, "data:") { @@ -503,6 +522,8 @@ func (a *Adaptor) handleClaudeNativeStream(ctx context.Context, resp *http.Respo } if _, err := writer.Write([]byte(line)); err != nil { + // 写入客户端失败：立即关闭上游连接，停止生成 + cleanup() info.StreamStatus.SetEndReason(common.StreamEndReasonClientGone, err) interruptedUsage := buildUsageFromClaude(&usage) helper.ApplyInterruptedUsageFallback(info, interruptedUsage, transferredTextLen) diff --git a/relay/channel/deepseek/adaptor.go b/relay/channel/deepseek/adaptor.go index 7577bc1..9b89957 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 端点，其余走标准 OpenAI 路径。 +// Completions(FIM) 模式使用 /beta/completions 端点，Responses 使用 /v1/responses，其余走标准 OpenAI 路径。 func (a *Adaptor) GetRequestURL(info *common.RelayInfo) (string, error) { baseURL := strings.TrimSuffix(info.ChannelMeta.BaseURL, "/") @@ -41,6 +41,8 @@ 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.RelayModeEmbeddings: return baseURL + "/v1/embeddings", nil default: @@ -56,13 +58,19 @@ func (a *Adaptor) SetupRequestHeader(header http.Header, info *common.RelayInfo) } // ConvertRequest 转换请求体。 -// Claude 入站直接透传到 Anthropic 兼容端点，其他格式先转为 OpenAI 再做 DeepSeek 特有适配。 +// Claude 入站直接透传到 Anthropic 兼容端点，Responses 模式处理 reasoning_effort 映射，其他格式先转为 OpenAI 再做 DeepSeek 特有适配。 func (a *Adaptor) ConvertRequest(ctx context.Context, info *common.RelayInfo, requestBody []byte) (io.Reader, error) { // Claude 入站：透传请求体，仅做模型映射和思考参数注入 if info.InboundFormat == constant.RelayFormatClaude { return convertClaudeRequestForDeepSeek(requestBody, info) } + // Responses 模式：处理 DeepSeek V4 的 reasoning_effort 映射 + if constant.RelayMode(info.RelayMode) == constant.RelayModeResponses || + constant.RelayMode(info.RelayMode) == constant.RelayModeResponsesCompact { + return convertResponsesRequestForDeepSeek(requestBody, info) + } + // 非 OpenAI 格式先转换为 OpenAI if info.InboundFormat != "" && info.InboundFormat != constant.RelayFormatOpenAI { converted, err := openai.ConvertToOpenAI(requestBody, info) @@ -262,3 +270,77 @@ func convertClaudeRequestForDeepSeek(requestBody []byte, info *common.RelayInfo) } return bytes.NewReader(result), nil } + +// convertResponsesRequestForDeepSeek 处理 Responses API 入站请求的 DeepSeek 兼容适配。 +// 功能：模型名映射 + 将 RelayInfo 中的思考后缀参数映射到 Responses API 的 reasoning.effort 字段。 +// 对齐 new-api commit 9724ef1b2 的实现逻辑。 +func convertResponsesRequestForDeepSeek(requestBody []byte, info *common.RelayInfo) (io.Reader, error) { + var rawMap map[string]json.RawMessage + if err := json.Unmarshal(requestBody, &rawMap); err != nil { + return bytes.NewReader(requestBody), nil + } + + // 模型名映射 + if info.ChannelMeta.IsModelMapped { + rawMap["model"] = json.RawMessage(`"` + info.ChannelMeta.UpstreamModelName + `"`) + } + + // 将思考后缀参数映射到 reasoning.effort（Responses API 格式） + rawMap = injectResponsesReasoningEffort(rawMap, info) + + result, err := json.Marshal(rawMap) + if err != nil { + return bytes.NewReader(requestBody), nil + } + return bytes.NewReader(result), nil +} + +// injectResponsesReasoningEffort 为 Responses API 请求注入 DeepSeek reasoning.effort 参数。 +// +// Responses API 用 reasoning.effort 而非 Chat API 的 reasoning_effort 顶层字段： +// - DeepSeek V4：仅支持 "max"，所有非空 effort 统一映射为 "max" +// - DeepSeek V3/R1：支持 "none"/"high"/"max"，做兼容映射 +// +// 注入优先级：客户端已显式设置 reasoning > 后缀路由注入 > 不干预（保留 DeepSeek 默认行为） +func injectResponsesReasoningEffort(rawMap map[string]json.RawMessage, info *common.RelayInfo) map[string]json.RawMessage { + // 客户端已显式设置 reasoning 字段 → 不干预 + if _, clientSet := rawMap["reasoning"]; clientSet { + return rawMap + } + + // -nothinking 后缀：显式关闭思考 + if info.ThinkingDisabled { + rawMap["reasoning"] = json.RawMessage(`{"effort":"none"}`) + return rawMap + } + + // 确定目标 effort（来自 -thinking 或 -high/-low/-medium 后缀） + effort := info.ReasoningEffort + if info.ThinkingEnabled && effort == "" { + effort = "max" + } + if effort == "" { + return rawMap // 无后缀，不干预，保留 DeepSeek 默认行为 + } + + modelName := info.ChannelMeta.UpstreamModelName + + // V4 模型仅支持 max + if isDeepSeekV4Model(modelName) { + rawMap["reasoning"] = json.RawMessage(`{"effort":"max"}`) + return rawMap + } + + // V3/R1：映射到支持的值（low/medium → high，xhigh/max → max） + switch effort { + case "low", "medium", "minimal": + effort = "high" + case "xhigh", "max": + effort = "max" + default: + effort = "high" + } + effortJSON, _ := json.Marshal(effort) + rawMap["reasoning"] = json.RawMessage(`{"effort":` + string(effortJSON) + `}`) + return rawMap +}