初始化导入 new-api 源码

relay/helper/common.go

@@ -0,0 +1,211 @@
+package helper
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+
+	"github.com/QuantumNous/new-api/common"
+	"github.com/QuantumNous/new-api/dto"
+	"github.com/QuantumNous/new-api/logger"
+	"github.com/QuantumNous/new-api/types"
+
+	"github.com/gin-gonic/gin"
+	"github.com/gorilla/websocket"
+)
+
+func FlushWriter(c *gin.Context) (err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			err = fmt.Errorf("flush panic recovered: %v", r)
+		}
+	}()
+
+	if c == nil || c.Writer == nil {
+		return nil
+	}
+
+	if c.Request != nil && c.Request.Context().Err() != nil {
+		return fmt.Errorf("request context done: %w", c.Request.Context().Err())
+	}
+
+	flusher, ok := c.Writer.(http.Flusher)
+	if !ok {
+		return errors.New("streaming error: flusher not found")
+	}
+
+	flusher.Flush()
+	return nil
+}
+
+func SetEventStreamHeaders(c *gin.Context) {
+	// 检查是否已经设置过头部
+	if _, exists := c.Get("event_stream_headers_set"); exists {
+		return
+	}
+
+	// 设置标志，表示头部已经设置过
+	c.Set("event_stream_headers_set", true)
+
+	c.Writer.Header().Set("Content-Type", "text/event-stream")
+	c.Writer.Header().Set("Cache-Control", "no-cache")
+	c.Writer.Header().Set("Connection", "keep-alive")
+	c.Writer.Header().Set("Transfer-Encoding", "chunked")
+	c.Writer.Header().Set("X-Accel-Buffering", "no")
+}
+
+func ClaudeData(c *gin.Context, resp dto.ClaudeResponse) error {
+	jsonData, err := common.Marshal(resp)
+	if err != nil {
+		common.SysError("error marshalling stream response: " + err.Error())
+	} else {
+		c.Render(-1, common.CustomEvent{Data: fmt.Sprintf("event: %s\n", resp.Type)})
+		c.Render(-1, common.CustomEvent{Data: "data: " + string(jsonData)})
+	}
+	_ = FlushWriter(c)
+	return nil
+}
+
+func ClaudeChunkData(c *gin.Context, resp dto.ClaudeResponse, data string) {
+	c.Render(-1, common.CustomEvent{Data: fmt.Sprintf("event: %s\n", resp.Type)})
+	c.Render(-1, common.CustomEvent{Data: fmt.Sprintf("data: %s\n", data)})
+	_ = FlushWriter(c)
+}
+
+func ResponseChunkData(c *gin.Context, resp dto.ResponsesStreamResponse, data string) {
+	c.Render(-1, common.CustomEvent{Data: fmt.Sprintf("event: %s\n", resp.Type)})
+	c.Render(-1, common.CustomEvent{Data: fmt.Sprintf("data: %s", data)})
+	_ = FlushWriter(c)
+}
+
+func StringData(c *gin.Context, str string) error {
+	if c == nil || c.Writer == nil {
+		return errors.New("context or writer is nil")
+	}
+
+	if c.Request != nil && c.Request.Context().Err() != nil {
+		return fmt.Errorf("request context done: %w", c.Request.Context().Err())
+	}
+
+	c.Render(-1, common.CustomEvent{Data: "data: " + str})
+	return FlushWriter(c)
+}
+
+func PingData(c *gin.Context) error {
+	if c == nil || c.Writer == nil {
+		return errors.New("context or writer is nil")
+	}
+
+	if c.Request != nil && c.Request.Context().Err() != nil {
+		return fmt.Errorf("request context done: %w", c.Request.Context().Err())
+	}
+
+	if _, err := c.Writer.Write([]byte(": PING\n\n")); err != nil {
+		return fmt.Errorf("write ping data failed: %w", err)
+	}
+	return FlushWriter(c)
+}
+
+func ObjectData(c *gin.Context, object interface{}) error {
+	if object == nil {
+		return errors.New("object is nil")
+	}
+	jsonData, err := common.Marshal(object)
+	if err != nil {
+		return fmt.Errorf("error marshalling object: %w", err)
+	}
+	return StringData(c, string(jsonData))
+}
+
+func Done(c *gin.Context) {
+	_ = StringData(c, "[DONE]")
+}
+
+func WssString(c *gin.Context, ws *websocket.Conn, str string) error {
+	if ws == nil {
+		logger.LogError(c, "websocket connection is nil")
+		return errors.New("websocket connection is nil")
+	}
+	//common.LogInfo(c, fmt.Sprintf("sending message: %s", str))
+	return ws.WriteMessage(1, []byte(str))
+}
+
+func WssObject(c *gin.Context, ws *websocket.Conn, object interface{}) error {
+	jsonData, err := common.Marshal(object)
+	if err != nil {
+		return fmt.Errorf("error marshalling object: %w", err)
+	}
+	if ws == nil {
+		logger.LogError(c, "websocket connection is nil")
+		return errors.New("websocket connection is nil")
+	}
+	//common.LogInfo(c, fmt.Sprintf("sending message: %s", jsonData))
+	return ws.WriteMessage(1, jsonData)
+}
+
+func WssError(c *gin.Context, ws *websocket.Conn, openaiError types.OpenAIError) {
+	if ws == nil {
+		return
+	}
+	errorObj := &dto.RealtimeEvent{
+		Type:    "error",
+		EventId: GetLocalRealtimeID(c),
+		Error:   &openaiError,
+	}
+	_ = WssObject(c, ws, errorObj)
+}
+
+func GetResponseID(c *gin.Context) string {
+	logID := c.GetString(common.RequestIdKey)
+	return fmt.Sprintf("chatcmpl-%s", logID)
+}
+
+func GetLocalRealtimeID(c *gin.Context) string {
+	logID := c.GetString(common.RequestIdKey)
+	return fmt.Sprintf("evt_%s", logID)
+}
+
+func GenerateStartEmptyResponse(id string, createAt int64, model string, systemFingerprint *string) *dto.ChatCompletionsStreamResponse {
+	return &dto.ChatCompletionsStreamResponse{
+		Id:                id,
+		Object:            "chat.completion.chunk",
+		Created:           createAt,
+		Model:             model,
+		SystemFingerprint: systemFingerprint,
+		Choices: []dto.ChatCompletionsStreamResponseChoice{
+			{
+				Delta: dto.ChatCompletionsStreamResponseChoiceDelta{
+					Role:    "assistant",
+					Content: common.GetPointer(""),
+				},
+			},
+		},
+	}
+}
+
+func GenerateStopResponse(id string, createAt int64, model string, finishReason string) *dto.ChatCompletionsStreamResponse {
+	return &dto.ChatCompletionsStreamResponse{
+		Id:                id,
+		Object:            "chat.completion.chunk",
+		Created:           createAt,
+		Model:             model,
+		SystemFingerprint: nil,
+		Choices: []dto.ChatCompletionsStreamResponseChoice{
+			{
+				FinishReason: &finishReason,
+			},
+		},
+	}
+}
+
+func GenerateFinalUsageResponse(id string, createAt int64, model string, usage dto.Usage) *dto.ChatCompletionsStreamResponse {
+	return &dto.ChatCompletionsStreamResponse{
+		Id:                id,
+		Object:            "chat.completion.chunk",
+		Created:           createAt,
+		Model:             model,
+		SystemFingerprint: nil,
+		Choices:           make([]dto.ChatCompletionsStreamResponseChoice, 0),
+		Usage:             &usage,
+	}
+}

