red-future/model-gateway
4
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Dev未优化 #4

Open
Ghost wants to merge 19 commits from dev未优化 into dev优化中
pull from: dev未优化
merge into: :dev优化中
Conversation 0 Commits 19 Files Changed 21
+1448 -123
21 changed files with 1448 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile
+2 -1
View File
@@ -1,7 +1,8 @@
# 阶段1: 构建
FROM golang:alpine AS builder
RUN apk add --no-cache git ca-certificates tzdata
RUN sed -i 's/dl-cdn.alpinelinux.org/mirrors.aliyun.com/g' /etc/apk/repositories && \
apk add --no-cache git ca-certificates tzdata
ENV TZ=Asia/Shanghai
RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
common/util/billing.go
+49 -27
View File
@@ -4,7 +4,6 @@ import (
"context"
"fmt"
"math"
"model-gateway/service/gateway"
"strings"
"github.com/gogf/gf/v2/encoding/gjson"
@@ -22,6 +21,8 @@ func CalculateBilling(config map[string]any, billingData map[string]any) map[str
return calculateInferenceTierBilling(config, billingData)
case "video_resolution": //视频模型计费
return calculateVideoResolutionBilling(config, billingData)
case "tts":
return calculateTTSBilling(config, billingData)
}
return nil
}
@@ -33,7 +34,6 @@ func calculateInferenceTierBilling(config map[string]any, data map[string]any) m
completionTokens := gconv.Int64(data["completion_tokens"])
hasAudio := gconv.Bool(data["has_audio"])
inputK := promptTokens / 1000
tiers := config["pricing"].(map[string]any)["tiers"].([]any)
var matched map[string]any
for _, t := range tiers {
@@ -54,7 +54,6 @@ func calculateInferenceTierBilling(config map[string]any, data map[string]any) m
inputPrice = gconv.Float64(matched["input_price"])
}
outputPrice := gconv.Float64(matched["output_price"])
inputCost := float64(promptTokens) * inputPrice / 1000000
outputCost := float64(completionTokens) * outputPrice / 1000000
@@ -116,6 +115,22 @@ func calculateVideoResolutionBilling(config map[string]any, data map[string]any)
}
}
func calculateTTSBilling(config map[string]any, data map[string]any) map[string]any {
usage := gconv.Float64(data["synthesize_text_length"])
unitPrice := gconv.Float64(config["pricing"])
totalFee := usage * unitPrice
return map[string]any{
"model_name": data["model_name"],
"total_tokens": int64(usage),
"total_fee": totalFee,
// 明细
"prompt_tokens": 0,
"completion_tokens": int64(usage),
"unit_price": unitPrice,
}
}
// ======================== 数据提取 ========================
func ExtractRequestBilling(ctx context.Context, config map[string]any, requestPayload map[string]any) map[string]any {
@@ -124,7 +139,15 @@ func ExtractRequestBilling(ctx context.Context, config map[string]any, requestPa
data := make(map[string]any)
for key, path := range fields {
data[key] = extractValue(requestPayload, gconv.String(path))
val := extractValue(requestPayload, gconv.String(path))
// TTS 模型：input_chars 自动计算字符数
if key == "input_chars" {
if s, ok := val.(string); ok {
data[key] = len([]rune(s))
continue
}
}
data[key] = val
}
if defaults, ok := config["defaults"].(map[string]any); ok {
@@ -135,29 +158,28 @@ func ExtractRequestBilling(ctx context.Context, config map[string]any, requestPa
}
}
// 处理 compute 字段
if compute, ok := config["compute"].(map[string]any); ok {
for targetField, rule := range compute {
r := rule.(map[string]any)
dependsOn := gconv.String(r["depends_on"])
dependsValue := gconv.Bool(r["depends_value"])
if gconv.Bool(data[dependsOn]) != dependsValue {
continue
}
switch r["service"] {
case "video_duration":
urls := extractVideoUrls(requestPayload)
if len(urls) > 0 {
resp, err := gateway.GetVideoDuration(ctx, urls)
if err == nil {
data[targetField] = resp.TotalDuration
}
}
}
}
}
//if compute, ok := config["compute"].(map[string]any); ok {
// for targetField, rule := range compute {
// r := rule.(map[string]any)
// dependsOn := gconv.String(r["depends_on"])
// dependsValue := gconv.Bool(r["depends_value"])
//
// if gconv.Bool(data[dependsOn]) != dependsValue {
// continue
// }
//
// switch r["service"] {
// case "video_duration":
// urls := extractVideoUrls(requestPayload)
// if len(urls) > 0 {
// resp, err := gateway.GetVideoDuration(ctx, urls)
// if err == nil {
// data[targetField] = resp.TotalDuration
// }
// }
// }
// }
//}
return data
}
common/util/mapping.go
+39 -11
View File
@@ -33,24 +33,45 @@ func ParseAndValidate(raw map[string]any, requiredFields []string) (map[string]a
return r
}, contentStr)
var arr []any
if err := json.Unmarshal([]byte(contentStr), &arr); err != nil {
// 第一步：先解析为通用 interface{}，判断是对象还是数组
var data any
if err := json.Unmarshal([]byte(contentStr), &data); err != nil {
return raw, fmt.Errorf("JSON解析失败: %w", err)
}
if len(arr) == 0 {
return raw, fmt.Errorf("解析后数组为空")
var arr []any
switch val := data.(type) {
case []any:
// 本身就是数组，直接赋值
arr = val
case map[string]any:
// 单个对象，包装成单元素数组，统一后续逻辑
arr = []any{val}
default:
return raw, fmt.Errorf("不支持的JSON类型，仅允许对象/数组")
}
if len(arr) == 0 {
return raw, fmt.Errorf("解析后数据数组为空")
}
// 校验每一项的必填字段
for _, field := range requiredFields {
for i, r := range arr {
round, _ := r.(map[string]any)
if round != nil && gjson.New(round).Get(field).IsNil() {
for i, item := range arr {
itemMap, ok := item.(map[string]any)
if !ok {
return raw, fmt.Errorf("rounds[%d] 不是合法JSON对象", i)
}
if gjson.New(itemMap).Get(field).IsNil() {
return raw, fmt.Errorf("rounds[%d] 缺少必填字段: %s", i, field)
}
}
}
return map[string]any{"total_rounds": len(arr), "rounds": arr}, nil
return map[string]any{
"total_rounds": len(arr),
"rounds": arr,
}, nil
}
// ParseStructResult 解析结构结果
@@ -263,10 +284,17 @@ 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("error.message").Val())
if errMsg == "" {
errMsg = gconv.String(gjson.New(result).Get("error").Val())
}
if errMsg == "" {
rawBytes, _ := json.Marshal(result)
errMsg = string(rawBytes)
}
g.Log().Errorf(ctx, "[PullTaskResult] 任务失败 taskID=%s err=%s", taskID, errMsg)
return result, fmt.Errorf("任务失败: %s", errMsg)
}
time.Sleep(time.Duration(interval) * time.Second)
}
}
consts/model/model_supplier.go
+108
View File
@@ -0,0 +1,108 @@
package model
import (
"model-gateway/consts/public"
"github.com/gogf/gf/v2/util/gconv"
)
// 供应商编码常量
const (
SupplierAliyun = 1
SupplierVolcengine = 2
SupplierTencent = 3
SupplierHuawei = 4
SupplierBaidu = 5
SupplierOpenAI = 6
SupplierAzure = 7
SupplierAWS = 8
SupplierGoogle = 9
SupplierDeepSeek = 10
SupplierMoonshot = 11
SupplierZhipu = 12
SupplierBaichuan = 13
SupplierMinimax = 14
SupplierXunfei = 15
SupplierOthers = 16
)
// SupplierType 供应商编码类型
type SupplierType *int8
// SupplierItem 供应商项
type SupplierItem struct {
Code SupplierType `json:"code"`
Desc string `json:"desc"`
}
// 名称映射【唯一文案维护】
var supplierNameMap = map[int]string{
SupplierAliyun: "阿里云百炼",
SupplierVolcengine: "火山引擎",
SupplierTencent: "腾讯云",
SupplierHuawei: "华为云",
SupplierBaidu: "百度智能云",
SupplierOpenAI: "OpenAI",
SupplierAzure: "Azure OpenAI",
SupplierAWS: "AWS Bedrock",
SupplierGoogle: "Google Cloud",
SupplierDeepSeek: "DeepSeek",
SupplierMoonshot: "Moonshot",
SupplierZhipu: "智谱AI",
SupplierBaichuan: "百川智能",
SupplierMinimax: "MiniMax",
SupplierXunfei: "科大讯飞",
SupplierOthers: "其他",
}
// 供应商展示顺序
var supplierOrder = []int{
SupplierAliyun, SupplierVolcengine, SupplierTencent, SupplierHuawei, SupplierBaidu,
SupplierOpenAI, SupplierAzure, SupplierAWS, SupplierGoogle, SupplierDeepSeek,
SupplierMoonshot, SupplierZhipu, SupplierBaichuan, SupplierMinimax, SupplierXunfei, SupplierOthers,
}
// 全局供应商实例
var (
SupplierItemAliyun = newSupplierItem(gconv.PtrInt8(SupplierAliyun))
SupplierItemVolcengine = newSupplierItem(gconv.PtrInt8(SupplierVolcengine))
SupplierItemTencent = newSupplierItem(gconv.PtrInt8(SupplierTencent))
SupplierItemHuawei = newSupplierItem(gconv.PtrInt8(SupplierHuawei))
SupplierItemBaidu = newSupplierItem(gconv.PtrInt8(SupplierBaidu))
SupplierItemOpenAI = newSupplierItem(gconv.PtrInt8(SupplierOpenAI))
SupplierItemAzure = newSupplierItem(gconv.PtrInt8(SupplierAzure))
SupplierItemAWS = newSupplierItem(gconv.PtrInt8(SupplierAWS))
SupplierItemGoogle = newSupplierItem(gconv.PtrInt8(SupplierGoogle))
SupplierItemDeepSeek = newSupplierItem(gconv.PtrInt8(SupplierDeepSeek))
SupplierItemMoonshot = newSupplierItem(gconv.PtrInt8(SupplierMoonshot))
SupplierItemZhipu = newSupplierItem(gconv.PtrInt8(SupplierZhipu))
SupplierItemBaichuan = newSupplierItem(gconv.PtrInt8(SupplierBaichuan))
SupplierItemMinimax = newSupplierItem(gconv.PtrInt8(SupplierMinimax))
SupplierItemXunfei = newSupplierItem(gconv.PtrInt8(SupplierXunfei))
SupplierItemOthers = newSupplierItem(gconv.PtrInt8(SupplierOthers))
)
func newSupplierItem(code SupplierType) SupplierItem {
val := int(*code)
return SupplierItem{
Code: code,
Desc: supplierNameMap[val],
}
}
// GetSupplierDescByCode 根据编码获取供应商名称
func GetSupplierDescByCode(code int) string {
return supplierNameMap[code]
}
// GetSupplierOptionList 获取供应商下拉列表
func GetSupplierOptionList() []*public.Option {
var list []*public.Option
for _, code := range supplierOrder {
list = append(list, &public.Option{
Value: code,
Label: supplierNameMap[code],
})
}
return list
}
consts/model/model_type.go
+186
View File
@@ -0,0 +1,186 @@
package model
import (
"model-gateway/consts/public"
"github.com/gogf/gf/v2/util/gconv"
)
// 模型类型编码常量
const (
TypeInference = 100 // 推理模型
TypeImage = 200 // 图片模型
TypeAudio = 300 // 音频模型
TypeVector = 400 // 向量化模型
TypeOmni = 500 // 全模态模型
TypeVideo = 600 // 视频模型
// 图片子类型
ImageSubTextToImage = 201
ImageSubImageToImage = 202
ImageSubImageEdit = 203
ImageSubImageVariation = 204
ImageSubImageTextToImage = 205
// 音频子类型
AudioSubTextToSpeech = 301
AudioSubSpeechToText = 302
AudioSubSpeechToSpeech = 303
// 向量化子类型
VectorSubEmbedding = 401
VectorSubRerank = 402
// 全模态子类型
OmniSubTextImageAudio = 501
OmniSubVision = 502
// 视频子类型
VideoSubTextToVideo = 601
VideoSubImageToVideo = 602
VideoSubImageTextToVideo = 603
VideoSubVideoToVideo = 604
)
// ModelType 编码类型
type ModelType *int
// ModelTypeItem 模型类型项
type ModelTypeItem struct {
Code ModelType `json:"code"`
Desc string `json:"desc"`
}
// TypeTree 树形结构
type TypeTree struct {
Value int `json:"value"`
Label string `json:"label"`
Children []*public.Option `json:"children"`
}
// 名称映射表【唯一文案维护入口】
var typeNameMap = map[int]string{
TypeInference: "推理模型",
TypeImage: "图片模型",
TypeAudio: "音频模型",
TypeVector: "向量化模型",
TypeOmni: "全模态模型",
TypeVideo: "视频模型",
ImageSubTextToImage: "文生图",
ImageSubImageToImage: "图生图",
ImageSubImageEdit: "图片编辑",
ImageSubImageVariation: "图片变体",
ImageSubImageTextToImage: "图文生图",
AudioSubTextToSpeech: "文生音",
AudioSubSpeechToText: "音生文",
AudioSubSpeechToSpeech: "音生音",
VectorSubEmbedding: "文本嵌入",
VectorSubRerank: "重排序",
OmniSubTextImageAudio: "文图音",
OmniSubVision: "视觉理解",
VideoSubTextToVideo: "文生视频",
VideoSubImageToVideo: "图生视频",
VideoSubImageTextToVideo: "图文生视频",
VideoSubVideoToVideo: "视频生视频",
}
// 父子级映射（仅存有子项的分类）
var parentChildMap = map[int][]int{
TypeImage: {ImageSubTextToImage, ImageSubImageToImage, ImageSubImageEdit, ImageSubImageVariation, ImageSubImageTextToImage},
TypeAudio: {AudioSubTextToSpeech, AudioSubSpeechToText, AudioSubSpeechToSpeech},
TypeVector: {VectorSubEmbedding, VectorSubRerank},
TypeOmni: {OmniSubTextImageAudio, OmniSubVision},
TypeVideo: {VideoSubTextToVideo, VideoSubImageToVideo, VideoSubImageTextToVideo, VideoSubVideoToVideo},
}
// 一级分类展示顺序
var parentTypeOrder = []int{
TypeInference, TypeImage, TypeAudio, TypeVector, TypeOmni, TypeVideo,
}
// 全局实例：一级 + 全部二级子类型，统一通过 newItem 构造，文案仅维护在 typeNameMap
var (
// 一级类型
ModelTypeInference = newItem(gconv.PtrInt(TypeInference))
ModelTypeImage = newItem(gconv.PtrInt(TypeImage))
ModelTypeAudio = newItem(gconv.PtrInt(TypeAudio))
ModelTypeVector = newItem(gconv.PtrInt(TypeVector))
ModelTypeOmni = newItem(gconv.PtrInt(TypeOmni))
ModelTypeVideo = newItem(gconv.PtrInt(TypeVideo))
// 图片二级子类型
ModelImageSubTextToImage = newItem(gconv.PtrInt(ImageSubTextToImage))
ModelImageSubImageToImage = newItem(gconv.PtrInt(ImageSubImageToImage))
ModelImageSubImageEdit = newItem(gconv.PtrInt(ImageSubImageEdit))
ModelImageSubImageVariation = newItem(gconv.PtrInt(ImageSubImageVariation))
ModelImageSubImageTextToImage = newItem(gconv.PtrInt(ImageSubImageTextToImage))
// 音频二级子类型
ModelAudioSubTextToSpeech = newItem(gconv.PtrInt(AudioSubTextToSpeech))
ModelAudioSubSpeechToText = newItem(gconv.PtrInt(AudioSubSpeechToText))
ModelAudioSubSpeechToSpeech = newItem(gconv.PtrInt(AudioSubSpeechToSpeech))
// 向量化二级子类型
ModelVectorSubEmbedding = newItem(gconv.PtrInt(VectorSubEmbedding))
ModelVectorSubRerank = newItem(gconv.PtrInt(VectorSubRerank))
// 全模态二级子类型
ModelOmniSubTextImageAudio = newItem(gconv.PtrInt(OmniSubTextImageAudio))
ModelOmniSubVision = newItem(gconv.PtrInt(OmniSubVision))
// 视频二级子类型
ModelVideoSubTextToVideo = newItem(gconv.PtrInt(VideoSubTextToVideo))
ModelVideoSubImageToVideo = newItem(gconv.PtrInt(VideoSubImageToVideo))
ModelVideoSubImageTextToVideo = newItem(gconv.PtrInt(VideoSubImageTextToVideo))
ModelVideoSubVideoToVideo = newItem(gconv.PtrInt(VideoSubVideoToVideo))
)
// newItem 构造方法：自动从 typeNameMap 读取描述
func newItem(code ModelType) ModelTypeItem {
val := int(*code)
return ModelTypeItem{
Code: code,
Desc: typeNameMap[val],
}
}
// GetDescByCode 根据编码获取名称
func GetDescByCode(code int) string {
return typeNameMap[code]
}
// GetTypeTreeList 生成树形数据
func GetTypeTreeList() []*TypeTree {
var list []*TypeTree
for _, parentCode := range parentTypeOrder {
tree := &TypeTree{
Value: parentCode,
Label: typeNameMap[parentCode],
Children: make([]*public.Option, 0),
}
if childCodes, ok := parentChildMap[parentCode]; ok {
for _, c := range childCodes {
tree.Children = append(tree.Children, &public.Option{
Value: c,
Label: typeNameMap[c],
})
}
}
list = append(list, tree)
}
return list
}
// GetAllTypeOption 全量平铺选项
func GetAllTypeOption() []*public.Option {
var list []*public.Option
for code, label := range typeNameMap {
list = append(list, &public.Option{Value: code, Label: label})
}
return list
}
consts/model/response_type.go
+27
View File
@@ -0,0 +1,27 @@
package model
import "github.com/gogf/gf/v2/util/gconv"
var (
ResponseTypeSync = newResponseType(gconv.PtrInt8(1), "sync") // 同步
ResponseTypeAsync = newResponseType(gconv.PtrInt8(2), "async") // 异步
ResponseTypeStream = newResponseType(gconv.PtrInt8(3), "stream") // 流
)
type ResponseType *int8
type responseType struct {
code ResponseType
desc string
}
func (s responseType) Code() ResponseType {
return s.code
}
func (s responseType) Desc() string {
return s.desc
}
func newResponseType(code ResponseType, desc string) responseType {
return responseType{code: code, desc: desc}
}
consts/public/public.go
+6
View File
@@ -1,5 +1,11 @@
package public
// Option 通用下拉选项
type Option struct {
Value int `json:"value"`
Label string `json:"label"`
}
const (
CallModeSync = 0 // 同步调用
CallModeAsync = 1 // 异步调用
consts/public/table_name.go
+5 -4
View File
@@ -5,8 +5,9 @@ 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" // 任务表
TableNameOpLog = "model_gateway_logs_op" // 操作日志表
TableNameStat = "model_gateway_logs_stat" // 按天统计表
TableNameModelManage = "model_gateway_model_manage"
)
controller/model_manage_controller.go
+56
View File
@@ -0,0 +1,56 @@
package controller
import (
"context"
"model-gateway/model/dto"
"model-gateway/service"
"gitea.redpowerfuture.com/red-future/common/beans"
)
// ModelManage 模型配置控制器
var ModelManage = new(modelManage)
type modelManage struct{}
// CreateModel 添加配置
func (c *modelManage) CreateModel(ctx context.Context, req *dto.CreateModelManageReq) (res *dto.CreateModelManageRes, err error) {
return service.ModelManage.Create(ctx, req)
}
// UpdateModel 更改配置
func (c *modelManage) UpdateModel(ctx context.Context, req *dto.UpdateModelManageReq) (res *beans.ResponseEmpty, err error) {
err = service.ModelManage.Update(ctx, req)
return
}
// DeleteModel 删除配置
func (c *modelManage) DeleteModel(ctx context.Context, req *dto.DeleteModelManageReq) (res *beans.ResponseEmpty, err error) {
err = service.ModelManage.Delete(ctx, req)
return
}
// GetModel 获取配置
func (c *modelManage) GetModel(ctx context.Context, req *dto.GetModelManageReq) (res *dto.GetModelManageRes, err error) {
return service.ModelManage.Get(ctx, req)
}
// ListModel 配置列表
func (c *modelManage) ListModel(ctx context.Context, req *dto.ListModelManageReq) (res *dto.ListModelManageRes, err error) {
return service.ModelManage.List(ctx, req)
}
// CheckChatModel 检查是否为聊天模型
func (c *modelManage) CheckChatModel(ctx context.Context, req *dto.CheckChatModelReq) (res *dto.CheckChatModelRes, err error) {
return service.ModelManage.CheckChatModel(ctx, req)
}
// ListType 模型类型列表
func (c *modelManage) ListType(ctx context.Context, req *dto.ModelTypeReq) (res *dto.ModelTypeRes, err error) {
return service.ModelManage.GetModelType(ctx, req)
}
// ListOperator 运营商列表
func (c *modelManage) ListOperator(ctx context.Context, req *dto.ModelSupplierReq) (res *dto.ModelSupplierRes, err error) {
return service.ModelManage.GetModelSupplier(ctx, req)
}
dao/model_manage_dao.go
+193
View File
@@ -0,0 +1,193 @@
package dao
import (
"context"
"fmt"
"model-gateway/consts/public"
"model-gateway/model/dto"
"model-gateway/model/entity"
"gitea.redpowerfuture.com/red-future/common/db/gfdb"
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/util/gconv"
)
var ModelManage = &modelManageDao{}
type modelManageDao struct{}
// Insert 插入
func (d *modelManageDao) Insert(ctx context.Context, req *dto.CreateModelManageReq) (id int64, err error) {
var e = new(entity.ModelManage)
err = gconv.Struct(req, &e)
if err != nil {
return
}
r, err := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameModelManage).Insert(e)
if err != nil {
return
}
return r.LastInsertId()
}
// Update 更新
func (d *modelManageDao) Update(ctx context.Context, req *dto.UpdateModelManageReq) (rows int64, err error) {
r, err := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameModelManage).OmitEmpty().Data(req).Where(entity.ModelManageCol.Id, req.Id).Update()
if err != nil {
return
}
return r.RowsAffected()
}
// Delete 删除
func (d *modelManageDao) Delete(ctx context.Context, req *dto.DeleteModelManageReq) (rows int64, err error) {
r, err := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameModelManage).Where(entity.ModelManageCol.Id, req.Id).Delete()
if err != nil {
return
}
return r.RowsAffected()
}
func (d *modelManageDao) Get(ctx context.Context, req *dto.GetModelManage, fields ...string) (res *entity.ModelManage, err error) {
r, err := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameModelManage).Cache(ctx).
OmitEmpty().
Where(entity.ModelManageCol.ModelName, req.ModelName).
Where(entity.ModelManageCol.ChatModel, req.ChatModel).
Where(entity.ModelManageCol.Creator, req.Creator).
Fields(fields).One()
if err != nil {
return
}
err = r.Struct(&res)
return
}
func (d *modelManageDao) GetNotTenantId(ctx context.Context, req *dto.GetModelManageReq, fields ...string) (res *entity.ModelManage, err error) {
// 获取表前缀
prefix := g.Cfg().MustGet(ctx, fmt.Sprintf("database.%s.0.prefix", public.DbNameModelGateway)).String()
table := prefix + public.TableNameModelManage
// 动态拼接 SELECT 列
var field string
if !g.IsEmpty(fields) {
for k, v := range fields {
if k == len(fields)-1 {
field = field + v
} else {
field = field + v + ","
}
}
} else {
field = "*"
}
// 动态拼接 WHERE 条件
var whereCondition string
var queryParams []interface{}
if !g.IsEmpty(req.Id) {
whereCondition = fmt.Sprintf(" AND %s=(?) ", entity.ModelManageCol.Id)
queryParams = append(queryParams, req.Id)
}
whereCondition = whereCondition + " AND " + entity.ModelManageCol.DeletedAt + " IS NULL "
sql := `SELECT ` + field + ` FROM ` + table + ` WHERE 1=1 ` + whereCondition + ``
// 执行查询
result, err := gfdb.DB(ctx, public.DbNameModelGateway).GetOne(ctx, sql, queryParams...)
if err != nil {
return nil, err
}
err = result.Struct(&res)
return
}
func (d *modelManageDao) ListNotTenantId(ctx context.Context, req *dto.ListModelManageReq, fields ...string) (res []*entity.ModelManage, total int, err error) {
// 获取表前缀
prefix := g.Cfg().MustGet(ctx, fmt.Sprintf("database.%s.0.prefix", public.DbNameModelGateway)).String()
table := prefix + public.TableNameModelManage
// 动态拼接 SELECT 列
var field string
if !g.IsEmpty(fields) {
for k, v := range fields {
if k == len(fields)-1 {
field = field + v
} else {
field = field + v + ","
}
}
} else {
field = "*"
}
// 动态拼接 WHERE 条件
var whereCondition string
var queryParams []interface{}
if !g.IsEmpty(req.ModelName) {
whereCondition += fmt.Sprintf(" AND %s=(?) ", entity.ModelManageCol.ModelName)
queryParams = append(queryParams, req.ModelName)
}
if !g.IsEmpty(req.ModelType) {
whereCondition += fmt.Sprintf(" AND %s=(?) ", entity.ModelManageCol.ModelType)
queryParams = append(queryParams, req.ModelType)
}
if !g.IsEmpty(req.Creator) {
whereCondition += fmt.Sprintf(" AND (%s=(?) OR %s=true) ", entity.ModelManageCol.Creator, entity.ModelManageCol.SystemModel)
queryParams = append(queryParams, req.Creator)
}
whereCondition = whereCondition + " AND " + entity.ModelManageCol.DeletedAt + " IS NULL "
// 1. 统计去重后总条数
countSql := fmt.Sprintf(
`SELECT COUNT(DISTINCT %s) FROM %s WHERE 1=1 %s`,
entity.ModelManageCol.ModelName,
table,
whereCondition,
)
countResult, err := gfdb.DB(ctx, public.DbNameModelGateway).GetOne(ctx, countSql, queryParams...)
if err != nil {
return nil, 0, err
}
type crr struct {
Count int64 `db:"count"`
}
var cr crr
if err = countResult.Struct(&cr); err != nil {
return nil, 0, err
}
total = int(cr.Count)
// 2. 分页处理
limitSql := ""
if req.Page != nil {
pageNum := int(req.Page.PageNum)
pageSize := int(req.Page.PageSize)
offset := (pageNum - 1) * pageSize
limitSql = fmt.Sprintf(" LIMIT ? OFFSET ? ")
// PG 语法 LIMIT 条数 OFFSET 偏移量
queryParams = append(queryParams, pageSize, offset)
}
// 排序优先级：1.分组字段ModelName 2.SystemModel升序(false在前，保留用户数据) 3.创建时间倒序
orderSql := fmt.Sprintf(
" ORDER BY %s, %s ASC, %s DESC ",
entity.ModelManageCol.ModelName,
entity.ModelManageCol.SystemModel,
entity.ModelManageCol.CreatedAt,
)
// PG DISTINCT ON 按模型名去重，同名只取第一条(用户数据)
sql := fmt.Sprintf(
`SELECT DISTINCT ON (%s) %s FROM %s WHERE 1=1 %s %s %s`,
entity.ModelManageCol.ModelName,
field,
table,
whereCondition,
orderSql,
limitSql,
)
// 执行查询
result, err := gfdb.DB(ctx, public.DbNameModelGateway).GetAll(ctx, sql, queryParams...)
if err != nil {
return nil, 0, err
}
err = result.Structs(&res)
return
}
go.mod
+1 -1
View File
@@ -3,7 +3,7 @@ module model-gateway
go 1.26.1
require (
gitea.redpowerfuture.com/red-future/common v0.0.29
gitea.redpowerfuture.com/red-future/common v0.0.30
github.com/gogf/gf/contrib/drivers/pgsql/v2 v2.10.2
github.com/gogf/gf/contrib/nosql/redis/v2 v2.10.2
github.com/gogf/gf/v2 v2.10.2
go.sum
+2
View File
@@ -1,6 +1,8 @@
cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
gitea.redpowerfuture.com/red-future/common v0.0.29 h1:5McaN5pSewvrLUHQzWMX6EaUvD+B5I5bMYoU+clHJk4=
gitea.redpowerfuture.com/red-future/common v0.0.29/go.mod h1:50U1Xi+Ie56z09S5LQbZvaken0Mxv3OeS9LgR7U/ZRY=
gitea.redpowerfuture.com/red-future/common v0.0.30 h1:UkWYubUsLPJQUhEhc9Ca2UPg5iLC6jzURo3ngztINYg=
gitea.redpowerfuture.com/red-future/common v0.0.30/go.mod h1:zuhqbWHd/YICalYJnmecY8Vqo5j4dtZOD63P96fiFeU=
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
github.com/BurntSushi/toml v1.6.0 h1:dRaEfpa2VI55EwlIW72hMRHdWouJeRF7TPYhI+AUQjk=
github.com/BurntSushi/toml v1.6.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho=
main.go
+4 -2
View File
@@ -26,6 +26,7 @@ func main() {
// 注册路由
http.RouteRegister([]interface{}{
controller.ModelManage,
controller.ModelGatewayModels,
controller.ModelGatewayTask,
controller.ModelGatewayLogsStat,
@@ -40,9 +41,10 @@ func main() {
<-quit
g.Log().Infof(ctx, "[main] 收到退出信号，开始优雅退出...")
cancel()
// 关闭 gateway server（RouteRegister 内部是 go Httpserver.Run() 启动的）
// 先关闭 gateway server，等待 in-flight 请求处理完成
_ = http.Httpserver.Shutdown()
// 再取消上下文，避免活跃请求被中断
cancel()
}
func startAutoRunner(ctx context.Context) {
model/dto/model_gateway_task_dto.go
+1
View File
@@ -14,6 +14,7 @@ type CreateTaskReq struct {
EpicycleId int64 `json:"epicycleId" dc:"轮次ID"`
BuildType int64 `json:"buildType" dc:"构建类型：1-提示词构建 2-节点构建"`
BuildModelName string `json:"buildModelName" json:"buildModelName" dc:"构建模型名称"`
TaskId string `json:"taskId" dc:"任务ID"`
}
type CreateTaskRes struct {
model/dto/model_manage_dto.go
+127
View File
@@ -0,0 +1,127 @@
package dto
import (
"model-gateway/consts/model"
"model-gateway/consts/public"
"model-gateway/model/entity"
"gitea.redpowerfuture.com/red-future/common/beans"
"github.com/gogf/gf/v2/frame/g"
)
// CreateModelManageReq 添加模型配置
type CreateModelManageReq struct {
g.Meta `path:"/createModelManage" method:"post" tags:"new模型管理" summary:"new创建模型配置" dc:"new添加新的模型配置"`
ModelSupplier model.SupplierType `json:"modelSupplier" v:"required#模型供应商不能为空" dc:"模型供应商"`
ModelName string `json:"modelName" v:"required#模型名称不能为空" dc:"模型名称"`
ModelType model.ModelType `json:"modelType" v:"required#模型类型不能为空" dc:"模型类型"`
BaseURL string `json:"baseUrl" v:"required#模型服务地址不能为空" dc:"模型服务地址"`
SystemModel *bool `json:"systemModel" dc:"系统模型"`
HttpMethod string `json:"httpMethod" dc:"请求方式：GET/POST" d:"POST"`
ChatModel *bool `json:"chatModel" dc:"对话模型"`
ResponseType model.ResponseType `json:"responseType" v:"required#调用模式不能为空" dc:"调用模式：0-同步 1-异步 2-流式"`
ApiKey string `json:"apiKey" dc:"调用凭证/密钥"`
Enabled *bool `json:"enabled" dc:"启用"`
RequestHeadMapping map[string]string `json:"requestHeadMapping" dc:"请求头映射"`
RequestBodyMapping map[string]any `json:"requestBodyMapping" dc:"请求体映射"`
ResponseMapping map[string]any `json:"responseMapping" dc:"返回映射"`
ResponseBodyMapping map[string]string `json:"responseBodyMapping" dc:"返回体映射"`
MaxConcurrency int `json:"maxConcurrency" dc:"最大并发数（默认10）"`
TokenMapping *entity.TokenMapping `json:"tokenMapping" dc:"token映射"`
AsyncTaskMapping *entity.AsyncTaskMapping `json:"asyncTaskMapping" dc:"异步任务映射"`
TokenPredictPrice float64 `json:"tokenPredictPrice" dc:"模型Token预估价格"`
TokenPredictPriceUnit string `json:"tokenPredictPriceUnit" dc:"模型Token预估价格单位"`
MaxTokens int `json:"maxTokens" dc:"最大token数"`
MaxDuration int `json:"maxDuration" dc:"最大时长"`
LastFrame string `json:"lastFrame" dc:"视频的尾帧图像"`
}
type CreateModelManageRes struct {
Id int64 `json:"id,string" dc:"配置ID"`
}
type UpdateModelManageReq struct {
g.Meta `path:"/updateModelManage" method:"put" tags:"new模型管理" summary:"new更新模型配置" dc:"new更新指定ID的模型配置"`
Id int64 `json:"id" v:"required#id不能为空" dc:"配置ID"`
ModelSupplier model.SupplierType `json:"modelSupplier" dc:"模型供应商"`
ModelName string `json:"modelName" dc:"模型名称"`
ModelType model.ModelType `json:"modelType" dc:"模型类型"`
BaseURL string `json:"baseUrl" dc:"模型服务地址"`
SystemModel *bool `json:"systemModel" dc:"系统模型"`
HttpMethod string `json:"httpMethod" dc:"请求方式：GET/POST"`
ChatModel *bool `json:"chatModel" dc:"对话模型"`
ResponseType model.ResponseType `json:"responseType" dc:"调用模式：0-同步 1-异步 2-流式"`
ApiKey string `json:"apiKey" dc:"调用凭证/密钥"`
Enabled *bool `json:"enabled" dc:"启用"`
RequestHeadMapping map[string]string `json:"requestHeadMapping" dc:"请求头映射"`
RequestBodyMapping map[string]any `json:"requestBodyMapping" dc:"请求体映射"`
ResponseMapping map[string]any `json:"responseMapping" dc:"返回映射"`
ResponseBodyMapping map[string]string `json:"responseBodyMapping" dc:"返回主体映射"`
MaxConcurrency int `json:"maxConcurrency" dc:"最大并发数（默认10）"`
TokenMapping *entity.TokenMapping `json:"tokenMapping" dc:"token映射"`
AsyncTaskMapping *entity.AsyncTaskMapping `json:"asyncTaskMapping" dc:"异步任务映射"`
TokenPredictPrice float64 `json:"tokenPredictPrice" dc:"模型Token预估价格"`
TokenPredictPriceUnit string `json:"tokenPredictPriceUnit" dc:"模型Token预估价格单位"`
MaxTokens int `json:"maxTokens" dc:"最大token数"`
MaxDuration int `json:"maxDuration" dc:"最大时长"`
LastFrame string `json:"lastFrame" dc:"视频的尾帧图像"`
}
type DeleteModelManageReq struct {
g.Meta `path:"/deleteModelManage" method:"delete" tags:"new模型管理" summary:"new删除模型配置" dc:"new删除指定ID的模型配置"`
Id int64 `p:"id" json:"id,string" v:"required#id不能为空" dc:"配置ID"`
}
type GetModelManage struct {
ChatModel *bool `json:"chatModel" dc:"对话模型"`
Creator string `json:"creator" dc:"创建人"`
ModelName string `json:"modelName" dc:"模型名称"`
}
type GetModelManageReq struct {
g.Meta `path:"/getModelManage" method:"get" tags:"new模型管理" summary:"new获取模型配置" dc:"new获取指定ID的模型配置"`
Id int64 `p:"id" json:"id,string" v:"required#id不能为空" dc:"配置ID"`
}
type GetModelManageRes struct {
*entity.ModelManage `json:"modelManage"`
}
// ListModelManageReq 配置列表
type ListModelManageReq struct {
g.Meta `path:"/listModelManage" method:"get" tags:"new模型管理" summary:"new模型配置列表" dc:"new分页获取模型配置列表"`
*beans.Page `json:"page"`
ModelName string `p:"modelName" json:"modelName" dc:"模型名称（模糊查询，可选）"`
ModelType int `p:"modelType" json:"modelType" dc:"模型类型"`
Creator string `json:"creator" dc:"创建人"`
}
type ListModelManageRes struct {
List []*entity.ModelManage `json:"list" dc:"列表数据"`
Total int `json:"total" dc:"总数"`
}
type CheckChatModelReq struct {
g.Meta `path:"/checkChatModel" method:"get" tags:"new模型管理" summary:"new检查是否为聊天模型" dc:"new检查是否为聊天模型"`
}
type CheckChatModelRes struct {
IsChatModel bool `json:"isChatModel" dc:"是否为聊天模型"`
}
// ModelTypeReq 模型类型列表（分页）
type ModelTypeReq struct {
g.Meta `path:"/modelType" method:"get" tags:"new模型管理" summary:"new模型类型列表" dc:"new分页获取模型类型列表"`
}
type ModelTypeRes struct {
List []*model.TypeTree `json:"list" dc:"模型类型ID到名称的映射"`
}
type ModelSupplierReq struct {
g.Meta `path:"/modelSupplier" method:"get" tags:"new模型管理" summary:"new获取运营商列表" dc:"new获取运营商列表"`
}
type ModelSupplierRes struct {
List []*public.Option `json:"list" dc:"运营商名称到ID的映射"`
}
model/entity/model_manage.go
+102
View File
@@ -0,0 +1,102 @@
package entity
import (
"model-gateway/consts/model"
"gitea.redpowerfuture.com/red-future/common/beans"
)
type modelManageCol struct {
beans.SQLBaseCol
ModelSupplier string
ModelName string
ModelType string
BaseURL string
SystemModel string
HttpMethod string
ChatModel string
ResponseType string
ApiKey string
Enabled string
RequestHeadMapping string
RequestBodyMapping string
ResponseMapping string
ResponseBodyMapping string
MaxConcurrency string
TokenPredictPrice string
TokenPredictPriceUnit string
MaxTokens string
MaxDuration string
LastFrame string
}
var ModelManageCol = modelManageCol{
SQLBaseCol: beans.DefSQLBaseCol,
ModelSupplier: "model_supplier",
ModelName: "model_name",
ModelType: "model_type",
BaseURL: "base_url",
SystemModel: "system_model",
HttpMethod: "http_method",
ChatModel: "chat_model",
ResponseType: "response_type",
ApiKey: "api_key",
Enabled: "enabled",
RequestHeadMapping: "request_head_mapping",
RequestBodyMapping: "request_body_mapping",
ResponseMapping: "response_mapping",
ResponseBodyMapping: "response_body_mapping",
MaxConcurrency: "max_concurrency",
TokenPredictPrice: "token_predict_price",
TokenPredictPriceUnit: "token_predict_price_unit",
MaxTokens: "max_tokens",
MaxDuration: "max_duration",
LastFrame: "last_frame",
}
type ModelManage struct {
beans.SQLBaseDO `orm:",inline"`
ModelSupplier model.SupplierType `orm:"model_supplier" json:"modelSupplier" description:"模型供应商"`
ModelName string `orm:"model_name" json:"modelName" description:"模型名称"`
ModelType model.ModelType `orm:"model_type" json:"modelType" description:"模型类型"`
BaseURL string `orm:"base_url" json:"baseUrl" description:"模型地址"`
SystemModel *bool `orm:"system_model" json:"systemModel" description:"系统模型"`
HttpMethod string `orm:"http_method" json:"httpMethod" description:"http方法"`
ChatModel *bool `orm:"chat_model" json:"ChatModel" description:"是否聊天模型"`
ResponseType model.ResponseType `orm:"response_type" json:"responseType" description:"返回类型：1同步，2异步，3流"`
ApiKey string `orm:"api_key" json:"apiKey" description:"api key"`
Enabled *bool `orm:"enabled" json:"enabled" description:"是否启用"`
RequestHeadMapping map[string]string `orm:"request_head_mapping" json:"requestHeadMapping" description:"请求头映射"`
RequestBodyMapping map[string]any `orm:"request_body_mapping" json:"requestBodyMapping" description:"请求体映射"`
ResponseMapping map[string]any `orm:"response_mapping" json:"responseMapping" description:"响应映射"`
ResponseBodyMapping map[string]string `orm:"response_body_mapping" json:"responseBodyMapping" description:"响应主体映射"`
MaxConcurrency int `orm:"max_concurrency" json:"maxConcurrency" description:"最大并发数"`
TokenMapping *TokenMapping `orm:"token_mapping" json:"tokenMapping" description:"token映射"`
AsyncTaskMapping *AsyncTaskMapping `orm:"async_task_mapping" json:"asyncTaskMapping" description:"异步任务映射"`
TokenPredictPrice float64 `orm:"token_predict_price" json:"tokenPredictPrice" description:"模型Token预估价格"`
TokenPredictPriceUnit string `orm:"token_predict_price_unit" json:"tokenPredictPriceUnit" description:"模型token预估价格单位（秒，百万Token，千Token，字数）"`
MaxTokens int `orm:"max_tokens" json:"maxTokens" description:"最大token数"`
MaxDuration int `orm:"max_duration" json:"maxDuration" description:"最大时长（秒）"`
LastFrame string `orm:"last_frame" json:"lastFrame" description:"视频的尾帧图像"`
}
type TokenMapping struct {
PromptTokens string `json:"promptTokens" dc:"输入token"`
CompletionTokens string `json:"completionTokens" dc:"输出token"`
TotalTokens string `json:"totalTokens" dc:"总token"`
}
type AsyncTaskMapping struct {
Url string `json:"url" dc:"url"`
HttpMethod string `json:"httpMethod" dc:"http方法" d:"POST"`
RequestHeadMapping map[string]string `json:"requestHeadMapping" description:"请求头映射"`
ResponseMapping map[string]any `json:"responseMapping" description:"响应映射"`
TaskId string `json:"taskId" dc:"任务id"`
TaskStatus string `json:"taskStatus" dc:"任务状态"`
TaskStatusPending string `json:"taskStatusPending" dc:"任务状态-待处理"`
TaskStatusRunning string `json:"taskStatusRunning" dc:"任务状态-运行中"`
TaskStatusSuccess string `json:"taskStatusSuccess" dc:"任务状态-成功"`
TaskStatusFailed string `json:"taskStatusFailed" dc:"任务状态-失败"`
TaskStatusCancel string `json:"taskStatusCancel" dc:"任务状态-取消"`
TaskStatusUnknown string `json:"taskStatusUnknown" dc:"任务状态-未知"`
}
service/gateway/gateway_http_service.go
+4 -2
View File
@@ -257,7 +257,9 @@ func DeductBalance(ctx context.Context, tenantId uint64, amount float64) error {
// TenantSurplusResp 租户余额返回
type TenantSurplusResp struct {
Surplus float64 `json:"surplus"`
Tenant struct {
Surplus float64 `json:"surplus"`
} `json:"tenant"`
}
// GetTenantSurplus 获取租户余额
@@ -278,7 +280,7 @@ func GetTenantSurplus(ctx context.Context, tenantId uint64) (float64, error) {
g.Log().Warningf(ctx, "[获取余额] 失败 tenantId=%d err=%v", tenantId, err)
return 0, err
}
return resp.Surplus, nil
return resp.Tenant.Surplus, nil
}
//// callback 向回调地址 POST 任务结果（与查询接口 GetTaskRes 出参一致）
service/model_manage_service.go
+204
View File
@@ -0,0 +1,204 @@
package service
import (
"context"
"fmt"
"model-gateway/consts/model"
"model-gateway/consts/public"
"model-gateway/dao"
"model-gateway/model/dto"
"model-gateway/model/entity"
"model-gateway/service/gateway"
"gitea.redpowerfuture.com/red-future/common/beans"
"gitea.redpowerfuture.com/red-future/common/db/gfdb"
"gitea.redpowerfuture.com/red-future/common/utils"
"github.com/gogf/gf/v2/database/gdb"
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/util/gconv"
)
var ModelManage = &modelManageService{}
type modelManageService struct{}
// Create 创建模型
func (s *modelManageService) Create(ctx context.Context, req *dto.CreateModelManageReq) (res *dto.CreateModelManageRes, err error) {
err = gfdb.DB(ctx, public.DbNameModelGateway).Transaction(ctx, func(ctx context.Context, tx gdb.TX) (err error) {
// 1）检查是否是超管
var isSuperAdmin bool
isSuperAdmin, err = gateway.IsSuperAdmin(ctx)
if err != nil {
return
}
req.SystemModel = &isSuperAdmin
// 1）如果设为会话模型，先把该用户旧会话模型取消
err = s.CancelChatModel(ctx, req.ModelType, req.ChatModel, isSuperAdmin)
if err != nil {
return
}
// 2）插入数据
id, err := dao.ModelManage.Insert(ctx, req)
if err != nil {
return
}
res = &dto.CreateModelManageRes{Id: id}
return
})
return
}
// Update 更新模型配置
func (s *modelManageService) Update(ctx context.Context, req *dto.UpdateModelManageReq) (err error) {
err = gfdb.DB(ctx, public.DbNameModelGateway).Transaction(ctx, func(ctx context.Context, tx gdb.TX) (err error) {
var get *entity.ModelManage
get, err = dao.ModelManage.GetNotTenantId(ctx, &dto.GetModelManageReq{
Id: req.Id,
})
if err != nil {
return
}
var user *beans.User
user, err = utils.GetUserInfo(ctx)
if err != nil {
return
}
// 1）如果不是创建者，且是系统模型，则需要拷贝
if get.Creator != user.UserName {
if get.SystemModel != nil && *get.SystemModel {
if g.IsEmpty(req.ApiKey) {
return fmt.Errorf("模型apiKey不能为空")
}
d := new(dto.CreateModelManageReq)
err = gconv.Struct(req, d)
if err != nil {
return
}
_, err = s.Create(ctx, d)
if err != nil {
return err
}
return
}
return fmt.Errorf("无权限操作")
}
// 1）检查是否是超管
var isSuperAdmin bool
isSuperAdmin, err = gateway.IsSuperAdmin(ctx)
if err != nil {
return
}
// 1）如果设为会话模型，先把该用户旧会话模型取消
err = s.CancelChatModel(ctx, req.ModelType, req.ChatModel, isSuperAdmin)
if err != nil {
return
}
// 2）更新数据
_, err = dao.ModelManage.Update(ctx, req)
return
})
return
}
func (s *modelManageService) CancelChatModel(ctx context.Context, modelType model.ModelType, chatModel *bool, isSuperAdmin bool) (err error) {
if !g.IsEmpty(chatModel) && *chatModel {
if *modelType == *model.ModelTypeInference.Code {
if isSuperAdmin {
return fmt.Errorf("超级管理员不能设置会话模型")
}
// 2）获取该用户信息
var user *beans.User
user, err = utils.GetUserInfo(ctx)
if err != nil {
return
}
// 3）取消该用户之前的会话模型
var get *entity.ModelManage
get, err = dao.ModelManage.Get(ctx, &dto.GetModelManage{
Creator: user.UserName,
ChatModel: chatModel,
})
if err != nil {
return
}
_, err = dao.ModelManage.Update(ctx, &dto.UpdateModelManageReq{
Id: get.Id,
ChatModel: gconv.PtrBool(false),
})
if err != nil {
return
}
} else {
return fmt.Errorf("只有推理模型可以设置成会话模型")
}
}
return
}
// Delete 删除模型
func (s *modelManageService) Delete(ctx context.Context, req *dto.DeleteModelManageReq) error {
_, err := dao.ModelManage.Delete(ctx, req)
return err
}
func (s *modelManageService) Get(ctx context.Context, req *dto.GetModelManageReq) (res *dto.GetModelManageRes, err error) {
get, err := dao.ModelManage.GetNotTenantId(ctx, req)
if err != nil {
return
}
err = gconv.Struct(get, &res)
return
}
// List 获取模型列表
func (s *modelManageService) List(ctx context.Context, req *dto.ListModelManageReq) (res *dto.ListModelManageRes, err error) {
var user *beans.User
user, err = utils.GetUserInfo(ctx)
if err != nil {
return
}
req.Creator = user.UserName
list, total, err := dao.ModelManage.ListNotTenantId(ctx, req)
if err != nil {
return
}
res = &dto.ListModelManageRes{
Total: total,
}
err = gconv.Struct(list, &res.List)
return
}
func (s *modelManageService) CheckChatModel(ctx context.Context, req *dto.CheckChatModelReq) (res *dto.CheckChatModelRes, err error) {
user, err := utils.GetUserInfo(ctx)
if err != nil {
return
}
get, err := dao.ModelManage.Get(ctx, &dto.GetModelManage{
Creator: user.UserName,
ChatModel: gconv.PtrBool(true),
})
if err != nil {
return
}
res = &dto.CheckChatModelRes{
IsChatModel: !g.IsEmpty(get),
}
return
}
// GetModelType 获取模型类型
func (s *modelManageService) GetModelType(ctx context.Context, req *dto.ModelTypeReq) (res *dto.ModelTypeRes, err error) {
res = &dto.ModelTypeRes{
List: model.GetTypeTreeList(),
}
return res, nil
}
// GetModelSupplier 获取运营商列表
func (s *modelManageService) GetModelSupplier(ctx context.Context, req *dto.ModelSupplierReq) (res *dto.ModelSupplierRes, err error) {
return &dto.ModelSupplierRes{
List: model.GetSupplierOptionList(),
}, nil
}
service/task/task_service.go
+190 -66
View File
@@ -15,7 +15,10 @@ 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/database/gredis"
"github.com/gogf/gf/v2/errors/gerror"
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/os/glog"
"github.com/gogf/gf/v2/util/gconv"
"github.com/google/uuid"
)
@@ -26,7 +29,10 @@ type taskService struct{}
// Create 创建任务
func (s *taskService) Create(ctx context.Context, req *dto.CreateTaskReq) (res *dto.CreateTaskRes, err error) {
taskID := uuid.NewString()
taskID := req.TaskId
if taskID == "" {
taskID = uuid.NewString()
}
startAt := time.Now()
// 1) 获取用户信息
@@ -49,81 +55,196 @@ func (s *taskService) Create(ctx context.Context, req *dto.CreateTaskReq) (res *
if model == nil || (model.Enabled != nil && *model.Enabled != 1) {
return nil, errors.New("模型不存在或未启用")
}
lockKey := fmt.Sprintf("lock:tenantId-%s:model-%s", gconv.String(userInfo.TenantId), req.ModelName)
success, e := Lock(ctx, lockKey, -1, int64(time.Minute.Seconds()*5), func(ctx context.Context) error {
const (
keyExpireSec = 600 // 计数Key兜底过期时间 10min
waitInterval = 10 * time.Second // 轮询等待间隔
)
// Redis 操作统一使用独立上下文，避免外部 ctx canceled
redisCtx := context.WithoutCancel(ctx)
// 模型并发计数Key
concurrencyKey := fmt.Sprintf("model:concurrency:%s", req.ModelName)
maxCon := gconv.Int64(model.MaxConcurrency)
// TODO: 排队控制暂时关闭，后续需要时取消注释
// limit := queue.GetRuntimeQueueLimit(ctx, req.ModelName, model.MaxConcurrency*2)
// if limit > 0 {
// ok, err := queue.AcquireQueueSlot(ctx, req.ModelName, taskID, limit, model.TimeoutSeconds)
// if err != nil {
// return nil, err
// }
// if !ok {
// return nil, errors.New("任务排队已满，请稍后再试")
// }
// }
// 循环尝试获取并发名额，超限则等待重试
var held bool // 标记当前是否持有未释放的计数
for {
// 检测全局上下文取消
if ctx.Err() != nil {
if held {
g.Redis().Decr(redisCtx, concurrencyKey)
}
return ctx.Err()
}
// 计数自增
currentCon, e := g.Redis().Incr(redisCtx, concurrencyKey)
if e != nil {
if held {
g.Redis().Decr(redisCtx, concurrencyKey)
}
glog.Errorf(ctx, "redis incr concurrency key err: %v", e)
return e
}
held = true
// 首次创建Key时设置过期时间（避免重复执行EXPIRE）
exists, errr := g.Redis().Exists(redisCtx, concurrencyKey)
if errr == nil && exists == 1 {
g.Redis().Expire(redisCtx, concurrencyKey, keyExpireSec)
}
// 3) 构建任务实体
task := &entity.ModelGatewayTask{
ModelName: model.ModelName,
TaskID: taskID,
State: public.TaskStatusRunning,
BizName: req.BizName,
CallbackURL: req.CallbackUrl,
RequestPayload: &entity.RequestPayload{
Body: req.RequestPayload,
Headers: util.ParseHeadMsgHeaders(model.HeadMsg),
},
EpicycleId: req.EpicycleId,
BuildModelName: req.BuildModelName,
}
// 未超限：跳出循环，执行业务
if currentCon <= maxCon {
glog.Infof(ctx, "并发数: %s %d/%d", concurrencyKey, currentCon, maxCon)
break
}
// 超限立刻回减，撤销本次计数
g.Redis().Decr(redisCtx, concurrencyKey)
held = false
glog.Infof(ctx, "并发超限等待: %s %d/%d", concurrencyKey, currentCon, maxCon)
time.Sleep(waitInterval)
}
// 4) 插入任务记录
id, err := dao.ModelGatewayTask.Insert(ctx, task)
if err != nil {
// TODO: 恢复排队逻辑后，此处需要回滚排队占位
// queue.ReleaseQueueSlot(ctx, req.ModelName, taskID)
return nil, err
}
task.Id = id
// 3) 构建任务实体
task := &entity.ModelGatewayTask{
ModelName: model.ModelName,
TaskID: taskID,
State: public.TaskStatusRunning,
BizName: req.BizName,
CallbackURL: req.CallbackUrl,
RequestPayload: &entity.RequestPayload{
Body: req.RequestPayload,
Headers: util.ParseHeadMsgHeaders(model.HeadMsg),
},
EpicycleId: req.EpicycleId,
BuildModelName: req.BuildModelName,
}
// 5) 记录操作日志（非关键路径，失败不影响主流程）
ip, ua := "", ""
if r := g.RequestFromCtx(ctx); r != nil {
ip = utils.GetLocalIP()
ua = r.UserAgent()
}
_, _ = dao.ModelGatewayLogsOp.Insert(ctx, &entity.ModelGatewayLogsOp{
IP: ip,
UserAgent: ua,
APIPath: "/task/createTask",
HttpMethod: "POST",
BizName: req.BizName,
ModelName: req.ModelName,
TaskID: taskID,
OpType: "createTask",
Success: 1,
CostMs: time.Since(startAt).Milliseconds(),
RequestPayload: task.RequestPayload,
ResponsePayload: gdb.Map{"taskId": taskID},
})
// 4) 插入任务记录
id, errr := dao.ModelGatewayTask.Insert(ctx, task)
if errr != nil {
g.Redis().Decr(redisCtx, concurrencyKey)
// TODO: 恢复排队逻辑后，此处需要回滚排队占位
//queue.ReleaseQueueSlot(ctx, req.ModelName, taskID)
return errr
}
task.Id = id
// 6) 模型计费
if len(model.BillingConfig) > 0 {
requestData := util.ExtractRequestBilling(ctx, model.BillingConfig, req.RequestPayload)
// 请求数据作为计费记录的基础字段，先存入数组
task.BillingData = append(task.BillingData, requestData)
_, _ = dao.ModelGatewayTask.Update(ctx, &entity.ModelGatewayTask{
SQLBaseDO: beans.SQLBaseDO{Id: task.Id},
BillingData: task.BillingData,
// 5) 记录操作日志（非关键路径，失败不影响主流程）
ip, ua := "", ""
if r := g.RequestFromCtx(ctx); r != nil {
ip = utils.GetLocalIP()
ua = r.UserAgent()
}
_, _ = dao.ModelGatewayLogsOp.Insert(ctx, &entity.ModelGatewayLogsOp{
IP: ip,
UserAgent: ua,
APIPath: "/task/createTask",
HttpMethod: "POST",
BizName: req.BizName,
ModelName: req.ModelName,
TaskID: taskID,
OpType: "createTask",
Success: 1,
CostMs: time.Since(startAt).Milliseconds(),
RequestPayload: task.RequestPayload,
ResponsePayload: gdb.Map{"taskId": taskID},
})
}
// 7) 异步执行任务
go AsyncWorker.handleOne(util.AsyncCtx(ctx), task, model, req)
// 6) 模型计费
if len(model.BillingConfig) > 0 {
requestData := util.ExtractRequestBilling(ctx, model.BillingConfig, req.RequestPayload)
// 请求数据作为计费记录的基础字段，先存入数组
task.BillingData = append(task.BillingData, requestData)
_, _ = dao.ModelGatewayTask.Update(ctx, &entity.ModelGatewayTask{
SQLBaseDO: beans.SQLBaseDO{Id: task.Id},
BillingData: task.BillingData,
})
}
// 7) 异步执行任务
go AsyncWorker.handleOne(util.AsyncCtx(ctx), task, model, req)
return nil
})
if e != nil {
err = e
return
}
if !success {
err = gerror.New("任务排队已满，请稍后再试")
return
}
return &dto.CreateTaskRes{TaskID: taskID}, nil
}
// Lock 分布式锁 纯原生命令、无Lua、隔离上下文防 context canceled
func Lock(ctx context.Context, key string, limit, expireSeconds int64, fn func(ctx context.Context) error) (success bool, err error) {
if limit <= 0 {
limit = -1
}
// 过期时间合法校验（单位：秒）
const maxExpireSec = 86400 * 7
if expireSeconds < 1 || expireSeconds > maxExpireSec {
glog.Warningf(ctx, "锁过期时间非法，原值:%d，兜底为60秒", expireSeconds)
expireSeconds = 60
}
lockVal := "1"
LOOP:
// 检测父级上下文取消，防止无限重试阻塞 goroutine
if ctx.Err() != nil {
return false, ctx.Err()
}
if limit != -1 {
if limit < 0 {
return false, errors.New("锁重试次数耗尽，获取锁失败")
}
limit--
}
// 核心：创建独立上下文，不受外部 ctx 取消影响
redisCtx := context.WithoutCancel(ctx)
// 加锁
val, err := g.Redis().Set(redisCtx, key, lockVal, gredis.SetOption{
TTLOption: gredis.TTLOption{
EX: &expireSeconds,
},
NX: true,
})
if err != nil {
glog.Errorf(ctx, "redis set lock failed: %v", err)
time.Sleep(time.Second)
goto LOOP
}
if val.Bool() {
// 执行业务逻辑（使用原上下文）
runErr := fn(ctx)
// 释放锁：同样使用独立上下文 + 先GET再DEL防误删
getRes, err := g.Redis().Get(redisCtx, key)
if err != nil {
glog.Errorf(ctx, "redis get lock value failed: %v", err)
} else if getRes.String() == lockVal {
_, delErr := g.Redis().Del(redisCtx, key)
if delErr != nil {
glog.Errorf(ctx, "redis del lock failed: %v", delErr)
}
}
return true, runErr
}
// 抢锁失败，休眠重试
time.Sleep(time.Second)
goto LOOP
}
// GetResult 获取任务结果
func (s *taskService) GetResult(ctx context.Context, taskID string) (res *dto.GetTaskResultRes, err error) {
t, err := dao.ModelGatewayTask.Get(ctx, &entity.ModelGatewayTask{
@@ -260,7 +381,10 @@ func (s *taskService) QueryPendingTasks(ctx context.Context, req *dto.QueryPendi
if err != nil || model == nil || model.QueryConfig == nil {
continue
}
result, err := util.PullTaskResult(ctx, nil, model.QueryConfig, model.HeadMsg)
// 每个任务使用独立的超时上下文，防止单个任务阻塞整个轮询
pullCtx, pullCancel := context.WithTimeout(ctx, 30*time.Second)
result, err := util.PullTaskResult(pullCtx, nil, model.QueryConfig, model.HeadMsg)
pullCancel()
if err != nil {
g.Log().Warningf(ctx, "[轮询] 查询失败 taskID=%s err=%v", t.TaskID, err)
continue
service/task/worker.go
+23 -8
View File
@@ -46,11 +46,11 @@ func (w *asyncWorker) handleOne(ctx context.Context, task *entity.ModelGatewayTa
// 1) 查询余额
// ============================================
surplus, _ = gateway.GetTenantSurplus(ctx, model.TenantId)
if surplus <= 0 {
if surplus <= 200 {
w.failTask(ctx, task, startTime, "租户余额不足")
return
}
g.Log().Infof(ctx, "[handleOne] 当前余额 tenantId=%d surplus=%.2f", task.TenantId, surplus)
g.Log().Infof(ctx, "[handleOne] 当前余额 tenantId=%d surplus=%.2f", model.TenantId, surplus)
// ============================================
// 2) 调用模型
@@ -83,6 +83,7 @@ func (w *asyncWorker) handleOne(ctx context.Context, task *entity.ModelGatewayTa
}
// 模型调用失败
if !strings.Contains(err.Error(), "Timeout") &&
!strings.Contains(err.Error(), "RequestCanceled") &&
!strings.Contains(err.Error(), "InternalServiceError") &&
!strings.Contains(err.Error(), "Invalid video_url") &&
!strings.Contains(err.Error(), "Invalid audio track") &&
@@ -122,12 +123,17 @@ func (w *asyncWorker) handleOne(ctx context.Context, task *entity.ModelGatewayTa
if billingResult != nil {
task.BillingData[0] = billingResult
}
if billingResult != nil {
task.BillingData[0] = billingResult
totalFee := gconv.Float64(billingResult["total_fee"])
if totalFee > 0 {
_ = gateway.DeductBalance(util.AsyncCtx(ctx), task.TenantId, -totalFee)
for attempt := 0; attempt <= maxRetry; attempt++ {
err = gateway.DeductBalance(util.AsyncCtx(ctx), model.TenantId, -totalFee)
if err == nil {
break
}
g.Log().Warningf(ctx, "[handleOne] 扣除余额失败 taskId=%s attempt=%d/%d err=%v", task.TaskID, attempt, maxRetry, err)
}
}
}
}
@@ -184,10 +190,11 @@ func (w *asyncWorker) handleOne(ctx context.Context, task *entity.ModelGatewayTa
g.Log().Errorf(ctx, "[handleOne] 更新DB失败 taskId=%s err=%v", task.TaskID, err)
return
}
go gateway.TriggerCallback(util.AsyncCtx(ctx), task)
concurrencyKey := fmt.Sprintf("model:concurrency:%s", req.ModelName)
g.Redis().Decr(ctx, concurrencyKey)
gateway.TriggerCallback(ctx, task)
if req.EpicycleId != 0 {
go gateway.TriggerPromptsCallback(util.AsyncCtx(ctx), task, req.EpicycleId)
gateway.TriggerPromptsCallback(ctx, task, req.EpicycleId)
}
g.Log().Infof(ctx, "[handleOne] 成功 taskId=%s duration=%ds fileType=%s",
@@ -338,7 +345,13 @@ func (w *asyncWorker) parseAndRetry(ctx context.Context, body map[string]any, mo
task.BillingData = append(task.BillingData, billingResult)
totalFee := gconv.Float64(billingResult["total_fee"])
if totalFee > 0 {
_ = gateway.DeductBalance(util.AsyncCtx(ctx), task.TenantId, -totalFee)
for a := 0; a <= maxRetry; a++ {
errr := gateway.DeductBalance(util.AsyncCtx(ctx), task.TenantId, -totalFee)
if errr == nil {
break
}
g.Log().Warningf(ctx, "[handleOne] 扣除余额失败 taskId=%s attempt=%d/%d err=%v", task.TaskID, a, maxRetry, errr)
}
}
}
@@ -539,6 +552,8 @@ func (w *asyncWorker) failTask(ctx context.Context, t *entity.ModelGatewayTask,
t.State = 3
t.ErrorMsg = errMsg
t.DurationSeconds = int64(time.Since(startTime).Seconds())
concurrencyKey := fmt.Sprintf("model:concurrency:%s", t.ModelName)
g.Redis().Decr(ctx, concurrencyKey)
_, _ = dao.ModelGatewayTask.Update(ctx, t) // 更新任务状态
go gateway.TriggerCallback(util.AsyncCtx(ctx), t) // 触发回调
}
update.sql
+119 -1
View File
@@ -230,4 +230,122 @@ COMMENT ON COLUMN model_gateway_logs_op.success IS '是否成功：1成功/0失
COMMENT ON COLUMN model_gateway_logs_op.error_msg IS '错误信息（失败时）';
COMMENT ON COLUMN model_gateway_logs_op.cost_ms IS '耗时（毫秒）';
COMMENT ON COLUMN model_gateway_logs_op.request_payload IS '请求 JSON';
COMMENT ON COLUMN model_gateway_logs_op.response_payload IS '响应 JSON';
COMMENT ON COLUMN model_gateway_logs_op.response_payload IS '响应 JSON';
-- =========================================================================================================================
CREATE TABLE "public"."model_gateway_model_manage" (
"id" int8 NOT NULL,
"tenant_id" int8 NOT NULL DEFAULT 0,
"creator" varchar(64) COLLATE "pg_catalog"."default" NOT NULL,
"created_at" timestamp(6) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updater" varchar(64) COLLATE "pg_catalog"."default" NOT NULL,
"updated_at" timestamp(6) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"deleted_at" timestamp(6),
"model_supplier" varchar(32) COLLATE "pg_catalog"."default" NOT NULL DEFAULT ''::character varying,
"model_name" varchar(128) COLLATE "pg_catalog"."default" NOT NULL DEFAULT ''::character varying,
"model_type" varchar(32) COLLATE "pg_catalog"."default" NOT NULL DEFAULT ''::character varying,
"base_url" varchar(512) COLLATE "pg_catalog"."default" NOT NULL DEFAULT ''::character varying,
"system_model" bool,
"http_method" varchar(32) COLLATE "pg_catalog"."default" NOT NULL DEFAULT ''::character varying,
"chat_model" bool,
"response_type" int2 NOT NULL DEFAULT 0,
"api_key" varchar(255) COLLATE "pg_catalog"."default" NOT NULL DEFAULT ''::character varying,
"enabled" bool,
"request_head_mapping" jsonb DEFAULT '{}'::jsonb,
"request_body_mapping" jsonb DEFAULT '{}'::jsonb,
"response_mapping" jsonb DEFAULT '{}'::jsonb,
"max_concurrency" int4 NOT NULL DEFAULT 0,
"token_mapping" jsonb,
"async_task_mapping" jsonb,
"token_predict_price" numeric(12,6) NOT NULL DEFAULT 0.000000,
"max_tokens" int4 NOT NULL DEFAULT 0,
"last_frame" varchar(512) COLLATE "pg_catalog"."default" NOT NULL DEFAULT ''::character varying,
"response_body_mapping" jsonb DEFAULT '{}'::jsonb,
"token_predict_price_unit" varchar(32) COLLATE "pg_catalog"."default",
"max_duration" int4,
CONSTRAINT "model_gateway_model_manage_pkey" PRIMARY KEY ("id")
)
;
ALTER TABLE "public"."model_gateway_model_manage"
OWNER TO "postgres";
CREATE INDEX "idx_model_manage_deleted_at" ON "public"."model_gateway_model_manage" USING btree (
"deleted_at" "pg_catalog"."timestamp_ops" ASC NULLS LAST
);
CREATE INDEX "idx_model_manage_model_type" ON "public"."model_gateway_model_manage" USING btree (
"model_type" COLLATE "pg_catalog"."default" "pg_catalog"."text_ops" ASC NULLS LAST
);
CREATE INDEX "idx_model_manage_response_type" ON "public"."model_gateway_model_manage" USING btree (
"response_type" "pg_catalog"."int2_ops" ASC NULLS LAST
);
CREATE INDEX "idx_model_manage_supplier" ON "public"."model_gateway_model_manage" USING btree (
"model_supplier" COLLATE "pg_catalog"."default" "pg_catalog"."text_ops" ASC NULLS LAST
);
CREATE INDEX "idx_model_manage_tenant_id" ON "public"."model_gateway_model_manage" USING btree (
"tenant_id" "pg_catalog"."int8_ops" ASC NULLS LAST
);
COMMENT ON COLUMN "public"."model_gateway_model_manage"."id" IS '主键ID';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."tenant_id" IS '租户ID';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."creator" IS '创建人';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."created_at" IS '创建时间';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."updater" IS '更新人';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."updated_at" IS '更新时间';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."deleted_at" IS '删除时间（软删）';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."model_supplier" IS '模型供应商';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."model_name" IS '模型名称';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."model_type" IS '模型类型';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."base_url" IS '模型地址';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."system_model" IS '是否系统模型';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."http_method" IS 'http请求方法';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."chat_model" IS '是否聊天模型';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."response_type" IS '返回类型：1同步，2异步，3流';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."api_key" IS '接口密钥';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."enabled" IS '是否启用';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."request_head_mapping" IS '请求头映射';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."request_body_mapping" IS '请求体映射';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."response_mapping" IS '响应映射';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."max_concurrency" IS '最大并发数';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."token_mapping" IS 'token映射';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."async_task_mapping" IS '异步任务映射';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."token_predict_price" IS '模型Token预估价格';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."max_tokens" IS '最大token数';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."last_frame" IS '视频尾帧图像地址';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."response_body_mapping" IS '响应主体映射';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."token_predict_price_unit" IS '模型token预估价格单位（秒，百万Token，千Token，字数）';
COMMENT ON COLUMN "public"."model_gateway_model_manage"."max_duration" IS '最大时长（秒）';
COMMENT ON TABLE "public"."model_gateway_model_manage" IS '模型管理表';