upload file cache

example/qwen_vl/stream_call.go

@@ -17,22 +17,24 @@ func main() {
 		panic("token is empty")
 	}
 
-	cli := dashscopego.NewTongyiClient(model, token)
+	cli := dashscopego.NewTongyiClient(model, token).
+		SetUploadCache(qwen.NewMemoryFileCache()) // 可以通过 UploadCacher 接口 自定义缓存实现 避免重复上传, 默认使用内存缓存
+
+	homedir, _ := os.UserHomeDir()
 
 	sysContent := qwen.VLContentList{
 		{
 			Text: "You are a helpful assistant.",
 		},
 	}
+
 	userContent := qwen.VLContentList{
 		{
-			// Text: "describe the image",
 			Text: "用唐诗体说明一下这张图片中的内容", //nolint:gosmopolitan
 		},
 		{
-			// Image: "https://pic.ntimg.cn/20140113/8800276_184351657000_2.jpg",
-			// Image: "file:///Users/xxxx/xxxx.png",
-			Image: "https://dashscope.oss-cn-beijing.aliyuncs.com/images/dog_and_girl.jpeg",
+			Image: "file://" + homedir + "/Downloads/pandas_img.jpg",
+			// Image: "https://dashscope.oss-cn-beijing.aliyuncs.com/images/dog_and_girl.jpeg",
 		},
 	}
 
@@ -42,12 +44,12 @@ func main() {
 			{Role: qwen.RoleUser, Content: &userContent},
 		},
 	}
-
 	// callback function:  print stream result
 	streamCallbackFn := func(ctx context.Context, chunk []byte) error {
 		log.Print(string(chunk))
 		return nil
 	}
