feat: 添加流模式下的SSE保活机制 #945

fix(relay): 优化数据流处理

.gitignore

@@ -9,4 +9,5 @@ logs
 web/dist
 .env
 one-api
-.DS_Store
+.DS_Store
+tiktoken_cache

common/init.go

@@ -73,25 +73,25 @@ func LoadEnv() {
 	DebugEnabled = os.Getenv("DEBUG") == "true"
 	MemoryCacheEnabled = os.Getenv("MEMORY_CACHE_ENABLED") == "true"
 	IsMasterNode = os.Getenv("NODE_TYPE") != "slave"
-	
+
 	// Parse requestInterval and set RequestInterval
 	requestInterval, _ = strconv.Atoi(os.Getenv("POLLING_INTERVAL"))
 	RequestInterval = time.Duration(requestInterval) * time.Second
-	
+
 	// Initialize variables with GetEnvOrDefault
 	SyncFrequency = GetEnvOrDefault("SYNC_FREQUENCY", 60)
 	BatchUpdateInterval = GetEnvOrDefault("BATCH_UPDATE_INTERVAL", 5)
 	RelayTimeout = GetEnvOrDefault("RELAY_TIMEOUT", 0)
-	
+
 	// Initialize string variables with GetEnvOrDefaultString
 	GeminiSafetySetting = GetEnvOrDefaultString("GEMINI_SAFETY_SETTING", "BLOCK_NONE")
 	CohereSafetySetting = GetEnvOrDefaultString("COHERE_SAFETY_SETTING", "NONE")
-	
+
 	// Initialize rate limit variables
 	GlobalApiRateLimitEnable = GetEnvOrDefaultBool("GLOBAL_API_RATE_LIMIT_ENABLE", true)
 	GlobalApiRateLimitNum = GetEnvOrDefault("GLOBAL_API_RATE_LIMIT", 180)
 	GlobalApiRateLimitDuration = int64(GetEnvOrDefault("GLOBAL_API_RATE_LIMIT_DURATION", 180))
-	
+
 	GlobalWebRateLimitEnable = GetEnvOrDefaultBool("GLOBAL_WEB_RATE_LIMIT_ENABLE", true)
 	GlobalWebRateLimitNum = GetEnvOrDefault("GLOBAL_WEB_RATE_LIMIT", 60)
 	GlobalWebRateLimitDuration = int64(GetEnvOrDefault("GLOBAL_WEB_RATE_LIMIT_DURATION", 180))

common/json.go

@@ -12,3 +12,7 @@ func DecodeJson(data []byte, v any) error {
 func DecodeJsonStr(data string, v any) error {
 	return DecodeJson(StringToByteSlice(data), v)
 }
+
+func EncodeJson(v any) ([]byte, error) {
+	return json.Marshal(v)
+}

docker-compose.yml

@@ -15,6 +15,7 @@ services:
       - SQL_DSN=root:123456@tcp(mysql:3306)/new-api  # Point to the mysql service
       - REDIS_CONN_STRING=redis://redis
       - TZ=Asia/Shanghai
+    #      - TIKTOKEN_CACHE_DIR=./tiktoken_cache  # 如果需要使用tiktoken_cache，请取消注释
     #      - SESSION_SECRET=random_string  # 多机部署时设置，必须修改这个随机字符串！！！！！！！
     #      - NODE_TYPE=slave  # Uncomment for slave node in multi-node deployment
     #      - SYNC_FREQUENCY=60  # Uncomment if regular database syncing is needed

dto/claude.go

@@ -7,7 +7,7 @@ type ClaudeMetadata struct {
 }
 
 type ClaudeMediaMessage struct {
-	Type        string               `json:"type"`
+	Type        string               `json:"type,omitempty"`
 	Text        *string              `json:"text,omitempty"`
 	Model       string               `json:"model,omitempty"`
 	Source      *ClaudeMessageSource `json:"source,omitempty"`
@@ -50,6 +50,11 @@ func (c *ClaudeMediaMessage) GetStringContent() string {
 	return ""
 }
 
+func (c *ClaudeMediaMessage) GetJsonRowString() string {
+	jsonContent, _ := json.Marshal(c)
+	return string(jsonContent)
+}
+
 func (c *ClaudeMediaMessage) SetContent(content any) {
 	jsonContent, _ := json.Marshal(content)
 	c.Content = jsonContent

dto/openai_request.go

@@ -111,6 +111,7 @@ type MediaContent struct {
 	Text       string `json:"text,omitempty"`
 	ImageUrl   any    `json:"image_url,omitempty"`
 	InputAudio any    `json:"input_audio,omitempty"`
+	File       any    `json:"file,omitempty"`
 }
 
 func (m *MediaContent) GetImageMedia() *MessageImageUrl {
@@ -120,6 +121,20 @@ func (m *MediaContent) GetImageMedia() *MessageImageUrl {
 	return nil
 }
 
+func (m *MediaContent) GetInputAudio() *MessageInputAudio {
+	if m.InputAudio != nil {
+		return m.InputAudio.(*MessageInputAudio)
+	}
+	return nil
+}
+
+func (m *MediaContent) GetFile() *MessageFile {
+	if m.File != nil {
+		return m.File.(*MessageFile)
+	}
+	return nil
+}
+
 type MessageImageUrl struct {
 	Url      string `json:"url"`
 	Detail   string `json:"detail"`
@@ -135,10 +150,17 @@ type MessageInputAudio struct {
 	Format string `json:"format"`
 }
 
+type MessageFile struct {
+	FileName string `json:"filename,omitempty"`
+	FileData string `json:"file_data,omitempty"`
+	FileId   string `json:"file_id,omitempty"`
+}
+
 const (
 	ContentTypeText       = "text"
 	ContentTypeImageURL   = "image_url"
 	ContentTypeInputAudio = "input_audio"
+	ContentTypeFile       = "file"
 )
 
 func (m *Message) GetPrefix() bool {
@@ -192,6 +214,12 @@ func (m *Message) StringContent() string {
 	return stringContent
 }
 
+func (m *Message) SetNullContent() {
+	m.Content = nil
+	m.parsedStringContent = nil
+	m.parsedContent = nil
+}
+
 func (m *Message) SetStringContent(content string) {
 	jsonContent, _ := json.Marshal(content)
 	m.Content = jsonContent
@@ -292,6 +320,30 @@ func (m *Message) ParseContent() []MediaContent {
 						})
 					}
 				}
+			case ContentTypeFile:
+				if fileData, ok := contentItem["file"].(map[string]interface{}); ok {
+					fileId, ok3 := fileData["file_id"].(string)
+					if ok3 {
+						contentList = append(contentList, MediaContent{
+							Type: ContentTypeFile,
+							File: &MessageFile{
+								FileId: fileId,
+							},
+						})
+					} else {
+						fileName, ok1 := fileData["filename"].(string)
+						fileDataStr, ok2 := fileData["file_data"].(string)
+						if ok1 && ok2 {
+							contentList = append(contentList, MediaContent{
+								Type: ContentTypeFile,
+								File: &MessageFile{
+									FileName: fileName,
+									FileData: fileDataStr,
+								},
+							})
+						}
+					}
+				}
 			}
 		}
 	}

dto/realtime.go

@@ -45,15 +45,16 @@ type RealtimeUsage struct {
 
 type InputTokenDetails struct {
 	CachedTokens         int `json:"cached_tokens"`
-	CachedCreationTokens int
+	CachedCreationTokens int `json:"-"`
 	TextTokens           int `json:"text_tokens"`
 	AudioTokens          int `json:"audio_tokens"`
 	ImageTokens          int `json:"image_tokens"`
 }
 
 type OutputTokenDetails struct {
-	TextTokens  int `json:"text_tokens"`
-	AudioTokens int `json:"audio_tokens"`
+	TextTokens      int `json:"text_tokens"`
+	AudioTokens     int `json:"audio_tokens"`
+	ReasoningTokens int `json:"reasoning_tokens"`
 }
 
 type RealtimeSession struct {

main.go

@@ -34,7 +34,7 @@ var indexPage []byte
 func main() {
 	err := godotenv.Load(".env")
 	if err != nil {
-		common.SysLog("Support for .env file is disabled")
+		common.SysLog("Support for .env file is disabled: " + err.Error())
 	}
 
 	common.LoadEnv()

relay/channel/dify/relay-dify.go

@@ -1,7 +1,6 @@
 package dify
 
 import (
-	"bufio"
 	"bytes"
 	"encoding/base64"
 	"encoding/json"
@@ -213,12 +212,8 @@ func streamResponseDify2OpenAI(difyResponse DifyChunkChatCompletionResponse) *dt
 func difyStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon.RelayInfo) (*dto.OpenAIErrorWithStatusCode, *dto.Usage) {
 	var responseText string
 	usage := &dto.Usage{}
-	scanner := bufio.NewScanner(resp.Body)
-	scanner.Split(bufio.ScanLines)
 	var nodeToken int
-
 	helper.SetEventStreamHeaders(c)
-
 	helper.StreamScannerHandler(c, resp, info, func(data string) bool {
 		var difyResponse DifyChunkChatCompletionResponse
 		err := json.Unmarshal([]byte(data), &difyResponse)
@@ -247,13 +242,10 @@ func difyStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon.Re
 		}
 		return true
 	})
-	if err := scanner.Err(); err != nil {
-		common.SysError("error reading stream: " + err.Error())
-	}
 	helper.Done(c)
 	err := resp.Body.Close()
 	if err != nil {
-		//return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil
+		// return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil
 		common.SysError("close_response_body_failed: " + err.Error())
 	}
 	if usage.TotalTokens == 0 {

relay/channel/gemini/relay-gemini.go

@@ -56,6 +56,7 @@ func CovertGemini2OpenAI(textRequest dto.GeneralOpenAIRequest) (*GeminiChatReque
 				continue
 			}
 			if tool.Function.Parameters != nil {
+
 				params, ok := tool.Function.Parameters.(map[string]interface{})
 				if ok {
 					if props, hasProps := params["properties"].(map[string]interface{}); hasProps {
@@ -65,6 +66,9 @@ func CovertGemini2OpenAI(textRequest dto.GeneralOpenAIRequest) (*GeminiChatReque
 					}
 				}
 			}
+			// Clean the parameters before appending
+			cleanedParams := cleanFunctionParameters(tool.Function.Parameters)
+			tool.Function.Parameters = cleanedParams
 			functions = append(functions, tool.Function)
 		}
 		if codeExecution {
@@ -86,11 +90,11 @@ func CovertGemini2OpenAI(textRequest dto.GeneralOpenAIRequest) (*GeminiChatReque
 		// json_data, _ := json.Marshal(geminiRequest.Tools)
 		// common.SysLog("tools_json: " + string(json_data))
 	} else if textRequest.Functions != nil {
-		geminiRequest.Tools = []GeminiChatTool{
-			{
-				FunctionDeclarations: textRequest.Functions,
-			},
-		}
+		//geminiRequest.Tools = []GeminiChatTool{
+		//	{
+		//		FunctionDeclarations: textRequest.Functions,
+		//	},
+		//}
 	}
 
 	if textRequest.ResponseFormat != nil && (textRequest.ResponseFormat.Type == "json_schema" || textRequest.ResponseFormat.Type == "json_object") {
@@ -204,6 +208,34 @@ func CovertGemini2OpenAI(textRequest dto.GeneralOpenAIRequest) (*GeminiChatReque
 						},
 					})
 				}
+			} else if part.Type == dto.ContentTypeFile {
+				if part.GetFile().FileId != "" {
+					return nil, fmt.Errorf("only base64 file is supported in gemini")
+				}
+				format, base64String, err := service.DecodeBase64FileData(part.GetFile().FileData)
+				if err != nil {
+					return nil, fmt.Errorf("decode base64 file data failed: %s", err.Error())
+				}
+				parts = append(parts, GeminiPart{
+					InlineData: &GeminiInlineData{
+						MimeType: format,
+						Data:     base64String,
+					},
+				})
+			} else if part.Type == dto.ContentTypeInputAudio {
+				if part.GetInputAudio().Data == "" {
+					return nil, fmt.Errorf("only base64 audio is supported in gemini")
+				}
+				format, base64String, err := service.DecodeBase64FileData(part.GetInputAudio().Data)
+				if err != nil {
+					return nil, fmt.Errorf("decode base64 audio data failed: %s", err.Error())
+				}
+				parts = append(parts, GeminiPart{
+					InlineData: &GeminiInlineData{
+						MimeType: format,
+						Data:     base64String,
+					},
+				})
 			}
 		}
 
@@ -229,6 +261,93 @@ func CovertGemini2OpenAI(textRequest dto.GeneralOpenAIRequest) (*GeminiChatReque
 	return &geminiRequest, nil
 }
 
+// cleanFunctionParameters recursively removes unsupported fields from Gemini function parameters.
+func cleanFunctionParameters(params interface{}) interface{} {
+	if params == nil {
+		return nil
+	}
+
+	paramMap, ok := params.(map[string]interface{})
+	if !ok {
+		// Not a map, return as is (e.g., could be an array or primitive)
+		return params
+	}
+
+	// Create a copy to avoid modifying the original
+	cleanedMap := make(map[string]interface{})
+	for k, v := range paramMap {
+		cleanedMap[k] = v
+	}
+
+	// Clean properties
+	if props, ok := cleanedMap["properties"].(map[string]interface{}); ok && props != nil {
+		cleanedProps := make(map[string]interface{})
+		for propName, propValue := range props {
+			propMap, ok := propValue.(map[string]interface{})
+			if !ok {
+				cleanedProps[propName] = propValue // Keep non-map properties
+				continue
+			}
+
+			// Create a copy of the property map
+			cleanedPropMap := make(map[string]interface{})
+			for k, v := range propMap {
+				cleanedPropMap[k] = v
+			}
+
+			// Remove unsupported fields
+			delete(cleanedPropMap, "default")
+			delete(cleanedPropMap, "exclusiveMaximum")
+			delete(cleanedPropMap, "exclusiveMinimum")
+
+			// Check and clean 'format' for string types
+			if propType, typeExists := cleanedPropMap["type"].(string); typeExists && propType == "string" {
+				if formatValue, formatExists := cleanedPropMap["format"].(string); formatExists {
+					if formatValue != "enum" && formatValue != "date-time" {
+						delete(cleanedPropMap, "format")
+					}
+				}
+			}
+
+			// Recursively clean nested properties within this property if it's an object/array
+			// Check the type before recursing
+			if propType, typeExists := cleanedPropMap["type"].(string); typeExists && (propType == "object" || propType == "array") {
+				cleanedProps[propName] = cleanFunctionParameters(cleanedPropMap)
+			} else {
+				cleanedProps[propName] = cleanedPropMap // Assign the cleaned map back if not recursing
+			}
+
+		}
+		cleanedMap["properties"] = cleanedProps
+	}
+
+	// Recursively clean items in arrays if needed (e.g., type: array, items: { ... })
+	if items, ok := cleanedMap["items"].(map[string]interface{}); ok && items != nil {
+		cleanedMap["items"] = cleanFunctionParameters(items)
+	}
+	// Also handle items if it's an array of schemas
+	if itemsArray, ok := cleanedMap["items"].([]interface{}); ok {
+		cleanedItemsArray := make([]interface{}, len(itemsArray))
+		for i, item := range itemsArray {
+			cleanedItemsArray[i] = cleanFunctionParameters(item)
+		}
+		cleanedMap["items"] = cleanedItemsArray
+	}
+
+	// Recursively clean other schema composition keywords if necessary
+	for _, field := range []string{"allOf", "anyOf", "oneOf"} {
+		if nested, ok := cleanedMap[field].([]interface{}); ok {
+			cleanedNested := make([]interface{}, len(nested))
+			for i, item := range nested {
+				cleanedNested[i] = cleanFunctionParameters(item)
+			}
+			cleanedMap[field] = cleanedNested
+		}
+	}
+
+	return cleanedMap
+}
+
 func removeAdditionalPropertiesWithDepth(schema interface{}, depth int) interface{} {
 	if depth >= 5 {
 		return schema

relay/channel/openai/adaptor.go

@@ -36,7 +36,7 @@ func (a *Adaptor) ConvertClaudeRequest(c *gin.Context, info *relaycommon.RelayIn
 	if !strings.Contains(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)
+	aiRequest, err := service.ClaudeToOpenAIRequest(*request, info)
 	if err != nil {
 		return nil, err
 	}

relay/channel/openai/helper.go

@@ -31,6 +31,9 @@ func handleClaudeFormat(c *gin.Context, data string, info *relaycommon.RelayInfo
 		return err
 	}
 
+	if streamResponse.Usage != nil {
+		info.ClaudeConvertInfo.Usage = streamResponse.Usage
+	}
 	claudeResponses := service.StreamResponseOpenAI2Claude(&streamResponse, info)
 	for _, resp := range claudeResponses {
 		helper.ClaudeData(c, *resp)
@@ -38,12 +41,7 @@ func handleClaudeFormat(c *gin.Context, data string, info *relaycommon.RelayInfo
 	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
-	}
-
+func ProcessStreamResponse(streamResponse dto.ChatCompletionsStreamResponse, responseTextBuilder *strings.Builder, toolCount *int) error {
 	for _, choice := range streamResponse.Choices {
 		responseTextBuilder.WriteString(choice.Delta.GetContentString())
 		responseTextBuilder.WriteString(choice.Delta.GetReasoningContent())
@@ -78,7 +76,11 @@ func processChatCompletions(streamResp string, streamItems []string, responseTex
 		// 一次性解析失败，逐个解析
 		common.SysError("error unmarshalling stream response: " + err.Error())
 		for _, item := range streamItems {
-			if err := processStreamResponse(item, responseTextBuilder, toolCount); err != nil {
+			var streamResponse dto.ChatCompletionsStreamResponse
+			if err := json.Unmarshal(common.StringToByteSlice(item), &streamResponse); err != nil {
+				return err
+			}
+			if err := ProcessStreamResponse(streamResponse, responseTextBuilder, toolCount); err != nil {
 				common.SysError("error processing stream response: " + err.Error())
 			}
 		}
@@ -170,15 +172,14 @@ func handleFinalResponse(c *gin.Context, info *relaycommon.RelayInfo, lastStream
 		helper.Done(c)
 
 	case relaycommon.RelayFormatClaude:
+		info.ClaudeConvertInfo.Done = true
 		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
-		}
+		info.ClaudeConvertInfo.Usage = usage
 
 		claudeResponses := service.StreamResponseOpenAI2Claude(&streamResponse, info)
 		for _, resp := range claudeResponses {

relay/channel/openai/relay-openai.go

@@ -117,6 +117,7 @@ func OaiStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rel
 	model := info.UpstreamModelName
 
 	var responseTextBuilder strings.Builder
+	var toolCount int
 	var usage = &dto.Usage{}
 	var streamItems []string // store stream items
 	var forceFormat bool
@@ -130,8 +131,6 @@ func OaiStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rel
 		thinkToContent = think2Content
 	}
 
-	toolCount := 0
-
 	var (
 		lastStreamData string
 	)
@@ -142,7 +141,6 @@ func OaiStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rel
 			if err != nil {
 				common.SysError("error handling stream format: " + err.Error())
 			}
-			info.SetFirstResponseTime()
 		}
 		lastStreamData = data
 		streamItems = append(streamItems, data)
@@ -170,8 +168,10 @@ func OaiStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rel
 			}
 		}
 	}
+
 	if shouldSendLastResp {
 		sendStreamData(c, info, lastStreamData, forceFormat, thinkToContent)
+		//err = handleStreamFormat(c, info, lastStreamData, forceFormat, thinkToContent)
 	}
 
 	// 处理token计算

relay/channel/xai/adaptor.go

@@ -8,7 +8,6 @@ import (
 	"net/http"
 	"one-api/dto"
 	"one-api/relay/channel"
-	"one-api/relay/channel/openai"
 	relaycommon "one-api/relay/common"
 	"strings"
 )
@@ -49,7 +48,6 @@ func (a *Adaptor) ConvertOpenAIRequest(c *gin.Context, info *relaycommon.RelayIn
 	if request == nil {
 		return nil, errors.New("request is nil")
 	}
-	request.StreamOptions = nil
 	if strings.HasPrefix(request.Model, "grok-3-mini") {
 		if request.MaxCompletionTokens == 0 && request.MaxTokens != 0 {
 			request.MaxCompletionTokens = request.MaxTokens
@@ -86,13 +84,13 @@ func (a *Adaptor) DoRequest(c *gin.Context, info *relaycommon.RelayInfo, request
 
 func (a *Adaptor) DoResponse(c *gin.Context, resp *http.Response, info *relaycommon.RelayInfo) (usage any, err *dto.OpenAIErrorWithStatusCode) {
 	if info.IsStream {
-		err, usage = openai.OaiStreamHandler(c, resp, info)
+		err, usage = xAIStreamHandler(c, resp, info)
 	} else {
-		err, usage = openai.OpenaiHandler(c, resp, info)
-	}
-	if _, ok := usage.(*dto.Usage); ok && usage != nil {
-		usage.(*dto.Usage).CompletionTokens = usage.(*dto.Usage).TotalTokens - usage.(*dto.Usage).PromptTokens
+		err, usage = xAIHandler(c, resp, info)
 	}
+	//if _, ok := usage.(*dto.Usage); ok && usage != nil {
+	//	usage.(*dto.Usage).CompletionTokens = usage.(*dto.Usage).TotalTokens - usage.(*dto.Usage).PromptTokens
+	//}
 
 	return
 }

relay/channel/xai/dto.go

@@ -0,0 +1,14 @@
+package xai
+
+import "one-api/dto"
+
+// ChatCompletionResponse represents the response from XAI chat completion API
+type ChatCompletionResponse struct {
+	Id                string `json:"id"`
+	Object            string `json:"object"`
+	Created           int64  `json:"created"`
+	Model             string `json:"model"`
+	Choices           []dto.ChatCompletionsStreamResponseChoice
+	Usage             *dto.Usage `json:"usage"`
+	SystemFingerprint string     `json:"system_fingerprint"`
+}

relay/channel/xai/text.go

@@ -0,0 +1,119 @@
+package xai
+
+import (
+	"bytes"
+	"encoding/json"
+	"github.com/gin-gonic/gin"
+	"io"
+	"net/http"
+	"one-api/common"
+	"one-api/dto"
+	"one-api/relay/channel/openai"
+	relaycommon "one-api/relay/common"
+	"one-api/relay/helper"
+	"one-api/service"
+	"strings"
+)
+
+func streamResponseXAI2OpenAI(xAIResp *dto.ChatCompletionsStreamResponse, usage *dto.Usage) *dto.ChatCompletionsStreamResponse {
+	if xAIResp == nil {
+		return nil
+	}
+	if xAIResp.Usage != nil {
+		xAIResp.Usage.CompletionTokens = usage.CompletionTokens
+	}
+	openAIResp := &dto.ChatCompletionsStreamResponse{
+		Id:      xAIResp.Id,
+		Object:  xAIResp.Object,
+		Created: xAIResp.Created,
+		Model:   xAIResp.Model,
+		Choices: xAIResp.Choices,
+		Usage:   xAIResp.Usage,
+	}
+
+	return openAIResp
+}
+
+func xAIStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon.RelayInfo) (*dto.OpenAIErrorWithStatusCode, *dto.Usage) {
+	usage := &dto.Usage{}
+	var responseTextBuilder strings.Builder
+	var toolCount int
+	var containStreamUsage bool
+
+	helper.SetEventStreamHeaders(c)
+
+	helper.StreamScannerHandler(c, resp, info, func(data string) bool {
+		var xAIResp *dto.ChatCompletionsStreamResponse
+		err := json.Unmarshal([]byte(data), &xAIResp)
+		if err != nil {
+			common.SysError("error unmarshalling stream response: " + err.Error())
+			return true
+		}
+
+		// 把 xAI 的usage转换为 OpenAI 的usage
+		if xAIResp.Usage != nil {
+			containStreamUsage = true
+			usage.PromptTokens = xAIResp.Usage.PromptTokens
+			usage.TotalTokens = xAIResp.Usage.TotalTokens
+			usage.CompletionTokens = usage.TotalTokens - usage.PromptTokens
+		}
+
+		openaiResponse := streamResponseXAI2OpenAI(xAIResp, usage)
+		_ = openai.ProcessStreamResponse(*openaiResponse, &responseTextBuilder, &toolCount)
+		err = helper.ObjectData(c, openaiResponse)
+		if err != nil {
+			common.SysError(err.Error())
+		}
+		return true
+	})
+
+	if !containStreamUsage {
+		usage, _ = service.ResponseText2Usage(responseTextBuilder.String(), info.UpstreamModelName, info.PromptTokens)
+		usage.CompletionTokens += toolCount * 7
+	}
+
+	helper.Done(c)
+	err := resp.Body.Close()
+	if err != nil {
+		//return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil
+		common.SysError("close_response_body_failed: " + err.Error())
+	}
+	return nil, usage
+}
+
+func xAIHandler(c *gin.Context, resp *http.Response, info *relaycommon.RelayInfo) (*dto.OpenAIErrorWithStatusCode, *dto.Usage) {
+	responseBody, err := io.ReadAll(resp.Body)
+	var response *dto.TextResponse
+	err = common.DecodeJson(responseBody, &response)
+	if err != nil {
+		common.SysError("error unmarshalling stream response: " + err.Error())
+		return nil, nil
+	}
+	response.Usage.CompletionTokens = response.Usage.TotalTokens - response.Usage.PromptTokens
+	response.Usage.CompletionTokenDetails.TextTokens = response.Usage.CompletionTokens - response.Usage.CompletionTokenDetails.ReasoningTokens
+
+	// new body
+	encodeJson, err := common.EncodeJson(response)
+	if err != nil {
+		common.SysError("error marshalling stream response: " + err.Error())
+		return nil, nil
+	}
+
+	// set new body
+	resp.Body = io.NopCloser(bytes.NewBuffer(encodeJson))
+
+	for k, v := range resp.Header {
+		c.Writer.Header().Set(k, v[0])
+	}
+	c.Writer.WriteHeader(resp.StatusCode)
+	_, err = io.Copy(c.Writer, resp.Body)
+	if err != nil {
+		return service.OpenAIErrorWrapper(err, "copy_response_body_failed", http.StatusInternalServerError), nil
+	}
+	err = resp.Body.Close()
+	if err != nil {
+		return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil
+	}
+
+	return nil, &response.Usage
+}

relay/channel/zhipu_4v/adaptor.go

@@ -10,6 +10,7 @@ import (
 	"one-api/relay/channel"
 	"one-api/relay/channel/openai"
 	relaycommon "one-api/relay/common"
+	relayconstant "one-api/relay/constant"
 )
 
 type Adaptor struct {
@@ -35,7 +36,13 @@ func (a *Adaptor) Init(info *relaycommon.RelayInfo) {
 }
 
 func (a *Adaptor) GetRequestURL(info *relaycommon.RelayInfo) (string, error) {
-	return fmt.Sprintf("%s/api/paas/v4/chat/completions", info.BaseUrl), nil
+	baseUrl := fmt.Sprintf("%s/api/paas/v4", info.BaseUrl)
+	switch info.RelayMode {
+	case relayconstant.RelayModeEmbeddings:
+		return fmt.Sprintf("%s/embeddings", baseUrl), nil
+	default:
+		return fmt.Sprintf("%s/chat/completions", baseUrl), nil
+	}
 }
 
 func (a *Adaptor) SetupRequestHeader(c *gin.Context, req *http.Header, info *relaycommon.RelayInfo) error {
@@ -60,8 +67,7 @@ func (a *Adaptor) ConvertRerankRequest(c *gin.Context, relayMode int, request dt
 }
 
 func (a *Adaptor) ConvertEmbeddingRequest(c *gin.Context, info *relaycommon.RelayInfo, request dto.EmbeddingRequest) (any, error) {
-	//TODO implement me
-	return nil, errors.New("not implemented")
+	return request, nil
 }
 
 func (a *Adaptor) DoRequest(c *gin.Context, info *relaycommon.RelayInfo, requestBody io.Reader) (any, error) {

relay/channel/zhipu_4v/relay-zhipu_v4.go

@@ -1,17 +1,9 @@
 package zhipu_4v
 
 import (
-	"bufio"
-	"bytes"
-	"encoding/json"
-	"github.com/gin-gonic/gin"
 	"github.com/golang-jwt/jwt"
-	"io"
-	"net/http"
 	"one-api/common"
 	"one-api/dto"
-	"one-api/relay/helper"
-	"one-api/service"
 	"strings"
 	"sync"
 	"time"
@@ -119,163 +111,3 @@ func requestOpenAI2Zhipu(request dto.GeneralOpenAIRequest) *dto.GeneralOpenAIReq
 		ToolChoice:  request.ToolChoice,
 	}
 }
-
-//func responseZhipu2OpenAI(response *dto.OpenAITextResponse) *dto.OpenAITextResponse {
-//	fullTextResponse := dto.OpenAITextResponse{
-//		Id:      response.Id,
-//		Object:  "chat.completion",
-//		Created: common.GetTimestamp(),
-//		Choices: make([]dto.OpenAITextResponseChoice, 0, len(response.TextResponseChoices)),
-//		Usage:   response.Usage,
-//	}
-//	for i, choice := range response.TextResponseChoices {
-//		content, _ := json.Marshal(strings.Trim(choice.Content, "\""))
-//		openaiChoice := dto.OpenAITextResponseChoice{
-//			Index: i,
-//			Message: dto.Message{
-//				Role:    choice.Role,
-//				Content: content,
-//			},
-//			FinishReason: "",
-//		}
-//		if i == len(response.TextResponseChoices)-1 {
-//			openaiChoice.FinishReason = "stop"
-//		}
-//		fullTextResponse.Choices = append(fullTextResponse.Choices, openaiChoice)
-//	}
-//	return &fullTextResponse
-//}
-
-func streamResponseZhipu2OpenAI(zhipuResponse *ZhipuV4StreamResponse) *dto.ChatCompletionsStreamResponse {
-	var choice dto.ChatCompletionsStreamResponseChoice
-	choice.Delta.Content = zhipuResponse.Choices[0].Delta.Content
-	choice.Delta.Role = zhipuResponse.Choices[0].Delta.Role
-	choice.Delta.ToolCalls = zhipuResponse.Choices[0].Delta.ToolCalls
-	choice.Index = zhipuResponse.Choices[0].Index
-	choice.FinishReason = zhipuResponse.Choices[0].FinishReason
-	response := dto.ChatCompletionsStreamResponse{
-		Id:      zhipuResponse.Id,
-		Object:  "chat.completion.chunk",
-		Created: zhipuResponse.Created,
-		Model:   "glm-4v",
-		Choices: []dto.ChatCompletionsStreamResponseChoice{choice},
-	}
-	return &response
-}
-
-func lastStreamResponseZhipuV42OpenAI(zhipuResponse *ZhipuV4StreamResponse) (*dto.ChatCompletionsStreamResponse, *dto.Usage) {
-	response := streamResponseZhipu2OpenAI(zhipuResponse)
-	return response, &zhipuResponse.Usage
-}
-
-func zhipuStreamHandler(c *gin.Context, resp *http.Response) (*dto.OpenAIErrorWithStatusCode, *dto.Usage) {
-	var usage *dto.Usage
-	scanner := bufio.NewScanner(resp.Body)
-	scanner.Split(func(data []byte, atEOF bool) (advance int, token []byte, err error) {
-		if atEOF && len(data) == 0 {
-			return 0, nil, nil
-		}
-		if i := strings.Index(string(data), "\n"); i >= 0 {
-			return i + 1, data[0:i], nil
-		}
-		if atEOF {
-			return len(data), data, nil
-		}
-		return 0, nil, nil
-	})
-	dataChan := make(chan string)
-	stopChan := make(chan bool)
-	go func() {
-		for scanner.Scan() {
-			data := scanner.Text()
-			if len(data) < 6 { // ignore blank line or wrong format
-				continue
-			}
-			if data[:6] != "data: " && data[:6] != "[DONE]" {
-				continue
-			}
-			dataChan <- data
-		}
-		stopChan <- true
-	}()
-	helper.SetEventStreamHeaders(c)
-	c.Stream(func(w io.Writer) bool {
-		select {
-		case data := <-dataChan:
-			if strings.HasPrefix(data, "data: [DONE]") {
-				data = data[:12]
-			}
-			// some implementations may add \r at the end of data
-			data = strings.TrimSuffix(data, "\r")
-
-			var streamResponse ZhipuV4StreamResponse
-			err := json.Unmarshal([]byte(data), &streamResponse)
-			if err != nil {
-				common.SysError("error unmarshalling stream response: " + err.Error())
-			}
-			var response *dto.ChatCompletionsStreamResponse
-			if strings.Contains(data, "prompt_tokens") {
-				response, usage = lastStreamResponseZhipuV42OpenAI(&streamResponse)
-			} else {
-				response = streamResponseZhipu2OpenAI(&streamResponse)
-			}
-			jsonResponse, err := json.Marshal(response)
-			if err != nil {
-				common.SysError("error marshalling stream response: " + err.Error())
-				return true
-			}
-			c.Render(-1, common.CustomEvent{Data: "data: " + string(jsonResponse)})
-			return true
-		case <-stopChan:
-			return false
-		}
-	})
-	err := resp.Body.Close()
-	if err != nil {
-		return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil
-	}
-	return nil, usage
-}
-
-func zhipuHandler(c *gin.Context, resp *http.Response) (*dto.OpenAIErrorWithStatusCode, *dto.Usage) {
-	var textResponse ZhipuV4Response
-	responseBody, err := io.ReadAll(resp.Body)
-	if err != nil {
-		return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil
-	}
-	err = resp.Body.Close()
-	if err != nil {
-		return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil
-	}
-	err = json.Unmarshal(responseBody, &textResponse)
-	if err != nil {
-		return service.OpenAIErrorWrapper(err, "unmarshal_response_body_failed", http.StatusInternalServerError), nil
-	}
-	if textResponse.Error.Type != "" {
-		return &dto.OpenAIErrorWithStatusCode{
-			Error:      textResponse.Error,
-			StatusCode: resp.StatusCode,
-		}, nil
-	}
-	// Reset response body
-	resp.Body = io.NopCloser(bytes.NewBuffer(responseBody))
-
-	// We shouldn't set the header before we parse the response body, because the parse part may fail.
-	// And then we will have to send an error response, but in this case, the header has already been set.
-	// So the HTTPClient will be confused by the response.
-	// For example, Postman will report error, and we cannot check the response at all.
-	for k, v := range resp.Header {
-		c.Writer.Header().Set(k, v[0])
-	}
-	c.Writer.WriteHeader(resp.StatusCode)
-	_, err = io.Copy(c.Writer, resp.Body)
-	if err != nil {
-		return service.OpenAIErrorWrapper(err, "copy_response_body_failed", http.StatusInternalServerError), nil
-	}
-	err = resp.Body.Close()
-	if err != nil {
-		return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil
-	}
-
-	return nil, &textResponse.Usage
-}

relay/common/relay_info.go

@@ -6,6 +6,7 @@ import (
 	"one-api/dto"
 	relayconstant "one-api/relay/constant"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/gin-gonic/gin"
@@ -19,13 +20,18 @@ type ThinkingContentInfo struct {
 }
 
 const (
-	LastMessageTypeText  = "text"
-	LastMessageTypeTools = "tools"
+	LastMessageTypeNone     = "none"
+	LastMessageTypeText     = "text"
+	LastMessageTypeTools    = "tools"
+	LastMessageTypeThinking = "thinking"
 )
 
 type ClaudeConvertInfo struct {
 	LastMessagesType string
 	Index            int
+	Usage            *dto.Usage
+	FinishReason     string
+	Done             bool
 }
 
 const (
@@ -49,6 +55,7 @@ type RelayInfo struct {
 	StartTime         time.Time
 	FirstResponseTime time.Time
 	isFirstResponse   bool
+	responseMutex     sync.Mutex // Add mutex for protecting concurrent access
 	//SendLastReasoningResponse bool
 	ApiType           int
 	IsStream          bool
@@ -97,6 +104,7 @@ var streamSupportedChannels = map[int]bool{
 	common.ChannelTypeAzure:      true,
 	common.ChannelTypeVolcEngine: true,
 	common.ChannelTypeOllama:     true,
+	common.ChannelTypeXai:        true,
 }
 
 func GenRelayInfoWs(c *gin.Context, ws *websocket.Conn) *RelayInfo {
@@ -113,7 +121,7 @@ func GenRelayInfoClaude(c *gin.Context) *RelayInfo {
 	info.RelayFormat = RelayFormatClaude
 	info.ShouldIncludeUsage = false
 	info.ClaudeConvertInfo = ClaudeConvertInfo{
-		LastMessagesType: LastMessageTypeText,
+		LastMessagesType: LastMessageTypeNone,
 	}
 	return info
 }
@@ -206,12 +214,19 @@ func (info *RelayInfo) SetIsStream(isStream bool) {
 }
 
 func (info *RelayInfo) SetFirstResponseTime() {
+	info.responseMutex.Lock()
+	defer info.responseMutex.Unlock()
+
 	if info.isFirstResponse {
 		info.FirstResponseTime = time.Now()
 		info.isFirstResponse = false
 	}
 }
 
+func (info *RelayInfo) HasSendResponse() bool {
+	return info.FirstResponseTime.After(info.StartTime)
+}
+
 type TaskRelayInfo struct {
 	*RelayInfo
 	Action       string

relay/helper/common.go

@@ -55,7 +55,20 @@ func StringData(c *gin.Context, str string) error {
 	return nil
 }
 
+func PingData(c *gin.Context) error {
+	c.Writer.Write([]byte(": PING\n\n"))
+	if flusher, ok := c.Writer.(http.Flusher); ok {
+		flusher.Flush()
+	} else {
+		return errors.New("streaming error: flusher not found")
+	}
+	return nil
+}
+
 func ObjectData(c *gin.Context, object interface{}) error {
+	if object == nil {
+		return errors.New("object is nil")
+	}
 	jsonData, err := json.Marshal(object)
 	if err != nil {
 		return fmt.Errorf("error marshalling object: %w", err)

relay/helper/stream_scanner.go

@@ -3,20 +3,29 @@ package helper
 import (
 	"bufio"
 	"context"
+	"github.com/bytedance/gopkg/util/gopool"
 	"io"
 	"net/http"
 	"one-api/common"
 	"one-api/constant"
 	relaycommon "one-api/relay/common"
+	"one-api/setting/operation_setting"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/gin-gonic/gin"
 )
 
+const (
+	InitialScannerBufferSize = 1 << 20  // 1MB (1*1024*1024)
+	MaxScannerBufferSize     = 10 << 20 // 10MB (10*1024*1024)
+	DefaultPingInterval      = 10 * time.Second
+)
+
 func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon.RelayInfo, dataHandler func(data string) bool) {
 
-	if resp == nil {
+	if resp == nil || dataHandler == nil {
 		return
 	}
 
@@ -29,16 +38,32 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon
 	}
 
 	var (
-		stopChan = make(chan bool, 2)
-		scanner  = bufio.NewScanner(resp.Body)
-		ticker   = time.NewTicker(streamingTimeout)
+		stopChan   = make(chan bool, 2)
+		scanner    = bufio.NewScanner(resp.Body)
+		ticker     = time.NewTicker(streamingTimeout)
+		pingTicker *time.Ticker
+		writeMutex sync.Mutex // Mutex to protect concurrent writes
 	)
 
+	generalSettings := operation_setting.GetGeneralSetting()
+	pingEnabled := generalSettings.PingIntervalEnabled
+	pingInterval := time.Duration(generalSettings.PingIntervalSeconds) * time.Second
+	if pingInterval <= 0 {
+		pingInterval = DefaultPingInterval
+	}
+
+	if pingEnabled {
+		pingTicker = time.NewTicker(pingInterval)
+	}
+
 	defer func() {
 		ticker.Stop()
+		if pingTicker != nil {
+			pingTicker.Stop()
+		}
 		close(stopChan)
 	}()
-
+	scanner.Buffer(make([]byte, InitialScannerBufferSize), MaxScannerBufferSize)
 	scanner.Split(bufio.ScanLines)
 	SetEventStreamHeaders(c)
 
@@ -46,6 +71,34 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon
 	defer cancel()
 
 	ctx = context.WithValue(ctx, "stop_chan", stopChan)
+
+	// Handle ping data sending
+	if pingEnabled && pingTicker != nil {
+		gopool.Go(func() {
+			for {
+				select {
+				case <-pingTicker.C:
+					writeMutex.Lock() // Lock before writing
+					err := PingData(c)
+					writeMutex.Unlock() // Unlock after writing
+					if err != nil {
+						common.LogError(c, "ping data error: "+err.Error())
+						common.SafeSendBool(stopChan, true)
+						return
+					}
+					if common.DebugEnabled {
+						println("ping data sent")
+					}
+				case <-ctx.Done():
+					if common.DebugEnabled {
+						println("ping data goroutine stopped")
+					}
+					return
+				}
+			}
+		})
+	}
+
 	common.RelayCtxGo(ctx, func() {
 		for scanner.Scan() {
 			ticker.Reset(streamingTimeout)
@@ -65,7 +118,9 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon
 			data = strings.TrimSuffix(data, "\"")
 			if !strings.HasPrefix(data, "[DONE]") {
 				info.SetFirstResponseTime()
+				writeMutex.Lock() // Lock before writing
 				success := dataHandler(data)
+				writeMutex.Unlock() // Unlock after writing
 				if !success {
 					break
 				}
@@ -85,7 +140,9 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon
 	case <-ticker.C:
 		// 超时处理逻辑
 		common.LogError(c, "streaming timeout")
+		common.SafeSendBool(stopChan, true)
 	case <-stopChan:
 		// 正常结束
+		common.LogInfo(c, "streaming finished")
 	}
 }

service/convert.go

@@ -6,9 +6,10 @@ import (
 	"one-api/common"
 	"one-api/dto"
 	relaycommon "one-api/relay/common"
+	"strings"
 )
 
-func ClaudeToOpenAIRequest(claudeRequest dto.ClaudeRequest) (*dto.GeneralOpenAIRequest, error) {
+func ClaudeToOpenAIRequest(claudeRequest dto.ClaudeRequest, info *relaycommon.RelayInfo) (*dto.GeneralOpenAIRequest, error) {
 	openAIRequest := dto.GeneralOpenAIRequest{
 		Model:       claudeRequest.Model,
 		MaxTokens:   claudeRequest.MaxTokens,
@@ -17,6 +18,13 @@ func ClaudeToOpenAIRequest(claudeRequest dto.ClaudeRequest) (*dto.GeneralOpenAIR
 		Stream:      claudeRequest.Stream,
 	}
 
+	if claudeRequest.Thinking != nil {
+		if strings.HasSuffix(info.OriginModelName, "-thinking") &&
+			!strings.HasSuffix(claudeRequest.Model, "-thinking") {
+			openAIRequest.Model = openAIRequest.Model + "-thinking"
+		}
+	}
+
 	// Convert stop sequences
 	if len(claudeRequest.StopSequences) == 1 {
 		openAIRequest.Stop = claudeRequest.StopSequences[0]
@@ -45,7 +53,7 @@ func ClaudeToOpenAIRequest(claudeRequest dto.ClaudeRequest) (*dto.GeneralOpenAIR
 
 	// Add system message if present
 	if claudeRequest.System != nil {
-		if claudeRequest.IsStringSystem() {
+		if claudeRequest.IsStringSystem() && claudeRequest.GetStringSystem() != "" {
 			openAIMessage := dto.Message{
 				Role: "system",
 			}
@@ -122,23 +130,22 @@ func ClaudeToOpenAIRequest(claudeRequest dto.ClaudeRequest) (*dto.GeneralOpenAIR
 						oaiToolMessage.SetStringContent(mediaMsg.GetStringContent())
 					} else {
 						mediaContents := mediaMsg.ParseMediaContent()
-						if len(mediaContents) > 0 && mediaContents[0].Text != nil {
-							oaiToolMessage.SetStringContent(*mediaContents[0].Text)
-						}
+						encodeJson, _ := common.EncodeJson(mediaContents)
+						oaiToolMessage.SetStringContent(string(encodeJson))
 					}
 					openAIMessages = append(openAIMessages, oaiToolMessage)
 				}
 			}
 
-			if len(mediaMessages) > 0 {
-				openAIMessage.SetMediaContent(mediaMessages)
-			}
-
 			if len(toolCalls) > 0 {
 				openAIMessage.SetToolCalls(toolCalls)
 			}
+
+			if len(mediaMessages) > 0 && len(toolCalls) == 0 {
+				openAIMessage.SetMediaContent(mediaMessages)
+			}
 		}
-		if len(openAIMessage.ParseContent()) > 0 {
+		if len(openAIMessage.ParseContent()) > 0 || len(openAIMessage.ToolCalls) > 0 {
 			openAIMessages = append(openAIMessages, openAIMessage)
 		}
 	}
@@ -211,15 +218,15 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon
 			resp.SetIndex(0)
 			claudeResponses = append(claudeResponses, resp)
 		} else {
-			resp := &dto.ClaudeResponse{
-				Type: "content_block_start",
-				ContentBlock: &dto.ClaudeMediaMessage{
-					Type: "text",
-					Text: common.GetPointer[string](""),
-				},
-			}
-			resp.SetIndex(0)
-			claudeResponses = append(claudeResponses, resp)
+			//resp := &dto.ClaudeResponse{
+			//	Type: "content_block_start",
+			//	ContentBlock: &dto.ClaudeMediaMessage{
+			//		Type: "text",
+			//		Text: common.GetPointer[string](""),
+			//	},
+			//}
+			//resp.SetIndex(0)
+			//claudeResponses = append(claudeResponses, resp)
 		}
 		return claudeResponses
 	}
@@ -232,16 +239,20 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon
 		chosenChoice := openAIResponse.Choices[0]
 		if chosenChoice.FinishReason != nil && *chosenChoice.FinishReason != "" {
 			// should be done
+			info.FinishReason = *chosenChoice.FinishReason
+			return claudeResponses
+		}
+		if info.Done {
 			claudeResponses = append(claudeResponses, generateStopBlock(info.ClaudeConvertInfo.Index))
-			if openAIResponse.Usage != nil {
+			if info.ClaudeConvertInfo.Usage != nil {
 				claudeResponses = append(claudeResponses, &dto.ClaudeResponse{
 					Type: "message_delta",
 					Usage: &dto.ClaudeUsage{
-						InputTokens:  openAIResponse.Usage.PromptTokens,
-						OutputTokens: openAIResponse.Usage.CompletionTokens,
+						InputTokens:  info.ClaudeConvertInfo.Usage.PromptTokens,
+						OutputTokens: info.ClaudeConvertInfo.Usage.CompletionTokens,
 					},
 					Delta: &dto.ClaudeMediaMessage{
-						StopReason: common.GetPointer[string](stopReasonOpenAI2Claude(*chosenChoice.FinishReason)),
+						StopReason: common.GetPointer[string](stopReasonOpenAI2Claude(info.FinishReason)),
 					},
 				})
 			}
@@ -250,10 +261,10 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon
 			})
 		} else {
 			var claudeResponse dto.ClaudeResponse
-			claudeResponse.SetIndex(0)
+			var isEmpty bool
 			claudeResponse.Type = "content_block_delta"
 			if len(chosenChoice.Delta.ToolCalls) > 0 {
-				if info.ClaudeConvertInfo.LastMessagesType == relaycommon.LastMessageTypeText {
+				if info.ClaudeConvertInfo.LastMessagesType != relaycommon.LastMessageTypeTools {
 					claudeResponses = append(claudeResponses, generateStopBlock(info.ClaudeConvertInfo.Index))
 					info.ClaudeConvertInfo.Index++
 					claudeResponses = append(claudeResponses, &dto.ClaudeResponse{
@@ -274,15 +285,57 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon
 					PartialJson: &chosenChoice.Delta.ToolCalls[0].Function.Arguments,
 				}
 			} else {
-				info.ClaudeConvertInfo.LastMessagesType = relaycommon.LastMessageTypeText
-				// text delta
-				claudeResponse.Delta = &dto.ClaudeMediaMessage{
-					Type: "text_delta",
-					Text: common.GetPointer[string](chosenChoice.Delta.GetContentString()),
+				reasoning := chosenChoice.Delta.GetReasoningContent()
+				textContent := chosenChoice.Delta.GetContentString()
+				if reasoning != "" || textContent != "" {
+					if reasoning != "" {
+						if info.ClaudeConvertInfo.LastMessagesType != relaycommon.LastMessageTypeThinking {
+							//info.ClaudeConvertInfo.Index++
+							claudeResponses = append(claudeResponses, &dto.ClaudeResponse{
+								Index: &info.ClaudeConvertInfo.Index,
+								Type:  "content_block_start",
+								ContentBlock: &dto.ClaudeMediaMessage{
+									Type:     "thinking",
+									Thinking: "",
+								},
+							})
+						}
+						info.ClaudeConvertInfo.LastMessagesType = relaycommon.LastMessageTypeThinking
+						// text delta
+						claudeResponse.Delta = &dto.ClaudeMediaMessage{
+							Type:     "thinking_delta",
+							Thinking: reasoning,
+						}
+					} else {
+						if info.ClaudeConvertInfo.LastMessagesType != relaycommon.LastMessageTypeText {
+							if info.LastMessagesType == relaycommon.LastMessageTypeThinking || info.LastMessagesType == relaycommon.LastMessageTypeTools {
+								claudeResponses = append(claudeResponses, generateStopBlock(info.ClaudeConvertInfo.Index))
+								info.ClaudeConvertInfo.Index++
+							}
+							claudeResponses = append(claudeResponses, &dto.ClaudeResponse{
+								Index: &info.ClaudeConvertInfo.Index,
+								Type:  "content_block_start",
+								ContentBlock: &dto.ClaudeMediaMessage{
+									Type: "text",
+									Text: common.GetPointer[string](""),
+								},
+							})
+						}
+						info.ClaudeConvertInfo.LastMessagesType = relaycommon.LastMessageTypeText
+						// text delta
+						claudeResponse.Delta = &dto.ClaudeMediaMessage{
+							Type: "text_delta",
+							Text: common.GetPointer[string](textContent),
+						}
+					}
+				} else {
+					isEmpty = true
 				}
 			}
 			claudeResponse.Index = &info.ClaudeConvertInfo.Index
-			claudeResponses = append(claudeResponses, &claudeResponse)
+			if !isEmpty {
+				claudeResponses = append(claudeResponses, &claudeResponse)
+			}
 		}
 	}
 

service/file_decoder.go

@@ -8,9 +8,9 @@ import (
 	"one-api/dto"
 )
 
-var maxFileSize = constant.MaxFileDownloadMB * 1024 * 1024
-
 func GetFileBase64FromUrl(url string) (*dto.LocalFileData, error) {
+	var maxFileSize = constant.MaxFileDownloadMB * 1024 * 1024
+
 	resp, err := DoDownloadRequest(url)
 	if err != nil {
 		return nil, err
@@ -22,7 +22,6 @@ func GetFileBase64FromUrl(url string) (*dto.LocalFileData, error) {
 	if err != nil {
 		return nil, err
 	}
-
 	// Check actual size after reading
 	if len(fileBytes) > maxFileSize {
 		return nil, fmt.Errorf("file size exceeds maximum allowed size: %dMB", constant.MaxFileDownloadMB)

service/token_counter.go

@@ -398,6 +398,8 @@ func CountTokenMessages(info *relaycommon.RelayInfo, messages []dto.Message, mod
 				} else if m.Type == dto.ContentTypeInputAudio {
 					// TODO: 音频token数量计算
 					tokenNum += 100
+				} else if m.Type == dto.ContentTypeFile {
+					tokenNum += 5000
 				} else {
 					tokenNum += getTokenNum(tokenEncoder, m.Text)
 				}

setting/operation_setting/general_setting.go

@@ -3,12 +3,16 @@ package operation_setting
 import "one-api/setting/config"
 
 type GeneralSetting struct {
-	DocsLink string `json:"docs_link"`
+	DocsLink            string `json:"docs_link"`
+	PingIntervalEnabled bool   `json:"ping_interval_enabled"`
+	PingIntervalSeconds int    `json:"ping_interval_seconds"`
 }
 
 // 默认配置
 var generalSetting = GeneralSetting{
-	DocsLink: "https://docs.newapi.pro",
+	DocsLink:            "https://docs.newapi.pro",
+	PingIntervalEnabled: false,
+	PingIntervalSeconds: 60,
 }
 
 func init() {

web/src/components/ModelSetting.js

@@ -18,6 +18,8 @@ const ModelSetting = () => {
     'claude.default_max_tokens': '',
     'claude.thinking_adapter_budget_tokens_percentage': 0.8,
     'global.pass_through_request_enabled': false,
+    'general_setting.ping_interval_enabled': false,
+    'general_setting.ping_interval_seconds': 60,
   });
 
   let [loading, setLoading] = useState(false);

web/src/components/PersonalSetting.js

@@ -793,23 +793,7 @@ const PersonalSetting = () => {
               </div>
             </Card>
             <Card style={{ marginTop: 10 }}>
-              <Tabs type="line" defaultActiveKey="price">
-                <TabPane tab={t('价格设置')} itemKey="price">
-                  <div style={{ marginTop: 20 }}>
-                    <Typography.Text strong>{t('接受未设置价格模型')}</Typography.Text>
-                    <div style={{ marginTop: 10 }}>
-                      <Checkbox
-                        checked={notificationSettings.acceptUnsetModelRatioModel}
-                        onChange={e => handleNotificationSettingChange('acceptUnsetModelRatioModel', e.target.checked)}
-                      >
-                        {t('接受未设置价格模型')}
-                      </Checkbox>
-                      <Typography.Text type="secondary" style={{ marginTop: 8, display: 'block' }}>
-                        {t('当模型没有设置价格时仍接受调用，仅当您信任该网站时使用，可能会产生高额费用')}
-                      </Typography.Text>
-                    </div>
-                  </div>
-                </TabPane>
+              <Tabs type="line" defaultActiveKey="notification">
                 <TabPane tab={t('通知设置')} itemKey="notification">
                   <div style={{ marginTop: 20 }}>
                     <Typography.Text strong>{t('通知方式')}</Typography.Text>
@@ -923,6 +907,23 @@ const PersonalSetting = () => {
                     </Typography.Text>
                   </div>
                 </TabPane>
+                <TabPane tab={t('价格设置')} itemKey="price">
+                  <div style={{ marginTop: 20 }}>
+                    <Typography.Text strong>{t('接受未设置价格模型')}</Typography.Text>
+                    <div style={{ marginTop: 10 }}>
+                      <Checkbox
+                        checked={notificationSettings.acceptUnsetModelRatioModel}
+                        onChange={e => handleNotificationSettingChange('acceptUnsetModelRatioModel', e.target.checked)}
+                      >
+                        {t('接受未设置价格模型')}
+                      </Checkbox>
+                      <Typography.Text type="secondary" style={{ marginTop: 8, display: 'block' }}>
+                        {t('当模型没有设置价格时仍接受调用，仅当您信任该网站时使用，可能会产生高额费用')}
+                      </Typography.Text>
+                    </div>
+                  </div>
+                </TabPane>
+                
               </Tabs>
               <div style={{ marginTop: 20 }}>
                 <Button type="primary" onClick={saveNotificationSettings}>

web/src/i18n/locales/en.json

@@ -492,7 +492,7 @@
   "请输入默认 API 版本，例如：2023-03-15-preview，该配置可以被实际的请求查询参数所覆盖": "Please enter the default API version, for example: 2023-03-15-preview, this configuration can be overridden by the actual request query parameters",
   "默认": "default",
   "图片演示": "Image demo",
-  "参数替换为你的部署名称（模型名称中的点会被剔除）": "Replace the parameter with your deployment name (dots in the model name will be removed)",
+  "注意，系统请求的时模型名称中的点会被剔除，例如：gpt-4.5-preview会请求为gpt-45-preview，所以部署的模型名称需要去掉点": "Note that the dot in the model name requested by the system will be removed, for example: gpt-4.5-preview will be requested as gpt-45-preview, so the deployed model name needs to remove the dot",
   "模型映射必须是合法的 JSON 格式！": "Model mapping must be in valid JSON format!",
   "取消无限额度": "Cancel unlimited quota",
   "取消": "Cancel",

web/src/pages/Channel/EditChannel.js

@@ -473,7 +473,7 @@ const EditChannel = (props) => {
               <div style={{ marginTop: 10 }}>
                 <Banner
                   type={'warning'}
-                  description={t('注意，模型部署名称必须和模型名称保持一致')}
+                  description={t('注意，系统请求的时模型名称中的点会被剔除，例如：gpt-4.5-preview会请求为gpt-45-preview，所以部署的模型名称需要去掉点')}
                 ></Banner>
               </div>
               <div style={{ marginTop: 10 }}>

web/src/pages/Setting/Model/SettingGlobalModel.js

@@ -1,5 +1,5 @@
 import React, { useEffect, useState, useRef } from 'react';
-import { Button, Col, Form, Row, Spin } from '@douyinfe/semi-ui';
+import { Button, Col, Form, Row, Spin, Banner } from '@douyinfe/semi-ui';
 import {
   compareObjects,
   API,
@@ -15,6 +15,8 @@ export default function SettingGlobalModel(props) {
   const [loading, setLoading] = useState(false);
   const [inputs, setInputs] = useState({
     'global.pass_through_request_enabled': false,
+    'general_setting.ping_interval_enabled': false,
+    'general_setting.ping_interval_seconds': 60,
   });
   const refForm = useRef();
   const [inputsRow, setInputsRow] = useState(inputs);
@@ -23,12 +25,8 @@ export default function SettingGlobalModel(props) {
     const updateArray = compareObjects(inputs, inputsRow);
     if (!updateArray.length) return showWarning(t('你似乎并没有修改什么'));
     const requestQueue = updateArray.map((item) => {
-      let value = '';
-      if (typeof inputs[item.key] === 'boolean') {
-        value = String(inputs[item.key]);
-      } else {
-        value = inputs[item.key];
-      }
+      let value = String(inputs[item.key]);
+
       return API.put('/api/option/', {
         key: item.key,
         value,
@@ -84,6 +82,36 @@ export default function SettingGlobalModel(props) {
                 />
               </Col>
             </Row>
+            
+            <Form.Section text={t('连接保活设置')}>
+            <Row style={{ marginTop: 10 }}>
+                  <Col span={24}>
+                    <Banner 
+                      type="warning"
+                      description="警告：启用保活后，如果已经写入保活数据后渠道出错，系统无法重试，如果必须开启，推荐设置尽可能大的Ping间隔"
+                    />
+                  </Col>
+                </Row>
+              <Row>
+                <Col xs={24} sm={12} md={8} lg={8} xl={8}>
+                  <Form.Switch
+                    label={t('启用Ping间隔')}
+                    field={'general_setting.ping_interval_enabled'}
+                    onChange={(value) => setInputs({ ...inputs, 'general_setting.ping_interval_enabled': value })}
+                    extraText={'开启后，将定期发送ping数据保持连接活跃'}
+                  />
+                </Col>
+                <Col xs={24} sm={12} md={8} lg={8} xl={8}>
+                  <Form.InputNumber
+                    label={t('Ping间隔（秒）')}
+                    field={'general_setting.ping_interval_seconds'}
+                    onChange={(value) => setInputs({ ...inputs, 'general_setting.ping_interval_seconds': value })}
+                    min={1}
+                    disabled={!inputs['general_setting.ping_interval_enabled']}
+                  />
+                </Col>
+              </Row>
+            </Form.Section>
 
             <Row>
               <Button size='default' onClick={onSubmit}>