Feat/doubao batch

.gitignore

@@ -16,4 +16,6 @@ script/
 *_test.go
 out.log
 out.log_2
-.env_3
+.env_3
+*.prof
+.env*

common/constants.go

@@ -239,6 +239,7 @@ const (
 	ChannelTypeVolcEngine     = 45
 	ChannelTypeBaiduV2        = 46
 	ChannelTypeXai            = 47
+	ChannelTypeDoubaoOffline  = 100
 	ChannelTypeDummy          // this one is only for count, do not add any channel after this
 
 )
@@ -293,4 +294,56 @@ var ChannelBaseURLs = []string{
 	"https://qianfan.baidubce.com",              //46
 	"",                                          //47
 	"https://api.x.ai",                          //48
+	"",                                          //49
+	"",                                          //50
+	"",                                          //51
+	"",                                          //52
+	"",                                          //53
+	"",                                          //54
+	"",                                          //55
+	"",                                          //56
+	"",                                          //57
+	"",                                          //58
+	"",                                          //59
+	"",                                          //60
+	"",                                          //61
+	"",                                          //62
+	"",                                          //63
+	"",                                          //64
+	"",                                          //65
+	"",                                          //66
+	"",                                          //67
+	"",                                          //68
+	"",                                          //69
+	"",                                          //70
+	"",                                          //71
+	"",                                          //72
+	"",                                          //73
+	"",                                          //74
+	"",                                          //75
+	"",                                          //76
+	"",                                          //77
+	"",                                          //78
+	"",                                          //79
+	"",                                          //80
+	"",                                          //81
+	"",                                          //82
+	"",                                          //83
+	"",                                          //84
+	"",                                          //85
+	"",                                          //86
+	"",                                          //87
+	"",                                          //88
+	"",                                          //89
+	"",                                          //90
+	"",                                          //91
+	"",                                          //92
+	"",                                          //93
+	"",                                          //94
+	"",                                          //95
+	"",                                          //96
+	"",                                          //97
+	"",                                          //98
+	"",                                          //99
+	"https://ark.cn-beijing.volces.com",         //100 - 豆包离线
 }

common/logger.go

@@ -10,9 +10,12 @@ import (
 	"path/filepath"
 	"runtime"
 	"sort"
+	"strings"
 	"sync"
 	"time"
 
+	"one-api/metrics"
+
 	"github.com/bytedance/gopkg/util/gopool"
 	"github.com/gin-gonic/gin"
 )
@@ -35,6 +38,20 @@ const (
 	maxLogCount = 1000000
 )
 
+// 错误类型常量
+const (
+	ErrorTypeOther              = "other"
+	ErrorTypeParameter          = "parameter_error"
+	ErrorTypeNoCandidates       = "no_candidates"
+	ErrorTypeRequestFailed      = "request_failed"
+	ErrorTypeBadGateway         = "bad_gateway"
+	ErrorTypeResponseFailed     = "response_failed"
+	ErrorTypeConnectionTimeout  = "connection_timeout"
+	ErrorTypeTokenUnavailable   = "token_unavailable"
+	ErrorTypeBadRequest         = "bad_request"
+	ErrorTypeNoAvailableChannel = "no_available_channel"
+)
+
 var logCount int
 var setupLogLock sync.Mutex
 var setupLogWorking bool
@@ -184,6 +201,42 @@ func LogError(ctx context.Context, msg string) {
 	logHelper(ctx, loggerError, msg)
 }
 