+
 	req := &dashscopego.VLRequest{
 		Input:       input,
 		StreamingFn: streamCallbackFn,

qwen/content.go

@@ -23,11 +23,17 @@ type IBlobContent interface {
 
 type IBlobListConvert interface {
 	ConvertToBlobList() []IBlobContent
+	ConvertBackFromBlobList(list []IBlobContent)
 }
 
 func popBlobContent(rawList IBlobListConvert) (IBlobContent, bool) {
+	// TODO: rawList must be a pointer, otherwise it will panic.
 	list := rawList.ConvertToBlobList()
-	return innerGetBlob(&list)
+	content, hasBlob := innerGetBlob(&list)
+
+	rawList.ConvertBackFromBlobList(list)
+
+	return content, hasBlob
 }
 
 func innerGetBlob(list *[]IBlobContent) (IBlobContent, bool) {
@@ -36,6 +42,7 @@ func innerGetBlob(list *[]IBlobContent) (IBlobContent, bool) {
 		if v.GetBlob() != "" {
 			hasBlob = true
 			preSlice := (*list)[:i]
+
 			if i == len(*list)-1 {
 				*list = preSlice
 			} else {

qwen/content_audio.go

@@ -81,3 +81,16 @@ func (acList *AudioContentList) ConvertToBlobList() []IBlobContent {
 	}
 	return list
 }
+
+func (acList *AudioContentList) ConvertBackFromBlobList(list []IBlobContent) {
+	if acList == nil {
+		panic("VLContentList is nil or empty")
+	}
+
+	*acList = make([]AudioContent, len(list))
+	for i, v := range list {
+		if content, ok := v.(AudioContent); ok {
+			(*acList)[i] = content
+		}
+	}
+}

qwen/content_file.go

@@ -78,3 +78,16 @@ func (fclist *FileContentList) ConvertToBlobList() []IBlobContent {
 	}
 	return list
 }
+
+func (fclist *FileContentList) ConvertBackFromBlobList(list []IBlobContent) {
+	if fclist == nil {
+		panic("VLContentList is nil or empty")
+	}
+
+	*fclist = make([]FileContent, len(list))
+	for i, v := range list {
+		if content, ok := v.(FileContent); ok {
+			(*fclist)[i] = content
+		}
+	}
+}

qwen/content_vl.go

@@ -80,3 +80,16 @@ func (vlist *VLContentList) ConvertToBlobList() []IBlobContent {
 	}
 	return list
 }
+
+func (vlist *VLContentList) ConvertBackFromBlobList(list []IBlobContent) {
+	if vlist == nil {
+		panic("VLContentList is nil or empty")
+	}
+
+	*vlist = make([]VLContent, len(list))
+	for i, v := range list {
+		if content, ok := v.(VLContent); ok {
+			(*vlist)[i] = content
+		}
+	}
+}

qwen/upload_cache.go

@@ -0,0 +1,82 @@
+package qwen
+
+import (
+	"crypto/sha1"
+	"encoding/hex"
+	"time"
+)
+
+// UploadCacher is an interface for caching uploaded file url to prevent duplicate upload.
+// By default we provide Sample MemoryFileCache as the implementation.
+// Customize your own cache manager by implementing this interface.
+type UploadCacher interface {
+	SaveCache(buf []byte, url string) error
+	GetCache(buf []byte) string
+}
+
+// ==================== Sample MemoryFileCache ====================
+type FileCacheInfo struct {
+	URL        string
+	UploadTime int64
+}
+
+// MemoryFileCache is a simple in-memory-cache implementation for UploadCacher interface.
+type MemoryFileCache struct {
+	MapFiles             map[string]*FileCacheInfo
+	MaxFileCacheLifeTime time.Duration
+}
+
+func NewMemoryFileCache() *MemoryFileCache {
+
+	mgr := &MemoryFileCache{
+		MapFiles:             make(map[string]*FileCacheInfo),
+		MaxFileCacheLifeTime: time.Hour*2 - time.Minute*5,
+	}
+
+	// cron job to clean up outdated memory cache.
+	go mgr.cronMemoryCleaner()
+
+	return mgr
+}
+
+func (mgr *MemoryFileCache) SaveCache(buf []byte, url string) error {
+	key := mgr.hash(buf)
+
+	mgr.MapFiles[key] = &FileCacheInfo{
+		URL:        url,
+		UploadTime: time.Now().Unix(),
+	}
+
+	return nil
+}
+
+func (mgr *MemoryFileCache) GetCache(buf []byte) string {
+	key := mgr.hash(buf)
+
+	cache, isok := mgr.MapFiles[key]
+	if isok {
+		return cache.URL
+	}
+
+	return ""
+}
+
+func (mgr *MemoryFileCache) cronMemoryCleaner() {
+	for {
+		time.Sleep(time.Minute * 5)
+
+		curtime := time.Now().Unix()
+
+		for k, v := range mgr.MapFiles {
+			if curtime-v.UploadTime > int64(mgr.MaxFileCacheLifeTime) {
+				delete(mgr.MapFiles, k)
+			}
+		}
+	}
+}
+
+func (mgr *MemoryFileCache) hash(buf []byte) string {
+	h := sha1.New()
+	h.Write(buf)
+	return hex.EncodeToString(h.Sum(nil))
+}

qwen/uploader.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"io"
+	"log"
 	"mime/multipart"
 	"net/http"
 	"os"
@@ -92,36 +93,60 @@ type UploadRequest struct {
 */
 
 // uploading local image to aliyun oss, a oss url will be returned.
-func UploadLocalFile(ctx context.Context, filePath, model, apiKey string) (string, error) {
+func UploadLocalFile(ctx context.Context, filePath, model, apiKey string, uploadCacher UploadCacher) (string, error) {
 	fileBytes, mimeType, err := loadLocalFileWithMimeType(filePath)
 	if err != nil {
 		return "", err
 	}
 
 	fileName := filepath.Base(filePath)
 
-	return uploadFIle(ctx, fileBytes, fileName, mimeType, model, apiKey)
+	if uploadCacher != nil {
+		return uploadFileWithCache(ctx, fileBytes, fileName, mimeType, model, apiKey, uploadCacher)
+	} else {
+		return uploadFile(ctx, fileBytes, fileName, mimeType, model, apiKey)
+	}
 }
 
 // download and uploading a online image to aliyun oss, a oss url will be returned.
-func UploadFileFromURL(ctx context.Context, fileURL, model, apiKey string) (string, error) {
+func UploadFileFromURL(ctx context.Context, fileURL, model, apiKey string, uploadCacher UploadCacher) (string, error) {
 	fileBytes, mimeType, err := downloadFileWithMimeType(fileURL)
 	if err != nil {
 		return "", err
 	}
 	fileName := filepath.Base(fileURL)
 
-	return uploadFIle(ctx, fileBytes, fileName, mimeType, model, apiKey)
+	if uploadCacher != nil {
+		return uploadFileWithCache(ctx, fileBytes, fileName, mimeType, model, apiKey, uploadCacher)
+	} else {
+		return uploadFile(ctx, fileBytes, fileName, mimeType, model, apiKey)
+	}
 }
 
-func uploadFIle(ctx context.Context, fileBytes []byte, fileName, mimeType, model, apiKey string) (string, error) {
-	if gFileCacheMgr != nil {
-		url := gFileCacheMgr.Get(fileBytes)
-		if url != "" {
-			return url, nil
-		}
+func uploadFileWithCache(ctx context.Context, fileBytes []byte, fileName, mimeType, model, apiKey string, uploadCacher UploadCacher) (string, error) {
+	var ossUrl string
+	var err error
+
+	ossUrl = uploadCacher.GetCache(fileBytes)
+	if ossUrl != "" {
+		return ossUrl, nil
+	}
+
+	ossUrl, err = uploadFile(ctx, fileBytes, fileName, mimeType, model, apiKey)
+	if err != nil {
+		return "", err
 	}
 
+	err = uploadCacher.SaveCache(fileBytes, ossUrl)
+	if err != nil {
+		log.Printf("save upload cache error: %v\n", err)
+	}
+
+	return ossUrl, nil
+}
+
+func uploadFile(ctx context.Context, fileBytes []byte, fileName, mimeType, model, apiKey string) (string, error) {
+
 	certInfo, err := getUploadCertificate(ctx, model, apiKey)
 	if err != nil {
 		return "", &WrapMessageError{Message: "upload Cert Error", Cause: err}
@@ -150,10 +175,6 @@ func uploadFIle(ctx context.Context, fileBytes []byte, fileName, mimeType, model
 
 	ossURL := "oss://" + ossKey
 
-	if gFileCacheMgr != nil {
-		gFileCacheMgr.Cache(fileBytes, ossURL)
-	}
-
 	return ossURL, nil
 }