feat: add Claude upstream stream aggregation

controller/model_list_test.go

@@ -220,10 +220,12 @@ func TestListModelsTokenLimitIncludesTieredBillingModel(t *testing.T) {
 		"zz-token-tiered-visible-model":    `tier("base", p * 1 + c * 2)`,
 		"zz-token-tiered-empty-expr-model": "",
 	})
+	setupModelListControllerTestDB(t)
 
 	recorder := httptest.NewRecorder()
 	ctx, _ := gin.CreateTestContext(recorder)
 	ctx.Request = httptest.NewRequest(http.MethodGet, "/v1/models", nil)
+	common.SetContextKey(ctx, constant.ContextKeyUserGroup, "default")
 	common.SetContextKey(ctx, constant.ContextKeyTokenModelLimitEnabled, true)
 	common.SetContextKey(ctx, constant.ContextKeyTokenModelLimit, map[string]bool{
 		"zz-token-tiered-visible-model":      true,

docs/channel/other_setting.md

@@ -1,6 +1,6 @@
-# 渠道而外设置说明
+# 渠道额外设置说明
 
-该配置用于设置一些额外的渠道参数，可以通过 JSON 对象进行配置。主要包含以下两个设置项：
+该配置用于设置一些额外的渠道参数，可以通过 JSON 对象进行配置。主要包含以下设置项：
 
 1. force_format
     - 用于标识是否对数据进行强制格式化为 OpenAI 格式
@@ -14,20 +14,33 @@
    - 用于标识是否将思考内容`reasoning_content`转换为`<think>`标签拼接到内容中返回
    - 类型为布尔值，设置为 true 时启用思考内容转换
 
+4. pass_through_body_enabled
+   - 用于标识是否启用请求体透传
+   - 类型为布尔值，设置为 true 时将尽量保持客户端请求体原样转发给上游
+   - 启用后，部分需要改写上游请求体的渠道功能会失效
+
+5. force_upstream_stream
+   - 用于标识是否在下游非流式请求时，强制 new-api 与上游之间使用流式通信，再由 new-api 聚合为非流式响应返回给下游
+   - 类型为布尔值，目前仅适用于支持该能力的 Claude 渠道
+   - 适用于上游域名经过 CDN 或负载均衡时，避免非流式长请求因长时间无响应数据被中间层断开
+   - 如果启用了 `pass_through_body_enabled`，该设置不会生效；前端会自动关闭并禁用该开关
+
 --------------------------------------------------------------
 
 ## JSON 格式示例
 
-以下是一个示例配置，启用强制格式化并设置了代理地址：
+以下是一个示例配置，启用强制格式化、思考内容转换、Claude 上游流式通信，并设置了代理地址：
 
 ```json
 {
-    "force_format": true,
-   "thinking_to_content": true,
-    "proxy": "socks5://xxxxxxx"
+  "force_format": true,
+  "thinking_to_content": true,
+  "pass_through_body_enabled": false,
+  "force_upstream_stream": true,
+  "proxy": "socks5://xxxxxxx"
 }
 ```
 
 --------------------------------------------------------------
 
-通过调整上述 JSON 配置中的值，可以灵活控制渠道的额外行为，比如是否进行格式化以及使用特定的网络代理。
+通过调整上述 JSON 配置中的值，可以灵活控制渠道的额外行为，比如是否进行格式化、是否使用特定的网络代理，以及是否在支持的渠道上使用上游流式通信。

dto/channel_settings.go

@@ -5,6 +5,7 @@ type ChannelSettings struct {
 	ThinkingToContent      bool   `json:"thinking_to_content,omitempty"`
 	Proxy                  string `json:"proxy"`
 	PassThroughBodyEnabled bool   `json:"pass_through_body_enabled,omitempty"`
+	ForceUpstreamStream    bool   `json:"force_upstream_stream,omitempty"`
 	SystemPrompt           string `json:"system_prompt,omitempty"`
 	SystemPromptOverride   bool   `json:"system_prompt_override,omitempty"`
 }

dto/claude.go

@@ -22,6 +22,7 @@ type ClaudeMediaMessage struct {
 	Source       *ClaudeMessageSource `json:"source,omitempty"`
 	Usage        *ClaudeUsage         `json:"usage,omitempty"`
 	StopReason   *string              `json:"stop_reason,omitempty"`
+	StopSequence *string              `json:"stop_sequence,omitempty"`
 	PartialJson  *string              `json:"partial_json,omitempty"`
 	Role         string               `json:"role,omitempty"`
 	Thinking     *string              `json:"thinking,omitempty"`
@@ -496,6 +497,7 @@ type ClaudeResponse struct {
 	Content      []ClaudeMediaMessage `json:"content,omitempty"`
 	Completion   string               `json:"completion,omitempty"`
 	StopReason   string               `json:"stop_reason,omitempty"`
+	StopSequence *string              `json:"stop_sequence,omitempty"`
 	Model        string               `json:"model,omitempty"`
 	Error        any                  `json:"error,omitempty"`
 	Usage        *ClaudeUsage         `json:"usage,omitempty"`
@@ -596,5 +598,6 @@ func (u *ClaudeUsage) GetCacheCreationTotalTokens() int {
 }
 
 type ClaudeServerToolUse struct {
+	// TODO: Preserve future Anthropic server_tool_use counters when new fields are introduced.
 	WebSearchRequests int `json:"web_search_requests"`
 }

relay/channel/claude/adaptor.go

@@ -7,9 +7,11 @@ import (
 	"net/http"
 	"net/url"
 
+	"github.com/QuantumNous/new-api/common"
 	"github.com/QuantumNous/new-api/dto"
 	"github.com/QuantumNous/new-api/relay/channel"
 	relaycommon "github.com/QuantumNous/new-api/relay/common"
+	"github.com/QuantumNous/new-api/service"
 	"github.com/QuantumNous/new-api/setting/model_setting"
 	"github.com/QuantumNous/new-api/types"
 
@@ -25,6 +27,7 @@ func (a *Adaptor) ConvertGeminiRequest(*gin.Context, *relaycommon.RelayInfo, *dt
 }
 
 func (a *Adaptor) ConvertClaudeRequest(c *gin.Context, info *relaycommon.RelayInfo, request *dto.ClaudeRequest) (any, error) {
+	enableClaudeUpstreamStreamIfNeeded(info, request)
 	return request, nil
 }
 
@@ -95,7 +98,12 @@ func (a *Adaptor) ConvertOpenAIRequest(c *gin.Context, info *relaycommon.RelayIn
 	if request == nil {
 		return nil, errors.New("request is nil")
 	}
-	return RequestOpenAI2ClaudeMessage(c, *request)
+	claudeReq, err := RequestOpenAI2ClaudeMessage(c, *request)
+	if err != nil {
+		return nil, err
+	}
+	enableClaudeUpstreamStreamIfNeeded(info, claudeReq)
+	return claudeReq, nil
 }
 
 func (a *Adaptor) ConvertRerankRequest(c *gin.Context, relayMode int, request dto.RerankRequest) (any, error) {
@@ -118,6 +126,18 @@ func (a *Adaptor) DoRequest(c *gin.Context, info *relaycommon.RelayInfo, request
 
 func (a *Adaptor) DoResponse(c *gin.Context, resp *http.Response, info *relaycommon.RelayInfo) (usage any, err *types.NewAPIError) {
 	info.FinalRequestRelayFormat = types.RelayFormatClaude
+	if info.UpstreamStream && !info.IsStream {
+		result, usage, err := AggregateClaudeStreamResponse(c, resp, info)
+		if err != nil {
+			return usage, err
+		}
+		if resultBody, marshalErr := common.Marshal(result); marshalErr != nil {
+			return usage, types.NewError(marshalErr, types.ErrorCodeBadResponseBody)
+		} else {
+			service.IOCopyBytesGracefully(c, nonStreamJSONResponse(resp), resultBody)
+		}
+		return usage, nil
+	}
 	if info.IsStream {
 		return ClaudeStreamHandler(c, resp, info)
 	} else {