relay/helper/model_mapped.go

@@ -0,0 +1,81 @@
+package helper
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/QuantumNous/new-api/dto"
+	"github.com/QuantumNous/new-api/relay/common"
+	relayconstant "github.com/QuantumNous/new-api/relay/constant"
+	"github.com/QuantumNous/new-api/setting/ratio_setting"
+	"github.com/gin-gonic/gin"
+)
+
+func ModelMappedHelper(c *gin.Context, info *common.RelayInfo, request dto.Request) error {
+	if info.ChannelMeta == nil {
+		info.ChannelMeta = &common.ChannelMeta{}
+	}
+
+	isResponsesCompact := info.RelayMode == relayconstant.RelayModeResponsesCompact
+	originModelName := info.OriginModelName
+	mappingModelName := originModelName
+	if isResponsesCompact && strings.HasSuffix(originModelName, ratio_setting.CompactModelSuffix) {
+		mappingModelName = strings.TrimSuffix(originModelName, ratio_setting.CompactModelSuffix)
+	}
+
+	// map model name
+	modelMapping := c.GetString("model_mapping")
+	if modelMapping != "" && modelMapping != "{}" {
+		modelMap := make(map[string]string)
+		err := json.Unmarshal([]byte(modelMapping), &modelMap)
+		if err != nil {
+			return fmt.Errorf("unmarshal_model_mapping_failed")
+		}
+
+		// 支持链式模型重定向，最终使用链尾的模型
+		currentModel := mappingModelName
+		visitedModels := map[string]bool{
+			currentModel: true,
+		}
+		for {
+			if mappedModel, exists := modelMap[currentModel]; exists && mappedModel != "" {
+				// 模型重定向循环检测，避免无限循环
+				if visitedModels[mappedModel] {
+					if mappedModel == currentModel {
+						if currentModel == info.OriginModelName {
+							info.IsModelMapped = false
+							return nil
+						} else {
+							info.IsModelMapped = true
+							break
+						}
+					}
+					return errors.New("model_mapping_contains_cycle")
+				}
+				visitedModels[mappedModel] = true
+				currentModel = mappedModel
+				info.IsModelMapped = true
+			} else {
+				break
+			}
+		}
+		if info.IsModelMapped {
+			info.UpstreamModelName = currentModel
+		}
+	}
+
+	if isResponsesCompact {
+		finalUpstreamModelName := mappingModelName
+		if info.IsModelMapped && info.UpstreamModelName != "" {
+			finalUpstreamModelName = info.UpstreamModelName
+		}
+		info.UpstreamModelName = finalUpstreamModelName
+		info.OriginModelName = ratio_setting.WithCompactModelSuffix(finalUpstreamModelName)
+	}
+	if request != nil {
+		request.SetModelName(info.UpstreamModelName)
+	}
+	return nil
+}

relay/helper/price.go

