feat: image edits/inpaiting 支持 replicate 的 flux remix

.github/workflows/ci.yml

@@ -1,13 +1,13 @@
 name: CI
 
 # This setup assumes that you run the unit tests with code coverage in the same
-# workflow that will also print the coverage report as comment to the pull request. 
+# workflow that will also print the coverage report as comment to the pull request.
 # Therefore, you need to trigger this workflow when a pull request is (re)opened or
 # when new code is pushed to the branch of the pull request. In addition, you also
-# need to trigger this workflow when new code is pushed to the main branch because 
+# need to trigger this workflow when new code is pushed to the main branch because
 # we need to upload the code coverage results as artifact for the main branch as
 # well since it will be the baseline code coverage.
-# 
+#
 # We do not want to trigger the workflow for pushes to *any* branch because this
 # would trigger our jobs twice on pull requests (once from "push" event and once
 # from "pull_request->synchronize")
@@ -31,7 +31,7 @@ jobs:
         with:
           go-version: ^1.22
 
-      # When you execute your unit tests, make sure to use the "-coverprofile" flag to write a 
+      # When you execute your unit tests, make sure to use the "-coverprofile" flag to write a
       # coverage profile to a file. You will need the name of the file (e.g. "coverage.txt")
       # in the next step as well as the next job.
       - name: Test

.gitignore

@@ -9,4 +9,5 @@ logs
 data
 /web/node_modules
 cmd.md
-.env
+.env
+/one-api

common/ctxkey/key.go

