Вы не можете выбрать более 25 тем Темы должны начинаться с буквы или цифры, могут содержать дефисы(-) и должны содержать не более 35 символов.
 
 
 

642 строки
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. if _, ok := c.Get("specific_channel_id"); ok {
  293. return false
  294. }
  295. code := openaiErr.StatusCode
  296. if code >= 200 && code < 300 {
  297. return false
  298. }
  299. if code < 100 || code > 599 {
  300. return true
  301. }
  302. return operation_setting.ShouldRetryByStatusCode(code)
  303. }
  304. func processChannelError(c *gin.Context, channelError types.ChannelError, err *types.NewAPIError) {
  305. logger.LogError(c, fmt.Sprintf("channel error (channel #%d, status code: %d): %s", channelError.ChannelId, err.StatusCode, err.Error()))
  306. // 不要使用context获取渠道信息,异步处理时可能会出现渠道信息不一致的情况
  307. // do not use context to get channel info, there may be inconsistent channel info when processing asynchronously
  308. if service.ShouldDisableChannel(channelError.ChannelType, err) && channelError.AutoBan {
  309. gopool.Go(func() {
  310. service.DisableChannel(channelError, err.ErrorWithStatusCode())
  311. })
  312. }
  313. if constant.ErrorLogEnabled && types.IsRecordErrorLog(err) {
  314. // 保存错误日志到mysql中
  315. userId := c.GetInt("id")
  316. tokenName := c.GetString("token_name")
  317. modelName := c.GetString("original_model")
  318. tokenId := c.GetInt("token_id")
  319. userGroup := c.GetString("group")
  320. channelId := c.GetInt("channel_id")
  321. other := make(map[string]interface{})
  322. if c.Request != nil && c.Request.URL != nil {
  323. other["request_path"] = c.Request.URL.Path
  324. }
  325. other["error_type"] = err.GetErrorType()
  326. other["error_code"] = err.GetErrorCode()
  327. other["status_code"] = err.StatusCode
  328. other["channel_id"] = channelId
  329. other["channel_name"] = c.GetString("channel_name")
  330. other["channel_type"] = c.GetInt("channel_type")
  331. adminInfo := make(map[string]interface{})
  332. adminInfo["use_channel"] = c.GetStringSlice("use_channel")
  333. isMultiKey := common.GetContextKeyBool(c, constant.ContextKeyChannelIsMultiKey)
  334. if isMultiKey {
  335. adminInfo["is_multi_key"] = true
  336. adminInfo["multi_key_index"] = common.GetContextKeyInt(c, constant.ContextKeyChannelMultiKeyIndex)
  337. }
  338. service.AppendChannelAffinityAdminInfo(c, adminInfo)
  339. other["admin_info"] = adminInfo
  340. startTime := common.GetContextKeyTime(c, constant.ContextKeyRequestStartTime)
  341. if startTime.IsZero() {
  342. startTime = time.Now()
  343. }
  344. useTimeSeconds := int(time.Since(startTime).Seconds())
  345. model.RecordErrorLog(c, userId, channelId, modelName, tokenName, err.MaskSensitiveErrorWithStatusCode(), tokenId, useTimeSeconds, false, userGroup, other)
  346. }
  347. }
  348. func RelayMidjourney(c *gin.Context) {
  349. relayInfo, err := relaycommon.GenRelayInfo(c, types.RelayFormatMjProxy, nil, nil)
  350. if err != nil {
  351. c.JSON(http.StatusInternalServerError, gin.H{
  352. "description": fmt.Sprintf("failed to generate relay info: %s", err.Error()),
  353. "type": "upstream_error",
  354. "code": 4,
  355. })
  356. return
  357. }
  358. var mjErr *dto.MidjourneyResponse
  359. switch relayInfo.RelayMode {
  360. case relayconstant.RelayModeMidjourneyNotify:
  361. mjErr = relay.RelayMidjourneyNotify(c)
  362. case relayconstant.RelayModeMidjourneyTaskFetch, relayconstant.RelayModeMidjourneyTaskFetchByCondition:
  363. mjErr = relay.RelayMidjourneyTask(c, relayInfo.RelayMode)
  364. case relayconstant.RelayModeMidjourneyTaskImageSeed:
  365. mjErr = relay.RelayMidjourneyTaskImageSeed(c)
  366. case relayconstant.RelayModeSwapFace:
  367. mjErr = relay.RelaySwapFace(c, relayInfo)
  368. default:
  369. mjErr = relay.RelayMidjourneySubmit(c, relayInfo)
  370. }
  371. //err = relayMidjourneySubmit(c, relayMode)
  372. log.Println(mjErr)
  373. if mjErr != nil {
  374. statusCode := http.StatusBadRequest
  375. if mjErr.Code == 30 {
  376. mjErr.Result = "当前分组负载已饱和,请稍后再试,或升级账户以提升服务质量。"
  377. statusCode = http.StatusTooManyRequests
  378. }
  379. c.JSON(statusCode, gin.H{
  380. "description": fmt.Sprintf("%s %s", mjErr.Description, mjErr.Result),
  381. "type": "upstream_error",
  382. "code": mjErr.Code,
  383. })
  384. channelId := c.GetInt("channel_id")
  385. logger.LogError(c, fmt.Sprintf("relay error (channel #%d, status code %d): %s", channelId, statusCode, fmt.Sprintf("%s %s", mjErr.Description, mjErr.Result)))
  386. }
  387. }
  388. func RelayNotImplemented(c *gin.Context) {
  389. err := types.OpenAIError{
  390. Message: "API not implemented",
  391. Type: "new_api_error",
  392. Param: "",
  393. Code: "api_not_implemented",
  394. }
  395. c.JSON(http.StatusNotImplemented, gin.H{
  396. "error": err,
  397. })
  398. }
  399. func RelayNotFound(c *gin.Context) {
  400. err := types.OpenAIError{
  401. Message: fmt.Sprintf("Invalid URL (%s %s)", c.Request.Method, c.Request.URL.Path),
  402. Type: "invalid_request_error",
  403. Param: "",
  404. Code: "",
  405. }
  406. c.JSON(http.StatusNotFound, gin.H{
  407. "error": err,
  408. })
  409. }
  410. func RelayTaskFetch(c *gin.Context) {
  411. relayInfo, err := relaycommon.GenRelayInfo(c, types.RelayFormatTask, nil, nil)
  412. if err != nil {
  413. c.JSON(http.StatusInternalServerError, &dto.TaskError{
  414. Code: "gen_relay_info_failed",
  415. Message: err.Error(),
  416. StatusCode: http.StatusInternalServerError,
  417. })
  418. return
  419. }
  420. if taskErr := relay.RelayTaskFetch(c, relayInfo.RelayMode); taskErr != nil {
  421. respondTaskError(c, taskErr)
  422. }
  423. }
  424. func RelayTask(c *gin.Context) {
  425. relayInfo, err := relaycommon.GenRelayInfo(c, types.RelayFormatTask, nil, nil)
  426. if err != nil {
  427. c.JSON(http.StatusInternalServerError, &dto.TaskError{
  428. Code: "gen_relay_info_failed",
  429. Message: err.Error(),
  430. StatusCode: http.StatusInternalServerError,
  431. })
  432. return
  433. }
  434. if taskErr := relay.ResolveOriginTask(c, relayInfo); taskErr != nil {
  435. respondTaskError(c, taskErr)
  436. return
  437. }
  438. var result *relay.TaskSubmitResult
  439. var taskErr *dto.TaskError
  440. defer func() {
  441. if taskErr != nil && relayInfo.Billing != nil {
  442. relayInfo.Billing.Refund(c)
  443. }
  444. }()
  445. retryParam := &service.RetryParam{
  446. Ctx: c,
  447. TokenGroup: relayInfo.TokenGroup,
  448. ModelName: relayInfo.OriginModelName,
  449. Retry: common.GetPointer(0),
  450. }
  451. for ; retryParam.GetRetry() <= common.RetryTimes; retryParam.IncreaseRetry() {
  452. var channel *model.Channel
  453. if lockedCh, ok := relayInfo.LockedChannel.(*model.Channel); ok && lockedCh != nil {
  454. channel = lockedCh
  455. if retryParam.GetRetry() > 0 {
  456. if setupErr := middleware.SetupContextForSelectedChannel(c, channel, relayInfo.OriginModelName); setupErr != nil {
  457. taskErr = service.TaskErrorWrapperLocal(setupErr.Err, "setup_locked_channel_failed", http.StatusInternalServerError)
  458. break
  459. }
  460. }
  461. } else {
  462. var channelErr *types.NewAPIError
  463. channel, channelErr = getChannel(c, relayInfo, retryParam)
  464. if channelErr != nil {
  465. logger.LogError(c, channelErr.Error())
  466. taskErr = service.TaskErrorWrapperLocal(channelErr.Err, "get_channel_failed", http.StatusInternalServerError)
  467. break
  468. }
  469. }
  470. addUsedChannel(c, channel.Id)
  471. bodyStorage, bodyErr := common.GetBodyStorage(c)
  472. if bodyErr != nil {
  473. if common.IsRequestBodyTooLargeError(bodyErr) || errors.Is(bodyErr, common.ErrRequestBodyTooLarge) {
  474. taskErr = service.TaskErrorWrapperLocal(bodyErr, "read_request_body_failed", http.StatusRequestEntityTooLarge)
  475. } else {
  476. taskErr = service.TaskErrorWrapperLocal(bodyErr, "read_request_body_failed", http.StatusBadRequest)
  477. }
  478. break
  479. }
  480. c.Request.Body = io.NopCloser(bodyStorage)
  481. result, taskErr = relay.RelayTaskSubmit(c, relayInfo)
  482. if taskErr == nil {
  483. break
  484. }
  485. if !taskErr.LocalError {
  486. processChannelError(c,
  487. *types.NewChannelError(channel.Id, channel.Type, channel.Name, channel.ChannelInfo.IsMultiKey,
  488. common.GetContextKeyString(c, constant.ContextKeyChannelKey), channel.GetAutoBan()),
  489. types.NewOpenAIError(taskErr.Error, types.ErrorCodeBadResponseStatusCode, taskErr.StatusCode))
  490. }
  491. if !shouldRetryTaskRelay(c, channel.Id, taskErr, common.RetryTimes-retryParam.GetRetry()) {
  492. break
  493. }
  494. }
  495. useChannel := c.GetStringSlice("use_channel")
  496. if len(useChannel) > 1 {
  497. retryLogStr := fmt.Sprintf("重试:%s", strings.Trim(strings.Join(strings.Fields(fmt.Sprint(useChannel)), "->"), "[]"))
  498. logger.LogInfo(c, retryLogStr)
  499. }
  500. // ── 成功:结算 + 日志 + 插入任务 ──
  501. if taskErr == nil {
  502. if settleErr := service.SettleBilling(c, relayInfo, result.Quota); settleErr != nil {
  503. common.SysError("settle task billing error: " + settleErr.Error())
  504. }
  505. service.LogTaskConsumption(c, relayInfo)
  506. task := model.InitTask(result.Platform, relayInfo)
  507. task.PrivateData.UpstreamTaskID = result.UpstreamTaskID
  508. task.PrivateData.BillingSource = relayInfo.BillingSource
  509. task.PrivateData.SubscriptionId = relayInfo.SubscriptionId
  510. task.PrivateData.TokenId = relayInfo.TokenId
  511. task.PrivateData.BillingContext = &model.TaskBillingContext{
  512. ModelPrice: relayInfo.PriceData.ModelPrice,
  513. GroupRatio: relayInfo.PriceData.GroupRatioInfo.GroupRatio,
  514. ModelRatio: relayInfo.PriceData.ModelRatio,
  515. OtherRatios: relayInfo.PriceData.OtherRatios,
  516. OriginModelName: relayInfo.OriginModelName,
  517. PerCallBilling: common.StringsContains(constant.TaskPricePatches, relayInfo.OriginModelName),
  518. }
  519. task.Quota = result.Quota
  520. task.Data = result.TaskData
  521. task.Action = relayInfo.Action
  522. if insertErr := task.Insert(); insertErr != nil {
  523. common.SysError("insert task error: " + insertErr.Error())
  524. }
  525. }
  526. if taskErr != nil {
  527. respondTaskError(c, taskErr)
  528. }
  529. }
  530. // respondTaskError 统一输出 Task 错误响应(含 429 限流提示改写)
  531. func respondTaskError(c *gin.Context, taskErr *dto.TaskError) {
  532. if taskErr.StatusCode == http.StatusTooManyRequests {
  533. taskErr.Message = "当前分组上游负载已饱和,请稍后再试"
  534. }
  535. c.JSON(taskErr.StatusCode, taskErr)
  536. }
  537. func shouldRetryTaskRelay(c *gin.Context, channelId int, taskErr *dto.TaskError, retryTimes int) bool {
  538. if taskErr == nil {
  539. return false
  540. }
  541. if service.ShouldSkipRetryAfterChannelAffinityFailure(c) {
  542. return false
  543. }
  544. if retryTimes <= 0 {
  545. return false
  546. }
  547. if _, ok := c.Get("specific_channel_id"); ok {
  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. }