19904408334
d699f7ce14
- 新增计费模块：执行开始建单、终态结算/取消/失败处理，支持按条/按秒/按token计费 - 新增执行生命周期跟踪：优雅关停时取消运行中执行并等待落库 - 新增异步任务等待/通知机制（Wait/Notify） - 重构执行记录落库与进度上报，统一失败分类与重试语义 - 重命名文件：async_task.go→async.go、flow_checkpoint_store.go→exec_checkpoint.go、flow_graph_util.go→exec_record.go - 更新 .gitignore 与数据库密码配置
210 lines
9.3 KiB
Go
210 lines
9.3 KiB
Go
// Package gateway ai-agent 对外部服务的调用层：模型网关（model-gateway）的查询/提交/聊天调用，以及文件下载/上传（OSS）。
|
||
// 从 workflow/service/flow 抽离为独立目录，供工作流节点与视频管线（video/）共用，业务侧不再直接依赖 workflow 引擎。
|
||
package gateway
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"net/http"
|
||
"time"
|
||
|
||
"ai-agent/workflow/consts/model"
|
||
"ai-agent/workflow/consts/public"
|
||
|
||
"gitea.redpowerfuture.com/red-future/common/beans"
|
||
commonHttp "gitea.redpowerfuture.com/red-future/common/http"
|
||
"gitea.redpowerfuture.com/red-future/common/utils"
|
||
gmq "github.com/bjang03/gmq/core/gmq"
|
||
"github.com/bjang03/gmq/mq"
|
||
"github.com/bjang03/gmq/types"
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
"github.com/gogf/gf/v2/util/gconv"
|
||
"github.com/google/uuid"
|
||
)
|
||
|
||
// ListModelManageReq 配置列表
|
||
type ListModelManageReq struct {
|
||
*beans.Page `json:"page"`
|
||
ModelName string `p:"modelName" json:"modelName" dc:"模型名称（模糊查询，可选）"`
|
||
ModelType int `p:"modelType" json:"modelType" dc:"模型类型"`
|
||
}
|
||
|
||
// ModelManageListItem 模型配置列表项。listModelManage 返回平铺的 model-gateway entity（id 在顶层），
|
||
// 与 getModelManage 的 {modelManage:{...}} 嵌套不同，故独立成结构而非复用 GetModelInfoByIdRes。
|
||
type ModelManageListItem struct {
|
||
Id int64 `json:"id" dc:"配置ID"`
|
||
ModelName string `json:"modelName" dc:"模型名称"`
|
||
}
|
||
|
||
type ListModelManageRes struct {
|
||
List []*ModelManageListItem `json:"list" dc:"列表数据"`
|
||
Total int `json:"total" dc:"总数"`
|
||
}
|
||
|
||
// GetModelInfoByIdReq 查询模型配置
|
||
type GetModelInfoByIdReq struct {
|
||
ModelId int64 `json:"id" dc:"模型ID"`
|
||
}
|
||
|
||
type GetModelInfoByIdRes struct {
|
||
ModelManage struct {
|
||
Id int64 `json:"id" dc:"模型ID"`
|
||
ModelType model.ModelType `json:"modelType" description:"模型类型"`
|
||
ModelName string `json:"modelName" description:"模型名称"`
|
||
MaxTokens int `json:"maxTokens" description:"最大token数"`
|
||
ChatModel *bool `json:"chatModel" description:"是否聊天模型"`
|
||
ResponseType model.ResponseType `json:"responseType" description:"返回类型：1同步，2异步，3流"`
|
||
Enabled *bool `json:"enabled" description:"是否启用"`
|
||
RequestBodyMapping map[string]any `json:"requestBodyMapping" description:"请求体映射（约等于视频模型 schema）"`
|
||
RequestBusinessFieldMapping map[string]string `json:"requestBusinessFieldMapping" description:"请求业务字段映射"`
|
||
MinDuration int `json:"minDuration" description:"最小时长（秒）"`
|
||
MaxDuration int `json:"maxDuration" description:"最大时长（秒）"`
|
||
} `json:"modelManage" dc:"模型配置"`
|
||
}
|
||
|
||
// modelCallTopic 生成异步模型消息主题：带业务标识（bizName/modelId/sessionId）便于排查，
|
||
// uuid 保证每次调用唯一，避免并发调用共用同一主题导致结果串扰
|
||
func modelCallTopic(bizName string, modelId int64, sessionId string) string {
|
||
return fmt.Sprintf("model-call-%s-%d-%s-%s", bizName, modelId, sessionId, uuid.NewString())
|
||
}
|
||
|
||
// ModelCallReq 调用模型网关
|
||
type ModelCallReq struct {
|
||
ModelId int64 `json:"modelId" dc:"模型ID"`
|
||
BizName string `json:"bizName" dc:"业务名称（调用方模块/系统，用于统计）"`
|
||
SessionId string `json:"sessionId" dc:"会话ID"`
|
||
RequestParams map[string]any `json:"requestParams" dc:"请求参数"`
|
||
BusinessParams map[string]any `json:"businessParams" dc:"业务参数"`
|
||
MsgTopic string `json:"msgTopic" dc:"消息主题（异步必要参数）"`
|
||
}
|
||
|
||
type ModelCallRes struct {
|
||
TaskId int64 `json:"id" dc:"任务ID"`
|
||
State int8 `json:"state" dc:"状态"`
|
||
ModelId int64 `json:"modelId" dc:"生效模型ID（引用行=解析后的系统模型ID，计价按此）"`
|
||
MediaType string `json:"mediaType" dc:"输入媒体类型（shop词汇: text/audio/video）"`
|
||
TotalTokens int64 `json:"totalTokens" dc:"总token"`
|
||
PromptTokens int64 `json:"promptTokens" dc:"输入token"`
|
||
CompletionTokens int64 `json:"completionTokens" dc:"输出token"`
|
||
Tools []ModelTool `json:"tools" dc:"工具"`
|
||
Content map[string]any `json:"content" dc:"内容"`
|
||
Cost float64 `json:"cost" dc:"费用（元）"`
|
||
Duration int64 `json:"duration" dc:"时长（秒）"`
|
||
ErrorMsg string `json:"errorMsg" dc:"错误消息"`
|
||
}
|
||
|
||
type ModelTool struct {
|
||
Id string `json:"id" dc:"工具ID"`
|
||
Type string `json:"type" dc:"工具类型"`
|
||
Function struct {
|
||
Name string `json:"name"`
|
||
Arguments string `json:"arguments"`
|
||
} `json:"function"`
|
||
}
|
||
|
||
// ListModelManage 配置列表
|
||
func ListModelManage(ctx context.Context, req *ListModelManageReq) (res *ListModelManageRes, err error) {
|
||
res = new(ListModelManageRes)
|
||
err = commonHttp.Get(ctx, "model-gateway/model/manage/listModelManage", utils.HeadersFromCtx(ctx, utils.HeadersOptions{TokenFromQuery: true}), res, req)
|
||
return
|
||
}
|
||
|
||
// GetModelInfoById 查询模型配置
|
||
func GetModelInfoById(ctx context.Context, req *GetModelInfoByIdReq) (res *GetModelInfoByIdRes, err error) {
|
||
res = new(GetModelInfoByIdRes)
|
||
err = commonHttp.Get(ctx, "model-gateway/model/manage/getModelManage", utils.HeadersFromCtx(ctx, utils.HeadersOptions{TokenFromQuery: true}), res, req)
|
||
return
|
||
}
|
||
|
||
// SubmitModelCall 提交模型调用（异步模型自动生成唯一 msgTopic），返回提交响应与消息主题。
|
||
// taskId 在 res.TaskId；同步模型 msgTopic 为空串。结果等待由 WaitModelCallResult 完成。
|
||
func SubmitModelCall(ctx context.Context, modelId int64, responseType model.ResponseType, sessionId string, requestParams map[string]any, businessParams map[string]any) (res *ModelCallRes, msgTopic string, err error) {
|
||
// 异步模型必须绑定消息主题接收结果：自动生成唯一主题，
|
||
// 带业务标识（bizName/modelId/sessionId）便于排查，每次调用唯一避免并发串结果
|
||
if responseType != nil && *responseType == *model.ResponseTypeAsync.Code() {
|
||
msgTopic = modelCallTopic(g.Cfg().MustGet(ctx, "server.name").String(), modelId, sessionId)
|
||
}
|
||
|
||
req := ModelCallReq{
|
||
ModelId: modelId,
|
||
BizName: g.Cfg().MustGet(ctx, "server.name").String(),
|
||
SessionId: sessionId,
|
||
RequestParams: requestParams,
|
||
BusinessParams: businessParams,
|
||
MsgTopic: msgTopic,
|
||
}
|
||
// 克隆 commonHttp 客户端（保留 Consul 服务发现），显式设置超时和 ResponseHeaderTimeout
|
||
client := commonHttp.Httpclient.Clone()
|
||
client.SetTimeout(30 * time.Minute)
|
||
if tr, ok := client.Transport.(*http.Transport); ok {
|
||
tr.ResponseHeaderTimeout = 30 * time.Minute
|
||
}
|
||
|
||
res = new(ModelCallRes)
|
||
err = commonHttp.Post(ctx, "model-gateway/model/call/modelCall", utils.HeadersFromCtx(ctx, utils.HeadersOptions{TokenFromQuery: true}), res, &req)
|
||
if err != nil {
|
||
return nil, "", err
|
||
}
|
||
if g.IsEmpty(res.TaskId) || !g.IsEmpty(res.ErrorMsg) {
|
||
return nil, "", fmt.Errorf("创建模型任务失败:%v", res.ErrorMsg)
|
||
}
|
||
return res, msgTopic, nil
|
||
}
|
||
|
||
// WaitModelCallResult 订阅 msgTopic 等待异步模型结果。
|
||
// 重订阅可拿回已发布结果：gmq NATS 侧 Retention=LimitsPolicy+MaxAge=7天（消息不随 ack 删除、保留 7 天），
|
||
// 订阅用 DeliverPolicy=DeliverAllPolicy（新订阅从 stream 序头重放），崩溃后重订阅同一 msgTopic 即可恢复。
|
||
func WaitModelCallResult(ctx context.Context, msgTopic string) (responseParams *ModelCallRes, err error) {
|
||
resultCh := make(chan *ModelCallRes, 1)
|
||
errCh := make(chan error, 1)
|
||
|
||
_, err = gmq.GetGmq(public.GmqMsgPluginsName).GmqSubscribe(ctx, &mq.NatsSubMessage{
|
||
SubMessage: types.SubMessage{
|
||
Topic: msgTopic,
|
||
ConsumerName: fmt.Sprintf("model-call-result-%s", uuid.NewString()),
|
||
AutoAck: false,
|
||
AutoUnsubscribe: true,
|
||
FetchCount: 1,
|
||
HandleFunc: func(ctx context.Context, msg any) error {
|
||
r := new(ModelCallRes)
|
||
if err := gconv.Struct(msg, r); err != nil {
|
||
errCh <- err
|
||
return nil
|
||
}
|
||
if g.IsEmpty(r.TaskId) || !g.IsEmpty(r.ErrorMsg) {
|
||
errCh <- fmt.Errorf("创建模型任务失败:%v", r.ErrorMsg)
|
||
return nil
|
||
}
|
||
resultCh <- r
|
||
return nil
|
||
},
|
||
},
|
||
Durable: true,
|
||
})
|
||
if err != nil {
|
||
return
|
||
}
|
||
|
||
select {
|
||
case responseParams = <-resultCh:
|
||
case err = <-errCh:
|
||
return nil, err
|
||
case <-ctx.Done():
|
||
return nil, ctx.Err()
|
||
}
|
||
return
|
||
}
|
||
|
||
// ModelCallResult 调模型网关生成一段内容并等待结果（内部复用 SubmitModelCall + WaitModelCallResult）。
|
||
// 同步模型提交即返回；异步模型为"提交+等待"一体。
|
||
func ModelCallResult(ctx context.Context, modelId int64, responseType model.ResponseType, sessionId string, requestParams map[string]any, businessParams map[string]any) (responseParams *ModelCallRes, err error) {
|
||
res, msgTopic, err := SubmitModelCall(ctx, modelId, responseType, sessionId, requestParams, businessParams)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if responseType != nil && *responseType == *model.ResponseTypeAsync.Code() {
|
||
return WaitModelCallResult(ctx, msgTopic)
|
||
}
|
||
return res, nil
|
||
}
|