diff --git a/common/util/config.go b/common/util/config.go new file mode 100644 index 0000000..cb9c70d --- /dev/null +++ b/common/util/config.go @@ -0,0 +1,14 @@ +package util + +import ( + "context" + + "github.com/gogf/gf/v2/frame/g" + "github.com/gogf/gf/v2/util/gconv" +) + +// GetModelPrompt 获取请求模型的提示词 +func GetModelPrompt(ctx context.Context, modelType int) string { + key := "modelPrompts.types." + gconv.String(modelType) + return g.Cfg().MustGet(ctx, key, "").String() +} diff --git a/common/util/files.go b/common/util/files.go index a5f57c4..adf9a29 100644 --- a/common/util/files.go +++ b/common/util/files.go @@ -1,10 +1,9 @@ package util import ( - "fmt" "net/http" - "os" "path/filepath" + "regexp" "strings" ) @@ -60,23 +59,91 @@ func DetectFileType(data []byte) (contentType string, ext string) { } } -// SaveTmpResult 将二进制数据写入临时文件 -func SaveTmpResult(taskID string, data []byte, ext string) (string, error) { - dir := filepath.Join(os.TempDir(), "model-asynch") - if err := os.MkdirAll(dir, 0o755); err != nil { - return "", fmt.Errorf("创建临时目录失败: %w", err) +var ( + // AllowedMIMEPrefixes 允许的文本类 MIME 类型前缀 + AllowedMIMEPrefixes = []string{ + "text/", + "application/json", + "application/xml", + "application/javascript", + "application/x-yaml", + "application/yaml", + "application/toml", + "application/x-httpd-php", + "application/x-sh", + "application/x-python", + "application/x-perl", + "application/x-ruby", } - if ext == "" { - ext = ".bin" - } - if ext[0] != '.' { - ext = "." + ext + // BannedExtensions 禁止的文件扩展名 + BannedExtensions = map[string]bool{ + ".png": true, ".jpg": true, ".jpeg": true, ".gif": true, ".bmp": true, + ".webp": true, ".svg": true, ".ico": true, ".tiff": true, ".tif": true, + ".mp3": true, ".wav": true, ".ogg": true, ".flac": true, ".aac": true, + ".wma": true, ".m4a": true, + ".mp4": true, ".avi": true, ".mkv": true, ".mov": true, ".wmv": true, + ".flv": true, ".webm": true, + ".tar": true, ".gz": true, ".rar": true, ".7z": true, + ".exe": true, ".dll": true, ".so": true, ".bin": true, ".dat": true, + ".class": true, ".pyc": true, + ".pdf": true, ".doc": true, ".docx": true, ".xls": true, ".xlsx": true, + ".ppt": true, ".pptx": true, } - path := filepath.Join(dir, fmt.Sprintf("%s%s", taskID, ext)) - if err := os.WriteFile(path, data, 0o644); err != nil { - return "", fmt.Errorf("写入临时文件失败: %w", err) + symbolCleaner = regexp.MustCompile(`[\x00-\x08\x0B\x0C\x0E-\x1F]`) + multiNewlines = regexp.MustCompile(`\n{3,}`) +) + +// SanitizeURL 清洗 URL 字符串 +func SanitizeURL(raw string) string { + s := strings.TrimSpace(raw) + s = strings.Trim(s, "`\"") + return s +} + +// CleanSymbols 清洗文本中的控制字符和多余空行 +func CleanSymbols(text string) string { + text = symbolCleaner.ReplaceAllString(text, "") + text = strings.ReplaceAll(text, "\r\n", "\n") + text = strings.ReplaceAll(text, "\r", "\n") + text = multiNewlines.ReplaceAllString(text, "\n\n") + return strings.TrimSpace(text) +} + +// IsBannedExtension 判断是否为禁止的文件扩展名 +func IsBannedExtension(url string) bool { + ext := extractExtension(url) + return BannedExtensions[ext] +} + +// IsZipExtension 判断是否为 zip 文件 +func IsZipExtension(url string) bool { + ext := extractExtension(url) + return ext == ".zip" +} + +// IsReadableContentType 判断是否为可读的文本类型 +func IsReadableContentType(contentType string) bool { + if contentType == "" { + return false } - return path, nil + + ct := strings.ToLower(contentType) + for _, prefix := range AllowedMIMEPrefixes { + if strings.HasPrefix(ct, prefix) { + return true + } + } + + return false +} + +// extractExtension 提取文件扩展名并清理查询参数 +func extractExtension(url string) string { + ext := strings.ToLower(filepath.Ext(url)) + if idx := strings.Index(ext, "?"); idx != -1 { + ext = ext[:idx] + } + return ext } diff --git a/common/util/headers.go b/common/util/headers.go index 446bf63..8ef0b22 100644 --- a/common/util/headers.go +++ b/common/util/headers.go @@ -2,6 +2,7 @@ package util import ( "context" + "net/url" "gitea.redpowerfuture.com/red-future/common/utils" "github.com/gogf/gf/v2/frame/g" @@ -77,3 +78,29 @@ func SetTaskHeadersToCtx(ctx context.Context, headers map[string]string) context } return ctx } + +// ======================== 请求工具 ======================== + +// ParseHeadMsgHeaders 从 head_msg 中提取 HTTP 请求头 +func ParseHeadMsgHeaders(headMsg map[string]any) map[string]string { + if len(headMsg) == 0 { + return nil + } + out := make(map[string]string, len(headMsg)) + for k, v := range headMsg { + out[k] = gconv.String(v) + } + return out +} + +// BodyToQuery 将 body 转为 URL 查询参数 +func BodyToQuery(payload map[string]any) (url.Values, error) { + q := url.Values{} + for k, v := range payload { + if v == nil { + continue + } + q.Set(k, gconv.String(v)) + } + return q, nil +} diff --git a/common/util/mapping.go b/common/util/mapping.go index b46de81..3f610e6 100644 --- a/common/util/mapping.go +++ b/common/util/mapping.go @@ -4,7 +4,6 @@ import ( "encoding/json" "fmt" "model-gateway/model/entity" - "net/url" "strings" "github.com/gogf/gf/v2/encoding/gjson" @@ -21,7 +20,6 @@ func ParseAndValidate(raw map[string]any, requiredFields []string) (map[string]a return raw, fmt.Errorf("字段 %s 为空", entity.ResponseBody) } - // 过滤控制字符 contentStr = cleanControlChars(contentStr) var arr []any @@ -32,7 +30,6 @@ func ParseAndValidate(raw map[string]any, requiredFields []string) (map[string]a return raw, fmt.Errorf("解析后数组为空") } - // 校验必填字段 if len(requiredFields) > 0 { for i, r := range arr { round, _ := r.(map[string]any) @@ -117,119 +114,6 @@ func MapResponsePayload(mapping map[string]any, result map[string]any) (map[stri return mapped, nil } -// ValidateAndParseForm 校验表单并转为嵌套 map -func ValidateAndParseForm(forms []entity.Form) (map[string]any, error) { - result := gjson.New("{}") - - for _, form := range forms { - if form.Key == "" { - continue - } - if form.Required && (form.Value == nil || gconv.String(form.Value) == "") { - return nil, fmt.Errorf("字段 %s 为必填", form.Label) - } - if form.Value == nil { - continue - } - - val, err := validateAndConvert(form) - if err != nil { - return nil, err - } - _ = result.Set(form.Key, val) - } - - return result.Map(), nil -} - -// validateAndConvert 验证表单字段并转为标准格式 -func validateAndConvert(form entity.Form) (any, error) { - val := form.Value - fc := form.FieldConstraint - switch form.Type { - case "string": - s := gconv.String(val) - if fc.MaxLength > 0 && len(s) > fc.MaxLength { - return nil, fmt.Errorf("字段 %s 超过最大长度 %d", form.Label, fc.MaxLength) - } - if fc.MinLength > 0 && len(s) < fc.MinLength { - return nil, fmt.Errorf("字段 %s 不足最小长度 %d", form.Label, fc.MinLength) - } - return s, nil - - case "number": - f := gconv.Float64(val) - if fc.Min != nil && f < gconv.Float64(fc.Min) { - return nil, fmt.Errorf("字段 %s 不能小于 %v", form.Label, fc.Min) - } - if fc.Max != nil && f > gconv.Float64(fc.Max) { - return nil, fmt.Errorf("字段 %s 不能大于 %v", form.Label, fc.Max) - } - // 根据 numberType 决定返回 int 还是 float64 - switch fc.NumberType { - case "float", "positiveFloat", "negativeFloat": - return f, nil - default: - return int(f), nil - } - - case "select", "radio": - v, ok := val.(map[string]any) - if !ok { - return nil, fmt.Errorf("字段 %s 格式错误", form.Label) - } - return v, nil - - case "upload": - var urls []string - switch v := val.(type) { - case []any: - for _, u := range v { - urls = append(urls, gconv.String(u)) - } - case []string: - urls = v - case string: - if v != "" { - urls = []string{v} - } - } - if fc.MaxCount > 0 && len(urls) > fc.MaxCount { - return nil, fmt.Errorf("字段 %s 上传数量超过上限 %d", form.Label, fc.MaxCount) - } - return urls, nil - - default: - return gconv.String(val), nil - } -} - -// ======================== 请求工具 ======================== - -// ParseHeadMsgHeaders 从 head_msg 中提取 HTTP 请求头 -func ParseHeadMsgHeaders(headMsg map[string]any) map[string]string { - if len(headMsg) == 0 { - return nil - } - out := make(map[string]string, len(headMsg)) - for k, v := range headMsg { - out[k] = gconv.String(v) - } - return out -} - -// BodyToQuery 将 body 转为 URL 查询参数 -func BodyToQuery(payload map[string]any) (url.Values, error) { - q := url.Values{} - for k, v := range payload { - if v == nil { - continue - } - q.Set(k, gconv.String(v)) - } - return q, nil -} - // ======================== 内部辅助 ======================== func cleanControlChars(s string) string { diff --git a/common/util/prompt.go b/common/util/prompt.go new file mode 100644 index 0000000..92eb04d --- /dev/null +++ b/common/util/prompt.go @@ -0,0 +1,391 @@ +package util + +import ( + "encoding/json" + "fmt" + "model-gateway/model/entity" + "model-gateway/service/gateway" + "path/filepath" + "strings" + + "github.com/gogf/gf/v2/encoding/gjson" + "github.com/gogf/gf/v2/util/gconv" +) + +// ======================== 表单校验与构建 ======================== + +// ValidateAndParseForm 校验表单并转为嵌套 map +func ValidateAndParseForm(forms []entity.Form) (map[string]any, error) { + result := gjson.New("{}") + for _, form := range forms { + if form.Key == "" { + continue + } + if form.Required && (form.Value == nil || gconv.String(form.Value) == "") { + return nil, fmt.Errorf("字段 %s 为必填", form.Label) + } + if form.Value == nil { + continue + } + + val, err := validateAndConvert(form) + if err != nil { + return nil, err + } + _ = result.Set(form.Key, val) + } + return result.Map(), nil +} + +// validateAndConvert 验证表单字段并转为标准格式 +func validateAndConvert(form entity.Form) (any, error) { + val := form.Value + fc := form.FieldConstraint + + switch form.Type { + case "string": + s := gconv.String(val) + if fc.MaxLength > 0 && len(s) > fc.MaxLength { + return nil, fmt.Errorf("字段 %s 超过最大长度 %d", form.Label, fc.MaxLength) + } + if fc.MinLength > 0 && len(s) < fc.MinLength { + return nil, fmt.Errorf("字段 %s 不足最小长度 %d", form.Label, fc.MinLength) + } + return s, nil + + case "number": + f := gconv.Float64(val) + if fc.Min != nil && f < gconv.Float64(fc.Min) { + return nil, fmt.Errorf("字段 %s 不能小于 %v", form.Label, fc.Min) + } + if fc.Max != nil && f > gconv.Float64(fc.Max) { + return nil, fmt.Errorf("字段 %s 不能大于 %v", form.Label, fc.Max) + } + switch fc.NumberType { + case "float", "positiveFloat", "negativeFloat": + return f, nil + default: + return int(f), nil + } + + case "select", "radio": + v, ok := val.(map[string]any) + if !ok { + return nil, fmt.Errorf("字段 %s 格式错误", form.Label) + } + return v, nil + + case "upload": + var urls []string + switch v := val.(type) { + case []any: + for _, u := range v { + urls = append(urls, gconv.String(u)) + } + case []string: + urls = v + case string: + if v != "" { + urls = []string{v} + } + } + if fc.MaxCount > 0 && len(urls) > fc.MaxCount { + return nil, fmt.Errorf("字段 %s 上传数量超过上限 %d", form.Label, fc.MaxCount) + } + if fc.Accept != "" && len(urls) > 0 { + allowed := strings.Split(fc.Accept, ",") + for _, fileUrl := range urls { + ext := strings.ToLower(strings.TrimPrefix(filepath.Ext(fileUrl), ".")) + if !containsExt(allowed, ext) { + return nil, fmt.Errorf("字段 %s 不支持的文件格式: %s", form.Label, ext) + } + } + } + return urls, nil + + default: + return gconv.String(val), nil + } +} + +func containsExt(allowed []string, ext string) bool { + for _, a := range allowed { + if strings.TrimSpace(a) == ext { + return true + } + } + return false +} + +// BuildTemplateFromForm 从 Form 构建模板 +func BuildTemplateFromForm(forms []entity.Form) map[string]any { + jsonObj := gjson.New("{}") + for _, form := range forms { + if form.Key == "" { + continue + } + if form.DefaultValue != nil { + _ = jsonObj.Set(form.Key, form.DefaultValue) + } else { + switch form.Type { + case "number": + _ = jsonObj.Set(form.Key, 0) + case "boolean": + _ = jsonObj.Set(form.Key, false) + default: + _ = jsonObj.Set(form.Key, "") + } + } + } + return jsonObj.Map() +} + +// ======================== 提示词拼接 ======================== + +// GetFormByRole 根据 role 获取单个 Form +func GetFormByRole(forms []entity.Form, role string) (entity.Form, bool) { + for _, f := range forms { + if f.Role == role { + return f, true + } + } + return entity.Form{}, false +} + +// GetRoleContentPath 从 Form 数组中提取指定 role 的 Key 路径 +func GetRoleContentPath(forms []entity.Form, role string) string { + for _, form := range forms { + if form.Role == role { + return form.Key + } + } + return "" +} + +// MergePrompt 将多个提示词拼接到指定路径的 content 中 +func MergePrompt(messages map[string]any, contentPath string, prompts ...string) map[string]any { + var parts []string + for _, p := range prompts { + if p != "" { + parts = append(parts, p) + } + } + if len(parts) == 0 { + return messages + } + + appendContent := strings.Join(parts, "\n") + msgJson := gjson.New(messages) + + existing := msgJson.Get(contentPath).String() + if existing != "" { + appendContent = existing + "\n" + appendContent + } + _ = msgJson.Set(contentPath, appendContent) + + return msgJson.Map() +} + +// ExtractUserContent 将 messages 转为自然语言文本 +func ExtractUserContent(messages map[string]any) string { + return formatMapToText(messages, "") +} + +func formatMapToText(v any, prefix string) string { + var b strings.Builder + switch val := v.(type) { + case map[string]any: + for k, vv := range val { + switch vv.(type) { + case map[string]any, []any: + b.WriteString(formatMapToText(vv, prefix)) + default: + b.WriteString(fmt.Sprintf("%s:%v,", k, vv)) + } + } + case []any: + for _, elem := range val { + b.WriteString(formatMapToText(elem, "")) + } + case string: + b.WriteString(fmt.Sprintf("%s,", val)) + } + return b.String() +} + +// InjectHistory 拼接历史提示词 +func InjectHistory(messages map[string]any, history []gateway.SessionHistoryItem, mergeOrder []string) map[string]any { + msgs, _ := messages["messages"].([]any) + + // 按 role 分组 + grouped := make(map[string][]any) + for _, m := range msgs { + if msg, ok := m.(map[string]any); ok { + role := gconv.String(msg["role"]) + grouped[role] = append(grouped[role], msg) + } + } + + result := make([]any, 0, len(msgs)+len(history)) + for _, part := range mergeOrder { + switch part { + case "history": + for _, h := range history { + result = append(result, map[string]any{ + "role": h.Role, + "content": h.Content, + }) + } + default: + if msgs, ok := grouped[part]; ok { + result = append(result, msgs...) + } + } + } + + messages["messages"] = result + return messages +} + +// SplitByAttachment 根据附件数量拆分多轮,未超出返回空 +func SplitByAttachment(messages map[string]any, forms []entity.Form) []map[string]any { + // 1) 获取约束 + maxVideo := getMaxCount(forms, "video") + maxImage := getMaxCount(forms, "image") + maxAudio := getMaxCount(forms, "audio") + + // 2) 提取 system 消息和 user content + msgs := gjson.New(messages).Get("messages").Array() + var systemMsg map[string]any + var userContent []any + + for _, m := range msgs { + msg := m.(map[string]any) + switch msg["role"] { + case "system": + systemMsg = msg + case "user": + if c, ok := msg["content"].([]any); ok { + userContent = c + } + } + } + if systemMsg == nil || len(userContent) == 0 { + return nil + } + + // 3) 分离 text 和各类附件 + var texts []any + var videos, images, audios []any + + for _, c := range userContent { + item := c.(map[string]any) + switch item["type"] { + case "video_url": + videos = append(videos, item) + case "image_url": + images = append(images, item) + case "input_audio": + audios = append(audios, item) + default: + texts = append(texts, item) + } + } + + // 4) 判断是否需要拆分 + if len(videos) <= maxVideo && len(images) <= maxImage && len(audios) <= maxAudio { + return nil + } + + // 5) 按约束分片 + videoChunks := chunk(videos, maxVideo) + imageChunks := chunk(images, maxImage) + audioChunks := chunk(audios, maxAudio) + + maxRounds := len(videoChunks) + if len(imageChunks) > maxRounds { + maxRounds = len(imageChunks) + } + if len(audioChunks) > maxRounds { + maxRounds = len(audioChunks) + } + + // 6) 构建每轮 messages + var rounds []map[string]any + for i := 0; i < maxRounds; i++ { + var newContent []any + newContent = append(newContent, texts...) + if i < len(videoChunks) { + newContent = append(newContent, videoChunks[i]...) + } else { + newContent = append(newContent, videos...) // 未超出,每轮都带 + } + if i < len(imageChunks) { + newContent = append(newContent, imageChunks[i]...) + } else { + newContent = append(newContent, images...) + } + if i < len(audioChunks) { + newContent = append(newContent, audioChunks[i]...) + } else { + newContent = append(newContent, audios...) + } + + rounds = append(rounds, map[string]any{ + "model": gjson.New(messages).Get("model").Val(), + "max_tokens": gjson.New(messages).Get("max_tokens").Val(), + "messages": []any{ + systemMsg, + map[string]any{"role": "user", "content": newContent}, + }, + }) + } + + return rounds +} + +// getMaxCount 从表单获取指定 role 的最大数量 +func getMaxCount(forms []entity.Form, role string) int { + for _, f := range forms { + if f.Role == role { + return f.FieldConstraint.MaxCount + } + } + return 0 +} + +// chunk 将数组按 size 分片 +func chunk(items []any, size int) [][]any { + if size <= 0 { + return nil + } + var chunks [][]any + for i := 0; i < len(items); i += size { + end := i + size + if end > len(items) { + end = len(items) + } + chunks = append(chunks, items[i:end]) + } + return chunks +} + +// ======================== 请求模板 ======================== + +// BuildRequestBody 替换模板占位符构建请求体 +func BuildRequestBody(template map[string]any, modelName, systemPrompt, userContent string) map[string]any { + escapedSystem := escapeJSON(systemPrompt) + escapedUser := escapeJSON(userContent) + + str := gjson.New(template).MustToJsonString() + str = strings.ReplaceAll(str, `"{{model}}"`, `"`+modelName+`"`) + str = strings.ReplaceAll(str, `"{{system}}"`, escapedSystem) + str = strings.ReplaceAll(str, `"{{user}}"`, escapedUser) + + return gjson.New(str).Map() +} + +func escapeJSON(s string) string { + b, _ := json.Marshal(s) + return string(b) +} diff --git a/common/util/pull_task.go b/common/util/pull_task.go index bb77128..2057fd9 100644 --- a/common/util/pull_task.go +++ b/common/util/pull_task.go @@ -16,7 +16,6 @@ import ( "github.com/gogf/gf/v2/util/gconv" ) -// PullTaskResult 轮询查询异步任务结果 func PullTaskResult(ctx context.Context, body map[string]any, queryConfig map[string]any, headMsg map[string]any) (map[string]any, error) { taskID, err := extractTaskID(body, queryConfig) if err != nil { @@ -40,6 +39,12 @@ func PullTaskResult(ctx context.Context, body map[string]any, queryConfig map[st } statusValues, _ := queryConfig["status_values"].(map[string]any) + // 失败信息路径 + errorPath := gconv.String(queryConfig["error_path"]) + if errorPath == "" { + errorPath = "error.message" + } + reqBodyMap := map[string]any{"task_id": taskID} for { @@ -70,8 +75,12 @@ func PullTaskResult(ctx context.Context, body map[string]any, queryConfig map[st } if matchStatus(statusStr, statusValues["failed"]) { - g.Log().Errorf(ctx, "[PullTaskResult] 任务失败 taskID=%s", taskID) - return result, fmt.Errorf("任务失败") + errMsg := gconv.String(gjson.New(result).Get(errorPath).Val()) + if errMsg == "" { + errMsg = "任务失败" + } + g.Log().Errorf(ctx, "[PullTaskResult] 任务失败 taskID=%s err=%s", taskID, errMsg) + return result, fmt.Errorf("任务失败: %s", errMsg) } time.Sleep(time.Duration(interval) * time.Second) diff --git a/config.yml b/config.yml index f2dd0a7..b297979 100644 --- a/config.yml +++ b/config.yml @@ -28,7 +28,7 @@ database: timeMaintainDisabled: false # (可选)是否完全关闭时间更新特性,为true时CreatedAt/UpdatedAt/DeletedAt都将失效 model_gateway: - type: "pgsql" - host: "192.168.3.30" + host: "192.168.3.8" port: "5432" user: "postgres" pass: "123456" @@ -67,4 +67,38 @@ queryPending: jobTask: intervalSeconds: 10 # 轮询间隔(秒) batchSize: 10 # 每批处理条数 - poolSize: 5 # 协程池大小 \ No newline at end of file + poolSize: 5 # 协程池大小 + +modelPrompts: + types: + 100: | + 你是一个智能文字处理助手,专注于文本理解、文本创作、文本优化与语言表达任务,能够根据不同场景完成文章撰写、商业文案、报告总结、邮件通知、脚本创作、内容改写、信息提炼、语言翻译等多种文字处理工作,并能够理解上下文语义关系,保持内容逻辑完整、结构清晰、表达自然。 + 在执行文本任务时,你需要以专业内容创作者、编辑顾问、语言优化专家的身份完成输出,严格保证语言准确性、逻辑连贯性、表达一致性与阅读体验,根据不同用户场景自动适配正式、口语化、专业化、营销化等表达风格,同时避免空洞表达、重复描述与机械化生成内容。 + 当用户提供具体需求时,需要结合用户输入、上下文信息、参数条件与目标场景生成最终文本结果;若涉及改写、扩写、摘要、总结、标题、营销内容等任务,需要保证核心语义不偏离,并根据用户真实目的完成结构化输出。 + 200: | + 你是一个智能图片处理助手,专注于视觉内容生成、图像编辑、画面分析与风格控制任务,能够根据文字描述生成不同风格的图片内容,包括写实、插画、动漫、水彩、电影感、商业海报等多种视觉形式,并支持图片局部修改、风格迁移、画面扩展、背景处理与视觉增强等操作。 + 在执行图片相关任务时,你需要以专业视觉设计师、插画师、摄影指导、美术导演的身份进行画面构建,重点关注主体构图、色彩关系、光影氛围、镜头语言、视觉层次与整体风格统一性,确保生成结果具备明确视觉主题与稳定审美表现,而不是简单关键词堆砌。 + 当用户提供图片需求时,需要结合用户描述、场景用途、风格方向、尺寸比例、主体元素、氛围要求等信息生成完整视觉方案;若存在图片编辑任务,则必须保留原图核心特征,仅对用户指定区域或效果进行修改。 + 300: | + 你是一个智能音频处理助手,专注于语音生成、语音识别、音频分析与声音编辑任务,能够完成文字转语音、语音转文本、多语言识别、音频降噪、音色处理、混音剪辑、情绪识别与声音特征分析等多种音频相关工作,并能够根据不同场景匹配对应语音风格与声音表现形式。 + 在执行音频任务时,你需要以专业配音导演、声音工程师、语音分析专家、后期音频制作人员的身份进行处理,重点保证语音自然度、情绪一致性、识别准确率、音频清晰度与输出稳定性,同时确保不同格式、采样率与播放场景下具备良好兼容性。 + 当用户提供具体音频需求时,需要结合音色、语速、语言类型、情绪风格、背景环境、输出格式等参数完成对应处理;若涉及语音识别或音频分析,则需要尽可能保留原始语义与声音特征,并明确标注不确定内容。 + 400: | + 你是一个智能向量化处理助手,专注于文本向量化、语义检索、知识索引、相似度计算与语义聚类任务,能够将文本内容转换为高维语义向量,并基于向量相似度完成语义搜索、知识召回、内容聚类、文档匹配与知识库构建等处理流程。 + 在执行向量化任务时,你需要以语义检索工程师、知识库架构师、AI检索系统专家的身份进行处理,重点保证语义表达准确性、向量一致性、检索稳定性与召回有效性,同时确保不同文本之间的语义关系能够被正确表达与计算。 + 当用户提供文本集合、知识内容或检索需求时,需要结合文本上下文、主题方向、检索目标、相似度要求与业务场景生成最终结果;若涉及聚类或知识库构建,则必须明确类别关系、索引结构与召回逻辑。 + 500: | + 你是一个全模态智能处理助手,能够同时理解、分析与生成文本、图片、音频、视频等多种模态内容,并支持跨模态转换、多模态融合推理、联合内容生成与复杂场景交互,能够根据不同输入形式自动匹配最合理的处理策略与输出方式。 + 在执行多模态任务时,你需要以全链路AI内容架构师、多模态交互专家、综合内容生成系统的身份完成处理,重点保证不同模态之间的语义一致性、风格统一性、信息完整性与交互连贯性,避免出现跨模态语义断裂或输出不一致的问题。 + 当用户提供混合输入内容时,需要结合文本、图片、音频、视频等多种信息共同分析用户真实目标,并根据任务场景自动决定最终输出形式;若涉及跨模态生成,则必须保证生成结果能够准确映射原始语义与核心信息。 + +nodePrompts: | + 你是流程路由助手,你的任务是根据上下文,选择一个正确的节点ID返回。 + 规则: + 1. 只允许从下面的可选节点ID列表中选择一个返回 + 2. 不要返回任何多余文字、标点、解释、标题 + 3. 只返回纯节点ID + 可选节点ID(ID: 节点描述): + %s + 上下文内容: + %s \ No newline at end of file diff --git a/consts/public/public.go b/consts/public/public.go index f20e1dc..5b849d7 100644 --- a/consts/public/public.go +++ b/consts/public/public.go @@ -15,9 +15,8 @@ const ( ) const ( - BuildTypePrompt = 1 //提示词构建 - BuildTypeNode = 2 //节点构建 - BuildTypeStruct = 3 //结构构建 + BuildTypeSingle = 1 // 单轮构建 + BuildTypeMulti = 2 // 多轮构建 ) // ModelType 模型类型常量 diff --git a/consts/public/table_name.go b/consts/public/table_name.go index a7f89cb..402f014 100644 --- a/consts/public/table_name.go +++ b/consts/public/table_name.go @@ -5,8 +5,10 @@ const ( ) const ( - TableNameModel = "model_gateway_models" // 模型表 - TableNameTask = "model_gateway_task" // 任务表 - TableNameOpLog = "model_gateway_logs_op" // 操作日志表 - TableNameStat = "model_gateway_logs_stat" // 按天统计表 + TableNameModel = "model_gateway_models" // 模型表 + TableNameTask = "model_gateway_task" // 任务表 + TableNameBuildRecord = "model_gateway_build_record" // 构建记录表 + TableNameOpLog = "model_gateway_logs_op" // 操作日志表 + TableNameStat = "model_gateway_logs_stat" // 按天统计表 + TableNameProtocol = "prompts_provider_protocol" // 模型协议表 ) diff --git a/controller/model_gateway_task_controller.go b/controller/model_gateway_task_controller.go index 909f67b..1300fcd 100644 --- a/controller/model_gateway_task_controller.go +++ b/controller/model_gateway_task_controller.go @@ -12,6 +12,11 @@ var ModelGatewayTask = new(task) type task struct{} +// BuildMessages 构建请求模型的数据结构 +func (c *task) BuildMessages(ctx context.Context, req *dto.BuildMessagesReq) (res *dto.BuildMessagesRes, err error) { + return taskService.ModelGatewayTask.BuildMessages(ctx, req) +} + // CreateTask 根据 modelName 创建异步任务,返回 taskId func (c *task) CreateTask(ctx context.Context, req *dto.CreateTaskReq) (res *dto.CreateTaskRes, err error) { return taskService.ModelGatewayTask.Create(ctx, req) diff --git a/dao/model_gateway_build_record_dao.go b/dao/model_gateway_build_record_dao.go new file mode 100644 index 0000000..028e9f7 --- /dev/null +++ b/dao/model_gateway_build_record_dao.go @@ -0,0 +1,88 @@ +package dao + +import ( + "context" + "model-gateway/consts/public" + "model-gateway/model/entity" + + "gitea.redpowerfuture.com/red-future/common/db/gfdb" + "github.com/gogf/gf/v2/util/gconv" +) + +var ModelGatewayBuildRecord = &buildRecordDao{} + +type buildRecordDao struct{} + +// Insert 插入构建记录 +func (d *buildRecordDao) Insert(ctx context.Context, req *entity.ModelGatewayBuildRecord) (id int64, err error) { + m := new(entity.ModelGatewayBuildRecord) + err = gconv.Struct(req, &m) + if err != nil { + return + } + r, err := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameBuildRecord).Insert(m) + if err != nil { + return + } + return r.LastInsertId() +} + +// Update 更新构建记录 +func (d *buildRecordDao) Update(ctx context.Context, req *entity.ModelGatewayBuildRecord) (rows int64, err error) { + r, err := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameBuildRecord). + OmitEmpty(). + Data(req). + Where(entity.ModelGatewayBuildRecordCol.Id, req.Id). + Where(entity.ModelGatewayBuildRecordCol.TaskID, req.TaskID). + Update() + if err != nil { + return + } + return r.RowsAffected() +} + +// Get 获取构建记录 +func (d *buildRecordDao) Get(ctx context.Context, req *entity.ModelGatewayBuildRecord) (m *entity.ModelGatewayBuildRecord, err error) { + r, err := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameBuildRecord). + OmitEmpty(). + Where(entity.ModelGatewayBuildRecordCol.TaskID, req.TaskID). + Where(entity.ModelGatewayBuildRecordCol.Id, req.Id). + One() + if err != nil { + return + } + err = r.Struct(&m) + return +} + +// List 分页查询 +func (d *buildRecordDao) List(ctx context.Context, pageNum, pageSize int, req *entity.ModelGatewayBuildRecord) (list []*entity.ModelGatewayBuildRecord, total int64, err error) { + model := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameBuildRecord). + OmitEmpty(). + Where(entity.ModelGatewayBuildRecordCol.Creator, req.Creator). + Where(entity.ModelGatewayBuildRecordCol.ModelName, req.ModelName). + Where(entity.ModelGatewayBuildRecordCol.Status, req.Status). + Where(entity.ModelGatewayBuildRecordCol.TaskID, req.TaskID). + OrderDesc(entity.ModelGatewayBuildRecordCol.CreatedAt) + if pageNum > 0 && pageSize > 0 { + model = model.Page(pageNum, pageSize) + } + r, totalInt, err := model.AllAndCount(false) + if err != nil { + return nil, 0, err + } + total = gconv.Int64(totalInt) + err = r.Structs(&list) + return +} + +// Delete 删除构建记录 +func (d *buildRecordDao) Delete(ctx context.Context, req *entity.ModelGatewayBuildRecord) (rows int64, err error) { + r, err := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameBuildRecord). + Where(entity.ModelGatewayBuildRecordCol.Id, req.Id). + Delete() + if err != nil { + return + } + return r.RowsAffected() +} diff --git a/dao/provider_protocol_dao.go b/dao/provider_protocol_dao.go new file mode 100644 index 0000000..6e0439c --- /dev/null +++ b/dao/provider_protocol_dao.go @@ -0,0 +1,91 @@ +package dao + +import ( + "context" + "model-gateway/consts/public" + "model-gateway/model/entity" + + "gitea.redpowerfuture.com/red-future/common/db/gfdb" +) + +var ProviderProtocol = &providerProtocolDao{} + +type providerProtocolDao struct{} + +// Insert 新增协议配置 +func (d *providerProtocolDao) Insert(ctx context.Context, req *entity.ProviderProtocol) (id int64, err error) { + r, err := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameProtocol).Insert(req) + if err != nil { + return 0, err + } + return r.LastInsertId() +} + +// Get 获取协议配置 +func (d *providerProtocolDao) Get(ctx context.Context, req *entity.ProviderProtocol) (res *entity.ProviderProtocol, err error) { + r, err := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameProtocol). + NoTenantId(ctx). + OmitEmpty(). + Where(entity.ProviderProtocolCol.Id, req.Id). + Where(entity.ProviderProtocolCol.ProviderName, req.ProviderName). + Where(entity.ProviderProtocolCol.Status, 1). + One() + if err != nil { + return nil, err + } + if r.IsEmpty() { + return nil, nil + } + err = r.Struct(&res) + return +} + +// List 列表查询 +func (d *providerProtocolDao) List(ctx context.Context, req *entity.ProviderProtocol, page, size int) (list []*entity.ProviderProtocol, total int, err error) { + if page <= 0 { + page = 1 + } + if size <= 0 { + size = 10 + } + model := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameProtocol).OmitEmpty() + if req.ProviderName != "" { + model = model.Where(entity.ProviderProtocolCol.ProviderName, req.ProviderName) + } + if req.Status > 0 { + model = model.Where(entity.ProviderProtocolCol.Status, req.Status) + } + model = model.OrderDesc(entity.ProviderProtocolCol.CreatedAt).Page(page, size) + r, totalInt, err := model.AllAndCount(false) + if err != nil { + return nil, 0, err + } + err = r.Structs(&list) + total = totalInt + return +} + +// Update 更新协议配置 +func (d *providerProtocolDao) Update(ctx context.Context, req *entity.ProviderProtocol) (rows int64, err error) { + r, err := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameProtocol). + OmitEmpty(). + Where(entity.ProviderProtocolCol.Id, req.Id). + Data(req). + Update() + if err != nil { + return 0, err + } + return r.RowsAffected() +} + +// Delete 软删除协议配置 +func (d *providerProtocolDao) Delete(ctx context.Context, id int64) (rows int64, err error) { + r, err := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameProtocol). + Where(entity.ProviderProtocolCol.Id, id). + Data(map[string]any{"deleted_at": "NOW()"}). + Update() + if err != nil { + return 0, err + } + return r.RowsAffected() +} diff --git a/model/dto/model_gateway_task_dto.go b/model/dto/model_gateway_task_dto.go index a0a2606..0b07f81 100644 --- a/model/dto/model_gateway_task_dto.go +++ b/model/dto/model_gateway_task_dto.go @@ -4,6 +4,32 @@ import ( "github.com/gogf/gf/v2/frame/g" ) +type BuildMessagesReq struct { + g.Meta `path:"/buildMessages" method:"post" tags:"提示词处理" summary:"拼接提示词"` + ModelName string `p:"modelName" json:"modelName" v:"required#modelName不能为空" dc:"实际请求的网关模型名称"` + BuildType int64 `p:"buildType" json:"buildType" v:"required#buildType不能为空" dc:"构建类型:1单轮 2多轮"` + SkillName string `p:"skillName" json:"skillName" dc:"技能名称"` + CallbackUrl string `p:"callbackUrl" json:"callbackUrl" dc:"回调地址"` + NodeId string `p:"nodeId" json:"nodeId" dc:"节点ID"` + SessionId string `p:"sessionId" json:"sessionId" dc:"会话ID"` + Messages map[string]any `json:"messages" dc:"前端构建好的消息结构"` + CustomPrompt string `p:"customPrompt" json:"customPrompt" dc:"用户提示词"` + Cause string `p:"cause" json:"cause" dc:"原因"` +} + +type BuildMessagesRes struct { + TaskId string `json:"taskId" dc:"任务ID"` +} + +type CallbackReq struct { + g.Meta `path:"/callback" method:"post" tags:"提示词处理" summary:"model-gateway 回调" dc:"model-gateway 成功后 POST 回调:callbackUrl/{bizName}"` + TaskId string `json:"task_id" v:"required#task_id不能为空" dc:"网关任务ID"` + State int `json:"state" dc:"网关任务状态"` + OssFile string `json:"oss_file" dc:"结果文件地址"` + FileType string `json:"file_type" dc:"结果文件类型"` + ErrorMsg string `json:"error_msg" dc:"错误信息"` +} + // CreateTaskReq 创建异步任务 type CreateTaskReq struct { g.Meta `path:"/createTask" method:"post" tags:"任务管理" summary:"创建异步任务" dc:"创建异步任务并返回任务ID;创建成功后会立即异步尝试执行当前任务,执行成功后按回调配置触发钩子"` diff --git a/model/entity/model_gateway_build_record.go b/model/entity/model_gateway_build_record.go new file mode 100644 index 0000000..b812442 --- /dev/null +++ b/model/entity/model_gateway_build_record.go @@ -0,0 +1,54 @@ +package entity + +import ( + "gitea.redpowerfuture.com/red-future/common/beans" +) + +type modelGatewayBuildRecordCol struct { + beans.SQLBaseCol + TaskID string + BuildType string + ModelName string + SkillName string + SessionID string + NodeID string + RequestMessages string + ResultMessages string + Status string + ErrorMsg string + DurationSeconds string + CallbackURL string +} + +var ModelGatewayBuildRecordCol = modelGatewayBuildRecordCol{ + SQLBaseCol: beans.DefSQLBaseCol, + TaskID: "task_id", + BuildType: "build_type", + ModelName: "model_name", + SkillName: "skill_name", + SessionID: "session_id", + NodeID: "node_id", + RequestMessages: "request_messages", + ResultMessages: "result_messages", + Status: "status", + ErrorMsg: "error_msg", + DurationSeconds: "duration_seconds", + CallbackURL: "callback_url", +} + +// ModelGatewayBuildRecord 构建记录 +type ModelGatewayBuildRecord struct { + beans.SQLBaseDO `orm:",inline"` + TaskID string `orm:"task_id" json:"taskId"` + BuildType int64 `orm:"build_type" json:"buildType"` + ModelName string `orm:"model_name" json:"modelName"` + SkillName string `orm:"skill_name" json:"skillName"` + SessionID string `orm:"session_id" json:"sessionId"` + NodeID string `orm:"node_id" json:"nodeId"` + RequestMessages map[string]any `orm:"request_messages" json:"requestMessages"` + ResultMessages []map[string]any `orm:"result_messages" json:"resultMessages"` + Status int `orm:"status" json:"status"` + ErrorMsg string `orm:"error_msg" json:"errorMsg"` + DurationSeconds int `orm:"duration_seconds" json:"durationSeconds"` + CallbackURL string `orm:"callback_url" json:"callbackUrl"` +} diff --git a/model/entity/model_gateway_model.go b/model/entity/model_gateway_model.go index c41117a..79707af 100644 --- a/model/entity/model_gateway_model.go +++ b/model/entity/model_gateway_model.go @@ -94,14 +94,15 @@ type ModelGatewayModel struct { } type Form struct { - Key string `json:"key"` // 字段名 - Value any `json:"value"` // 值 - Label string `json:"label"` // 标签 - Type string `json:"type"` // 类型:string / number / boolean / select / radio / upload / json / array - DefaultValue any `json:"defaultValue"` // 默认值 - Required bool `json:"required"` // 是否必填 - IsForm bool `json:"isForm"` // 是否作为表单(用作工作流展示) - Options []map[string]any `json:"options"` // 选项(下拉/单选) + Key string `json:"key"` // 字段名 + Value any `json:"value"` // 值 + Label string `json:"label"` // 标签 + Type string `json:"type"` // 类型:string / number / boolean / select / radio / upload / json / array + DefaultValue any `json:"defaultValue"` // 默认值 + Required bool `json:"required"` // 是否必填 + IsForm bool `json:"isForm"` // 是否作为表单(用作工作流展示) + Options []map[string]any `json:"options"` // 选项(下拉/单选) + Role string `json:"role"` FieldConstraint FieldConstraint `json:"fieldConstraint"` // 字段约束 } diff --git a/model/entity/prompts_provider_protocol.go b/model/entity/prompts_provider_protocol.go new file mode 100644 index 0000000..01eed9d --- /dev/null +++ b/model/entity/prompts_provider_protocol.go @@ -0,0 +1,30 @@ +package entity + +import "gitea.redpowerfuture.com/red-future/common/beans" + +// ProviderProtocol 模型协议映射配置 +type ProviderProtocol struct { + beans.SQLBaseDO `orm:",inherit"` + ProviderName string `orm:"provider_name" json:"providerName"` + RequestTemplate map[string]any `orm:"request_template" json:"requestTemplate"` + SystemPromptTemplate string `orm:"system_prompt_template" json:"systemPromptTemplate"` + Status int `orm:"status" json:"status"` +} + +// providerProtocolCol 列名 +type providerProtocolCol struct { + beans.SQLBaseCol + ProviderName string + RequestTemplate string + SystemPromptTemplate string + Status string +} + +// ProviderProtocolCol 列名常量 +var ProviderProtocolCol = providerProtocolCol{ + SQLBaseCol: beans.DefSQLBaseCol, + ProviderName: "provider_name", + RequestTemplate: "request_template", + SystemPromptTemplate: "system_prompt_template", + Status: "status", +} diff --git a/service/gateway/gateway_http_service.go b/service/gateway/gateway_http_service.go index 96d7eff..3d64287 100644 --- a/service/gateway/gateway_http_service.go +++ b/service/gateway/gateway_http_service.go @@ -12,6 +12,7 @@ import ( commonHttp "gitea.redpowerfuture.com/red-future/common/http" "github.com/gogf/gf/v2/frame/g" + "github.com/gogf/gf/v2/os/gtime" "github.com/gogf/gf/v2/util/guid" ) @@ -166,6 +167,40 @@ func TriggerPromptsCallback(ctx context.Context, t *entity.ModelGatewayTask) { g.Log().Infof(ctx, "[提示词回调] 发送成功 epicycleId=%d 回调地址=%s 消息体大小=%d字节", t.EpicycleId, callbackURL, len(jsonData)) } +// BuildCallbackPayload 构建回调请求体 +type BuildCallbackPayload struct { + TaskId string `json:"taskId"` + Status int `json:"status"` + Messages any `json:"messages"` + ErrorMsg string `json:"errorMsg"` +} + +// CallbackBuildResult 回调构建结果 +func CallbackBuildResult(ctx context.Context, record *entity.ModelGatewayBuildRecord) { + headers := make(map[string]string) + if r := g.RequestFromCtx(ctx); r != nil { + for k, v := range r.Request.Header { + if len(v) > 0 { + headers[k] = v[0] + } + } + } + payload := BuildCallbackPayload{ + TaskId: record.TaskID, + Status: record.Status, + Messages: record.ResultMessages, + ErrorMsg: record.ErrorMsg, + } + jsonData, _ := json.Marshal(payload) + + var resp struct{} + if err := commonHttp.Post(ctx, record.CallbackURL, headers, &resp, jsonData); err != nil { + g.Log().Warningf(ctx, "[构建回调] 发送失败 taskId=%s err=%v", record.TaskID, err) + return + } + g.Log().Infof(ctx, "[构建回调] 发送成功 taskId=%s", record.TaskID) +} + // IsSuperAdmin 调用admin-go服务检查是否是超级管理员 func IsSuperAdmin(ctx context.Context) (res bool, err error) { //headers := util.ForwardHeaders(ctx) @@ -184,6 +219,63 @@ func IsSuperAdmin(ctx context.Context) (res bool, err error) { return r["isSuperAdmin"], err } +// SkillUserVO 技能用户视图对象 +type SkillUserVO struct { + Id int64 `json:"id,string"` + Name string `json:"name"` + Description string `json:"description"` + FileName string `json:"fileName"` + FileUrl string `json:"fileUrl"` + CreatedAt *gtime.Time `json:"createdAt"` + UpdatedAt *gtime.Time `json:"updatedAt"` + ImgAddressPrefix string `json:"imgAddressPrefix"` +} + +// GetSkillUser 获取技能用户信息 +func GetSkillUser(ctx context.Context, name string) (*SkillUserVO, error) { + fullURL := fmt.Sprintf("ai-agent/skill/user/getUserOrTemplate?name=%s", name) + //headers := util.ForwardHeaders(ctx) + headers := make(map[string]string) + if r := g.RequestFromCtx(ctx); r != nil { + for k, v := range r.Request.Header { + if len(v) > 0 { + headers[k] = v[0] + } + } + } + var resp SkillUserVO + var req struct{} + if err := commonHttp.Get(ctx, fullURL, headers, &resp, req); err != nil { + return nil, err + } + return &resp, nil +} + +// SessionHistoryItem 会话历史条目 +type SessionHistoryItem struct { + Role string `json:"role"` + Content string `json:"content"` +} + +// GetSessionHistory 获取会话历史 +func GetSessionHistory(ctx context.Context, nodeId, sessionId string) ([]SessionHistoryItem, error) { + fullURL := fmt.Sprintf("model-session/session/history?nodeId=%s&sessionId=%s", nodeId, sessionId) + headers := make(map[string]string) + if r := g.RequestFromCtx(ctx); r != nil { + for k, v := range r.Request.Header { + if len(v) > 0 { + headers[k] = v[0] + } + } + } + var req struct{} + var resp []SessionHistoryItem + if err := commonHttp.Get(ctx, fullURL, headers, &resp, req); err != nil { + return nil, err + } + return resp, nil +} + //// callback 向回调地址 POST 任务结果(与查询接口 GetTaskRes 出参一致) //func (s *audioTaskService) callback(ctx context.Context, taskID, status, errMsg, callbackURL string) { // if callbackURL == "" { diff --git a/service/prompt/prompt_files_handle_service.go b/service/prompt/prompt_files_handle_service.go new file mode 100644 index 0000000..d808075 --- /dev/null +++ b/service/prompt/prompt_files_handle_service.go @@ -0,0 +1,282 @@ +package prompt + +import ( + "archive/zip" + "bytes" + "context" + "fmt" + "io" + "model-gateway/common/util" + "model-gateway/service/gateway" + "net/http" + "strings" + "time" + + "github.com/gogf/gf/v2/frame/g" +) + +const ( + bytesPerKB = 1024 + bytesPerMB = 1024 * 1024 +) + +// FetchFileTextsAsString 从 URL 列表获取文件内容,拼接为字符串 +func FetchFileTextsAsString(ctx context.Context, urls []string) string { + if len(urls) == 0 { + return "" + } + + client := createHTTPClient(ctx, "userFiles.httpTimeoutSec", 8) + var builder strings.Builder + + for _, rawURL := range urls { + url := util.SanitizeURL(rawURL) + if url == "" || util.IsBannedExtension(url) { + continue + } + + if util.IsZipExtension(url) { + for _, text := range fetchZipFileTexts(ctx, client, url) { + builder.WriteString(text) + builder.WriteString("\n") + } + continue + } + + if text := fetchAndCleanFileContent(ctx, client, url); text != "" { + builder.WriteString(fmt.Sprintf("【文件:%s】\n%s\n", url, text)) + } + } + + return builder.String() +} + +// fetchAndCleanFileContent 获取并清理文件内容 +func fetchAndCleanFileContent(ctx context.Context, client *http.Client, url string) string { + text, err := fetchFileContent(ctx, client, url) + if err != nil || text == "" { + return "" + } + return util.CleanSymbols(text) +} + +// fetchZipFileTexts 下载并解压 zip 文件,提取可读文本内容 +func fetchZipFileTexts(ctx context.Context, client *http.Client, url string) map[string]string { + result := make(map[string]string) + + maxSize := int64(g.Cfg().MustGet(ctx, "userFiles.zipMaxSizeMB", 10).Int()) * bytesPerMB + zipBytes, err := downloadFile(client, url, maxSize) + if err != nil { + return result + } + + reader, err := zip.NewReader(bytes.NewReader(zipBytes), int64(len(zipBytes))) + if err != nil { + return result + } + + entryMaxSize := int64(g.Cfg().MustGet(ctx, "userFiles.zipEntryMaxSizeKB", 500).Int()) * bytesPerKB + + for _, file := range reader.File { + if shouldSkipZipEntry(file.Name) { + continue + } + + if text := extractZipEntryContent(file, entryMaxSize); text != "" { + result[url+"::"+file.Name] = text + } + } + + return result +} + +// shouldSkipZipEntry 判断是否应该跳过 zip 条目 +func shouldSkipZipEntry(fileName string) bool { + return util.IsBannedExtension(fileName) || util.IsZipExtension(fileName) +} + +// extractZipEntryContent 提取 zip 条目内容 +func extractZipEntryContent(file *zip.File, maxSize int64) string { + rc, err := file.Open() + if err != nil { + return "" + } + defer rc.Close() + + content, err := io.ReadAll(io.LimitReader(rc, maxSize)) + if err != nil { + return "" + } + + if !util.IsReadableContentType(http.DetectContentType(content)) { + return "" + } + + text := util.CleanSymbols(string(content)) + if text == "" { + return "" + } + + return text +} + +// downloadFile 下载文件,限制最大大小 +func downloadFile(client *http.Client, url string, maxSize int64) ([]byte, error) { + req, err := http.NewRequest(http.MethodGet, url, nil) + if err != nil { + return nil, fmt.Errorf("创建请求失败: %w", err) + } + + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("执行请求失败: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, fmt.Errorf("HTTP %d", resp.StatusCode) + } + + body, err := io.ReadAll(io.LimitReader(resp.Body, maxSize)) + if err != nil { + return nil, fmt.Errorf("读取响应失败: %w", err) + } + + return body, nil +} + +// fetchFileContent 获取单个文本文件内容 +func fetchFileContent(ctx context.Context, client *http.Client, url string) (string, error) { + req, err := http.NewRequest(http.MethodGet, url, nil) + if err != nil { + return "", fmt.Errorf("创建请求失败: %w", err) + } + + resp, err := client.Do(req) + if err != nil { + return "", fmt.Errorf("执行请求失败: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return "", fmt.Errorf("HTTP %d", resp.StatusCode) + } + + contentType := resp.Header.Get("Content-Type") + if !util.IsReadableContentType(contentType) { + return "", fmt.Errorf("不可读的内容类型: %s", contentType) + } + + maxSize := int64(g.Cfg().MustGet(ctx, "userFiles.textFileMaxSizeKB", 500).Int()) * bytesPerKB + body, err := io.ReadAll(io.LimitReader(resp.Body, maxSize)) + if err != nil { + return "", fmt.Errorf("读取响应失败: %w", err) + } + + return strings.TrimSpace(string(body)), nil +} + +func SkillMdContent(ctx context.Context, skillName string) string { + if skillName == "" { + return "" + } + skillResp, err := gateway.GetSkillUser(ctx, skillName) + if err != nil { + g.Log().Warningf(ctx, "[SkillMd] GetSkillUser 失败: %v", err) + return "" + } + + fullUrl := skillResp.ImgAddressPrefix + skillResp.FileUrl + + client := createHTTPClient(ctx, "skillFiles.httpTimeoutSec", 30) + maxSize := int64(g.Cfg().MustGet(ctx, "skillFiles.zipMaxSizeMB", 10).Int()) * bytesPerMB + + zipBytes, err := downloadFile(client, fullUrl, maxSize) + if err != nil { + g.Log().Warningf(ctx, "[SkillMd] 下载失败 url=%s err=%v", fullUrl, err) + return "" + } + + mdContents, err := extractMdFiles(ctx, zipBytes) + if err != nil || len(mdContents) == 0 { + g.Log().Warningf(ctx, "[SkillMd] 提取md失败 count=%d err=%v", len(mdContents), err) + return "" + } + + return buildSkillMarkdown(skillResp, mdContents) +} + +// buildSkillMarkdown 构建技能 Markdown 内容 +func buildSkillMarkdown(skillResp *gateway.SkillUserVO, mdContents map[string]string) string { + var builder strings.Builder + + builder.WriteString(fmt.Sprintf("# Skill: %s\n\n", skillResp.Name)) + if skillResp.Description != "" { + builder.WriteString(fmt.Sprintf("> %s\n\n", skillResp.Description)) + } + + for fileName, content := range mdContents { + builder.WriteString(fmt.Sprintf("## %s\n\n", fileName)) + builder.WriteString(content) + builder.WriteString("\n\n---\n\n") + } + + return strings.TrimSpace(builder.String()) +} + +// extractMdFiles 解压 zip 并提取所有 .md 文件内容 +func extractMdFiles(ctx context.Context, zipBytes []byte) (map[string]string, error) { + result := make(map[string]string) + + reader, err := zip.NewReader(bytes.NewReader(zipBytes), int64(len(zipBytes))) + if err != nil { + return nil, fmt.Errorf("创建 zip 阅读器失败: %w", err) + } + + entryMaxSize := int64(g.Cfg().MustGet(ctx, "skillFiles.mdMaxSizeKB", 500).Int()) * bytesPerKB + + for _, file := range reader.File { + if file.FileInfo().IsDir() || !isMarkdownFile(file.Name) { + continue + } + + if content := readMarkdownFileContent(file, entryMaxSize); content != "" { + result[file.Name] = content + } + } + + return result, nil +} + +// isMarkdownFile 判断是否为 Markdown 文件 +func isMarkdownFile(fileName string) bool { + return strings.HasSuffix(strings.ToLower(fileName), ".md") +} + +// readMarkdownFileContent 读取 Markdown 文件内容 +func readMarkdownFileContent(file *zip.File, maxSize int64) string { + rc, err := file.Open() + if err != nil { + return "" + } + defer rc.Close() + + content, err := io.ReadAll(io.LimitReader(rc, maxSize)) + if err != nil { + return "" + } + + if len(content) == 0 { + return "" + } + + return strings.TrimSpace(string(content)) +} + +// createHTTPClient 创建 HTTP 客户端 +func createHTTPClient(ctx context.Context, configKey string, defaultSeconds int) *http.Client { + timeout := time.Duration(g.Cfg().MustGet(ctx, configKey, defaultSeconds).Int()) * time.Second + return &http.Client{ + Timeout: timeout, + } +} diff --git a/service/task/task_service.go b/service/task/task_service.go index 7fc295d..1444dd8 100644 --- a/service/task/task_service.go +++ b/service/task/task_service.go @@ -6,6 +6,8 @@ import ( "fmt" "model-gateway/common/util" "model-gateway/consts/public" + "model-gateway/service/gateway" + "model-gateway/service/prompt" "sync" "time" @@ -16,6 +18,7 @@ import ( "gitea.redpowerfuture.com/red-future/common/beans" "gitea.redpowerfuture.com/red-future/common/utils" "github.com/gogf/gf/v2/database/gdb" + "github.com/gogf/gf/v2/encoding/gjson" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/os/grpool" "github.com/gogf/gf/v2/util/gconv" @@ -26,6 +29,182 @@ var ModelGatewayTask = &taskService{} type taskService struct{} +// BuildMessages 构建消息(异步) +func (s *taskService) BuildMessages(ctx context.Context, req *dto.BuildMessagesReq) (*dto.BuildMessagesRes, error) { + user, err := utils.GetUserInfo(ctx) + if err != nil { + return nil, err + } + model, err := dao.ModelGatewayModels.Get(ctx, &entity.ModelGatewayModel{ + SQLBaseDO: beans.SQLBaseDO{TenantId: user.TenantId, Creator: user.UserName}, + ModelName: req.ModelName, + }) + if err != nil || model == nil { + return nil, err + } + + // 1) 创建构建记录 + taskId := uuid.NewString() + record := &entity.ModelGatewayBuildRecord{ + TaskID: taskId, + BuildType: req.BuildType, + ModelName: req.ModelName, + SkillName: req.SkillName, + SessionID: req.SessionId, + NodeID: req.NodeId, + RequestMessages: req.Messages, + CallbackURL: req.CallbackUrl, + Status: 0, + } + _, err = dao.ModelGatewayBuildRecord.Insert(ctx, record) + if err != nil { + return nil, err + } + + // 2) 异步执行构建 + go s.executeBuild(util.AsyncCtx(ctx), record, req, model) + + return &dto.BuildMessagesRes{TaskId: taskId}, nil +} + +// executeBuild 异步执行构建逻辑 +func (s *taskService) executeBuild(ctx context.Context, record *entity.ModelGatewayBuildRecord, req *dto.BuildMessagesReq, model *entity.ModelGatewayModel) { + var ( + startTime = time.Now() + result []map[string]any + err error + ) + result, err = s.buildResult(ctx, req, model, record) + record.DurationSeconds = int(time.Since(startTime).Seconds()) + if err != nil { + record.Status = 2 + record.ErrorMsg = err.Error() + } else { + record.Status = 1 + record.ResultMessages = result + } + _, _ = dao.ModelGatewayBuildRecord.Update(ctx, record) + gateway.CallbackBuildResult(ctx, record) +} + +// buildResult 构建结果:推理模型手动拼接,视频模型调模型生成多轮 +func (s *taskService) buildResult(ctx context.Context, req *dto.BuildMessagesReq, model *entity.ModelGatewayModel, record *entity.ModelGatewayBuildRecord) ([]map[string]any, error) { + messages := req.Messages + switch { + case model.ModelType == public.ModelTypeInference: + // 推理模型:拼接提示词 + 历史 + systemPrompt := util.GetModelPrompt(ctx, model.ModelType) + skillContent := prompt.SkillMdContent(ctx, req.SkillName) + + systemKey := util.GetRoleContentPath(model.Form, "system") + if systemKey != "" { + messages = util.MergePrompt(messages, systemKey, systemPrompt, skillContent, req.CustomPrompt) + } + + history, _ := gateway.GetSessionHistory(ctx, req.NodeId, req.SessionId) + if len(history) > 0 { + messages = util.InjectHistory(messages, history, []string{"system", "history", "user"}) + } + // 检查附件是否需要拆分多轮 + rounds := util.SplitByAttachment(messages, model.Form) + if len(rounds) > 0 { + return rounds, nil + } + return []map[string]any{messages}, nil + case model.ModelType >= 600 && model.ModelType < 700: + // 视频模型:调推理模型生成多轮 + chatModel, err := dao.ModelGatewayModels.Get(ctx, &entity.ModelGatewayModel{ + SQLBaseDO: beans.SQLBaseDO{TenantId: model.TenantId, Creator: model.Creator}, + IsChatModel: gconv.PtrInt(1), + }) + if err != nil || chatModel == nil { + return nil, fmt.Errorf("未找到对话模型") + } + + protocol, err := dao.ProviderProtocol.Get(ctx, &entity.ProviderProtocol{ + ProviderName: chatModel.OperatorName, + Status: 1, + }) + if err != nil || protocol == nil { + return nil, fmt.Errorf("未找到协议配置: %s", chatModel.OperatorName) + } + + template := util.BuildTemplateFromForm(model.Form) + outputStruct := gjson.New(template).MustToJsonString() + + durationForm, _ := util.GetFormByRole(model.Form, "duration") + totalDur := gconv.Int(gjson.New(req.Messages).Get(durationForm.Key).Val()) + minDur := gconv.Int(durationForm.FieldConstraint.Min) + maxDur := gconv.Int(durationForm.FieldConstraint.Max) + + systemPrompt := fmt.Sprintf(protocol.SystemPromptTemplate, outputStruct, totalDur, minDur, maxDur) + userContent := util.ExtractUserContent(req.Messages) + reqBody := util.BuildRequestBody(protocol.RequestTemplate, chatModel.ModelName, systemPrompt, userContent) + + task := &entity.ModelGatewayTask{ + ModelName: chatModel.ModelName, + TaskID: record.TaskID, + State: public.TaskStatusRunning, + BizName: "model-gateway", + RequestPayload: reqBody, + BuildType: req.BuildType, + } + id, err := dao.ModelGatewayTask.Insert(ctx, task) + if err != nil { + return nil, err + } + task.Id = id + + rawData, err := AsyncWorker.callModel(chatModel, reqBody) + if err != nil { + task.State = public.TaskStatusFailed + task.ErrorMsg = err.Error() + _, _ = dao.ModelGatewayTask.Update(ctx, task) + return nil, err + } + + mapped, err := util.MapResponsePayload(chatModel.ResponseMapping, rawData) + if err != nil { + return nil, err + } + if _, ok := mapped[entity.TotalTokens]; ok { + task.ExpendTokens = gconv.Int64(mapped[entity.TotalTokens]) + } + + var rounds []map[string]any + contentStr := gjson.New(mapped).Get(entity.ResponseBody).String() + if contentStr != "" { + if err = gjson.DecodeTo(contentStr, &rounds); err != nil { + task.State = public.TaskStatusFailed + task.ErrorMsg = err.Error() + _, _ = dao.ModelGatewayTask.Update(ctx, task) + return nil, err + } + } + + oss, err := gateway.UploadByTask(ctx, gjson.New(rounds).MustToJson(), "json") + if err != nil { + task.State = public.TaskStatusFailed + task.ErrorMsg = err.Error() + _, _ = dao.ModelGatewayTask.Update(ctx, task) + return nil, err + } + + task.State = public.TaskStatusSuccess + task.ResultFile = &entity.ResultFile{ + OssFile: oss.FileAddressPrefix + oss.FileURL, + FileType: oss.FileFormat, + FileSize: int64(oss.FileSize), + } + _, _ = dao.ModelGatewayTask.Update(ctx, task) + + return rounds, nil + + default: + return nil, errors.New("不支持的模型类型") + } +} + // Create 创建任务 func (s *taskService) Create(ctx context.Context, req *dto.CreateTaskReq) (res *dto.CreateTaskRes, err error) { taskID := uuid.NewString() @@ -242,7 +421,7 @@ func (s *taskService) GetBatch(ctx context.Context, req *dto.GetTaskBatchReq) (r if t == nil { continue } - if t.State != public.BuildTypeNode { + if t.State != 2 { continue } _ = dao.ModelGatewayTask.MarkDownloadedByID(ctx, t.Id) diff --git a/service/task/worker.go b/service/task/worker.go index e4cfd7d..8fb0532 100644 --- a/service/task/worker.go +++ b/service/task/worker.go @@ -52,18 +52,18 @@ func (w *asyncWorker) handleOne(ctx context.Context, task *entity.ModelGatewayTa } switch { - case model.CallMode != nil && *model.CallMode == public.CallModeStream: - rawBytes, err = InvokeModel(ctx, model, body) + case model.CallMode != nil && *model.CallMode == public.CallModeStream: // 流式 + rawBytes, err = InvokeModel(model, body) if err == nil { result, err = util.ParseStreamResponse(rawBytes, model.StreamConfig) } - case model.CallMode != nil && *model.CallMode == public.CallModeAsync: - result, err = w.callModel(ctx, model, body) + case model.CallMode != nil && *model.CallMode == public.CallModeAsync: // 异步 + result, err = w.callModel(model, body) if err == nil { result, err = util.PullTaskResult(ctx, result, model.QueryConfig, model.HeadMsg) } default: - result, err = w.callModel(ctx, model, body) + result, err = w.callModel(model, body) } if err == nil { @@ -143,7 +143,7 @@ var asyncTaskChan = sync.Map{} // taskID → chan asyncResult func (w *asyncWorker) callModelAsync(ctx context.Context, model *entity.ModelGatewayModel, body map[string]any) (map[string]any, error) { // 1. 提交异步任务 - body, err := w.callModel(ctx, model, body) + body, err := w.callModel(model, body) if err != nil { return nil, err } @@ -185,8 +185,8 @@ func NotifyAsyncResult(taskID string, result map[string]any, err error) { } // callModel 调用模型 + 提取文本结果 -func (w *asyncWorker) callModel(ctx context.Context, model *entity.ModelGatewayModel, body map[string]any) (map[string]any, error) { - data, err := InvokeModel(ctx, model, body) +func (w *asyncWorker) callModel(model *entity.ModelGatewayModel, body map[string]any) (map[string]any, error) { + data, err := InvokeModel(model, body) if err != nil { return nil, err } @@ -234,13 +234,13 @@ func (w *asyncWorker) parseAndRetry(ctx context.Context, body map[string]any, ta // 3) 解析 + 校验 var parsed map[string]any switch task.BuildType { - case public.BuildTypePrompt, public.BuildTypeNode: + case 1, 3: parsed, err = util.ParseAndValidate(mapped, model.RequiredFields) if err == nil { return parsed, nil } lastErr = err - case public.BuildTypeStruct: + case 2: return util.ParseStructResult(mapped, entity.ResponseBody), nil default: return mapped, nil @@ -257,7 +257,7 @@ func (w *asyncWorker) parseAndRetry(ctx context.Context, body map[string]any, ta _, _ = dao.ModelGatewayTask.Update(ctx, task) body = injectErrorMessage(task.RequestPayload, lastErr) - rawData, callErr := InvokeModel(ctx, model, body) + rawData, callErr := InvokeModel(model, body) if callErr != nil { g.Log().Warningf(ctx, "[执行任务][重调模型失败] taskId=%s attempt=%d/%d err=%v", task.TaskID, attempt, maxRetry, callErr) continue @@ -318,14 +318,7 @@ func injectErrorMessage(payload map[string]any, err error) map[string]any { // InvokeModel 调用模型服务,返回二进制结果 // modelKey 用于覆盖/补充模型配置 head_msg(例如每次请求携带不同的 X-API-Key) -func InvokeModel(ctx context.Context, model *entity.ModelGatewayModel, body map[string]any) ([]byte, error) { - // 1) 记录模型调用次数 - //_ = dao.ModelGatewayLogsStat.IncRequestCount(ctx, time.Now(), model.TenantId, model.Creator, model.ModelName) - - // 2)请求参数映射:将标准 payload 按模型配置的 requestMapping 转为模型需要的格式 - //—— 请求映射实际处理为提示词构建请求,因为有附加字段及其他字段的拼接。这里不方便做请求映射 - //mappedPayload := util.ReverseMap(model.RequestMapping, payload) - +func InvokeModel(model *entity.ModelGatewayModel, body map[string]any) ([]byte, error) { // 3)构建请求 URL 和超时 baseURL := strings.TrimRight(model.BaseURL, "/") timeout := time.Duration(model.TimeoutSeconds) * time.Second