From 9328b907f26261e9d1b540709fe93a49e1c9e395 Mon Sep 17 00:00:00 2001
From: quran <quran@xiaomi.com>
Date: Thu, 10 Apr 2025 16:56:16 +0800
Subject: [PATCH] =?UTF-8?q?fix(relay):=20=E4=BC=98=E5=8C=96=E6=95=B0?=
 =?UTF-8?q?=E6=8D=AE=E6=B5=81=E5=A4=84=E7=90=86?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 移除了 bufio 的无效使用
- 在 StreamScannerHandler 中增加了初始和最大缓冲区大小的常量设置
- 调整 StreamScannerHandler 中缓冲区大小，避免出现token too long报错
---
 relay/channel/dify/relay-dify.go | 10 +---------
 relay/helper/stream_scanner.go   |  7 ++++++-
 2 files changed, 7 insertions(+), 10 deletions(-)

diff --git a/relay/channel/dify/relay-dify.go b/relay/channel/dify/relay-dify.go
index b806aa18..b58fbe53 100644
--- a/relay/channel/dify/relay-dify.go
+++ b/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 {
diff --git a/relay/helper/stream_scanner.go b/relay/helper/stream_scanner.go
index 7a7507f5..24b0b6d4 100644
--- a/relay/helper/stream_scanner.go
+++ b/relay/helper/stream_scanner.go
@@ -14,6 +14,11 @@ import (
 	"github.com/gin-gonic/gin"
 )
 
+const (
+	InitialScannerBufferSize = 1 << 20  // 1MB (1*1024*1024)
+	MaxScannerBufferSize     = 10 << 20 // 10MB (10*1024*1024)
+)
+
 func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon.RelayInfo, dataHandler func(data string) bool) {
 
 	if resp == nil {
@@ -38,7 +43,7 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon
 		ticker.Stop()
 		close(stopChan)
 	}()
-
+	scanner.Buffer(make([]byte, InitialScannerBufferSize), MaxScannerBufferSize)
 	scanner.Split(bufio.ScanLines)
 	SetEventStreamHeaders(c)