@@ -0,0 +1,199 @@
+package helper
+
+import (
+	"fmt"
+
+	"github.com/QuantumNous/new-api/common"
+	"github.com/QuantumNous/new-api/logger"
+	relaycommon "github.com/QuantumNous/new-api/relay/common"
+	"github.com/QuantumNous/new-api/setting/operation_setting"
+	"github.com/QuantumNous/new-api/setting/ratio_setting"
+	"github.com/QuantumNous/new-api/types"
+
+	"github.com/gin-gonic/gin"
+)
+
+// https://docs.claude.com/en/docs/build-with-claude/prompt-caching#1-hour-cache-duration
+const claudeCacheCreation1hMultiplier = 6 / 3.75
+
+// HandleGroupRatio checks for "auto_group" in the context and updates the group ratio and relayInfo.UsingGroup if present
+func HandleGroupRatio(ctx *gin.Context, relayInfo *relaycommon.RelayInfo) types.GroupRatioInfo {
+	groupRatioInfo := types.GroupRatioInfo{
+		GroupRatio:        1.0, // default ratio
+		GroupSpecialRatio: -1,
+	}
+
+	// check auto group
+	autoGroup, exists := ctx.Get("auto_group")
+	if exists {
+		logger.LogDebug(ctx, fmt.Sprintf("final group: %s", autoGroup))
+		relayInfo.UsingGroup = autoGroup.(string)
+	}
+
+	// check user group special ratio
+	userGroupRatio, ok := ratio_setting.GetGroupGroupRatio(relayInfo.UserGroup, relayInfo.UsingGroup)
+	if ok {
+		// user group special ratio
+		groupRatioInfo.GroupSpecialRatio = userGroupRatio
+		groupRatioInfo.GroupRatio = userGroupRatio
+		groupRatioInfo.HasSpecialRatio = true
+	} else {
+		// normal group ratio
+		groupRatioInfo.GroupRatio = ratio_setting.GetGroupRatio(relayInfo.UsingGroup)
+	}
+
+	return groupRatioInfo
+}
+
+func ModelPriceHelper(c *gin.Context, info *relaycommon.RelayInfo, promptTokens int, meta *types.TokenCountMeta) (types.PriceData, error) {
+	modelPrice, usePrice := ratio_setting.GetModelPrice(info.OriginModelName, false)
+
+	groupRatioInfo := HandleGroupRatio(c, info)
+
+	var preConsumedQuota int
+	var modelRatio float64
+	var completionRatio float64
+	var cacheRatio float64
+	var imageRatio float64
+	var cacheCreationRatio float64
+	var cacheCreationRatio5m float64
+	var cacheCreationRatio1h float64
+	var audioRatio float64
+	var audioCompletionRatio float64
+	var freeModel bool
+	if !usePrice {
+		preConsumedTokens := common.Max(promptTokens, common.PreConsumedQuota)
+		if meta.MaxTokens != 0 {
+			preConsumedTokens += meta.MaxTokens
+		}
+		var success bool
+		var matchName string
+		modelRatio, success, matchName = ratio_setting.GetModelRatio(info.OriginModelName)
+		if !success {
+			acceptUnsetRatio := false
+			if info.UserSetting.AcceptUnsetRatioModel {
+				acceptUnsetRatio = true
+			}
+			if !acceptUnsetRatio {
+				return types.PriceData{}, fmt.Errorf("模型 %s 倍率或价格未配置，请联系管理员设置或开始自用模式；Model %s ratio or price not set, please set or start self-use mode", matchName, matchName)
+			}
+		}
+		completionRatio = ratio_setting.GetCompletionRatio(info.OriginModelName)
+		cacheRatio, _ = ratio_setting.GetCacheRatio(info.OriginModelName)
+		cacheCreationRatio, _ = ratio_setting.GetCreateCacheRatio(info.OriginModelName)
+		cacheCreationRatio5m = cacheCreationRatio
+		// 固定1h和5min缓存写入价格的比例
+		cacheCreationRatio1h = cacheCreationRatio * claudeCacheCreation1hMultiplier
+		imageRatio, _ = ratio_setting.GetImageRatio(info.OriginModelName)
+		audioRatio = ratio_setting.GetAudioRatio(info.OriginModelName)
+		audioCompletionRatio = ratio_setting.GetAudioCompletionRatio(info.OriginModelName)
+		ratio := modelRatio * groupRatioInfo.GroupRatio
+		preConsumedQuota = int(float64(preConsumedTokens) * ratio)
+	} else {
+		if meta.ImagePriceRatio != 0 {
+			modelPrice = modelPrice * meta.ImagePriceRatio
+		}
+		preConsumedQuota = int(modelPrice * common.QuotaPerUnit * groupRatioInfo.GroupRatio)
+	}
+
+	// check if free model pre-consume is disabled
+	if !operation_setting.GetQuotaSetting().EnableFreeModelPreConsume {
+		// if model price or ratio is 0, do not pre-consume quota
+		if groupRatioInfo.GroupRatio == 0 {
+			preConsumedQuota = 0
+			freeModel = true
+		} else if usePrice {
+			if modelPrice == 0 {
+				preConsumedQuota = 0
+				freeModel = true
+			}
+		} else {
+			if modelRatio == 0 {
+				preConsumedQuota = 0
+				freeModel = true
+			}
+		}
+	}
+
+	priceData := types.PriceData{
+		FreeModel:            freeModel,
+		ModelPrice:           modelPrice,
+		ModelRatio:           modelRatio,
+		CompletionRatio:      completionRatio,
+		GroupRatioInfo:       groupRatioInfo,
+		UsePrice:             usePrice,
+		CacheRatio:           cacheRatio,
+		ImageRatio:           imageRatio,
+		AudioRatio:           audioRatio,
+		AudioCompletionRatio: audioCompletionRatio,
+		CacheCreationRatio:   cacheCreationRatio,
+		CacheCreation5mRatio: cacheCreationRatio5m,
+		CacheCreation1hRatio: cacheCreationRatio1h,
+		QuotaToPreConsume:    preConsumedQuota,
+	}
+
+	if common.DebugEnabled {
+		println(fmt.Sprintf("model_price_helper result: %s", priceData.ToSetting()))
+	}
+	info.PriceData = priceData
+	return priceData, nil
+}
+
+// ModelPriceHelperPerCall 按次计费的 PriceHelper (MJ、Task)
+func ModelPriceHelperPerCall(c *gin.Context, info *relaycommon.RelayInfo) (types.PriceData, error) {
+	groupRatioInfo := HandleGroupRatio(c, info)
+
+	modelPrice, success := ratio_setting.GetModelPrice(info.OriginModelName, true)
+	// 如果没有配置价格，检查模型倍率配置
+	if !success {
+
+		// 没有配置费用，也要使用默认费用,否则按费率计费模型无法使用
+		defaultPrice, ok := ratio_setting.GetDefaultModelPriceMap()[info.OriginModelName]
+		if ok {
+			modelPrice = defaultPrice
+		} else {
+			// 没有配置倍率也不接受没配置,那就返回错误
+			_, ratioSuccess, matchName := ratio_setting.GetModelRatio(info.OriginModelName)
+			acceptUnsetRatio := false
+			if info.UserSetting.AcceptUnsetRatioModel {
+				acceptUnsetRatio = true
+			}
+			if !ratioSuccess && !acceptUnsetRatio {
+				return types.PriceData{}, fmt.Errorf("模型 %s 倍率或价格未配置，请联系管理员设置或开始自用模式；Model %s ratio or price not set, please set or start self-use mode", matchName, matchName)
+			}
+			// 未配置价格但配置了倍率，使用默认预扣价格
+			modelPrice = float64(common.PreConsumedQuota) / common.QuotaPerUnit
+		}
+
+	}
+	quota := int(modelPrice * common.QuotaPerUnit * groupRatioInfo.GroupRatio)
+
+	// 免费模型检测（与 ModelPriceHelper 对齐）
+	freeModel := false
+	if !operation_setting.GetQuotaSetting().EnableFreeModelPreConsume {
+		if groupRatioInfo.GroupRatio == 0 || modelPrice == 0 {
+			quota = 0
+			freeModel = true
+		}
+	}
+
+	priceData := types.PriceData{
+		FreeModel:      freeModel,
+		ModelPrice:     modelPrice,
+		Quota:          quota,
+		GroupRatioInfo: groupRatioInfo,
+	}
+	return priceData, nil
+}
+
+func ContainPriceOrRatio(modelName string) bool {
+	_, ok := ratio_setting.GetModelPrice(modelName, false)
+	if ok {
+		return true
+	}
+	_, ok, _ = ratio_setting.GetModelRatio(modelName)
+	if ok {
+		return true
+	}
+	return false
+}

relay/helper/stream_scanner.go

