You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

657 line
21 KiB

  1. package controller
  2. import (
  3. "errors"
  4. "fmt"
  5. "io"
  6. "log"
  7. "net/http"
  8. "strconv"
  9. "strings"
  10. "time"
  11. "github.com/QuantumNous/new-api/common"
  12. "github.com/QuantumNous/new-api/common/metrics"
  13. "github.com/QuantumNous/new-api/constant"
  14. "github.com/QuantumNous/new-api/dto"
  15. "github.com/QuantumNous/new-api/logger"
  16. "github.com/QuantumNous/new-api/middleware"
  17. "github.com/QuantumNous/new-api/model"
  18. "github.com/QuantumNous/new-api/relay"
  19. relaycommon "github.com/QuantumNous/new-api/relay/common"
  20. relayconstant "github.com/QuantumNous/new-api/relay/constant"
  21. "github.com/QuantumNous/new-api/relay/helper"
  22. "github.com/QuantumNous/new-api/service"
  23. "github.com/QuantumNous/new-api/setting"
  24. "github.com/QuantumNous/new-api/setting/operation_setting"
  25. "github.com/QuantumNous/new-api/types"
  26. "github.com/bytedance/gopkg/util/gopool"
  27. "github.com/gin-gonic/gin"
  28. "github.com/gorilla/websocket"
  29. )
  30. func relayHandler(c *gin.Context, info *relaycommon.RelayInfo) *types.NewAPIError {
  31. var err *types.NewAPIError
  32. switch info.RelayMode {
  33. case relayconstant.RelayModeImagesGenerations, relayconstant.RelayModeImagesEdits:
  34. err = relay.ImageHelper(c, info)
  35. case relayconstant.RelayModeAudioSpeech:
  36. fallthrough
  37. case relayconstant.RelayModeAudioTranslation:
  38. fallthrough
  39. case relayconstant.RelayModeAudioTranscription:
  40. err = relay.AudioHelper(c, info)
  41. case relayconstant.RelayModeRerank:
  42. err = relay.RerankHelper(c, info)
  43. case relayconstant.RelayModeEmbeddings:
  44. err = relay.EmbeddingHelper(c, info)
  45. case relayconstant.RelayModeResponses, relayconstant.RelayModeResponsesCompact:
  46. err = relay.ResponsesHelper(c, info)
  47. default:
  48. err = relay.TextHelper(c, info)
  49. }
  50. return err
  51. }
  52. func geminiRelayHandler(c *gin.Context, info *relaycommon.RelayInfo) *types.NewAPIError {
  53. var err *types.NewAPIError
  54. if strings.Contains(c.Request.URL.Path, "embed") {
  55. err = relay.GeminiEmbeddingHandler(c, info)
  56. } else {
  57. err = relay.GeminiHelper(c, info)
  58. }
  59. return err
  60. }
  61. func Relay(c *gin.Context, relayFormat types.RelayFormat) {
  62. requestId := c.GetString(common.RequestIdKey)
  63. //group := common.GetContextKeyString(c, constant.ContextKeyUsingGroup)
  64. //originalModel := common.GetContextKeyString(c, constant.ContextKeyOriginalModel)
  65. var (
  66. newAPIError *types.NewAPIError
  67. ws *websocket.Conn
  68. )
  69. if relayFormat == types.RelayFormatOpenAIRealtime {
  70. var err error
  71. ws, err = upgrader.Upgrade(c.Writer, c.Request, nil)
  72. if err != nil {
  73. helper.WssError(c, ws, types.NewError(err, types.ErrorCodeGetChannelFailed, types.ErrOptionWithSkipRetry()).ToOpenAIError())
  74. return
  75. }
  76. defer ws.Close()
  77. }
  78. defer func() {
  79. if newAPIError != nil {
  80. logger.LogError(c, fmt.Sprintf("relay error: %s", newAPIError.Error()))
  81. // Prometheus: 错误计数
  82. if metrics.IsEnabled() {
  83. channelId := strconv.Itoa(c.GetInt("channel_id"))
  84. modelName := c.GetString("original_model")
  85. errorType, errorCode := metrics.ClassifyError(newAPIError.StatusCode, newAPIError.Error())
  86. metrics.GetMetrics().RequestErrorsTotal.WithLabelValues(channelId, modelName, errorType, errorCode).Inc()
  87. }
  88. newAPIError.SetMessage(common.MessageWithRequestId(newAPIError.Error(), requestId))
  89. switch relayFormat {
  90. case types.RelayFormatOpenAIRealtime:
  91. helper.WssError(c, ws, newAPIError.ToOpenAIError())
  92. case types.RelayFormatClaude:
  93. c.JSON(newAPIError.StatusCode, gin.H{
  94. "type": "error",
  95. "error": newAPIError.ToClaudeError(),
  96. })
  97. default:
  98. c.JSON(newAPIError.StatusCode, gin.H{
  99. "error": newAPIError.ToOpenAIError(),
  100. })
  101. }
  102. }
  103. }()
  104. request, err := helper.GetAndValidateRequest(c, relayFormat)
  105. if err != nil {
  106. // Map "request body too large" to 413 so clients can handle it correctly
  107. if common.IsRequestBodyTooLargeError(err) || errors.Is(err, common.ErrRequestBodyTooLarge) {
  108. newAPIError = types.NewErrorWithStatusCode(err, types.ErrorCodeReadRequestBodyFailed, http.StatusRequestEntityTooLarge, types.ErrOptionWithSkipRetry())
  109. } else {
  110. newAPIError = types.NewError(err, types.ErrorCodeInvalidRequest)
  111. }
  112. return
  113. }
  114. relayInfo, err := relaycommon.GenRelayInfo(c, relayFormat, request, ws)
  115. if err != nil {
  116. newAPIError = types.NewError(err, types.ErrorCodeGenRelayInfoFailed)
  117. return
  118. }
  119. needSensitiveCheck := setting.ShouldCheckPromptSensitive()
  120. needCountToken := constant.CountToken
  121. // Avoid building huge CombineText (strings.Join) when token counting and sensitive check are both disabled.
  122. var meta *types.TokenCountMeta
  123. if needSensitiveCheck || needCountToken {
  124. meta = request.GetTokenCountMeta()
  125. } else {
  126. meta = fastTokenCountMetaForPricing(request)
  127. }
  128. if needSensitiveCheck && meta != nil {
  129. contains, words := service.CheckSensitiveText(meta.CombineText)
  130. if contains {
  131. logger.LogWarn(c, fmt.Sprintf("user sensitive words detected: %s", strings.Join(words, ", ")))
  132. newAPIError = types.NewError(err, types.ErrorCodeSensitiveWordsDetected)
  133. return
  134. }
  135. }
  136. tokens, err := service.EstimateRequestToken(c, meta, relayInfo)
  137. if err != nil {
  138. newAPIError = types.NewError(err, types.ErrorCodeCountTokenFailed)
  139. return
  140. }
  141. relayInfo.SetEstimatePromptTokens(tokens)
  142. priceData, err := helper.ModelPriceHelper(c, relayInfo, tokens, meta)
  143. if err != nil {
  144. newAPIError = types.NewError(err, types.ErrorCodeModelPriceError)
  145. return
  146. }
  147. // common.SetContextKey(c, constant.ContextKeyTokenCountMeta, meta)
  148. if priceData.FreeModel {
  149. logger.LogInfo(c, fmt.Sprintf("模型 %s 免费,跳过预扣费", relayInfo.OriginModelName))
  150. } else {
  151. common.SysLog(fmt.Sprintf("[RegionSync] relay PreConsumeBilling: userId=%d, QuotaToPreConsume=%d, FreeModel=%v, UsePrice=%v, ModelRatio=%.4f, GroupRatio=%.4f, ModelPrice=%.4f, model=%s",
  152. relayInfo.UserId, priceData.QuotaToPreConsume, priceData.FreeModel, priceData.UsePrice, priceData.ModelRatio, priceData.GroupRatioInfo.GroupRatio, priceData.ModelPrice, relayInfo.OriginModelName))
  153. newAPIError = service.PreConsumeBilling(c, priceData.QuotaToPreConsume, relayInfo)
  154. if newAPIError != nil {
  155. return
  156. }
  157. }
  158. defer func() {
  159. // Only return quota if downstream failed and quota was actually pre-consumed
  160. if newAPIError != nil {
  161. newAPIError = service.NormalizeViolationFeeError(newAPIError)
  162. if relayInfo.Billing != nil {
  163. relayInfo.Billing.Refund(c)
  164. }
  165. service.ChargeViolationFeeIfNeeded(c, relayInfo, newAPIError)
  166. }
  167. }()
  168. retryParam := &service.RetryParam{
  169. Ctx: c,
  170. TokenGroup: relayInfo.TokenGroup,
  171. ModelName: relayInfo.OriginModelName,
  172. Retry: common.GetPointer(0),
  173. }
  174. for ; retryParam.GetRetry() <= common.RetryTimes; retryParam.IncreaseRetry() {
  175. channel, channelErr := getChannel(c, relayInfo, retryParam)
  176. if channelErr != nil {
  177. logger.LogError(c, channelErr.Error())
  178. newAPIError = channelErr
  179. break
  180. }
  181. addUsedChannel(c, channel.Id)
  182. bodyStorage, bodyErr := common.GetBodyStorage(c)
  183. if bodyErr != nil {
  184. // Ensure consistent 413 for oversized bodies even when error occurs later (e.g., retry path)
  185. if common.IsRequestBodyTooLargeError(bodyErr) || errors.Is(bodyErr, common.ErrRequestBodyTooLarge) {
  186. newAPIError = types.NewErrorWithStatusCode(bodyErr, types.ErrorCodeReadRequestBodyFailed, http.StatusRequestEntityTooLarge, types.ErrOptionWithSkipRetry())
  187. } else {
  188. newAPIError = types.NewErrorWithStatusCode(bodyErr, types.ErrorCodeReadRequestBodyFailed, http.StatusBadRequest, types.ErrOptionWithSkipRetry())
  189. }
  190. break
  191. }
  192. c.Request.Body = io.NopCloser(bodyStorage)
  193. switch relayFormat {
  194. case types.RelayFormatOpenAIRealtime:
  195. newAPIError = relay.WssHelper(c, relayInfo)
  196. case types.RelayFormatClaude:
  197. newAPIError = relay.ClaudeHelper(c, relayInfo)
  198. case types.RelayFormatGemini:
  199. newAPIError = geminiRelayHandler(c, relayInfo)
  200. default:
  201. newAPIError = relayHandler(c, relayInfo)
  202. }
  203. if newAPIError == nil {
  204. return
  205. }
  206. newAPIError = service.NormalizeViolationFeeError(newAPIError)
  207. processChannelError(c, *types.NewChannelError(channel.Id, channel.Type, channel.Name, channel.ChannelInfo.IsMultiKey, common.GetContextKeyString(c, constant.ContextKeyChannelKey), channel.GetAutoBan()), newAPIError)
  208. if !shouldRetry(c, newAPIError, common.RetryTimes-retryParam.GetRetry()) {
  209. break
  210. }
  211. // Prometheus: 重试计数
  212. if metrics.IsEnabled() {
  213. channelIdStr := strconv.Itoa(channel.Id)
  214. metrics.GetMetrics().RequestRetriesTotal.WithLabelValues(channelIdStr, relayInfo.OriginModelName).Inc()
  215. }
  216. }
  217. useChannel := c.GetStringSlice("use_channel")
  218. if len(useChannel) > 1 {
  219. retryLogStr := fmt.Sprintf("重试:%s", strings.Trim(strings.Join(strings.Fields(fmt.Sprint(useChannel)), "->"), "[]"))
  220. logger.LogInfo(c, retryLogStr)
  221. }
  222. }
  223. var upgrader = websocket.Upgrader{
  224. Subprotocols: []string{"realtime"}, // WS 握手支持的协议,如果有使用 Sec-WebSocket-Protocol,则必须在此声明对应的 Protocol TODO add other protocol
  225. CheckOrigin: func(r *http.Request) bool {
  226. return true // 允许跨域
  227. },
  228. }
  229. func addUsedChannel(c *gin.Context, channelId int) {
  230. useChannel := c.GetStringSlice("use_channel")
  231. useChannel = append(useChannel, fmt.Sprintf("%d", channelId))
  232. c.Set("use_channel", useChannel)
  233. }
  234. func fastTokenCountMetaForPricing(request dto.Request) *types.TokenCountMeta {
  235. if request == nil {
  236. return &types.TokenCountMeta{}
  237. }
  238. meta := &types.TokenCountMeta{
  239. TokenType: types.TokenTypeTokenizer,
  240. }
  241. switch r := request.(type) {
  242. case *dto.GeneralOpenAIRequest:
  243. if r.MaxCompletionTokens > r.MaxTokens {
  244. meta.MaxTokens = int(r.MaxCompletionTokens)
  245. } else {
  246. meta.MaxTokens = int(r.MaxTokens)
  247. }
  248. case *dto.OpenAIResponsesRequest:
  249. meta.MaxTokens = int(r.MaxOutputTokens)
  250. case *dto.ClaudeRequest:
  251. meta.MaxTokens = int(r.MaxTokens)
  252. case *dto.ImageRequest:
  253. // Pricing for image requests depends on ImagePriceRatio; safe to compute even when CountToken is disabled.
  254. return r.GetTokenCountMeta()
  255. default:
  256. // Best-effort: leave CombineText empty to avoid large allocations.
  257. }
  258. return meta
  259. }
  260. func getChannel(c *gin.Context, info *relaycommon.RelayInfo, retryParam *service.RetryParam) (*model.Channel, *types.NewAPIError) {
  261. if info.ChannelMeta == nil {
  262. autoBan := c.GetBool("auto_ban")
  263. autoBanInt := 1
  264. if !autoBan {
  265. autoBanInt = 0
  266. }
  267. return &model.Channel{
  268. Id: c.GetInt("channel_id"),
  269. Type: c.GetInt("channel_type"),
  270. Name: c.GetString("channel_name"),
  271. AutoBan: &autoBanInt,
  272. }, nil
  273. }
  274. channel, selectGroup, err := service.CacheGetRandomSatisfiedChannel(retryParam)
  275. info.PriceData.GroupRatioInfo = helper.HandleGroupRatio(c, info)
  276. if err != nil {
  277. return nil, types.NewError(fmt.Errorf("获取分组 %s 下模型 %s 的可用渠道失败(retry): %s", selectGroup, info.OriginModelName, err.Error()), types.ErrorCodeGetChannelFailed, types.ErrOptionWithSkipRetry())
  278. }
  279. if channel == nil {
  280. return nil, types.NewError(fmt.Errorf("分组 %s 下模型 %s 的可用渠道不存在(retry)", selectGroup, info.OriginModelName), types.ErrorCodeGetChannelFailed, types.ErrOptionWithSkipRetry())
  281. }
  282. newAPIError := middleware.SetupContextForSelectedChannel(c, channel, info.OriginModelName)
  283. if newAPIError != nil {
  284. return nil, newAPIError
  285. }
  286. return channel, nil
  287. }
  288. func shouldRetry(c *gin.Context, openaiErr *types.NewAPIError, retryTimes int) bool {
  289. if openaiErr == nil {
  290. return false
  291. }
  292. if service.ShouldSkipRetryAfterChannelAffinityFailure(c) {
  293. return false
  294. }
  295. if types.IsChannelError(openaiErr) {
  296. return true
  297. }
  298. if types.IsSkipRetryError(openaiErr) {
  299. return false
  300. }
  301. if retryTimes <= 0 {
  302. return false
  303. }
  304. code := openaiErr.StatusCode
  305. if code >= 200 && code < 300 {
  306. return false
  307. }
  308. if code < 100 || code > 599 {
  309. return true
  310. }
  311. return operation_setting.ShouldRetryByStatusCode(code)
  312. }
  313. func processChannelError(c *gin.Context, channelError types.ChannelError, err *types.NewAPIError) {
  314. logger.LogError(c, fmt.Sprintf("channel error (channel #%d, status code: %d): %s", channelError.ChannelId, err.StatusCode, err.Error()))
  315. // 不要使用context获取渠道信息,异步处理时可能会出现渠道信息不一致的情况
  316. // do not use context to get channel info, there may be inconsistent channel info when processing asynchronously
  317. if service.ShouldDisableChannel(channelError.ChannelType, err) && channelError.AutoBan {
  318. gopool.Go(func() {
  319. service.DisableChannel(channelError, err.ErrorWithStatusCode())
  320. })
  321. }
  322. if constant.ErrorLogEnabled && types.IsRecordErrorLog(err) {
  323. // 保存错误日志到mysql中
  324. userId := c.GetInt("id")
  325. tokenName := c.GetString("token_name")
  326. modelName := c.GetString("original_model")
  327. tokenId := c.GetInt("token_id")
  328. userGroup := c.GetString("group")
  329. channelId := c.GetInt("channel_id")
  330. other := make(map[string]interface{})
  331. if c.Request != nil && c.Request.URL != nil {
  332. other["request_path"] = c.Request.URL.Path
  333. }
  334. other["error_type"] = err.GetErrorType()
  335. other["error_code"] = err.GetErrorCode()
  336. other["status_code"] = err.StatusCode
  337. other["channel_id"] = channelId
  338. other["channel_name"] = c.GetString("channel_name")
  339. other["channel_type"] = c.GetInt("channel_type")
  340. if err.UpstreamRequestId != "" {
  341. other["upstream_request_id"] = err.UpstreamRequestId
  342. }
  343. if err.UpstreamBody != "" {
  344. other["upstream_body"] = err.UpstreamBody
  345. }
  346. adminInfo := make(map[string]interface{})
  347. adminInfo["use_channel"] = c.GetStringSlice("use_channel")
  348. isMultiKey := common.GetContextKeyBool(c, constant.ContextKeyChannelIsMultiKey)
  349. if isMultiKey {
  350. adminInfo["is_multi_key"] = true
  351. adminInfo["multi_key_index"] = common.GetContextKeyInt(c, constant.ContextKeyChannelMultiKeyIndex)
  352. }
  353. service.AppendChannelAffinityAdminInfo(c, adminInfo)
  354. other["admin_info"] = adminInfo
  355. startTime := common.GetContextKeyTime(c, constant.ContextKeyRequestStartTime)
  356. if startTime.IsZero() {
  357. startTime = time.Now()
  358. }
  359. useTimeSeconds := int(time.Since(startTime).Seconds())
  360. model.RecordErrorLog(c, userId, channelId, modelName, tokenName, err.MaskSensitiveErrorWithStatusCode(), tokenId, useTimeSeconds, false, userGroup, other)
  361. }
  362. }
  363. func RelayMidjourney(c *gin.Context) {
  364. relayInfo, err := relaycommon.GenRelayInfo(c, types.RelayFormatMjProxy, nil, nil)
  365. if err != nil {
  366. c.JSON(http.StatusInternalServerError, gin.H{
  367. "description": fmt.Sprintf("failed to generate relay info: %s", err.Error()),
  368. "type": "upstream_error",
  369. "code": 4,
  370. })
  371. return
  372. }
  373. var mjErr *dto.MidjourneyResponse
  374. switch relayInfo.RelayMode {
  375. case relayconstant.RelayModeMidjourneyNotify:
  376. mjErr = relay.RelayMidjourneyNotify(c)
  377. case relayconstant.RelayModeMidjourneyTaskFetch, relayconstant.RelayModeMidjourneyTaskFetchByCondition:
  378. mjErr = relay.RelayMidjourneyTask(c, relayInfo.RelayMode)
  379. case relayconstant.RelayModeMidjourneyTaskImageSeed:
  380. mjErr = relay.RelayMidjourneyTaskImageSeed(c)
  381. case relayconstant.RelayModeSwapFace:
  382. mjErr = relay.RelaySwapFace(c, relayInfo)
  383. default:
  384. mjErr = relay.RelayMidjourneySubmit(c, relayInfo)
  385. }
  386. //err = relayMidjourneySubmit(c, relayMode)
  387. log.Println(mjErr)
  388. if mjErr != nil {
  389. statusCode := http.StatusBadRequest
  390. if mjErr.Code == 30 {
  391. mjErr.Result = "当前分组负载已饱和,请稍后再试,或升级账户以提升服务质量。"
  392. statusCode = http.StatusTooManyRequests
  393. }
  394. c.JSON(statusCode, gin.H{
  395. "description": fmt.Sprintf("%s %s", mjErr.Description, mjErr.Result),
  396. "type": "upstream_error",
  397. "code": mjErr.Code,
  398. })
  399. channelId := c.GetInt("channel_id")
  400. logger.LogError(c, fmt.Sprintf("relay error (channel #%d, status code %d): %s", channelId, statusCode, fmt.Sprintf("%s %s", mjErr.Description, mjErr.Result)))
  401. }
  402. }
  403. func RelayNotImplemented(c *gin.Context) {
  404. err := types.OpenAIError{
  405. Message: "API not implemented",
  406. Type: "new_api_error",
  407. Param: "",
  408. Code: "api_not_implemented",
  409. }
  410. c.JSON(http.StatusNotImplemented, gin.H{
  411. "error": err,
  412. })
  413. }
  414. func RelayNotFound(c *gin.Context) {
  415. err := types.OpenAIError{
  416. Message: fmt.Sprintf("Invalid URL (%s %s)", c.Request.Method, c.Request.URL.Path),
  417. Type: "invalid_request_error",
  418. Param: "",
  419. Code: "",
  420. }
  421. c.JSON(http.StatusNotFound, gin.H{
  422. "error": err,
  423. })
  424. }
  425. func RelayTaskFetch(c *gin.Context) {
  426. relayInfo, err := relaycommon.GenRelayInfo(c, types.RelayFormatTask, nil, nil)
  427. if err != nil {
  428. c.JSON(http.StatusInternalServerError, &dto.TaskError{
  429. Code: "gen_relay_info_failed",
  430. Message: err.Error(),
  431. StatusCode: http.StatusInternalServerError,
  432. })
  433. return
  434. }
  435. if taskErr := relay.RelayTaskFetch(c, relayInfo.RelayMode); taskErr != nil {
  436. respondTaskError(c, taskErr)
  437. }
  438. }
  439. func RelayTask(c *gin.Context) {
  440. relayInfo, err := relaycommon.GenRelayInfo(c, types.RelayFormatTask, nil, nil)
  441. if err != nil {
  442. c.JSON(http.StatusInternalServerError, &dto.TaskError{
  443. Code: "gen_relay_info_failed",
  444. Message: err.Error(),
  445. StatusCode: http.StatusInternalServerError,
  446. })
  447. return
  448. }
  449. if taskErr := relay.ResolveOriginTask(c, relayInfo); taskErr != nil {
  450. respondTaskError(c, taskErr)
  451. return
  452. }
  453. var result *relay.TaskSubmitResult
  454. var taskErr *dto.TaskError
  455. defer func() {
  456. if taskErr != nil && relayInfo.Billing != nil {
  457. relayInfo.Billing.Refund(c)
  458. }
  459. }()
  460. retryParam := &service.RetryParam{
  461. Ctx: c,
  462. TokenGroup: relayInfo.TokenGroup,
  463. ModelName: relayInfo.OriginModelName,
  464. Retry: common.GetPointer(0),
  465. }
  466. for ; retryParam.GetRetry() <= common.RetryTimes; retryParam.IncreaseRetry() {
  467. var channel *model.Channel
  468. if lockedCh, ok := relayInfo.LockedChannel.(*model.Channel); ok && lockedCh != nil {
  469. channel = lockedCh
  470. if retryParam.GetRetry() > 0 {
  471. if setupErr := middleware.SetupContextForSelectedChannel(c, channel, relayInfo.OriginModelName); setupErr != nil {
  472. taskErr = service.TaskErrorWrapperLocal(setupErr.Err, "setup_locked_channel_failed", http.StatusInternalServerError)
  473. break
  474. }
  475. }
  476. } else {
  477. var channelErr *types.NewAPIError
  478. channel, channelErr = getChannel(c, relayInfo, retryParam)
  479. if channelErr != nil {
  480. logger.LogError(c, channelErr.Error())
  481. taskErr = service.TaskErrorWrapperLocal(channelErr.Err, "get_channel_failed", http.StatusInternalServerError)
  482. break
  483. }
  484. }
  485. addUsedChannel(c, channel.Id)
  486. bodyStorage, bodyErr := common.GetBodyStorage(c)
  487. if bodyErr != nil {
  488. if common.IsRequestBodyTooLargeError(bodyErr) || errors.Is(bodyErr, common.ErrRequestBodyTooLarge) {
  489. taskErr = service.TaskErrorWrapperLocal(bodyErr, "read_request_body_failed", http.StatusRequestEntityTooLarge)
  490. } else {
  491. taskErr = service.TaskErrorWrapperLocal(bodyErr, "read_request_body_failed", http.StatusBadRequest)
  492. }
  493. break
  494. }
  495. c.Request.Body = io.NopCloser(bodyStorage)
  496. result, taskErr = relay.RelayTaskSubmit(c, relayInfo)
  497. if taskErr == nil {
  498. break
  499. }
  500. if !taskErr.LocalError {
  501. processChannelError(c,
  502. *types.NewChannelError(channel.Id, channel.Type, channel.Name, channel.ChannelInfo.IsMultiKey,
  503. common.GetContextKeyString(c, constant.ContextKeyChannelKey), channel.GetAutoBan()),
  504. types.NewOpenAIError(taskErr.Error, types.ErrorCodeBadResponseStatusCode, taskErr.StatusCode))
  505. }
  506. if !shouldRetryTaskRelay(c, channel.Id, taskErr, common.RetryTimes-retryParam.GetRetry()) {
  507. break
  508. }
  509. }
  510. useChannel := c.GetStringSlice("use_channel")
  511. if len(useChannel) > 1 {
  512. retryLogStr := fmt.Sprintf("重试:%s", strings.Trim(strings.Join(strings.Fields(fmt.Sprint(useChannel)), "->"), "[]"))
  513. logger.LogInfo(c, retryLogStr)
  514. }
  515. // ── 成功:结算 + 日志 + 插入任务 ──
  516. if taskErr == nil {
  517. if settleErr := service.SettleBilling(c, relayInfo, result.Quota); settleErr != nil {
  518. common.SysError("settle task billing error: " + settleErr.Error())
  519. }
  520. service.LogTaskConsumption(c, relayInfo)
  521. task := model.InitTask(result.Platform, relayInfo)
  522. task.PrivateData.UpstreamTaskID = result.UpstreamTaskID
  523. task.PrivateData.BillingSource = relayInfo.BillingSource
  524. task.PrivateData.SubscriptionId = relayInfo.SubscriptionId
  525. task.PrivateData.TokenId = relayInfo.TokenId
  526. task.PrivateData.BillingContext = &model.TaskBillingContext{
  527. ModelPrice: relayInfo.PriceData.ModelPrice,
  528. GroupRatio: relayInfo.PriceData.GroupRatioInfo.GroupRatio,
  529. ModelRatio: relayInfo.PriceData.ModelRatio,
  530. OtherRatios: relayInfo.PriceData.OtherRatios,
  531. OriginModelName: relayInfo.OriginModelName,
  532. PerCallBilling: common.StringsContains(constant.TaskPricePatches, relayInfo.OriginModelName),
  533. }
  534. task.Quota = result.Quota
  535. task.Data = result.TaskData
  536. task.Action = relayInfo.Action
  537. if insertErr := task.Insert(); insertErr != nil {
  538. common.SysError("insert task error: " + insertErr.Error())
  539. }
  540. }
  541. if taskErr != nil {
  542. respondTaskError(c, taskErr)
  543. }
  544. }
  545. // respondTaskError 统一输出 Task 错误响应(含 429 限流提示改写)
  546. func respondTaskError(c *gin.Context, taskErr *dto.TaskError) {
  547. if taskErr.StatusCode == http.StatusTooManyRequests {
  548. taskErr.Message = "当前分组上游负载已饱和,请稍后再试"
  549. }
  550. c.JSON(taskErr.StatusCode, taskErr)
  551. }
  552. func shouldRetryTaskRelay(c *gin.Context, channelId int, taskErr *dto.TaskError, retryTimes int) bool {
  553. if taskErr == nil {
  554. return false
  555. }
  556. if service.ShouldSkipRetryAfterChannelAffinityFailure(c) {
  557. return false
  558. }
  559. if retryTimes <= 0 {
  560. return false
  561. }
  562. if taskErr.StatusCode == http.StatusTooManyRequests {
  563. return true
  564. }
  565. if taskErr.StatusCode == 307 {
  566. return true
  567. }
  568. if taskErr.StatusCode/100 == 5 {
  569. // 超时不重试
  570. if operation_setting.IsAlwaysSkipRetryStatusCode(taskErr.StatusCode) {
  571. return false
  572. }
  573. return true
  574. }
  575. if taskErr.StatusCode == http.StatusBadRequest {
  576. return false
  577. }
  578. if taskErr.StatusCode == 408 {
  579. // azure处理超时不重试
  580. return false
  581. }
  582. if taskErr.LocalError {
  583. return false
  584. }
  585. if taskErr.StatusCode/100 == 2 {
  586. return false
  587. }
  588. return true
  589. }