feat: 初步兼容流模式下openai渠道类型转为claude格式访问 #862

relay/channel/openai/adaptor.go

@@ -21,6 +21,7 @@ import (
 	"one-api/relay/channel/xinference"
 	relaycommon "one-api/relay/common"
 	"one-api/relay/constant"
+	"one-api/service"
 	"strings"
 )
 
@@ -29,10 +30,20 @@ type Adaptor struct {
 	ResponseFormat string
 }
 
-func (a *Adaptor) ConvertClaudeRequest(*gin.Context, *relaycommon.RelayInfo, *dto.ClaudeRequest) (any, error) {
-	//TODO implement me
-	panic("implement me")
-	return nil, nil
+func (a *Adaptor) ConvertClaudeRequest(c *gin.Context, info *relaycommon.RelayInfo, request *dto.ClaudeRequest) (any, error) {
+	if !strings.HasPrefix(request.Model, "claude") {
+		return nil, fmt.Errorf("you are using openai channel type with path /v1/messages, only claude model supported convert, but got %s", request.Model)
+	}
+	aiRequest, err := service.ClaudeToOpenAIRequest(*request)
+	if err != nil {
+		return nil, err
+	}
+	if info.SupportStreamOptions {
+		aiRequest.StreamOptions = &dto.StreamOptions{
+			IncludeUsage: true,
+		}
+	}
+	return a.ConvertOpenAIRequest(c, info, aiRequest)
 }
 
 func (a *Adaptor) Init(info *relaycommon.RelayInfo) {
@@ -40,6 +51,9 @@ func (a *Adaptor) Init(info *relaycommon.RelayInfo) {
 }
 
 func (a *Adaptor) GetRequestURL(info *relaycommon.RelayInfo) (string, error) {
+	if info.RelayFormat == relaycommon.RelayFormatClaude {
+		return fmt.Sprintf("%s/v1/chat/completions", info.BaseUrl), nil
+	}
 	if info.RelayMode == constant.RelayModeRealtime {
 		if strings.HasPrefix(info.BaseUrl, "https://") {
 			baseUrl := strings.TrimPrefix(info.BaseUrl, "https://")
@@ -115,7 +129,7 @@ func (a *Adaptor) SetupRequestHeader(c *gin.Context, header *http.Header, info *
 	return nil
 }
 
-func (a *Adaptor) ConvertRequest(c *gin.Context, info *relaycommon.RelayInfo, request *dto.GeneralOpenAIRequest) (any, error) {
+func (a *Adaptor) ConvertOpenAIRequest(c *gin.Context, info *relaycommon.RelayInfo, request *dto.GeneralOpenAIRequest) (any, error) {
 	if request == nil {
 		return nil, errors.New("request is nil")
 	}

relay/channel/openai/helper.go

@@ -0,0 +1,188 @@
+package openai
+
+import (
+	"encoding/json"
+	"one-api/common"
+	"one-api/dto"
+	relaycommon "one-api/relay/common"
+	relayconstant "one-api/relay/constant"
+	"one-api/relay/helper"
+	"one-api/service"
+	"strings"
+
+	"github.com/gin-gonic/gin"
+)
+
+// 辅助函数
+func handleStreamFormat(c *gin.Context, info *relaycommon.RelayInfo, data string, forceFormat bool, thinkToContent bool) error {
+	info.SendResponseCount++
+	switch info.RelayFormat {
+	case relaycommon.RelayFormatOpenAI:
+		return sendStreamData(c, info, data, forceFormat, thinkToContent)
+	case relaycommon.RelayFormatClaude:
+		return handleClaudeFormat(c, data, info)
+	}
+	return nil
+}
+
+func handleClaudeFormat(c *gin.Context, data string, info *relaycommon.RelayInfo) error {
+	var streamResponse dto.ChatCompletionsStreamResponse
+	if err := json.Unmarshal(common.StringToByteSlice(data), &streamResponse); err != nil {
+		return err
+	}
+
+	claudeResponses := service.StreamResponseOpenAI2Claude(&streamResponse, info)
+	for _, resp := range claudeResponses {
+		helper.ClaudeData(c, *resp)
+	}
+	return nil
+}
+
+func processStreamResponse(item string, responseTextBuilder *strings.Builder, toolCount *int) error {
+	var streamResponse dto.ChatCompletionsStreamResponse
+	if err := json.Unmarshal(common.StringToByteSlice(item), &streamResponse); err != nil {
+		return err
+	}
+
+	for _, choice := range streamResponse.Choices {
+		responseTextBuilder.WriteString(choice.Delta.GetContentString())
+		responseTextBuilder.WriteString(choice.Delta.GetReasoningContent())
+		if choice.Delta.ToolCalls != nil {
+			if len(choice.Delta.ToolCalls) > *toolCount {
+				*toolCount = len(choice.Delta.ToolCalls)
+			}
+			for _, tool := range choice.Delta.ToolCalls {
+				responseTextBuilder.WriteString(tool.Function.Name)
+				responseTextBuilder.WriteString(tool.Function.Arguments)
+			}
+		}
+	}
+	return nil
+}
+
+func processTokens(relayMode int, streamItems []string, responseTextBuilder *strings.Builder, toolCount *int) error {
+	streamResp := "[" + strings.Join(streamItems, ",") + "]"
+
+	switch relayMode {
+	case relayconstant.RelayModeChatCompletions:
+		return processChatCompletions(streamResp, streamItems, responseTextBuilder, toolCount)
+	case relayconstant.RelayModeCompletions:
+		return processCompletions(streamResp, streamItems, responseTextBuilder)
+	}
+	return nil
+}
+
+func processChatCompletions(streamResp string, streamItems []string, responseTextBuilder *strings.Builder, toolCount *int) error {
+	var streamResponses []dto.ChatCompletionsStreamResponse
+	if err := json.Unmarshal(common.StringToByteSlice(streamResp), &streamResponses); err != nil {
+		// 一次性解析失败，逐个解析
+		common.SysError("error unmarshalling stream response: " + err.Error())
+		for _, item := range streamItems {
+			if err := processStreamResponse(item, responseTextBuilder, toolCount); err != nil {
+				common.SysError("error processing stream response: " + err.Error())
+			}
+		}
+		return nil
+	}
+
+	// 批量处理所有响应
+	for _, streamResponse := range streamResponses {
+		for _, choice := range streamResponse.Choices {
+			responseTextBuilder.WriteString(choice.Delta.GetContentString())
+			responseTextBuilder.WriteString(choice.Delta.GetReasoningContent())
+			if choice.Delta.ToolCalls != nil {
+				if len(choice.Delta.ToolCalls) > *toolCount {
+					*toolCount = len(choice.Delta.ToolCalls)
+				}
+				for _, tool := range choice.Delta.ToolCalls {
+					responseTextBuilder.WriteString(tool.Function.Name)
+					responseTextBuilder.WriteString(tool.Function.Arguments)
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func processCompletions(streamResp string, streamItems []string, responseTextBuilder *strings.Builder) error {
+	var streamResponses []dto.CompletionsStreamResponse
+	if err := json.Unmarshal(common.StringToByteSlice(streamResp), &streamResponses); err != nil {
+		// 一次性解析失败，逐个解析
+		common.SysError("error unmarshalling stream response: " + err.Error())
+		for _, item := range streamItems {
+			var streamResponse dto.CompletionsStreamResponse
+			if err := json.Unmarshal(common.StringToByteSlice(item), &streamResponse); err != nil {
+				continue
+			}
+			for _, choice := range streamResponse.Choices {
+				responseTextBuilder.WriteString(choice.Text)
+			}
+		}
+		return nil
+	}
+
+	// 批量处理所有响应
+	for _, streamResponse := range streamResponses {
+		for _, choice := range streamResponse.Choices {
+			responseTextBuilder.WriteString(choice.Text)
+		}
+	}
+	return nil
+}
+
+func handleLastResponse(lastStreamData string, responseId *string, createAt *int64,
+	systemFingerprint *string, model *string, usage **dto.Usage,
+	containStreamUsage *bool, info *relaycommon.RelayInfo,
+	shouldSendLastResp *bool) error {
+
+	var lastStreamResponse dto.ChatCompletionsStreamResponse
+	if err := json.Unmarshal(common.StringToByteSlice(lastStreamData), &lastStreamResponse); err != nil {
+		return err
+	}
+
+	*responseId = lastStreamResponse.Id
+	*createAt = lastStreamResponse.Created
+	*systemFingerprint = lastStreamResponse.GetSystemFingerprint()
+	*model = lastStreamResponse.Model
+
+	if service.ValidUsage(lastStreamResponse.Usage) {
+		*containStreamUsage = true
+		*usage = lastStreamResponse.Usage
+		if !info.ShouldIncludeUsage {
+			*shouldSendLastResp = false
+		}
+	}
+
+	return nil
+}
+
+func handleFinalResponse(c *gin.Context, info *relaycommon.RelayInfo, lastStreamData string,
+	responseId string, createAt int64, model string, systemFingerprint string,
+	usage *dto.Usage, containStreamUsage bool) {
+
+	switch info.RelayFormat {
+	case relaycommon.RelayFormatOpenAI:
+		if info.ShouldIncludeUsage && !containStreamUsage {
+			response := helper.GenerateFinalUsageResponse(responseId, createAt, model, *usage)
+			response.SetSystemFingerprint(systemFingerprint)
+			helper.ObjectData(c, response)
+		}
+		helper.Done(c)
+
+	case relaycommon.RelayFormatClaude:
+		var streamResponse dto.ChatCompletionsStreamResponse
+		if err := json.Unmarshal(common.StringToByteSlice(lastStreamData), &streamResponse); err != nil {
+			common.SysError("error unmarshalling stream response: " + err.Error())
+			return
+		}
+
+		if !containStreamUsage {
+			streamResponse.Usage = usage
+		}
+
+		claudeResponses := service.StreamResponseOpenAI2Claude(&streamResponse, info)
+		for _, resp := range claudeResponses {
+			helper.ClaudeData(c, *resp)
+		}
+	}
+}

relay/channel/openai/relay-openai.go

@@ -12,7 +12,6 @@ import (
 	"one-api/constant"
 	"one-api/dto"
 	relaycommon "one-api/relay/common"
-	relayconstant "one-api/relay/constant"
 	"one-api/relay/helper"
 	"one-api/service"
 	"os"
@@ -137,10 +136,11 @@ func OaiStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rel
 
 	helper.StreamScannerHandler(c, resp, info, func(data string) bool {
 		if lastStreamData != "" {
-			err := sendStreamData(c, info, lastStreamData, forceFormat, thinkToContent)
+			err := handleStreamFormat(c, info, lastStreamData, forceFormat, thinkToContent)
 			if err != nil {
-				common.LogError(c, "streaming error: "+err.Error())
+				common.SysError("error handling stream format: " + err.Error())
 			}
+			info.SetFirstResponseTime()
 		}
 		lastStreamData = data
 		streamItems = append(streamItems, data)
@@ -172,83 +172,9 @@ func OaiStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rel
 		sendStreamData(c, info, lastStreamData, forceFormat, thinkToContent)
 	}
 
-	// 计算token
-	streamResp := "[" + strings.Join(streamItems, ",") + "]"
-	switch info.RelayMode {
-	case relayconstant.RelayModeChatCompletions:
-		var streamResponses []dto.ChatCompletionsStreamResponse
-		err := json.Unmarshal(common.StringToByteSlice(streamResp), &streamResponses)
-		if err != nil {
-			// 一次性解析失败，逐个解析
-			common.SysError("error unmarshalling stream response: " + err.Error())
-			for _, item := range streamItems {
-				var streamResponse dto.ChatCompletionsStreamResponse
-				err := json.Unmarshal(common.StringToByteSlice(item), &streamResponse)
-				if err == nil {
-					//if service.ValidUsage(streamResponse.Usage) {
-					//	usage = streamResponse.Usage
-					//}
-					for _, choice := range streamResponse.Choices {
-						responseTextBuilder.WriteString(choice.Delta.GetContentString())
-
-						// handle both reasoning_content and reasoning
-						responseTextBuilder.WriteString(choice.Delta.GetReasoningContent())
-
-						if choice.Delta.ToolCalls != nil {
-							if len(choice.Delta.ToolCalls) > toolCount {
-								toolCount = len(choice.Delta.ToolCalls)
-							}
-							for _, tool := range choice.Delta.ToolCalls {
-								responseTextBuilder.WriteString(tool.Function.Name)
-								responseTextBuilder.WriteString(tool.Function.Arguments)
-							}
-						}
-					}
-				}
-			}
-		} else {
-			for _, streamResponse := range streamResponses {
-				//if service.ValidUsage(streamResponse.Usage) {
-				//	usage = streamResponse.Usage
-				//	containStreamUsage = true
-				//}
-				for _, choice := range streamResponse.Choices {
-					responseTextBuilder.WriteString(choice.Delta.GetContentString())
-					responseTextBuilder.WriteString(choice.Delta.GetReasoningContent()) // This will handle both reasoning_content and reasoning
-					if choice.Delta.ToolCalls != nil {
-						if len(choice.Delta.ToolCalls) > toolCount {
-							toolCount = len(choice.Delta.ToolCalls)
-						}
-						for _, tool := range choice.Delta.ToolCalls {
-							responseTextBuilder.WriteString(tool.Function.Name)
-							responseTextBuilder.WriteString(tool.Function.Arguments)
-						}
-					}
-				}
-			}
-		}
-	case relayconstant.RelayModeCompletions:
-		var streamResponses []dto.CompletionsStreamResponse
-		err := json.Unmarshal(common.StringToByteSlice(streamResp), &streamResponses)
-		if err != nil {
-			// 一次性解析失败，逐个解析
-			common.SysError("error unmarshalling stream response: " + err.Error())
-			for _, item := range streamItems {
-				var streamResponse dto.CompletionsStreamResponse
-				err := json.Unmarshal(common.StringToByteSlice(item), &streamResponse)
-				if err == nil {
-					for _, choice := range streamResponse.Choices {
-						responseTextBuilder.WriteString(choice.Text)
-					}
-				}
-			}
-		} else {
-			for _, streamResponse := range streamResponses {
-				for _, choice := range streamResponse.Choices {
-					responseTextBuilder.WriteString(choice.Text)
-				}
-			}
-		}
+	// 处理token计算
+	if err := processTokens(info.RelayMode, streamItems, &responseTextBuilder, &toolCount); err != nil {
+		common.SysError("error processing tokens: " + err.Error())
 	}
 
 	if !containStreamUsage {
@@ -262,15 +188,8 @@ func OaiStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rel
 		}
 	}
 
-	if info.ShouldIncludeUsage && !containStreamUsage {
-		response := helper.GenerateFinalUsageResponse(responseId, createAt, model, *usage)
-		response.SetSystemFingerprint(systemFingerprint)
-		helper.ObjectData(c, response)
-	}
+	handleFinalResponse(c, info, lastStreamData, responseId, createAt, model, systemFingerprint, usage, containStreamUsage)
 
-	helper.Done(c)
-
-	//resp.Body.Close()
 	return nil, usage
 }