@@ -0,0 +1,283 @@
+package helper
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/QuantumNous/new-api/common"
+	"github.com/QuantumNous/new-api/constant"
+	"github.com/QuantumNous/new-api/logger"
+	relaycommon "github.com/QuantumNous/new-api/relay/common"
+	"github.com/QuantumNous/new-api/setting/operation_setting"
+
+	"github.com/bytedance/gopkg/util/gopool"
+
+	"github.com/gin-gonic/gin"
+)
+
+const (
+	InitialScannerBufferSize    = 64 << 10 // 64KB (64*1024)
+	DefaultMaxScannerBufferSize = 64 << 20 // 64MB (64*1024*1024) default SSE buffer size
+	DefaultPingInterval         = 10 * time.Second
+)
+
+func getScannerBufferSize() int {
+	if constant.StreamScannerMaxBufferMB > 0 {
+		return constant.StreamScannerMaxBufferMB << 20
+	}
+	return DefaultMaxScannerBufferSize
+}
+
+func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon.RelayInfo, dataHandler func(data string) bool) {
+
+	if resp == nil || dataHandler == nil {
+		return
+	}
+
+	// 确保响应体总是被关闭
+	defer func() {
+		if resp.Body != nil {
+			resp.Body.Close()
+		}
+	}()
+
+	streamingTimeout := time.Duration(constant.StreamingTimeout) * time.Second
+
+	var (
+		stopChan   = make(chan bool, 3) // 增加缓冲区避免阻塞
+		scanner    = bufio.NewScanner(resp.Body)
+		ticker     = time.NewTicker(streamingTimeout)
+		pingTicker *time.Ticker
+		writeMutex sync.Mutex     // Mutex to protect concurrent writes
+		wg         sync.WaitGroup // 用于等待所有 goroutine 退出
+	)
+
+	generalSettings := operation_setting.GetGeneralSetting()
+	pingEnabled := generalSettings.PingIntervalEnabled && !info.DisablePing
+	pingInterval := time.Duration(generalSettings.PingIntervalSeconds) * time.Second
+	if pingInterval <= 0 {
+		pingInterval = DefaultPingInterval
+	}
+
+	if pingEnabled {
+		pingTicker = time.NewTicker(pingInterval)
+	}
+
+	if common.DebugEnabled {
+		// print timeout and ping interval for debugging
+		println("relay timeout seconds:", common.RelayTimeout)
+		println("relay max idle conns:", common.RelayMaxIdleConns)
+		println("relay max idle conns per host:", common.RelayMaxIdleConnsPerHost)
+		println("streaming timeout seconds:", int64(streamingTimeout.Seconds()))
+		println("ping interval seconds:", int64(pingInterval.Seconds()))
+	}
+
+	// 改进资源清理，确保所有 goroutine 正确退出
+	defer func() {
+		// 通知所有 goroutine 停止
+		common.SafeSendBool(stopChan, true)
+
+		ticker.Stop()
+		if pingTicker != nil {
+			pingTicker.Stop()
+		}
+
+		// 等待所有 goroutine 退出，最多等待5秒
+		done := make(chan struct{})
+		gopool.Go(func() {
+			wg.Wait()
+			close(done)
+		})
+
+		select {
+		case <-done:
+		case <-time.After(5 * time.Second):
+			logger.LogError(c, "timeout waiting for goroutines to exit")
+		}
+
+		close(stopChan)
+	}()
+
+	scanner.Buffer(make([]byte, InitialScannerBufferSize), getScannerBufferSize())
+	scanner.Split(bufio.ScanLines)
+	SetEventStreamHeaders(c)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	ctx = context.WithValue(ctx, "stop_chan", stopChan)
+
+	// Handle ping data sending with improved error handling
+	if pingEnabled && pingTicker != nil {
+		wg.Add(1)
+		gopool.Go(func() {
+			defer func() {
+				wg.Done()
+				if r := recover(); r != nil {
+					logger.LogError(c, fmt.Sprintf("ping goroutine panic: %v", r))
+					common.SafeSendBool(stopChan, true)
+				}
+				if common.DebugEnabled {
+					println("ping goroutine exited")
+				}
+			}()
+
+			// 添加超时保护，防止 goroutine 无限运行
+			maxPingDuration := 30 * time.Minute // 最大 ping 持续时间
+			pingTimeout := time.NewTimer(maxPingDuration)
+			defer pingTimeout.Stop()
+
+			for {
+				select {
+				case <-pingTicker.C:
+					// 使用超时机制防止写操作阻塞
+					done := make(chan error, 1)
+					gopool.Go(func() {
+						writeMutex.Lock()
+						defer writeMutex.Unlock()
+						done <- PingData(c)
+					})
+
+					select {
+					case err := <-done:
+						if err != nil {
+							logger.LogError(c, "ping data error: "+err.Error())
+							return
+						}
+						if common.DebugEnabled {
+							println("ping data sent")
+						}
+					case <-time.After(10 * time.Second):
+						logger.LogError(c, "ping data send timeout")
+						return
+					case <-ctx.Done():
+						return
+					case <-stopChan:
+						return
+					}
+				case <-ctx.Done():
+					return
+				case <-stopChan:
+					return
+				case <-c.Request.Context().Done():
+					// 监听客户端断开连接
+					return
+				case <-pingTimeout.C:
+					logger.LogError(c, "ping goroutine max duration reached")
+					return
+				}
+			}
+		})
+	}
+
+	dataChan := make(chan string, 10)
+
+	wg.Add(1)
+	gopool.Go(func() {
+		defer func() {
+			wg.Done()
+			if r := recover(); r != nil {
+				logger.LogError(c, fmt.Sprintf("data handler goroutine panic: %v", r))
+			}
+			common.SafeSendBool(stopChan, true)
+		}()
+		for data := range dataChan {
+			writeMutex.Lock()
+			success := dataHandler(data)
+			writeMutex.Unlock()
+			if !success {
+				return
+			}
+		}
+	})
+
+	// Scanner goroutine with improved error handling
+	wg.Add(1)
+	common.RelayCtxGo(ctx, func() {
+		defer func() {
+			close(dataChan)
+			wg.Done()
+			if r := recover(); r != nil {
+				logger.LogError(c, fmt.Sprintf("scanner goroutine panic: %v", r))
+			}
+			common.SafeSendBool(stopChan, true)
+			if common.DebugEnabled {
+				println("scanner goroutine exited")
+			}
+		}()
+
+		for scanner.Scan() {
+			// 检查是否需要停止
+			select {
+			case <-stopChan:
+				return
+			case <-ctx.Done():
+				return
+			case <-c.Request.Context().Done():
+				return
+			default:
+			}
+
+			ticker.Reset(streamingTimeout)
+			data := scanner.Text()
+			if common.DebugEnabled {
+				println(data)
+			}
+
+			if len(data) < 6 {
+				continue
+			}
+			if data[:5] != "data:" && data[:6] != "[DONE]" {
+				continue
+			}
+			data = data[5:]
+			data = strings.TrimSpace(data)
+			if data == "" {
+				continue
+			}
+			if !strings.HasPrefix(data, "[DONE]") {
+				info.SetFirstResponseTime()
+				info.ReceivedResponseCount++
+
+				select {
+				case dataChan <- data:
+				case <-ctx.Done():
+					return
+				case <-stopChan:
+					return
+				}
+			} else {
+				// done, 处理完成标志，直接退出停止读取剩余数据防止出错
+				if common.DebugEnabled {
+					println("received [DONE], stopping scanner")
+				}
+				return
+			}
+		}
+
+		if err := scanner.Err(); err != nil {
+			if err != io.EOF {
+				logger.LogError(c, "scanner error: "+err.Error())
+			}
+		}
+	})
+
+	// 主循环等待完成或超时
+	select {
+	case <-ticker.C:
+		// 超时处理逻辑
+		logger.LogError(c, "streaming timeout")
+	case <-stopChan:
+		// 正常结束
+		logger.LogInfo(c, "streaming finished")
+	case <-c.Request.Context().Done():
+		// 客户端断开连接
+		logger.LogInfo(c, "client disconnected")
+	}
+}

relay/helper/stream_scanner_test.go

