Nevar pievienot vairāk kā 25 tēmas Tēmai ir jāsākas ar burtu vai ciparu, tā var saturēt domu zīmes ('-') un var būt līdz 35 simboliem gara.
 
 
 

648 rindas
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. if err.UpstreamRequestId != "" {
  332. other["upstream_request_id"] = err.UpstreamRequestId
  333. }
  334. if err.UpstreamBody != "" {
  335. other["upstream_body"] = err.UpstreamBody
  336. }
  337. adminInfo := make(map[string]interface{})
  338. adminInfo["use_channel"] = c.GetStringSlice("use_channel")
  339. isMultiKey := common.GetContextKeyBool(c, constant.ContextKeyChannelIsMultiKey)
  340. if isMultiKey {
  341. adminInfo["is_multi_key"] = true
  342. adminInfo["multi_key_index"] = common.GetContextKeyInt(c, constant.ContextKeyChannelMultiKeyIndex)
  343. }
  344. service.AppendChannelAffinityAdminInfo(c, adminInfo)
  345. other["admin_info"] = adminInfo
  346. startTime := common.GetContextKeyTime(c, constant.ContextKeyRequestStartTime)
  347. if startTime.IsZero() {
  348. startTime = time.Now()
  349. }
  350. useTimeSeconds := int(time.Since(startTime).Seconds())
  351. model.RecordErrorLog(c, userId, channelId, modelName, tokenName, err.MaskSensitiveErrorWithStatusCode(), tokenId, useTimeSeconds, false, userGroup, other)
  352. }
  353. }
  354. func RelayMidjourney(c *gin.Context) {
  355. relayInfo, err := relaycommon.GenRelayInfo(c, types.RelayFormatMjProxy, nil, nil)
  356. if err != nil {
  357. c.JSON(http.StatusInternalServerError, gin.H{
  358. "description": fmt.Sprintf("failed to generate relay info: %s", err.Error()),
  359. "type": "upstream_error",
  360. "code": 4,
  361. })
  362. return
  363. }
  364. var mjErr *dto.MidjourneyResponse
  365. switch relayInfo.RelayMode {
  366. case relayconstant.RelayModeMidjourneyNotify:
  367. mjErr = relay.RelayMidjourneyNotify(c)
  368. case relayconstant.RelayModeMidjourneyTaskFetch, relayconstant.RelayModeMidjourneyTaskFetchByCondition:
  369. mjErr = relay.RelayMidjourneyTask(c, relayInfo.RelayMode)
  370. case relayconstant.RelayModeMidjourneyTaskImageSeed:
  371. mjErr = relay.RelayMidjourneyTaskImageSeed(c)
  372. case relayconstant.RelayModeSwapFace:
  373. mjErr = relay.RelaySwapFace(c, relayInfo)
  374. default:
  375. mjErr = relay.RelayMidjourneySubmit(c, relayInfo)
  376. }
  377. //err = relayMidjourneySubmit(c, relayMode)
  378. log.Println(mjErr)
  379. if mjErr != nil {
  380. statusCode := http.StatusBadRequest
  381. if mjErr.Code == 30 {
  382. mjErr.Result = "当前分组负载已饱和,请稍后再试,或升级账户以提升服务质量。"
  383. statusCode = http.StatusTooManyRequests
  384. }
  385. c.JSON(statusCode, gin.H{
  386. "description": fmt.Sprintf("%s %s", mjErr.Description, mjErr.Result),
  387. "type": "upstream_error",
  388. "code": mjErr.Code,
  389. })
  390. channelId := c.GetInt("channel_id")
  391. logger.LogError(c, fmt.Sprintf("relay error (channel #%d, status code %d): %s", channelId, statusCode, fmt.Sprintf("%s %s", mjErr.Description, mjErr.Result)))
  392. }
  393. }
  394. func RelayNotImplemented(c *gin.Context) {
  395. err := types.OpenAIError{
  396. Message: "API not implemented",
  397. Type: "new_api_error",
  398. Param: "",
  399. Code: "api_not_implemented",
  400. }
  401. c.JSON(http.StatusNotImplemented, gin.H{
  402. "error": err,
  403. })
  404. }
  405. func RelayNotFound(c *gin.Context) {
  406. err := types.OpenAIError{
  407. Message: fmt.Sprintf("Invalid URL (%s %s)", c.Request.Method, c.Request.URL.Path),
  408. Type: "invalid_request_error",
  409. Param: "",
  410. Code: "",
  411. }
  412. c.JSON(http.StatusNotFound, gin.H{
  413. "error": err,
  414. })
  415. }
  416. func RelayTaskFetch(c *gin.Context) {
  417. relayInfo, err := relaycommon.GenRelayInfo(c, types.RelayFormatTask, nil, nil)
  418. if err != nil {
  419. c.JSON(http.StatusInternalServerError, &dto.TaskError{
  420. Code: "gen_relay_info_failed",
  421. Message: err.Error(),
  422. StatusCode: http.StatusInternalServerError,
  423. })
  424. return
  425. }
  426. if taskErr := relay.RelayTaskFetch(c, relayInfo.RelayMode); taskErr != nil {
  427. respondTaskError(c, taskErr)
  428. }
  429. }
  430. func RelayTask(c *gin.Context) {
  431. relayInfo, err := relaycommon.GenRelayInfo(c, types.RelayFormatTask, nil, nil)
  432. if err != nil {
  433. c.JSON(http.StatusInternalServerError, &dto.TaskError{
  434. Code: "gen_relay_info_failed",
  435. Message: err.Error(),
  436. StatusCode: http.StatusInternalServerError,
  437. })
  438. return
  439. }
  440. if taskErr := relay.ResolveOriginTask(c, relayInfo); taskErr != nil {
  441. respondTaskError(c, taskErr)
  442. return
  443. }
  444. var result *relay.TaskSubmitResult
  445. var taskErr *dto.TaskError
  446. defer func() {
  447. if taskErr != nil && relayInfo.Billing != nil {
  448. relayInfo.Billing.Refund(c)
  449. }
  450. }()
  451. retryParam := &service.RetryParam{
  452. Ctx: c,
  453. TokenGroup: relayInfo.TokenGroup,
  454. ModelName: relayInfo.OriginModelName,
  455. Retry: common.GetPointer(0),
  456. }
  457. for ; retryParam.GetRetry() <= common.RetryTimes; retryParam.IncreaseRetry() {
  458. var channel *model.Channel
  459. if lockedCh, ok := relayInfo.LockedChannel.(*model.Channel); ok && lockedCh != nil {
  460. channel = lockedCh
  461. if retryParam.GetRetry() > 0 {
  462. if setupErr := middleware.SetupContextForSelectedChannel(c, channel, relayInfo.OriginModelName); setupErr != nil {
  463. taskErr = service.TaskErrorWrapperLocal(setupErr.Err, "setup_locked_channel_failed", http.StatusInternalServerError)
  464. break
  465. }
  466. }
  467. } else {
  468. var channelErr *types.NewAPIError
  469. channel, channelErr = getChannel(c, relayInfo, retryParam)
  470. if channelErr != nil {
  471. logger.LogError(c, channelErr.Error())
  472. taskErr = service.TaskErrorWrapperLocal(channelErr.Err, "get_channel_failed", http.StatusInternalServerError)
  473. break
  474. }
  475. }
  476. addUsedChannel(c, channel.Id)
  477. bodyStorage, bodyErr := common.GetBodyStorage(c)
  478. if bodyErr != nil {
  479. if common.IsRequestBodyTooLargeError(bodyErr) || errors.Is(bodyErr, common.ErrRequestBodyTooLarge) {
  480. taskErr = service.TaskErrorWrapperLocal(bodyErr, "read_request_body_failed", http.StatusRequestEntityTooLarge)
  481. } else {
  482. taskErr = service.TaskErrorWrapperLocal(bodyErr, "read_request_body_failed", http.StatusBadRequest)
  483. }
  484. break
  485. }
  486. c.Request.Body = io.NopCloser(bodyStorage)
  487. result, taskErr = relay.RelayTaskSubmit(c, relayInfo)
  488. if taskErr == nil {
  489. break
  490. }
  491. if !taskErr.LocalError {
  492. processChannelError(c,
  493. *types.NewChannelError(channel.Id, channel.Type, channel.Name, channel.ChannelInfo.IsMultiKey,
  494. common.GetContextKeyString(c, constant.ContextKeyChannelKey), channel.GetAutoBan()),
  495. types.NewOpenAIError(taskErr.Error, types.ErrorCodeBadResponseStatusCode, taskErr.StatusCode))
  496. }
  497. if !shouldRetryTaskRelay(c, channel.Id, taskErr, common.RetryTimes-retryParam.GetRetry()) {
  498. break
  499. }
  500. }
  501. useChannel := c.GetStringSlice("use_channel")
  502. if len(useChannel) > 1 {
  503. retryLogStr := fmt.Sprintf("重试:%s", strings.Trim(strings.Join(strings.Fields(fmt.Sprint(useChannel)), "->"), "[]"))
  504. logger.LogInfo(c, retryLogStr)
  505. }
  506. // ── 成功:结算 + 日志 + 插入任务 ──
  507. if taskErr == nil {
  508. if settleErr := service.SettleBilling(c, relayInfo, result.Quota); settleErr != nil {
  509. common.SysError("settle task billing error: " + settleErr.Error())
  510. }
  511. service.LogTaskConsumption(c, relayInfo)
  512. task := model.InitTask(result.Platform, relayInfo)
  513. task.PrivateData.UpstreamTaskID = result.UpstreamTaskID
  514. task.PrivateData.BillingSource = relayInfo.BillingSource
  515. task.PrivateData.SubscriptionId = relayInfo.SubscriptionId
  516. task.PrivateData.TokenId = relayInfo.TokenId
  517. task.PrivateData.BillingContext = &model.TaskBillingContext{
  518. ModelPrice: relayInfo.PriceData.ModelPrice,
  519. GroupRatio: relayInfo.PriceData.GroupRatioInfo.GroupRatio,
  520. ModelRatio: relayInfo.PriceData.ModelRatio,
  521. OtherRatios: relayInfo.PriceData.OtherRatios,
  522. OriginModelName: relayInfo.OriginModelName,
  523. PerCallBilling: common.StringsContains(constant.TaskPricePatches, relayInfo.OriginModelName),
  524. }
  525. task.Quota = result.Quota
  526. task.Data = result.TaskData
  527. task.Action = relayInfo.Action
  528. if insertErr := task.Insert(); insertErr != nil {
  529. common.SysError("insert task error: " + insertErr.Error())
  530. }
  531. }
  532. if taskErr != nil {
  533. respondTaskError(c, taskErr)
  534. }
  535. }
  536. // respondTaskError 统一输出 Task 错误响应(含 429 限流提示改写)
  537. func respondTaskError(c *gin.Context, taskErr *dto.TaskError) {
  538. if taskErr.StatusCode == http.StatusTooManyRequests {
  539. taskErr.Message = "当前分组上游负载已饱和,请稍后再试"
  540. }
  541. c.JSON(taskErr.StatusCode, taskErr)
  542. }
  543. func shouldRetryTaskRelay(c *gin.Context, channelId int, taskErr *dto.TaskError, retryTimes int) bool {
  544. if taskErr == nil {
  545. return false
  546. }
  547. if service.ShouldSkipRetryAfterChannelAffinityFailure(c) {
  548. return false
  549. }
  550. if retryTimes <= 0 {
  551. return false
  552. }
  553. if _, ok := c.Get("specific_channel_id"); ok {
  554. return false
  555. }
  556. if taskErr.StatusCode == http.StatusTooManyRequests {
  557. return true
  558. }
  559. if taskErr.StatusCode == 307 {
  560. return true
  561. }
  562. if taskErr.StatusCode/100 == 5 {
  563. // 超时不重试
  564. if operation_setting.IsAlwaysSkipRetryStatusCode(taskErr.StatusCode) {
  565. return false
  566. }
  567. return true
  568. }
  569. if taskErr.StatusCode == http.StatusBadRequest {
  570. return false
  571. }
  572. if taskErr.StatusCode == 408 {
  573. // azure处理超时不重试
  574. return false
  575. }
  576. if taskErr.LocalError {
  577. return false
  578. }
  579. if taskErr.StatusCode/100 == 2 {
  580. return false
  581. }
  582. return true
  583. }