365 lines
14 KiB
Go
365 lines
14 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"model-gateway/consts/model"
|
||
"model-gateway/dao"
|
||
"model-gateway/model/dto"
|
||
"model-gateway/model/entity"
|
||
modelUtils "model-gateway/service/utils"
|
||
"net/http"
|
||
"time"
|
||
|
||
"gitea.redpowerfuture.com/red-future/common/utils"
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
"github.com/gogf/gf/v2/os/glog"
|
||
"github.com/gogf/gf/v2/util/gconv"
|
||
)
|
||
|
||
var ModelCall = &modelCallService{}
|
||
|
||
type modelCallService struct{}
|
||
|
||
func (s *modelCallService) ModelCall(ctx context.Context, req *dto.ModelCallReq) (res *dto.ModelCallRes, err error) {
|
||
// 1) 检查模型配置
|
||
var modelInfo *entity.ModelManage
|
||
modelInfo, err = dao.ModelManage.GetNotTenantId(ctx, &dto.GetModelManageReq{
|
||
Id: req.ModelId,
|
||
})
|
||
if err != nil {
|
||
return nil, fmt.Errorf("获取模型配置失败: %v", err)
|
||
}
|
||
if modelInfo == nil {
|
||
return nil, fmt.Errorf("模型不存在")
|
||
}
|
||
// 引用行 → 解析为系统模型配置+本人 apiKey；系统模型已删除等解析失败 → 阻塞调用
|
||
modelInfo, err = modelUtils.ResolveModelConfig(ctx, modelInfo)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if modelInfo.Enabled != nil && !*modelInfo.Enabled {
|
||
return nil, fmt.Errorf("模型不存在或未启用")
|
||
}
|
||
|
||
now := time.Now()
|
||
userInfo, err := utils.GetUserInfo(ctx)
|
||
if err != nil {
|
||
return
|
||
}
|
||
if !g.IsEmpty(modelInfo.RefSystemModelId) {
|
||
// 调用前检查模型计价配置（shop-user-trade）：未配置/未启用 → 阻塞调用（subject=解析后的系统模型 id）
|
||
if err = modelBillable(ctx, modelInfo.Id); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
err = queue(ctx, modelInfo.ModelName, userInfo.TenantId, gconv.Int64(modelInfo.MaxConcurrency), func(ctx context.Context) (err error) {
|
||
if *modelInfo.ResponseType == *model.ResponseTypeSync.Code() || *modelInfo.ResponseType == *model.ResponseTypeStream.Code() {
|
||
var newRequestParams map[string]any
|
||
var id int64
|
||
id, newRequestParams, err = s.saveModelRequestParams(ctx, now, modelInfo, req)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if g.IsEmpty(id) || g.IsEmpty(newRequestParams) {
|
||
return fmt.Errorf("保存模型请求参数失败")
|
||
}
|
||
if *modelInfo.ResponseType == *model.ResponseTypeSync.Code() {
|
||
res, err = ModelSession.CreateSession(ctx, &dto.CallModelSessionReq{
|
||
Id: id,
|
||
ModelInfo: modelInfo,
|
||
RequestParams: newRequestParams,
|
||
})
|
||
} else {
|
||
res, err = ModelSession.CreateSessionStreamOnce(ctx, &dto.CallModelSessionReq{
|
||
Id: id,
|
||
ModelInfo: modelInfo,
|
||
RequestParams: newRequestParams,
|
||
})
|
||
}
|
||
}
|
||
if *modelInfo.ResponseType == *model.ResponseTypeAsync.Code() {
|
||
if g.IsEmpty(req.MsgTopic) {
|
||
return fmt.Errorf("请指定消息主题")
|
||
}
|
||
var newRequestParams map[string]any
|
||
var id int64
|
||
id, newRequestParams, err = s.saveModelRequestParams(ctx, now, modelInfo, req)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if g.IsEmpty(id) || g.IsEmpty(newRequestParams) {
|
||
return fmt.Errorf("保存模型请求参数失败")
|
||
}
|
||
res, err = ModelTaskStart.CreateTask(ctx, &dto.CallModelTaskStartReq{
|
||
Id: id,
|
||
ModelInfo: modelInfo,
|
||
RequestParams: newRequestParams,
|
||
})
|
||
}
|
||
return
|
||
})
|
||
return
|
||
}
|
||
|
||
func (s *modelCallService) ModelCallStream(ctx context.Context, w http.ResponseWriter, req *dto.ModelCallStreamReq) (err error) {
|
||
// 1) 检查模型配置
|
||
var modelInfo *entity.ModelManage
|
||
modelInfo, err = dao.ModelManage.GetNotTenantId(ctx, &dto.GetModelManageReq{
|
||
Id: req.ModelId,
|
||
})
|
||
if err != nil {
|
||
return fmt.Errorf("获取模型配置失败: %v", err)
|
||
}
|
||
if modelInfo == nil {
|
||
return fmt.Errorf("模型不存在")
|
||
}
|
||
// 引用行 → 解析为系统模型配置+本人 apiKey；系统模型已删除等解析失败 → 阻塞调用
|
||
modelInfo, err = modelUtils.ResolveModelConfig(ctx, modelInfo)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if modelInfo.Enabled != nil && !*modelInfo.Enabled {
|
||
return fmt.Errorf("模型不存在或未启用")
|
||
}
|
||
if *modelInfo.ResponseType == *model.ResponseTypeStream.Code() {
|
||
now := time.Now()
|
||
userInfo, err := utils.GetUserInfo(ctx)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !g.IsEmpty(modelInfo.RefSystemModelId) {
|
||
// 调用前检查模型计价配置（shop-user-trade）：未配置/未启用 → 阻塞调用（subject=解析后的系统模型 id）
|
||
if err = modelBillable(ctx, modelInfo.Id); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
err = queue(ctx, modelInfo.ModelName, userInfo.TenantId, gconv.Int64(modelInfo.MaxConcurrency), func(ctx context.Context) (err error) {
|
||
var newRequestParams map[string]any
|
||
var id int64
|
||
id, newRequestParams, err = s.saveModelRequestParams(ctx, now, modelInfo, &dto.ModelCallReq{
|
||
ModelId: req.ModelId,
|
||
RequestParams: req.RequestParams,
|
||
BusinessParams: req.BusinessParams,
|
||
SessionId: req.SessionId,
|
||
BizName: req.BizName,
|
||
})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if g.IsEmpty(id) || g.IsEmpty(newRequestParams) {
|
||
return fmt.Errorf("保存模型请求参数失败")
|
||
}
|
||
_, err = ModelSession.CreateSessionStream(ctx, w, &dto.CallModelSessionReq{
|
||
Id: id,
|
||
ModelInfo: modelInfo,
|
||
RequestParams: newRequestParams,
|
||
})
|
||
return
|
||
})
|
||
} else {
|
||
return fmt.Errorf("模型响应类型错误")
|
||
}
|
||
return err
|
||
}
|
||
|
||
// saveModelRequestParams 保存模型请求参数
|
||
func (s *modelCallService) saveModelRequestParams(ctx context.Context, now time.Time, modelInfo *entity.ModelManage, req *dto.ModelCallReq) (id int64, newRequestParams map[string]any, err error) {
|
||
|
||
// 统一走模板校验+构建：requestParams 只装模板字段，businessParams 只装业务字段
|
||
out, err := buildChatRequestParams(modelInfo, req.RequestParams, req.BusinessParams)
|
||
if err != nil {
|
||
return 0, nil, err
|
||
}
|
||
|
||
// 1) 上传模型原始请求参数文件（requestParams + businessParams 合并，保证审计完整）
|
||
originalParams := make(map[string]any, len(req.RequestParams)+len(req.BusinessParams))
|
||
for k, v := range req.RequestParams {
|
||
originalParams[k] = v
|
||
}
|
||
for k, v := range req.BusinessParams {
|
||
originalParams[k] = v
|
||
}
|
||
uploadOriginalReq, err := Upload(ctx, &dto.UploadFileBytesReq{
|
||
FileBytes: gconv.Bytes(gconv.String(originalParams)),
|
||
FileName: fmt.Sprintf("modelRequestParams:%v.json", now.UnixMilli()),
|
||
})
|
||
if err != nil {
|
||
return 0, nil, fmt.Errorf("上传模型原始请求参数文件失败: %v", err)
|
||
}
|
||
|
||
// 2) 上传模型解析成功的请求参数文件
|
||
uploadNewReq, err := Upload(ctx, &dto.UploadFileBytesReq{
|
||
FileBytes: gconv.Bytes(gconv.String(out)),
|
||
FileName: fmt.Sprintf("modelNewRequestParams:%v.json", now.UnixMilli()),
|
||
})
|
||
if err != nil {
|
||
return 0, nil, fmt.Errorf("上传模型解析请求参数文件失败：%v", err)
|
||
}
|
||
|
||
// 3) 保存模型请求信息（快照媒体类型=shop 计费词汇 audio/video，空=无媒体引用，任务完成时直接用于算费；模型计费配置任务完成时按 modelId 现查）
|
||
if *modelInfo.ResponseType == *model.ResponseTypeAsync.Code() {
|
||
id, err = dao.ModelTaskStart.Insert(ctx, &dto.CreateModelTaskStartReq{
|
||
ModelId: req.ModelId,
|
||
BizName: req.BizName,
|
||
MsgTopic: req.MsgTopic,
|
||
RequestPath: uploadNewReq.FileURL,
|
||
OriginalRequestPath: uploadOriginalReq.FileURL,
|
||
MediaType: modelUtils.DetectMediaType(modelInfo.RequestBusinessFieldMapping, out),
|
||
})
|
||
if err != nil {
|
||
return 0, nil, fmt.Errorf("保存模型请求信息失败: %v", err)
|
||
}
|
||
} else {
|
||
id, err = dao.ModelSession.Insert(ctx, &dto.CreateModelSessionReq{
|
||
ModelId: req.ModelId,
|
||
BizName: req.BizName,
|
||
SessionId: req.SessionId,
|
||
RequestPath: uploadNewReq.FileURL,
|
||
OriginalRequestPath: uploadOriginalReq.FileURL,
|
||
})
|
||
if err != nil {
|
||
return 0, nil, fmt.Errorf("保存模型请求信息失败: %v", err)
|
||
}
|
||
}
|
||
return id, out, nil
|
||
}
|
||
|
||
func queue(ctx context.Context, modelName string, tenantId uint64, maxCon int64, f func(ctx context.Context) (err error)) (err error) {
|
||
const (
|
||
keyExpireSec = 600 // 名额Key兜底过期时间 10min（进程崩溃后自愈）
|
||
refreshStep = keyExpireSec / 3 // 执行期间续期间隔
|
||
waitInterval = 10 * time.Second // 超限轮询等待间隔
|
||
)
|
||
// Redis 操作统一使用独立上下文，避免外部 ctx canceled
|
||
redisCtx := context.WithoutCancel(ctx)
|
||
concurrencyKey := fmt.Sprintf("model:concurrency:%d:%s", tenantId, modelName)
|
||
|
||
// 1) 原子占用并发名额：utils.SemaphoreAcquire 在 WATCH 事务内完成 判满→INCR→首设EXPIRE→超限不写，
|
||
// 不再需要旧 reserveSlot 的「分布式锁 + Incr + 回滚」组合（组合已事务化，外层锁冗余）。
|
||
// 超限（false）按 waitInterval 轮询重试；max<=0 视为不限制（SemaphoreAcquire 内部直接放行）。
|
||
for {
|
||
if ctx.Err() != nil {
|
||
return ctx.Err()
|
||
}
|
||
ok, e := utils.SemaphoreAcquire(redisCtx, concurrencyKey, int(maxCon), keyExpireSec)
|
||
if e != nil {
|
||
return e
|
||
}
|
||
if ok {
|
||
// 展示当前并发数（占用后 GET，与旧 reserveSlot 的 Incr 后计数值语义一致）
|
||
if v, e := g.Redis().Get(redisCtx, concurrencyKey); e == nil {
|
||
glog.Infof(ctx, "并发数: %s %d/%d", concurrencyKey, v.Int64(), maxCon)
|
||
}
|
||
break
|
||
}
|
||
glog.Infof(ctx, "并发超限等待: %s max=%d", concurrencyKey, maxCon)
|
||
time.Sleep(waitInterval)
|
||
}
|
||
|
||
// 2) 执行业务期间周期续期名额Key：SemaphoreAcquire 仅在首次占用（计数从 0 起）时设 TTL，
|
||
// 长耗时调用靠本循环持续保活——Key 过期后计数归零会突破 max 并发上限造成超发。
|
||
stop := make(chan struct{})
|
||
go refreshTTL(redisCtx, concurrencyKey, keyExpireSec, refreshStep, stop)
|
||
// 3) 无论业务正常返回还是 panic，都停掉续期并释放名额（幂等，计数归零自动删除 key）
|
||
defer func() {
|
||
close(stop)
|
||
_ = utils.SemaphoreRelease(redisCtx, concurrencyKey)
|
||
}()
|
||
|
||
// 4) 执行业务
|
||
return f(ctx)
|
||
}
|
||
|
||
// refreshTTL 周期给名额 Key 续期，直到 stop 关闭；防止长耗时执行期间 Key 提前过期。
|
||
func refreshTTL(redisCtx context.Context, concurrencyKey string, keyExpireSec, step int64, stop <-chan struct{}) {
|
||
interval := step
|
||
if interval < 1 {
|
||
interval = 1
|
||
}
|
||
ticker := time.NewTicker(time.Duration(interval) * time.Second)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case <-ticker.C:
|
||
if _, err := g.Redis().Expire(redisCtx, concurrencyKey, keyExpireSec); err != nil {
|
||
glog.Errorf(context.TODO(), "redis refresh concurrency ttl err: %v", err)
|
||
}
|
||
case <-stop:
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
// buildChatRequestParams 按模型配置的请求模板 + 业务字段映射构建请求体（ModelCall 请求路径共用）：
|
||
// 1. requestParams 只装模板字段，按 requestBodyMapping 模板校验（CheckParams）+ 构建（ParseConfigTemplate）；
|
||
// 未配置映射的字段（如未配置映射的 messages/tools）会被模板拒绝，明确报错
|
||
// 2. requestParams 为空时按配置模板构建请求结构（模板 defaultValue 生效），
|
||
// 避免"结构由模板声明、值全走业务字段"的场景因请求体为空而构建失败
|
||
// 3. businessParams 只装业务字段，按业务字段名（RequestBusinessFieldMapping 的 key）传值，
|
||
// TakeBusinessFields 解析为写入路径，构建完成后由 WriteBusinessFields 按路径写入最终请求体
|
||
func buildChatRequestParams(modelInfo *entity.ModelManage, requestParams, businessParams map[string]any) (map[string]any, error) {
|
||
|
||
// requestParams 可能混有扁平路径 key（messages.enumValues...）与已是对象/数组的值（stream）。
|
||
// IsFlatMap 遇 map/slice 值即整体返回 false 会跳过 unflatten；sjson.Set 能处理任意值类型作为子树，
|
||
// 带点 key 按路径展开、无点 key 直接赋值，故始终 unflatten
|
||
var err error
|
||
requestParams, err = utils.UnFlatBySjson(requestParams)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
rest := make(map[string]any, len(requestParams))
|
||
for k, v := range requestParams {
|
||
rest[k] = v
|
||
}
|
||
// 请求结构源：模板字段兜底（模板 value/defaultValue 生效）+ requestParams 覆盖同名；
|
||
// 保证模板声明的结构字段（如 stream_options 对象）即使 requestParams 未传也进请求体
|
||
// 用户已传字段按模板类型元数据递归合并（补 type 包装、补默认字段），数组字段仅在用户提供时才合并
|
||
src := rest
|
||
for k, v := range modelInfo.RequestBodyMapping {
|
||
tmplMap, _ := v.(map[string]any)
|
||
if tmplMap != nil {
|
||
if t, _ := tmplMap["type"].(string); t == "array" {
|
||
if _, has := src[k]; !has {
|
||
continue
|
||
}
|
||
src[k] = modelUtils.MergeNode(src[k], v)
|
||
continue
|
||
}
|
||
}
|
||
if userVal, has := src[k]; has {
|
||
src[k] = modelUtils.MergeNode(userVal, v)
|
||
} else {
|
||
src[k] = modelUtils.DeepCopyNode(v)
|
||
}
|
||
}
|
||
if len(requestParams) > 0 {
|
||
// requestParams 非空才按模板严格校验模板字段（空值回填 default）；
|
||
// 为空时跳过，避免业务字段未写入就误报必填缺失
|
||
if err := modelUtils.CheckParams(src, modelInfo.RequestBodyMapping); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
|
||
out := modelUtils.ParseConfigTemplate(src)
|
||
if !g.IsEmpty(businessParams) {
|
||
// 业务字段：businessParams 按业务字段名传值，解析为映射路径后写入
|
||
bizValues, err := modelUtils.TakeBusinessFields(businessParams, modelInfo.RequestBusinessFieldMapping)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
// 业务字段按映射路径写入最终请求体（写 out 而非 rest：rest 只是模板字段容器，out 才是下发模型的请求体）
|
||
if err = modelUtils.WriteBusinessFields(out, bizValues); err != nil {
|
||
return nil, err
|
||
}
|
||
// 合并后按模板约束整体校验：必填/长度/范围
|
||
if err = modelUtils.CheckBody(out, modelInfo.RequestBodyMapping); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
// 按模板声明的 type 归一字段值类型（模板字段 value / 业务字段写入值都可能与声明类型不符）
|
||
out = modelUtils.CoerceBodyTypes(out, modelInfo.RequestBodyMapping)
|
||
return out, nil
|
||
}
|