@@ -0,0 +1,521 @@
+package helper
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/QuantumNous/new-api/constant"
+	relaycommon "github.com/QuantumNous/new-api/relay/common"
+	"github.com/QuantumNous/new-api/setting/operation_setting"
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func init() {
+	gin.SetMode(gin.TestMode)
+}
+
+func setupStreamTest(t *testing.T, body io.Reader) (*gin.Context, *http.Response, *relaycommon.RelayInfo) {
+	t.Helper()
+
+	oldTimeout := constant.StreamingTimeout
+	constant.StreamingTimeout = 30
+	t.Cleanup(func() {
+		constant.StreamingTimeout = oldTimeout
+	})
+
+	recorder := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(recorder)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil)
+
+	resp := &http.Response{
+		Body: io.NopCloser(body),
+	}
+
+	info := &relaycommon.RelayInfo{
+		ChannelMeta: &relaycommon.ChannelMeta{},
+	}
+
+	return c, resp, info
+}
+
+func buildSSEBody(n int) string {
+	var b strings.Builder
+	for i := 0; i < n; i++ {
+		fmt.Fprintf(&b, "data: {\"id\":%d,\"choices\":[{\"delta\":{\"content\":\"token_%d\"}}]}\n", i, i)
+	}
+	b.WriteString("data: [DONE]\n")
+	return b.String()
+}
+
+// slowReader wraps a reader and injects a delay before each Read call,
+// simulating a slow upstream that trickles data.
+type slowReader struct {
+	r     io.Reader
+	delay time.Duration
+}
+
+func (s *slowReader) Read(p []byte) (int, error) {
+	time.Sleep(s.delay)
+	return s.r.Read(p)
+}
+
+// ---------- Basic correctness ----------
+
+func TestStreamScannerHandler_NilInputs(t *testing.T) {
+	t.Parallel()
+
+	recorder := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(recorder)
+	c.Request = httptest.NewRequest(http.MethodPost, "/", nil)
+
+	info := &relaycommon.RelayInfo{ChannelMeta: &relaycommon.ChannelMeta{}}
+
+	StreamScannerHandler(c, nil, info, func(data string) bool { return true })
+	StreamScannerHandler(c, &http.Response{Body: io.NopCloser(strings.NewReader(""))}, info, nil)
+}
+
+func TestStreamScannerHandler_EmptyBody(t *testing.T) {
+	t.Parallel()
+
+	c, resp, info := setupStreamTest(t, strings.NewReader(""))
+
+	var called atomic.Bool
+	StreamScannerHandler(c, resp, info, func(data string) bool {
+		called.Store(true)
+		return true
+	})
+
+	assert.False(t, called.Load(), "handler should not be called for empty body")
+}
+
+func TestStreamScannerHandler_1000Chunks(t *testing.T) {
+	t.Parallel()
+
+	const numChunks = 1000
+	body := buildSSEBody(numChunks)
+	c, resp, info := setupStreamTest(t, strings.NewReader(body))
+
+	var count atomic.Int64
+	StreamScannerHandler(c, resp, info, func(data string) bool {
+		count.Add(1)
+		return true
+	})
+
+	assert.Equal(t, int64(numChunks), count.Load())
+	assert.Equal(t, numChunks, info.ReceivedResponseCount)
+}
+
+func TestStreamScannerHandler_10000Chunks(t *testing.T) {
+	t.Parallel()
+
+	const numChunks = 10000
+	body := buildSSEBody(numChunks)
+	c, resp, info := setupStreamTest(t, strings.NewReader(body))
+
+	var count atomic.Int64
+	start := time.Now()
+
+	StreamScannerHandler(c, resp, info, func(data string) bool {
+		count.Add(1)
+		return true
+	})
+
+	elapsed := time.Since(start)
+	assert.Equal(t, int64(numChunks), count.Load())
+	assert.Equal(t, numChunks, info.ReceivedResponseCount)
+	t.Logf("10000 chunks processed in %v", elapsed)
+}
+
+func TestStreamScannerHandler_OrderPreserved(t *testing.T) {
+	t.Parallel()
+
+	const numChunks = 500
+	body := buildSSEBody(numChunks)
+	c, resp, info := setupStreamTest(t, strings.NewReader(body))
+
+	var mu sync.Mutex
+	received := make([]string, 0, numChunks)
+
+	StreamScannerHandler(c, resp, info, func(data string) bool {
+		mu.Lock()
+		received = append(received, data)
+		mu.Unlock()
+		return true
+	})
+
+	require.Equal(t, numChunks, len(received))
+	for i := 0; i < numChunks; i++ {
+		expected := fmt.Sprintf("{\"id\":%d,\"choices\":[{\"delta\":{\"content\":\"token_%d\"}}]}", i, i)
+		assert.Equal(t, expected, received[i], "chunk %d out of order", i)
+	}
+}
+
+func TestStreamScannerHandler_DoneStopsScanner(t *testing.T) {
+	t.Parallel()
+
+	body := buildSSEBody(50) + "data: should_not_appear\n"
+	c, resp, info := setupStreamTest(t, strings.NewReader(body))
+
+	var count atomic.Int64
+	StreamScannerHandler(c, resp, info, func(data string) bool {
+		count.Add(1)
+		return true
+	})
+
+	assert.Equal(t, int64(50), count.Load(), "data after [DONE] must not be processed")
+}
+
+func TestStreamScannerHandler_HandlerFailureStops(t *testing.T) {
+	t.Parallel()
+
+	const numChunks = 200
+	body := buildSSEBody(numChunks)
+	c, resp, info := setupStreamTest(t, strings.NewReader(body))
+
+	const failAt = 50
+	var count atomic.Int64
+	StreamScannerHandler(c, resp, info, func(data string) bool {
+		n := count.Add(1)
+		return n < failAt
+	})
+
+	// The worker stops at failAt; the scanner may have read ahead,
+	// but the handler should not be called beyond failAt.
+	assert.Equal(t, int64(failAt), count.Load())
+}
+
+func TestStreamScannerHandler_SkipsNonDataLines(t *testing.T) {
+	t.Parallel()
+
+	var b strings.Builder
+	b.WriteString(": comment line\n")
+	b.WriteString("event: message\n")
+	b.WriteString("id: 12345\n")
+	b.WriteString("retry: 5000\n")
+	for i := 0; i < 100; i++ {
+		fmt.Fprintf(&b, "data: payload_%d\n", i)
+		b.WriteString(": interleaved comment\n")
+	}
+	b.WriteString("data: [DONE]\n")
+
+	c, resp, info := setupStreamTest(t, strings.NewReader(b.String()))
+
+	var count atomic.Int64
+	StreamScannerHandler(c, resp, info, func(data string) bool {
+		count.Add(1)
+		return true
+	})
+
+	assert.Equal(t, int64(100), count.Load())
+}
+
+func TestStreamScannerHandler_DataWithExtraSpaces(t *testing.T) {
+	t.Parallel()
+
+	body := "data:   {\"trimmed\":true}  \ndata: [DONE]\n"
+	c, resp, info := setupStreamTest(t, strings.NewReader(body))
+
+	var got string
+	StreamScannerHandler(c, resp, info, func(data string) bool {
+		got = data
+		return true
+	})
+
+	assert.Equal(t, "{\"trimmed\":true}", got)
+}
+
+// ---------- Decoupling: scanner not blocked by slow handler ----------
+
+func TestStreamScannerHandler_ScannerDecoupledFromSlowHandler(t *testing.T) {
+	t.Parallel()
+
+	// Strategy: use a slow upstream (io.Pipe, 10ms per chunk) AND a slow handler (20ms per chunk).
+	// If the scanner were synchronously coupled to the handler, total time would be
+	// ~numChunks * (10ms + 20ms) = 30ms * 50 = 1500ms.
+	// With decoupling, total time should be closer to
+	// ~numChunks * max(10ms, 20ms) = 20ms * 50 = 1000ms
+	// because the scanner reads ahead into the buffer while the handler processes.
+	const numChunks = 50
+	const upstreamDelay = 10 * time.Millisecond
+	const handlerDelay = 20 * time.Millisecond
+
+	pr, pw := io.Pipe()
+	go func() {
+		defer pw.Close()
+		for i := 0; i < numChunks; i++ {
+			fmt.Fprintf(pw, "data: {\"id\":%d}\n", i)
+			time.Sleep(upstreamDelay)
+		}
+		fmt.Fprint(pw, "data: [DONE]\n")
+	}()
+
+	recorder := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(recorder)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil)
+
+	oldTimeout := constant.StreamingTimeout
+	constant.StreamingTimeout = 30
+	t.Cleanup(func() { constant.StreamingTimeout = oldTimeout })
+
+	resp := &http.Response{Body: pr}
+	info := &relaycommon.RelayInfo{ChannelMeta: &relaycommon.ChannelMeta{}}
+
+	var count atomic.Int64
+	start := time.Now()
+	done := make(chan struct{})
+	go func() {
+		StreamScannerHandler(c, resp, info, func(data string) bool {
+			time.Sleep(handlerDelay)
+			count.Add(1)
+			return true
+		})
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(15 * time.Second):
+		t.Fatal("StreamScannerHandler did not complete in time")
+	}
+
+	elapsed := time.Since(start)
+	assert.Equal(t, int64(numChunks), count.Load())
+
+	coupledTime := time.Duration(numChunks) * (upstreamDelay + handlerDelay)
+	t.Logf("elapsed=%v, coupled_estimate=%v", elapsed, coupledTime)
+
+	// If decoupled, elapsed should be well under the coupled estimate.
+	assert.Less(t, elapsed, coupledTime*85/100,
+		"decoupled elapsed time (%v) should be significantly less than coupled estimate (%v)", elapsed, coupledTime)
+}
+
+func TestStreamScannerHandler_SlowUpstreamFastHandler(t *testing.T) {
+	t.Parallel()
+
+	const numChunks = 50
+	body := buildSSEBody(numChunks)
+	reader := &slowReader{r: strings.NewReader(body), delay: 2 * time.Millisecond}
+	c, resp, info := setupStreamTest(t, reader)
+
+	var count atomic.Int64
+	start := time.Now()
+
+	done := make(chan struct{})
+	go func() {
+		StreamScannerHandler(c, resp, info, func(data string) bool {
+			count.Add(1)
+			return true
+		})
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(15 * time.Second):
+		t.Fatal("timed out with slow upstream")
+	}
+
+	elapsed := time.Since(start)
+	assert.Equal(t, int64(numChunks), count.Load())
+	t.Logf("slow upstream (%d chunks, 2ms/read): %v", numChunks, elapsed)
+}
+
+// ---------- Ping tests ----------
+
+func TestStreamScannerHandler_PingSentDuringSlowUpstream(t *testing.T) {
+	t.Parallel()
+
+	setting := operation_setting.GetGeneralSetting()
+	oldEnabled := setting.PingIntervalEnabled
+	oldSeconds := setting.PingIntervalSeconds
+	setting.PingIntervalEnabled = true
+	setting.PingIntervalSeconds = 1
+	t.Cleanup(func() {
+		setting.PingIntervalEnabled = oldEnabled
+		setting.PingIntervalSeconds = oldSeconds
+	})
+
+	// Create a reader that delivers data slowly: one chunk every 500ms over 3.5 seconds.
+	// The ping interval is 1s, so we should see at least 2 pings.
+	pr, pw := io.Pipe()
+	go func() {
+		defer pw.Close()
+		for i := 0; i < 7; i++ {
+			fmt.Fprintf(pw, "data: chunk_%d\n", i)
+			time.Sleep(500 * time.Millisecond)
+		}
+		fmt.Fprint(pw, "data: [DONE]\n")
+	}()
+
+	recorder := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(recorder)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil)
+
+	oldTimeout := constant.StreamingTimeout
+	constant.StreamingTimeout = 30
+	t.Cleanup(func() {
+		constant.StreamingTimeout = oldTimeout
+	})
+
+	resp := &http.Response{Body: pr}
+	info := &relaycommon.RelayInfo{ChannelMeta: &relaycommon.ChannelMeta{}}
+
+	var count atomic.Int64
+	done := make(chan struct{})
+	go func() {
+		StreamScannerHandler(c, resp, info, func(data string) bool {
+			count.Add(1)
+			return true
+		})
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(15 * time.Second):
+		t.Fatal("timed out waiting for stream to finish")
+	}
+
+	assert.Equal(t, int64(7), count.Load())
+
+	body := recorder.Body.String()
+	pingCount := strings.Count(body, ": PING")
+	t.Logf("received %d pings in response body", pingCount)
+	assert.GreaterOrEqual(t, pingCount, 2,
+		"expected at least 2 pings during 3.5s stream with 1s interval; got %d", pingCount)
+}
+
+func TestStreamScannerHandler_PingDisabledByRelayInfo(t *testing.T) {
+	t.Parallel()
+
+	setting := operation_setting.GetGeneralSetting()
+	oldEnabled := setting.PingIntervalEnabled
+	oldSeconds := setting.PingIntervalSeconds
+	setting.PingIntervalEnabled = true
+	setting.PingIntervalSeconds = 1
+	t.Cleanup(func() {
+		setting.PingIntervalEnabled = oldEnabled
+		setting.PingIntervalSeconds = oldSeconds
+	})
+
+	pr, pw := io.Pipe()
+	go func() {
+		defer pw.Close()
+		for i := 0; i < 5; i++ {
+			fmt.Fprintf(pw, "data: chunk_%d\n", i)
+			time.Sleep(500 * time.Millisecond)
+		}
+		fmt.Fprint(pw, "data: [DONE]\n")
+	}()
+
+	recorder := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(recorder)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil)
+
+	oldTimeout := constant.StreamingTimeout
+	constant.StreamingTimeout = 30
+	t.Cleanup(func() {
+		constant.StreamingTimeout = oldTimeout
+	})
+
+	resp := &http.Response{Body: pr}
+	info := &relaycommon.RelayInfo{
+		DisablePing: true,
+		ChannelMeta: &relaycommon.ChannelMeta{},
+	}
+
+	var count atomic.Int64
+	done := make(chan struct{})
+	go func() {
+		StreamScannerHandler(c, resp, info, func(data string) bool {
+			count.Add(1)
+			return true
+		})
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(15 * time.Second):
+		t.Fatal("timed out")
+	}
+
+	assert.Equal(t, int64(5), count.Load())
+
+	body := recorder.Body.String()
+	pingCount := strings.Count(body, ": PING")
+	assert.Equal(t, 0, pingCount, "pings should be disabled when DisablePing=true")
+}
+
+func TestStreamScannerHandler_PingInterleavesWithSlowUpstream(t *testing.T) {
+	t.Parallel()
+
+	setting := operation_setting.GetGeneralSetting()
+	oldEnabled := setting.PingIntervalEnabled
+	oldSeconds := setting.PingIntervalSeconds
+	setting.PingIntervalEnabled = true
+	setting.PingIntervalSeconds = 1
+	t.Cleanup(func() {
+		setting.PingIntervalEnabled = oldEnabled
+		setting.PingIntervalSeconds = oldSeconds
+	})
+
+	// Slow upstream + slow handler. Total stream takes ~5 seconds.
+	// The ping goroutine stays alive as long as the scanner is reading,
+	// so pings should fire between data writes.
+	pr, pw := io.Pipe()
+	go func() {
+		defer pw.Close()
+		for i := 0; i < 10; i++ {
+			fmt.Fprintf(pw, "data: chunk_%d\n", i)
+			time.Sleep(500 * time.Millisecond)
+		}
+		fmt.Fprint(pw, "data: [DONE]\n")
+	}()
+
+	recorder := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(recorder)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil)
+
+	oldTimeout := constant.StreamingTimeout
+	constant.StreamingTimeout = 30
+	t.Cleanup(func() {
+		constant.StreamingTimeout = oldTimeout
+	})
+
+	resp := &http.Response{Body: pr}
+	info := &relaycommon.RelayInfo{ChannelMeta: &relaycommon.ChannelMeta{}}
+
+	var count atomic.Int64
+	done := make(chan struct{})
+	go func() {
+		StreamScannerHandler(c, resp, info, func(data string) bool {
+			count.Add(1)
+			return true
+		})
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(15 * time.Second):
+		t.Fatal("timed out")
+	}
+
+	assert.Equal(t, int64(10), count.Load())
+
+	body := recorder.Body.String()
+	pingCount := strings.Count(body, ": PING")
+	t.Logf("received %d pings interleaved with 10 chunks over 5s", pingCount)
+	assert.GreaterOrEqual(t, pingCount, 3,
+		"expected at least 3 pings during 5s stream with 1s ping interval; got %d", pingCount)
+}