+// 获取错误类型
+func getErrorType(msg string) (string, string) {
+	// 提取错误码（如果有）
+	errorCode := "unknown"
+	if strings.Contains(msg, "status code:") {
+		parts := strings.Split(msg, "status code:")
+		if len(parts) > 1 {
+			errorCode = strings.TrimSpace(parts[1])
+		}
+	}
+
+	// 根据错误消息内容判断错误类型
+	switch {
+	case strings.Contains(msg, "One or more parameter"):
+		return ErrorTypeParameter, errorCode
+	case strings.Contains(msg, "No candidates"):
+		return ErrorTypeNoCandidates, errorCode
+	case strings.Contains(msg, "do request failed"):
+		return ErrorTypeRequestFailed, errorCode
+	case strings.Contains(msg, "status code: 502"):
+		return ErrorTypeBadGateway, errorCode
+	case strings.Contains(msg, "doResponse failed"):
+		return ErrorTypeResponseFailed, errorCode
+	case strings.Contains(msg, "write: connection timed out"):
+		return ErrorTypeConnectionTimeout, errorCode
+	case strings.Contains(msg, "该令牌状态不可用"):
+		return ErrorTypeTokenUnavailable, errorCode
+	case strings.Contains(msg, "bad response status code 400"):
+		return ErrorTypeBadRequest, errorCode
+	case strings.Contains(msg, "无可用渠道"):
+		return ErrorTypeNoAvailableChannel, errorCode
+	default:
+		return ErrorTypeOther, errorCode
+	}
+}
+
 func logHelper(ctx context.Context, level string, msg string) {
 	// 获取请求ID
 	var requestId string
@@ -209,6 +262,38 @@ func logHelper(ctx context.Context, level string, msg string) {
 	now := time.Now()
 	caller := getCallerInfo()
 	_, _ = fmt.Fprintf(writer, "[%s] %v | %s | %s | %s \n", level, now.Format("2006/01/02 - 15:04:05"), requestId, caller, msg)
+
+	// 如果是错误日志，增加错误计数
+	if level == loggerError {
+		errorType, errorCode := getErrorType(msg)
+		// 从上下文中获取相关信息
+		channel := "unknown"
+		channelName := "unknown"
+		model := "unknown"
+		group := "unknown"
+		tokenName := "unknown"
+
+		if ginCtx, ok := ctx.Value("gin_context").(*gin.Context); ok {
+			if ch := ginCtx.GetString("channel"); ch != "" {
+				channel = ch
+			}
+			if chName := ginCtx.GetString("channel_name"); chName != "" {
+				channelName = chName
+			}
+			if m := ginCtx.GetString("model"); m != "" {
+				model = m
+			}
+			if g := ginCtx.GetString("group"); g != "" {
+				group = g
+			}
+			if tn := ginCtx.GetString("token_name"); tn != "" {
+				tokenName = tn
+			}
+		}
+
+		metrics.IncrementErrorLog(channel, channelName, errorCode, errorType, model, group, tokenName, 1.0)
+	}
+
 	logCount++ // we don't need accurate count, so no lock here
 	if logCount > maxLogCount && !setupLogWorking {
 		logCount = 0

controller/misc.go

@@ -270,3 +270,13 @@ func ResetPassword(c *gin.Context) {
 	})
 	return
 }
+
+func Ping(c *gin.Context) {
+	c.Writer.Header().Set("Retry_request_id", "Retry_request_id")
+	c.JSON(http.StatusOK, gin.H{
+		"success":   true,
+		"message":   "pong",
+		"timestamp": common.GetTimestamp(),
+	})
+	return
+}

controller/relay.go

@@ -4,8 +4,6 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
-	"github.com/gin-gonic/gin"
-	"github.com/gorilla/websocket"
 	"io"
 	"log"
 	"net/http"
@@ -23,6 +21,9 @@ import (
 	"strconv"
 	"strings"
 	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/gorilla/websocket"
 )
 
 func relayInfoHandler(c *gin.Context, relayMode int) (*relaycommon.RelayInfo, interface{}, string, *dto.OpenAIErrorWithStatusCode) {
@@ -121,6 +122,9 @@ func Relay(c *gin.Context) {
 			openaiErr = service.OpenAIErrorWrapperLocal(err, "get_channel_failed", http.StatusInternalServerError)
 			break
 		}
+		// 设置 channel 信息到上下文
+		c.Set("channel", strconv.Itoa(channel.Id))
+		c.Set("channel_name", channel.Name)
 		fillRelayRequest(c, channel)
 		var (
 			relayInfo    *relaycommon.RelayInfo
@@ -130,20 +134,20 @@ func Relay(c *gin.Context) {
 		relayInfo, request, requestModel, openaiErr = relayInfoHandler(c, relayMode)
 		if i == 0 {
 			// e2e 用户请求计数
-			metrics.IncrementRelayRequestE2ETotalCounter(strconv.Itoa(channel.Id), requestModel, group, tokenKey, tokenName, 1)
+			metrics.IncrementRelayRequestE2ETotalCounter(strconv.Itoa(channel.Id), channel.Name, requestModel, group, tokenKey, tokenName, 1)
 		} else {
 			// 重试计数
 			channelTag := ""
 			if channel.Tag != nil {
 				channelTag = *channel.Tag
 			}
-			metrics.IncrementRelayRetryCounter(strconv.Itoa(channel.Id), channelTag, channel.GetBaseURL(), requestModel, group, 1)
+			metrics.IncrementRelayRetryCounter(strconv.Itoa(channel.Id), channel.Name, channelTag, channel.GetBaseURL(), requestModel, group, 1)
 		}
 		if openaiErr == nil {
 			openaiErr = executeRelayRequest(c, relayMode, relayInfo, request)
 			if openaiErr == nil {
-				metrics.IncrementRelayRequestE2ESuccessCounter(strconv.Itoa(channel.Id), requestModel, group, tokenKey, tokenName, 1)
-				metrics.ObserveRelayRequestE2EDuration(strconv.Itoa(channel.Id), requestModel, group, tokenKey, tokenName, time.Since(startTime).Seconds())
+				metrics.IncrementRelayRequestE2ESuccessCounter(strconv.Itoa(channel.Id), channel.Name, requestModel, group, tokenKey, tokenName, 1)
+				metrics.ObserveRelayRequestE2EDuration(strconv.Itoa(channel.Id), channel.Name, requestModel, group, tokenKey, tokenName, time.Since(startTime).Seconds())
 				return
 			}
 		}
@@ -152,7 +156,7 @@ func Relay(c *gin.Context) {
 
 		if !shouldRetry(c, openaiErr, common.RetryTimes-i) {
 			// e2e 失败计数
-			metrics.IncrementRelayRequestE2EFailedCounter(strconv.Itoa(channel.Id), requestModel, group, strconv.Itoa(openaiErr.StatusCode), tokenKey, tokenName, 1)
+			metrics.IncrementRelayRequestE2EFailedCounter(strconv.Itoa(channel.Id), channel.Name, requestModel, group, strconv.Itoa(openaiErr.StatusCode), tokenKey, tokenName, 1)
 			break
 		}
 	}
@@ -167,6 +171,33 @@ func Relay(c *gin.Context) {
 			common.LogError(c, fmt.Sprintf("origin 429 error: %s", openaiErr.Error.Message))
 			openaiErr.Error.Message = "当前分组上游负载已饱和，请稍后再试"
 		}
+
+		// 处理自定义的 NewAPI batch 错误码
+		if openaiErr.StatusCode == dto.StatusNewAPIBatchRateLimitExceeded {
+			common.LogError(c, fmt.Sprintf("origin %d error: %s", openaiErr.StatusCode, openaiErr.Error.Message))
+			openaiErr.Error.Message = "当前服务端限速已满，请稍后再试"
+		}
+		if openaiErr.StatusCode == dto.StatusNewAPIBatchTimeout {
+			common.LogError(c, fmt.Sprintf("origin %d error: %s", openaiErr.StatusCode, openaiErr.Error.Message))
+			openaiErr.Error.Message = "未等待到结果，请稍后使用Retry_request_id再次查询"
+		}
+		if openaiErr.StatusCode == dto.StatusNewAPIBatchInternal {
+			common.LogError(c, fmt.Sprintf("origin %d error: %s", openaiErr.StatusCode, openaiErr.Error.Message))
+			openaiErr.Error.Message = "服务内部错误，请稍后再试"
+		}
+		if openaiErr.StatusCode == dto.StatusNewAPIBatchSubmitted {
+			common.LogError(c, fmt.Sprintf("origin %d error: %s", openaiErr.StatusCode, openaiErr.Error.Message))
+			openaiErr.Error.Message = "批量请求已提交，但是结果还未出来，请使用Retry_request_id查询结果"
+		}
+		if openaiErr.StatusCode == dto.StatusNewAPIBatchAccepted {
+			common.LogError(c, fmt.Sprintf("origin %d error: %s", openaiErr.StatusCode, openaiErr.Error.Message))
+			openaiErr.Error.Message = "批量请求已接受，正在处理中，请稍后使用Retry_request_id查询结果"
+		}
+		if openaiErr.StatusCode == dto.StatusRequestConflict {
+			common.LogError(c, fmt.Sprintf("origin %d error: %s", openaiErr.StatusCode, openaiErr.Error.Message))
+			openaiErr.Error.Message = "请求冲突，有其他请求使用了这个Retry_request_id，请稍后再试"
+		}
+
 		openaiErr.Error.Message = common.MessageWithRequestId(openaiErr.Error.Message, requestId)
 		c.JSON(openaiErr.StatusCode, gin.H{
 			"error": openaiErr.Error,
@@ -230,6 +261,31 @@ func WssRelay(c *gin.Context) {
 		if openaiErr.StatusCode == http.StatusTooManyRequests {
 			openaiErr.Error.Message = "当前分组上游负载已饱和，请稍后再试"
 		}
+		// 处理自定义的 NewAPI batch 错误码
+		if openaiErr.StatusCode == dto.StatusNewAPIBatchRateLimitExceeded {
+			common.LogError(c, fmt.Sprintf("origin %d error: %s", openaiErr.StatusCode, openaiErr.Error.Message))
+			openaiErr.Error.Message = "当前服务端限速已满，请稍后再试"
+		}
+		if openaiErr.StatusCode == dto.StatusNewAPIBatchTimeout {
+			common.LogError(c, fmt.Sprintf("origin %d error: %s", openaiErr.StatusCode, openaiErr.Error.Message))
+			openaiErr.Error.Message = "未等待到结果，请稍后使用Retry_request_id再次查询"
+		}
+		if openaiErr.StatusCode == dto.StatusNewAPIBatchInternal {
+			common.LogError(c, fmt.Sprintf("origin %d error: %s", openaiErr.StatusCode, openaiErr.Error.Message))
+			openaiErr.Error.Message = "服务内部错误，请稍后再试"
+		}
+		if openaiErr.StatusCode == dto.StatusNewAPIBatchSubmitted {
+			common.LogError(c, fmt.Sprintf("origin %d error: %s", openaiErr.StatusCode, openaiErr.Error.Message))
+			openaiErr.Error.Message = "批量请求已提交，但是结果还未出来，请使用Retry_request_id查询结果"
+		}
+		if openaiErr.StatusCode == dto.StatusNewAPIBatchAccepted {
+			common.LogError(c, fmt.Sprintf("origin %d error: %s", openaiErr.StatusCode, openaiErr.Error.Message))
+			openaiErr.Error.Message = "批量请求已接受，正在处理中，请稍后使用Retry_request_id查询结果"
+		}
+		if openaiErr.StatusCode == dto.StatusRequestConflict {
+			common.LogError(c, fmt.Sprintf("origin %d error: %s", openaiErr.StatusCode, openaiErr.Error.Message))
+			openaiErr.Error.Message = "请求冲突，有其他请求使用了这个Retry_request_id，请稍后再试"
+		}
 		openaiErr.Error.Message = common.MessageWithRequestId(openaiErr.Error.Message, requestId)
 		helper.WssError(c, ws, openaiErr.Error)
 	}
@@ -298,6 +354,26 @@ func shouldRetry(c *gin.Context, openaiErr *dto.OpenAIErrorWithStatusCode, retry
 	if openaiErr.StatusCode == http.StatusTooManyRequests {
 		return true
 	}
+	// 处理自定义的 NewAPI batch 错误码
+	if openaiErr.StatusCode == dto.StatusNewAPIBatchRateLimitExceeded {
+		return false
+	}
+	if openaiErr.StatusCode == dto.StatusNewAPIBatchTimeout {
+		return false
+	}
+	if openaiErr.StatusCode == dto.StatusNewAPIBatchInternal {
+		return false
+	}
+	if openaiErr.StatusCode == dto.StatusNewAPIBatchSubmitted {
+		return false
+	}
+	if openaiErr.StatusCode == dto.StatusNewAPIBatchAccepted {
+		return false
+	}
+	if openaiErr.StatusCode == dto.StatusRequestConflict {
+		return false
+	}
+
 	if openaiErr.StatusCode == 307 {
 		return true
 	}

dto/error.go

@@ -53,3 +53,13 @@ func (e GeneralErrorResponse) ToMessage() string {
 	}
 	return ""
 }
+
+// 自定义HTTP状态码 (使用非标准状态码范围)
+const (
+	StatusNewAPIBatchRateLimitExceeded = 499 // 自定义限流状态码
+	StatusNewAPIBatchTimeout           = 598 // 自定义超时状态码
+	StatusNewAPIBatchInternal          = 599 // 自定义内部错误状态码
+	StatusNewAPIBatchSubmitted         = 203 // 批量请求已提交，需要重试获取结果
+	StatusNewAPIBatchAccepted          = 202 // 批量请求已接受，正在处理中
+	StatusRequestConflict              = 409 // 请求冲突，如分布式锁获取失败
+)