Não pode escolher mais do que 25 tópicos Os tópicos devem começar com uma letra ou um número, podem incluir traços ('-') e podem ter até 35 caracteres.
 
 
 

642 linhas
21 KiB

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