relay/helper/valid_request.go

@@ -0,0 +1,341 @@
+package helper
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"math"
+	"strings"
+
+	"github.com/QuantumNous/new-api/common"
+	"github.com/QuantumNous/new-api/dto"
+	"github.com/QuantumNous/new-api/logger"
+	relayconstant "github.com/QuantumNous/new-api/relay/constant"
+	"github.com/QuantumNous/new-api/types"
+	"github.com/samber/lo"
+
+	"github.com/gin-gonic/gin"
+)
+
+func GetAndValidateRequest(c *gin.Context, format types.RelayFormat) (request dto.Request, err error) {
+	relayMode := relayconstant.Path2RelayMode(c.Request.URL.Path)
+
+	switch format {
+	case types.RelayFormatOpenAI:
+		request, err = GetAndValidateTextRequest(c, relayMode)
+	case types.RelayFormatGemini:
+		if strings.Contains(c.Request.URL.Path, ":embedContent") {
+			request, err = GetAndValidateGeminiEmbeddingRequest(c)
+		} else if strings.Contains(c.Request.URL.Path, ":batchEmbedContents") {
+			request, err = GetAndValidateGeminiBatchEmbeddingRequest(c)
+		} else {
+			request, err = GetAndValidateGeminiRequest(c)
+		}
+	case types.RelayFormatClaude:
+		request, err = GetAndValidateClaudeRequest(c)
+	case types.RelayFormatOpenAIResponses:
+		request, err = GetAndValidateResponsesRequest(c)
+	case types.RelayFormatOpenAIResponsesCompaction:
+		request, err = GetAndValidateResponsesCompactionRequest(c)
+
+	case types.RelayFormatOpenAIImage:
+		request, err = GetAndValidOpenAIImageRequest(c, relayMode)
+	case types.RelayFormatEmbedding:
+		request, err = GetAndValidateEmbeddingRequest(c, relayMode)
+	case types.RelayFormatRerank:
+		request, err = GetAndValidateRerankRequest(c)
+	case types.RelayFormatOpenAIAudio:
+		request, err = GetAndValidAudioRequest(c, relayMode)
+	case types.RelayFormatOpenAIRealtime:
+		request = &dto.BaseRequest{}
+	default:
+		return nil, fmt.Errorf("unsupported relay format: %s", format)
+	}
+	return request, err
+}
+
+func GetAndValidAudioRequest(c *gin.Context, relayMode int) (*dto.AudioRequest, error) {
+	audioRequest := &dto.AudioRequest{}
+	err := common.UnmarshalBodyReusable(c, audioRequest)
+	if err != nil {
+		return nil, err
+	}
+	switch relayMode {
+	case relayconstant.RelayModeAudioSpeech:
+		if audioRequest.Model == "" {
+			return nil, errors.New("model is required")
+		}
+	default:
+		if audioRequest.Model == "" {
+			return nil, errors.New("model is required")
+		}
+		if audioRequest.ResponseFormat == "" {
+			audioRequest.ResponseFormat = "json"
+		}
+	}
+	return audioRequest, nil
+}
+
+func GetAndValidateRerankRequest(c *gin.Context) (*dto.RerankRequest, error) {
+	var rerankRequest *dto.RerankRequest
+	err := common.UnmarshalBodyReusable(c, &rerankRequest)
+	if err != nil {
+		logger.LogError(c, fmt.Sprintf("getAndValidateTextRequest failed: %s", err.Error()))
+		return nil, types.NewError(err, types.ErrorCodeInvalidRequest, types.ErrOptionWithSkipRetry())
+	}
+
+	if rerankRequest.Query == "" {
+		return nil, types.NewError(fmt.Errorf("query is empty"), types.ErrorCodeInvalidRequest, types.ErrOptionWithSkipRetry())
+	}
+	if len(rerankRequest.Documents) == 0 {
+		return nil, types.NewError(fmt.Errorf("documents is empty"), types.ErrorCodeInvalidRequest, types.ErrOptionWithSkipRetry())
+	}
+	return rerankRequest, nil
+}
+
+func GetAndValidateEmbeddingRequest(c *gin.Context, relayMode int) (*dto.EmbeddingRequest, error) {
+	var embeddingRequest *dto.EmbeddingRequest
+	err := common.UnmarshalBodyReusable(c, &embeddingRequest)
+	if err != nil {
+		logger.LogError(c, fmt.Sprintf("getAndValidateTextRequest failed: %s", err.Error()))
+		return nil, types.NewError(err, types.ErrorCodeInvalidRequest, types.ErrOptionWithSkipRetry())
+	}
+
+	if embeddingRequest.Input == nil {
+		return nil, fmt.Errorf("input is empty")
+	}
+	if relayMode == relayconstant.RelayModeModerations && embeddingRequest.Model == "" {
+		embeddingRequest.Model = "omni-moderation-latest"
+	}
+	if relayMode == relayconstant.RelayModeEmbeddings && embeddingRequest.Model == "" {
+		embeddingRequest.Model = c.Param("model")
+	}
+	return embeddingRequest, nil
+}
+
+func GetAndValidateResponsesRequest(c *gin.Context) (*dto.OpenAIResponsesRequest, error) {
+	request := &dto.OpenAIResponsesRequest{}
+	err := common.UnmarshalBodyReusable(c, request)
+	if err != nil {
+		return nil, err
+	}
+	if request.Model == "" {
+		return nil, errors.New("model is required")
+	}
+	if request.Input == nil {
+		return nil, errors.New("input is required")
+	}
+	return request, nil
+}
+
+func GetAndValidateResponsesCompactionRequest(c *gin.Context) (*dto.OpenAIResponsesCompactionRequest, error) {
+	request := &dto.OpenAIResponsesCompactionRequest{}
+	if err := common.UnmarshalBodyReusable(c, request); err != nil {
+		return nil, err
+	}
+	if request.Model == "" {
+		return nil, errors.New("model is required")
+	}
+	return request, nil
+}
+
+func GetAndValidOpenAIImageRequest(c *gin.Context, relayMode int) (*dto.ImageRequest, error) {
+	imageRequest := &dto.ImageRequest{}
+
+	switch relayMode {
+	case relayconstant.RelayModeImagesEdits:
+		if strings.Contains(c.Request.Header.Get("Content-Type"), "multipart/form-data") {
+			_, err := c.MultipartForm()
+			if err != nil {
+				return nil, fmt.Errorf("failed to parse image edit form request: %w", err)
+			}
+			formData := c.Request.PostForm
+			imageRequest.Prompt = formData.Get("prompt")
+			imageRequest.Model = formData.Get("model")
+			imageRequest.N = common.GetPointer(uint(common.String2Int(formData.Get("n"))))
+			imageRequest.Quality = formData.Get("quality")
+			imageRequest.Size = formData.Get("size")
+			if imageValue := formData.Get("image"); imageValue != "" {
+				imageRequest.Image, _ = json.Marshal(imageValue)
+			}
+
+			if imageRequest.Model == "gpt-image-1" {
+				if imageRequest.Quality == "" {
+					imageRequest.Quality = "standard"
+				}
+			}
+			if imageRequest.N == nil || *imageRequest.N == 0 {
+				imageRequest.N = common.GetPointer(uint(1))
+			}
+
+			hasWatermark := formData.Has("watermark")
+			if hasWatermark {
+				watermark := formData.Get("watermark") == "true"
+				imageRequest.Watermark = &watermark
+			}
+			break
+		}
+		fallthrough
+	default:
+		err := common.UnmarshalBodyReusable(c, imageRequest)
+		if err != nil {
+			return nil, err
+		}
+
+		if imageRequest.Model == "" {
+			//imageRequest.Model = "dall-e-3"
+			return nil, errors.New("model is required")
+		}
+
+		if strings.Contains(imageRequest.Size, "×") {
+			return nil, errors.New("size an unexpected error occurred in the parameter, please use 'x' instead of the multiplication sign '×'")
+		}
+
+		// Not "256x256", "512x512", or "1024x1024"
+		if imageRequest.Model == "dall-e-2" || imageRequest.Model == "dall-e" {
+			if imageRequest.Size != "" && imageRequest.Size != "256x256" && imageRequest.Size != "512x512" && imageRequest.Size != "1024x1024" {
+				return nil, errors.New("size must be one of 256x256, 512x512, or 1024x1024 for dall-e-2 or dall-e")
+			}
+			if imageRequest.Size == "" {
+				imageRequest.Size = "1024x1024"
+			}
+		} else if imageRequest.Model == "dall-e-3" {
+			if imageRequest.Size != "" && imageRequest.Size != "1024x1024" && imageRequest.Size != "1024x1792" && imageRequest.Size != "1792x1024" {
+				return nil, errors.New("size must be one of 1024x1024, 1024x1792 or 1792x1024 for dall-e-3")
+			}
+			if imageRequest.Quality == "" {
+				imageRequest.Quality = "standard"
+			}
+			if imageRequest.Size == "" {
+				imageRequest.Size = "1024x1024"
+			}
+		} else if imageRequest.Model == "gpt-image-1" {
+			if imageRequest.Quality == "" {
+				imageRequest.Quality = "auto"
+			}
+		}
+
+		//if imageRequest.Prompt == "" {
+		//	return nil, errors.New("prompt is required")
+		//}
+
+		if imageRequest.N == nil || *imageRequest.N == 0 {
+			imageRequest.N = common.GetPointer(uint(1))
+		}
+	}
+
+	return imageRequest, nil
+}
+
+func GetAndValidateClaudeRequest(c *gin.Context) (textRequest *dto.ClaudeRequest, err error) {
+	textRequest = &dto.ClaudeRequest{}
+	err = common.UnmarshalBodyReusable(c, textRequest)
+	if err != nil {
+		return nil, err
+	}
+	if textRequest.Messages == nil || len(textRequest.Messages) == 0 {
+		return nil, errors.New("field messages is required")
+	}
+	if textRequest.Model == "" {
+		return nil, errors.New("field model is required")
+	}
+
+	//if textRequest.Stream {
+	//	relayInfo.IsStream = true
+	//}
+
+	return textRequest, nil
+}
+
+func GetAndValidateTextRequest(c *gin.Context, relayMode int) (*dto.GeneralOpenAIRequest, error) {
+	textRequest := &dto.GeneralOpenAIRequest{}
+	err := common.UnmarshalBodyReusable(c, textRequest)
+	if err != nil {
+		return nil, err
+	}
+
+	if relayMode == relayconstant.RelayModeModerations && textRequest.Model == "" {
+		textRequest.Model = "text-moderation-latest"
+	}
+	if relayMode == relayconstant.RelayModeEmbeddings && textRequest.Model == "" {
+		textRequest.Model = c.Param("model")
+	}
+
+	if lo.FromPtrOr(textRequest.MaxTokens, uint(0)) > math.MaxInt32/2 {
+		return nil, errors.New("max_tokens is invalid")
+	}
+	if textRequest.Model == "" {
+		return nil, errors.New("model is required")
+	}
+	if textRequest.WebSearchOptions != nil {
+		if textRequest.WebSearchOptions.SearchContextSize != "" {
+			validSizes := map[string]bool{
+				"high":   true,
+				"medium": true,
+				"low":    true,
+			}
+			if !validSizes[textRequest.WebSearchOptions.SearchContextSize] {
+				return nil, errors.New("invalid search_context_size, must be one of: high, medium, low")
+			}
+		} else {
+			textRequest.WebSearchOptions.SearchContextSize = "medium"
+		}
+	}
+	switch relayMode {
+	case relayconstant.RelayModeCompletions:
+		if textRequest.Prompt == "" {
+			return nil, errors.New("field prompt is required")
+		}
+	case relayconstant.RelayModeChatCompletions:
+		// For FIM (Fill-in-the-middle) requests with prefix/suffix, messages is optional
+		// It will be filled by provider-specific adaptors if needed (e.g., SiliconFlow)。Or it is allowed by model vendor(s) (e.g., DeepSeek)
+		if len(textRequest.Messages) == 0 && textRequest.Prefix == nil && textRequest.Suffix == nil {
+			return nil, errors.New("field messages is required")
+		}
+	case relayconstant.RelayModeEmbeddings:
+	case relayconstant.RelayModeModerations:
+		if textRequest.Input == nil || textRequest.Input == "" {
+			return nil, errors.New("field input is required")
+		}
+	case relayconstant.RelayModeEdits:
+		if textRequest.Instruction == "" {
+			return nil, errors.New("field instruction is required")
+		}
+	}
+	return textRequest, nil
+}
+
+func GetAndValidateGeminiRequest(c *gin.Context) (*dto.GeminiChatRequest, error) {
+	request := &dto.GeminiChatRequest{}
+	err := common.UnmarshalBodyReusable(c, request)
+	if err != nil {
+		return nil, err
+	}
+	if len(request.Contents) == 0 && len(request.Requests) == 0 {
+		return nil, errors.New("contents is required")
+	}
+
+	//if c.Query("alt") == "sse" {
+	//	relayInfo.IsStream = true
+	//}
+
+	return request, nil
+}
+
+func GetAndValidateGeminiEmbeddingRequest(c *gin.Context) (*dto.GeminiEmbeddingRequest, error) {
+	request := &dto.GeminiEmbeddingRequest{}
+	err := common.UnmarshalBodyReusable(c, request)
+	if err != nil {
+		return nil, err
+	}
+	return request, nil
+}
+
+func GetAndValidateGeminiBatchEmbeddingRequest(c *gin.Context) (*dto.GeminiBatchEmbeddingRequest, error) {
+	request := &dto.GeminiBatchEmbeddingRequest{}
+	err := common.UnmarshalBodyReusable(c, request)
+	if err != nil {
+		return nil, err
+	}
+	return request, nil
+}