Merge pull request #2087 from feitianbubu/pr/doubao-tts-stream

feat: doubao tts support streaming realtime audio

relay/channel/volcengine/adaptor.go

@@ -23,6 +23,11 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/gin-gonic/gin"
 )
 )
 
 
+const (
+	contextKeyTTSRequest     = "volcengine_tts_request"
+	contextKeyResponseFormat = "response_format"
+)
+
 type Adaptor struct {
 type Adaptor struct {
 }
 }
 
 
@@ -50,7 +55,7 @@ func (a *Adaptor) ConvertAudioRequest(c *gin.Context, info *relaycommon.RelayInf
 	speedRatio := request.Speed
 	speedRatio := request.Speed
 	encoding := mapEncoding(request.ResponseFormat)
 	encoding := mapEncoding(request.ResponseFormat)
 
 
-	c.Set("response_format", encoding)
+	c.Set(contextKeyResponseFormat, encoding)
 
 
 	volcRequest := VolcengineTTSRequest{
 	volcRequest := VolcengineTTSRequest{
 		App: VolcengineTTSApp{
 		App: VolcengineTTSApp{
@@ -70,18 +75,23 @@ func (a *Adaptor) ConvertAudioRequest(c *gin.Context, info *relaycommon.RelayInf
 		Request: VolcengineTTSReqInfo{
 		Request: VolcengineTTSReqInfo{
 			ReqID:     generateRequestID(),
 			ReqID:     generateRequestID(),
 			Text:      request.Input,
 			Text:      request.Input,
-			Operation: "query",
+			Operation: "submit",
 			Model:     info.OriginModelName,
 			Model:     info.OriginModelName,
 		},
 		},
 	}
 	}
 
 
-	// 同步扩展字段的厂商自定义metadata
 	if len(request.Metadata) > 0 {
 	if len(request.Metadata) > 0 {
 		if err = json.Unmarshal(request.Metadata, &volcRequest); err != nil {
 		if err = json.Unmarshal(request.Metadata, &volcRequest); err != nil {
 			return nil, fmt.Errorf("error unmarshalling metadata to volcengine request: %w", err)
 			return nil, fmt.Errorf("error unmarshalling metadata to volcengine request: %w", err)
 		}
 		}
 	}
 	}
 
 
+	c.Set(contextKeyTTSRequest, volcRequest)
+
+	if volcRequest.Request.Operation == "submit" {
+		info.IsStream = true
+	}
+
 	jsonData, err := json.Marshal(volcRequest)
 	jsonData, err := json.Marshal(volcRequest)
 	if err != nil {
 	if err != nil {
 		return nil, fmt.Errorf("error marshalling volcengine request: %w", err)
 		return nil, fmt.Errorf("error marshalling volcengine request: %w", err)
@@ -100,9 +110,8 @@ func (a *Adaptor) ConvertImageRequest(c *gin.Context, info *relaycommon.RelayInf
 		writer := multipart.NewWriter(&requestBody)
 		writer := multipart.NewWriter(&requestBody)
 
 
 		writer.WriteField("model", request.Model)
 		writer.WriteField("model", request.Model)
-		// 获取所有表单字段
+
 		formData := c.Request.PostForm
 		formData := c.Request.PostForm
-		// 遍历表单字段并打印输出
 		for key, values := range formData {
 		for key, values := range formData {
 			if key == "model" {
 			if key == "model" {
 				continue
 				continue
@@ -112,21 +121,16 @@ func (a *Adaptor) ConvertImageRequest(c *gin.Context, info *relaycommon.RelayInf
 			}
 			}
 		}
 		}
 
 
-		// Parse the multipart form to handle both single image and multiple images
-		if err := c.Request.ParseMultipartForm(32 << 20); err != nil { // 32MB max memory
+		if err := c.Request.ParseMultipartForm(32 << 20); err != nil {
 			return nil, errors.New("failed to parse multipart form")
 			return nil, errors.New("failed to parse multipart form")
 		}
 		}
 
 
 		if c.Request.MultipartForm != nil && c.Request.MultipartForm.File != nil {
 		if c.Request.MultipartForm != nil && c.Request.MultipartForm.File != nil {
-			// Check if "image" field exists in any form, including array notation
 			var imageFiles []*multipart.FileHeader
 			var imageFiles []*multipart.FileHeader
 			var exists bool
 			var exists bool
 
 
-			// First check for standard "image" field
 			if imageFiles, exists = c.Request.MultipartForm.File["image"]; !exists || len(imageFiles) == 0 {
 			if imageFiles, exists = c.Request.MultipartForm.File["image"]; !exists || len(imageFiles) == 0 {
-				// If not found, check for "image[]" field
 				if imageFiles, exists = c.Request.MultipartForm.File["image[]"]; !exists || len(imageFiles) == 0 {
 				if imageFiles, exists = c.Request.MultipartForm.File["image[]"]; !exists || len(imageFiles) == 0 {
-					// If still not found, iterate through all fields to find any that start with "image["
 					foundArrayImages := false
 					foundArrayImages := false
 					for fieldName, files := range c.Request.MultipartForm.File {
 					for fieldName, files := range c.Request.MultipartForm.File {
 						if strings.HasPrefix(fieldName, "image[") && len(files) > 0 {
 						if strings.HasPrefix(fieldName, "image[") && len(files) > 0 {
@@ -137,14 +141,12 @@ func (a *Adaptor) ConvertImageRequest(c *gin.Context, info *relaycommon.RelayInf
 						}
 						}
 					}
 					}
 
 
-					// If no image fields found at all
 					if !foundArrayImages && (len(imageFiles) == 0) {
 					if !foundArrayImages && (len(imageFiles) == 0) {
 						return nil, errors.New("image is required")
 						return nil, errors.New("image is required")
 					}
 					}
 				}
 				}
 			}
 			}
 
 
-			// Process all image files
 			for i, fileHeader := range imageFiles {
 			for i, fileHeader := range imageFiles {
 				file, err := fileHeader.Open()
 				file, err := fileHeader.Open()
 				if err != nil {
 				if err != nil {
@@ -152,16 +154,13 @@ func (a *Adaptor) ConvertImageRequest(c *gin.Context, info *relaycommon.RelayInf
 				}
 				}
 				defer file.Close()
 				defer file.Close()
 
 
-				// If multiple images, use image[] as the field name
 				fieldName := "image"
 				fieldName := "image"
 				if len(imageFiles) > 1 {
 				if len(imageFiles) > 1 {
 					fieldName = "image[]"
 					fieldName = "image[]"
 				}
 				}
 
 
-				// Determine MIME type based on file extension
 				mimeType := detectImageMimeType(fileHeader.Filename)
 				mimeType := detectImageMimeType(fileHeader.Filename)
 
 
-				// Create a form file with the appropriate content type
 				h := make(textproto.MIMEHeader)
 				h := make(textproto.MIMEHeader)
 				h.Set("Content-Disposition", fmt.Sprintf(`form-data; name="%s"; filename="%s"`, fieldName, fileHeader.Filename))
 				h.Set("Content-Disposition", fmt.Sprintf(`form-data; name="%s"; filename="%s"`, fieldName, fileHeader.Filename))
 				h.Set("Content-Type", mimeType)
 				h.Set("Content-Type", mimeType)
@@ -176,7 +175,6 @@ func (a *Adaptor) ConvertImageRequest(c *gin.Context, info *relaycommon.RelayInf
 				}
 				}
 			}
 			}
 
 
-			// Handle mask file if present
 			if maskFiles, exists := c.Request.MultipartForm.File["mask"]; exists && len(maskFiles) > 0 {
 			if maskFiles, exists := c.Request.MultipartForm.File["mask"]; exists && len(maskFiles) > 0 {
 				maskFile, err := maskFiles[0].Open()
 				maskFile, err := maskFiles[0].Open()
 				if err != nil {
 				if err != nil {
@@ -184,10 +182,8 @@ func (a *Adaptor) ConvertImageRequest(c *gin.Context, info *relaycommon.RelayInf
 				}
 				}
 				defer maskFile.Close()
 				defer maskFile.Close()
 
 
-				// Determine MIME type for mask file
 				mimeType := detectImageMimeType(maskFiles[0].Filename)
 				mimeType := detectImageMimeType(maskFiles[0].Filename)
 
 
-				// Create a form file with the appropriate content type
 				h := make(textproto.MIMEHeader)
 				h := make(textproto.MIMEHeader)
 				h.Set("Content-Disposition", fmt.Sprintf(`form-data; name="mask"; filename="%s"`, maskFiles[0].Filename))
 				h.Set("Content-Disposition", fmt.Sprintf(`form-data; name="mask"; filename="%s"`, maskFiles[0].Filename))
 				h.Set("Content-Type", mimeType)
 				h.Set("Content-Type", mimeType)
@@ -205,7 +201,6 @@ func (a *Adaptor) ConvertImageRequest(c *gin.Context, info *relaycommon.RelayInf
 			return nil, errors.New("no multipart form data found")
 			return nil, errors.New("no multipart form data found")
 		}
 		}
 
 
-		// 关闭 multipart 编写器以设置分界线
 		writer.Close()
 		writer.Close()
 		c.Request.Header.Set("Content-Type", writer.FormDataContentType())
 		c.Request.Header.Set("Content-Type", writer.FormDataContentType())
 		return bytes.NewReader(requestBody.Bytes()), nil
 		return bytes.NewReader(requestBody.Bytes()), nil
@@ -215,7 +210,6 @@ func (a *Adaptor) ConvertImageRequest(c *gin.Context, info *relaycommon.RelayInf
 	}
 	}
 }
 }
 
 
-// detectImageMimeType determines the MIME type based on the file extension
 func detectImageMimeType(filename string) string {
 func detectImageMimeType(filename string) string {
 	ext := strings.ToLower(filepath.Ext(filename))
 	ext := strings.ToLower(filepath.Ext(filename))
 	switch ext {
 	switch ext {
@@ -226,11 +220,9 @@ func detectImageMimeType(filename string) string {
 	case ".webp":
 	case ".webp":
 		return "image/webp"
 		return "image/webp"
 	default:
 	default:
-		// Try to detect from extension if possible
 		if strings.HasPrefix(ext, ".jp") {
 		if strings.HasPrefix(ext, ".jp") {
 			return "image/jpeg"
 			return "image/jpeg"
 		}
 		}
-		// Default to png as a fallback
 		return "image/png"
 		return "image/png"
 	}
 	}
 }
 }
@@ -266,9 +258,8 @@ func (a *Adaptor) GetRequestURL(info *relaycommon.RelayInfo) (string, error) {
 		case constant.RelayModeRerank:
 		case constant.RelayModeRerank:
 			return fmt.Sprintf("%s/api/v3/rerank", baseUrl), nil
 			return fmt.Sprintf("%s/api/v3/rerank", baseUrl), nil
 		case constant.RelayModeAudioSpeech:
 		case constant.RelayModeAudioSpeech:
-			// 只有当 baseUrl 是火山默认的官方Url时才改为官方的的TTS接口，否则走透传的New接口
 			if baseUrl == channelconstant.ChannelBaseURLs[channelconstant.ChannelTypeVolcEngine] {
 			if baseUrl == channelconstant.ChannelBaseURLs[channelconstant.ChannelTypeVolcEngine] {
-				return "https://openspeech.bytedance.com/api/v1/tts", nil
+				return "wss://openspeech.bytedance.com/api/v1/tts/ws_binary", nil
 			}
 			}
 			return fmt.Sprintf("%s/v1/audio/speech", baseUrl), nil
 			return fmt.Sprintf("%s/v1/audio/speech", baseUrl), nil
 		default:
 		default:
@@ -297,7 +288,7 @@ func (a *Adaptor) ConvertOpenAIRequest(c *gin.Context, info *relaycommon.RelayIn
 	if request == nil {
 	if request == nil {
 		return nil, errors.New("request is nil")
 		return nil, errors.New("request is nil")
 	}
 	}
-	// 适配 方舟deepseek混合模型 的 thinking 后缀
+
 	if strings.HasSuffix(info.UpstreamModelName, "-thinking") && strings.HasPrefix(info.UpstreamModelName, "deepseek") {
 	if strings.HasSuffix(info.UpstreamModelName, "-thinking") && strings.HasPrefix(info.UpstreamModelName, "deepseek") {
 		info.UpstreamModelName = strings.TrimSuffix(info.UpstreamModelName, "-thinking")
 		info.UpstreamModelName = strings.TrimSuffix(info.UpstreamModelName, "-thinking")
 		request.Model = info.UpstreamModelName
 		request.Model = info.UpstreamModelName
@@ -315,17 +306,58 @@ func (a *Adaptor) ConvertEmbeddingRequest(c *gin.Context, info *relaycommon.Rela
 }
 }
 
 
 func (a *Adaptor) ConvertOpenAIResponsesRequest(c *gin.Context, info *relaycommon.RelayInfo, request dto.OpenAIResponsesRequest) (any, error) {
 func (a *Adaptor) ConvertOpenAIResponsesRequest(c *gin.Context, info *relaycommon.RelayInfo, request dto.OpenAIResponsesRequest) (any, error) {
-	// TODO implement me
 	return nil, errors.New("not implemented")
 	return nil, errors.New("not implemented")
 }
 }
 
 
 func (a *Adaptor) DoRequest(c *gin.Context, info *relaycommon.RelayInfo, requestBody io.Reader) (any, error) {
 func (a *Adaptor) DoRequest(c *gin.Context, info *relaycommon.RelayInfo, requestBody io.Reader) (any, error) {
+	if info.RelayMode == constant.RelayModeAudioSpeech {
+		baseUrl := info.ChannelBaseUrl
+		if baseUrl == "" {
+			baseUrl = channelconstant.ChannelBaseURLs[channelconstant.ChannelTypeVolcEngine]
+		}
+
+		if baseUrl == channelconstant.ChannelBaseURLs[channelconstant.ChannelTypeVolcEngine] {
+			if info.IsStream {
+				return nil, nil
+			}
+		}
+	}
 	return channel.DoApiRequest(a, c, info, requestBody)
 	return channel.DoApiRequest(a, c, info, requestBody)
 }
 }
 
 
 func (a *Adaptor) DoResponse(c *gin.Context, resp *http.Response, info *relaycommon.RelayInfo) (usage any, err *types.NewAPIError) {
 func (a *Adaptor) DoResponse(c *gin.Context, resp *http.Response, info *relaycommon.RelayInfo) (usage any, err *types.NewAPIError) {
 	if info.RelayMode == constant.RelayModeAudioSpeech {
 	if info.RelayMode == constant.RelayModeAudioSpeech {
-		encoding := mapEncoding(c.GetString("response_format"))
+		encoding := mapEncoding(c.GetString(contextKeyResponseFormat))
+		if info.IsStream {
+			volcRequestInterface, exists := c.Get(contextKeyTTSRequest)
+			if !exists {
+				return nil, types.NewErrorWithStatusCode(
+					errors.New("volcengine TTS request not found in context"),
+					types.ErrorCodeBadRequestBody,
+					http.StatusInternalServerError,
+				)
+			}
+
+			volcRequest, ok := volcRequestInterface.(VolcengineTTSRequest)
+			if !ok {
+				return nil, types.NewErrorWithStatusCode(
+					errors.New("invalid volcengine TTS request type"),
+					types.ErrorCodeBadRequestBody,
+					http.StatusInternalServerError,
+				)
+			}
+
+			// Get the WebSocket URL
+			requestURL, urlErr := a.GetRequestURL(info)
+			if urlErr != nil {
+				return nil, types.NewErrorWithStatusCode(
+					urlErr,
+					types.ErrorCodeBadRequestBody,
+					http.StatusInternalServerError,
+				)
+			}
+			return handleTTSWebSocketResponse(c, requestURL, volcRequest, info, encoding)
+		}
 		return handleTTSResponse(c, resp, info, encoding)
 		return handleTTSResponse(c, resp, info, encoding)
 	}
 	}
 
 

relay/channel/volcengine/tts.go

 package volcengine
 package volcengine
 
 
 import (
 import (
+	"context"
 	"encoding/base64"
 	"encoding/base64"
 	"encoding/json"
 	"encoding/json"
 	"errors"
 	"errors"
+	"fmt"
 	"io"
 	"io"
 	"net/http"
 	"net/http"
 	"strings"
 	"strings"
@@ -13,6 +15,7 @@ import (
 	"github.com/QuantumNous/new-api/types"
 	"github.com/QuantumNous/new-api/types"
 	"github.com/gin-gonic/gin"
 	"github.com/gin-gonic/gin"
 	"github.com/google/uuid"
 	"github.com/google/uuid"
+	"github.com/gorilla/websocket"
 )
 )
 
 
 type VolcengineTTSRequest struct {
 type VolcengineTTSRequest struct {
@@ -192,3 +195,111 @@ func handleTTSResponse(c *gin.Context, resp *http.Response, info *relaycommon.Re
 func generateRequestID() string {
 func generateRequestID() string {
 	return uuid.New().String()
 	return uuid.New().String()
 }
 }
+
+func handleTTSWebSocketResponse(c *gin.Context, requestURL string, volcRequest VolcengineTTSRequest, info *relaycommon.RelayInfo, encoding string) (usage any, err *types.NewAPIError) {
+	_, token, parseErr := parseVolcengineAuth(info.ApiKey)
+	if parseErr != nil {
+		return nil, types.NewErrorWithStatusCode(
+			parseErr,
+			types.ErrorCodeChannelInvalidKey,
+			http.StatusUnauthorized,
+		)
+	}
+
+	header := http.Header{}
+	header.Set("Authorization", fmt.Sprintf("Bearer;%s", token))
+
+	conn, resp, dialErr := websocket.DefaultDialer.DialContext(context.Background(), requestURL, header)
+	if dialErr != nil {
+		if resp != nil {
+			return nil, types.NewErrorWithStatusCode(
+				fmt.Errorf("failed to connect to websocket: %w, status: %d", dialErr, resp.StatusCode),
+				types.ErrorCodeBadResponseStatusCode,
+				http.StatusBadGateway,
+			)
+		}
+		return nil, types.NewErrorWithStatusCode(
+			fmt.Errorf("failed to connect to websocket: %w", dialErr),
+			types.ErrorCodeBadResponseStatusCode,
+			http.StatusBadGateway,
+		)
+	}
+	defer conn.Close()
+
+	payload, marshalErr := json.Marshal(volcRequest)
+	if marshalErr != nil {
+		return nil, types.NewErrorWithStatusCode(
+			fmt.Errorf("failed to marshal request: %w", marshalErr),
+			types.ErrorCodeBadRequestBody,
+			http.StatusInternalServerError,
+		)
+	}
+
+	if sendErr := FullClientRequest(conn, payload); sendErr != nil {
+		return nil, types.NewErrorWithStatusCode(
+			fmt.Errorf("failed to send request: %w", sendErr),
+			types.ErrorCodeBadRequestBody,
+			http.StatusInternalServerError,
+		)
+	}
+
+	contentType := getContentTypeByEncoding(encoding)
+	c.Header("Content-Type", contentType)
+	c.Header("Transfer-Encoding", "chunked")
+
+	for {
+		msg, recvErr := ReceiveMessage(conn)
+		if recvErr != nil {
+			if websocket.IsCloseError(recvErr, websocket.CloseNormalClosure, websocket.CloseGoingAway) {
+				break
+			}
+			return nil, types.NewErrorWithStatusCode(
+				fmt.Errorf("failed to receive message: %w", recvErr),
+				types.ErrorCodeBadResponse,
+				http.StatusInternalServerError,
+			)
+		}
+
+		switch msg.MsgType {
+		case MsgTypeError:
+			return nil, types.NewErrorWithStatusCode(
+				fmt.Errorf("received error from server: code=%d, %s", msg.ErrorCode, string(msg.Payload)),
+				types.ErrorCodeBadResponse,
+				http.StatusBadRequest,
+			)
+		case MsgTypeFrontEndResultServer:
+			continue
+		case MsgTypeAudioOnlyServer:
+			if len(msg.Payload) > 0 {
+				if _, writeErr := c.Writer.Write(msg.Payload); writeErr != nil {
+					return nil, types.NewErrorWithStatusCode(
+						fmt.Errorf("failed to write audio data: %w", writeErr),
+						types.ErrorCodeBadResponse,
+						http.StatusInternalServerError,
+					)
+				}
+				c.Writer.Flush()
+			}
+
+			if msg.Sequence < 0 {
+				c.Status(http.StatusOK)
+				usage = &dto.Usage{
+					PromptTokens:     info.PromptTokens,
+					CompletionTokens: 0,
+					TotalTokens:      info.PromptTokens,
+				}
+				return usage, nil
+			}
+		default:
+			continue
+		}
+	}
+
+	c.Status(http.StatusOK)
+	usage = &dto.Usage{
+		PromptTokens:     info.PromptTokens,
+		CompletionTokens: 0,
+		TotalTokens:      info.PromptTokens,
+	}
+	return usage, nil
+}