@@ -3,6 +3,7 @@ package ctxkey
 const (
 	Config            = "config"
 	Id                = "id"
+	RequestId         = "X-Oneapi-Request-Id"
 	Username          = "username"
 	Role              = "role"
 	Status            = "status"
@@ -15,6 +16,7 @@ const (
 	Group             = "group"
 	ModelMapping      = "model_mapping"
 	ChannelName       = "channel_name"
+	ContentType       = "content_type"
 	TokenId           = "token_id"
 	TokenName         = "token_name"
 	BaseURL           = "base_url"

common/gin.go

@@ -3,43 +3,53 @@
 import (
 	"bytes"
 	"encoding/json"
-	"github.com/gin-gonic/gin"
-	"github.com/songquanpeng/one-api/common/ctxkey"
 	"io"
+	"reflect"
 	"strings"
+
+	"github.com/gin-gonic/gin"
+	"github.com/pkg/errors"
+	"github.com/songquanpeng/one-api/common/ctxkey"
 )
 
-func GetRequestBody(c *gin.Context) ([]byte, error) {
-	requestBody, _ := c.Get(ctxkey.KeyRequestBody)
-	if requestBody != nil {
-		return requestBody.([]byte), nil
+func GetRequestBody(c *gin.Context) (requestBody []byte, err error) {
+	if requestBodyCache, _ := c.Get(ctxkey.KeyRequestBody); requestBodyCache != nil {
+		return requestBodyCache.([]byte), nil
 	}
-	requestBody, err := io.ReadAll(c.Request.Body)
+	requestBody, err = io.ReadAll(c.Request.Body)
 	if err != nil {
 		return nil, err
 	}
 	_ = c.Request.Body.Close()
 	c.Set(ctxkey.KeyRequestBody, requestBody)
-	return requestBody.([]byte), nil
+
+	return requestBody, nil
 }
 
 func UnmarshalBodyReusable(c *gin.Context, v any) error {
 	requestBody, err := GetRequestBody(c)
 	if err != nil {
 		return err
 	}
+
+	// check v should be a pointer
+	if v == nil || reflect.TypeOf(v).Kind() != reflect.Ptr {
+		return errors.Errorf("UnmarshalBodyReusable only accept pointer, got %v", reflect.TypeOf(v))
+	}
+
 	contentType := c.Request.Header.Get("Content-Type")
 	if strings.HasPrefix(contentType, "application/json") {
-		err = json.Unmarshal(requestBody, &v)
-		c.Request.Body = io.NopCloser(bytes.NewBuffer(requestBody))
+		err = json.Unmarshal(requestBody, v)
 	} else {
 		c.Request.Body = io.NopCloser(bytes.NewBuffer(requestBody))
-		err = c.ShouldBind(&v)
+		err = c.ShouldBind(v)
 	}
 	if err != nil {
-		return err
+		return errors.Wrap(err, "unmarshal request body failed")
 	}
 	// Reset request body
+	c.Request.Body = io.NopCloser(bytes.NewBuffer(requestBody))
+
 	return nil
 }
 

controller/relay.go

@@ -26,7 +26,8 @@
 func relayHelper(c *gin.Context, relayMode int) *model.ErrorWithStatusCode {
 	var err *model.ErrorWithStatusCode
 	switch relayMode {
-	case relaymode.ImagesGenerations:
+	case relaymode.ImagesGenerations,
+		relaymode.ImagesEdits:
 		err = controller.RelayImageHelper(c, relayMode)
 	case relaymode.AudioSpeech:
 		fallthrough
@@ -45,10 +46,6 @@
 func Relay(c *gin.Context) {
 	ctx := c.Request.Context()
 	relayMode := relaymode.GetByPath(c.Request.URL.Path)
-	if config.DebugEnabled {
-		requestBody, _ := common.GetRequestBody(c)
-		logger.Debugf(ctx, "request body: %s", string(requestBody))
-	}
 	channelId := c.GetInt(ctxkey.ChannelId)
 	userId := c.GetInt(ctxkey.Id)
 	bizErr := relayHelper(c, relayMode)
@@ -60,7 +57,7 @@
 	channelName := c.GetString(ctxkey.ChannelName)
 	group := c.GetString(ctxkey.Group)
 	originalModel := c.GetString(ctxkey.OriginalModel)
-	go processChannelRelayError(ctx, userId, channelId, channelName, bizErr)
+	go processChannelRelayError(ctx, userId, channelId, channelName, *bizErr)
 	requestId := c.GetString(helper.RequestIdKey)
 	retryTimes := config.RetryTimes
 	if !shouldRetry(c, bizErr.StatusCode) {
@@ -87,9 +84,9 @@
 		channelId := c.GetInt(ctxkey.ChannelId)
 		lastFailedChannelId = channelId
 		channelName := c.GetString(ctxkey.ChannelName)
-		// BUG: bizErr is in race condition
-		go processChannelRelayError(ctx, userId, channelId, channelName, bizErr)
+		go processChannelRelayError(ctx, userId, channelId, channelName, *bizErr)
 	}
+
 	if bizErr != nil {
 		if bizErr.StatusCode == http.StatusTooManyRequests {
 			bizErr.Error.Message = "当前分组上游负载已饱和，请稍后再试"
@@ -122,7 +119,10 @@
 	return true
 }
 
-func processChannelRelayError(ctx context.Context, userId int, channelId int, channelName string, err *model.ErrorWithStatusCode) {
+func processChannelRelayError(ctx context.Context,
+	userId int, channelId int, channelName string,
+	// FIX: err should not use a pointer to avoid data race in concurrent situations
+	err model.ErrorWithStatusCode) {
 	logger.Errorf(ctx, "relay error (channel id %d, user id: %d): %s", channelId, userId, err.Message)
 	// https://platform.openai.com/docs/guides/error-codes/api-errors
 	if monitor.ShouldDisableChannel(&err.Error, err.StatusCode) {

go.mod

@@ -27,6 +27,7 @@ require (
 	github.com/stretchr/testify v1.9.0
 	golang.org/x/crypto v0.24.0
 	golang.org/x/image v0.18.0
+	golang.org/x/sync v0.7.0
 	google.golang.org/api v0.187.0
 	gorm.io/driver/mysql v1.5.6
 	gorm.io/driver/postgres v1.5.7
@@ -99,7 +100,6 @@ require (
 	golang.org/x/arch v0.8.0 // indirect
 	golang.org/x/net v0.26.0 // indirect
 	golang.org/x/oauth2 v0.21.0 // indirect
-	golang.org/x/sync v0.7.0 // indirect
 	golang.org/x/sys v0.21.0 // indirect
 	golang.org/x/text v0.16.0 // indirect
 	golang.org/x/time v0.5.0 // indirect

middleware/distributor.go

@@ -64,6 +64,7 @@
 	if channel.SystemPrompt != nil && *channel.SystemPrompt != "" {
 		c.Set(ctxkey.SystemPrompt, *channel.SystemPrompt)
 	}
+	c.Set(ctxkey.ContentType, c.Request.Header.Get("Content-Type"))
 	c.Set(ctxkey.ModelMapping, channel.GetModelMapping())
 	c.Set(ctxkey.OriginalModel, modelName) // for retry
 	c.Request.Header.Set("Authorization", fmt.Sprintf("Bearer %s", channel.Key))

middleware/logger.go

@@ -2,6 +2,7 @@ package middleware
 
 import (
 	"fmt"
+
 	"github.com/gin-gonic/gin"
 	"github.com/songquanpeng/one-api/common/helper"
 )

middleware/request-id.go

@@ -2,6 +2,7 @@ package middleware
 
 import (
 	"context"
+
 	"github.com/gin-gonic/gin"
 	"github.com/songquanpeng/one-api/common/helper"
 )

middleware/utils.go

@@ -1,12 +1,13 @@
 package middleware
 
 import (
-	"fmt"
+	"strings"
+
 	"github.com/gin-gonic/gin"
+	"github.com/pkg/errors"
 	"github.com/songquanpeng/one-api/common"
 	"github.com/songquanpeng/one-api/common/helper"
 	"github.com/songquanpeng/one-api/common/logger"
-	"strings"
 )
 
 func abortWithMessage(c *gin.Context, statusCode int, message string) {
@@ -24,28 +25,30 @@
 	var modelRequest ModelRequest
 	err := common.UnmarshalBodyReusable(c, &modelRequest)
 	if err != nil {
-		return "", fmt.Errorf("common.UnmarshalBodyReusable failed: %w", err)
+		return "", errors.Wrap(err, "common.UnmarshalBodyReusable failed")
 	}
-	if strings.HasPrefix(c.Request.URL.Path, "/v1/moderations") {
+
+	switch {
+	case strings.HasPrefix(c.Request.URL.Path, "/v1/moderations"):
 		if modelRequest.Model == "" {
 			modelRequest.Model = "text-moderation-stable"
 		}
-	}
-	if strings.HasSuffix(c.Request.URL.Path, "embeddings") {
+	case strings.HasSuffix(c.Request.URL.Path, "embeddings"):
 		if modelRequest.Model == "" {
 			modelRequest.Model = c.Param("model")
 		}
-	}
-	if strings.HasPrefix(c.Request.URL.Path, "/v1/images/generations") {
+	case strings.HasPrefix(c.Request.URL.Path, "/v1/images/generations"),
+		strings.HasPrefix(c.Request.URL.Path, "/v1/images/edits"):
 		if modelRequest.Model == "" {
 			modelRequest.Model = "dall-e-2"
 		}
-	}
-	if strings.HasPrefix(c.Request.URL.Path, "/v1/audio/transcriptions") || strings.HasPrefix(c.Request.URL.Path, "/v1/audio/translations") {
+	case strings.HasPrefix(c.Request.URL.Path, "/v1/audio/transcriptions"),
+		strings.HasPrefix(c.Request.URL.Path, "/v1/audio/translations"):
 		if modelRequest.Model == "" {
 			modelRequest.Model = "whisper-1"
 		}
 	}
+
 	return modelRequest.Model, nil
 }
 

monitor/manage.go

@@ -34,7 +34,7 @@
 		strings.Contains(lowerMessage, "credit") ||
 		strings.Contains(lowerMessage, "balance") ||
 		strings.Contains(lowerMessage, "permission denied") ||
-  	strings.Contains(lowerMessage, "organization has been restricted") || // groq
+		strings.Contains(lowerMessage, "organization has been restricted") || // groq
 		strings.Contains(lowerMessage, "已欠费") {
 		return true
 	}

relay/adaptor.go

@@ -16,6 +16,7 @@ import (
 	"github.com/songquanpeng/one-api/relay/adaptor/openai"
 	"github.com/songquanpeng/one-api/relay/adaptor/palm"
 	"github.com/songquanpeng/one-api/relay/adaptor/proxy"
+	"github.com/songquanpeng/one-api/relay/adaptor/replicate"
 	"github.com/songquanpeng/one-api/relay/adaptor/tencent"
 	"github.com/songquanpeng/one-api/relay/adaptor/vertexai"
 	"github.com/songquanpeng/one-api/relay/adaptor/xunfei"
@@ -61,6 +62,8 @@ func GetAdaptor(apiType int) adaptor.Adaptor {
 		return &vertexai.Adaptor{}
 	case apitype.Proxy:
 		return &proxy.Adaptor{}
+	case apitype.Replicate:
+		return &replicate.Adaptor{}
 	}
 	return nil
 }

relay/adaptor/common.go

@@ -3,11 +3,13 @@
 import (
 	"errors"
 	"fmt"
+	"io"
+	"net/http"
+
 	"github.com/gin-gonic/gin"
 	"github.com/songquanpeng/one-api/common/client"
+	"github.com/songquanpeng/one-api/common/ctxkey"
 	"github.com/songquanpeng/one-api/relay/meta"
-	"io"
-	"net/http"
 )
 
 func SetupCommonRequestHeader(c *gin.Context, req *http.Request, meta *meta.Meta) {
@@ -27,6 +29,9 @@
 	if err != nil {
 		return nil, fmt.Errorf("new request failed: %w", err)
 	}
+
+	req.Header.Set("Content-Type", c.GetString(ctxkey.ContentType))
+
 	err = a.SetupRequestHeader(c, req, meta)
 	if err != nil {
 		return nil, fmt.Errorf("setup request header failed: %w", err)

relay/adaptor/ollama/main.go

@@ -31,8 +31,8 @@
 			TopP:             request.TopP,
 			FrequencyPenalty: request.FrequencyPenalty,
 			PresencePenalty:  request.PresencePenalty,
-			NumPredict:  	  request.MaxTokens,
-			NumCtx:  	  request.NumCtx,
+			NumPredict:       request.MaxTokens,
+			NumCtx:           request.NumCtx,
 		},
 		Stream: request.Stream,
 	}
@@ -122,7 +122,7 @@
 	for scanner.Scan() {
 		data := scanner.Text()
 		if strings.HasPrefix(data, "}") {
-		    data = strings.TrimPrefix(data, "}") + "}"
+			data = strings.TrimPrefix(data, "}") + "}"
 		}
 
 		var ollamaResponse ChatResponse

relay/adaptor/openai/adaptor.go

@@ -111,10 +111,13 @@
 		switch meta.Mode {
 		case relaymode.ImagesGenerations:
 			err, _ = ImageHandler(c, resp)
+		case relaymode.ImagesEdits:
+			err, _ = ImagesEditsHandler(c, resp)
 		default:
 			err, usage = Handler(c, resp, meta.PromptTokens, meta.ActualModelName)
 		}
 	}
+
 	return
 }
 

relay/adaptor/openai/image.go

@@ -3,12 +3,30 @@
 import (
 	"bytes"
 	"encoding/json"
-	"github.com/gin-gonic/gin"
-	"github.com/songquanpeng/one-api/relay/model"
 	"io"
 	"net/http"
+
+	"github.com/gin-gonic/gin"
+	"github.com/songquanpeng/one-api/relay/model"
 )
 
+// ImagesEditsHandler just copy response body to client
+//
+// https://platform.openai.com/docs/api-reference/images/createEdit
+func ImagesEditsHandler(c *gin.Context, resp *http.Response) (*model.ErrorWithStatusCode, *model.Usage) {
+	c.Writer.WriteHeader(resp.StatusCode)
+	for k, v := range resp.Header {
+		c.Writer.Header().Set(k, v[0])
+	}
+
+	if _, err := io.Copy(c.Writer, resp.Body); err != nil {
+		return ErrorWrapper(err, "copy_response_body_failed", http.StatusInternalServerError), nil
+	}
+	defer resp.Body.Close()
+
+	return nil, nil
+}
+
 func ImageHandler(c *gin.Context, resp *http.Response) (*model.ErrorWithStatusCode, *model.Usage) {
 	var imageResponse ImageResponse
 	responseBody, err := io.ReadAll(resp.Body)