diff --git a/workflow/dao/node/node_execution_dao.go b/workflow/dao/node/node_execution_dao.go index 8933965..3080940 100644 --- a/workflow/dao/node/node_execution_dao.go +++ b/workflow/dao/node/node_execution_dao.go @@ -86,6 +86,7 @@ func (d *nodeExecutionDao) Get(ctx context.Context, req *nodeDto.GetNodeExecutio func (d *nodeExecutionDao) ListByFlowExecutionId(ctx context.Context, req *nodeDto.ListNodeExecutionByFlowReq, fields ...string) (res []*entity.NodeExecution, total int, err error) { model := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameNodeExecution).NoTenantId(ctx).Fields(fields).OmitEmpty() model.Where(entity.NodeExecutionCol.FlowExecutionId, req.FlowExecutionId) + model.Where(entity.NodeExecutionCol.NodeGroupId, req.NodeGroupId) model.OrderAsc(entity.NodeExecutionCol.CreatedAt) if req.Page != nil { model.Page(int(req.Page.PageNum), int(req.Page.PageSize)) diff --git a/workflow/model/dto/flow/flow_execution_dto.go b/workflow/model/dto/flow/flow_execution_dto.go index 0bedc82..8fb2a8a 100644 --- a/workflow/model/dto/flow/flow_execution_dto.go +++ b/workflow/model/dto/flow/flow_execution_dto.go @@ -124,17 +124,19 @@ type ComposeCallbackReq struct { TotalRounds int `json:"total_rounds"` // 总轮数 Rounds []map[string]any `json:"rounds"` // 每轮详情（动态类型） } `json:"messages,omitempty"` - EpicycleId int64 `json:"epicycleId"` - ErrorMsg string `json:"errorMsg,omitempty"` + EpicycleId int64 `json:"epicycleId"` + ErrorMsg string `json:"errorMsg,omitempty"` + BillingData []map[string]any `json:"billing_data"` } type ModelCallbackReq struct { - g.Meta `path:"/modelCallback" method:"post" tags:"提示词处理" summary:"model-gateway 回调" dc:"model-gateway 成功后 GET 回调：callbackUrl/{bizName}"` - TaskId string `p:"task_id" json:"task_id" v:"required#task_id不能为空" dc:"网关任务ID"` - State int `p:"state" json:"state" dc:"网关任务状态"` - OssFile string `p:"oss_file" json:"oss_file" dc:"结果文件地址"` - FileType string `p:"file_type" json:"file_type" dc:"结果文件类型"` - ErrorMsg string `json:"error_msg"` + g.Meta `path:"/modelCallback" method:"post" tags:"提示词处理" summary:"model-gateway 回调" dc:"model-gateway 成功后 GET 回调：callbackUrl/{bizName}"` + TaskId string `p:"task_id" json:"task_id" v:"required#task_id不能为空" dc:"网关任务ID"` + State int `p:"state" json:"state" dc:"网关任务状态"` + OssFile string `p:"oss_file" json:"oss_file" dc:"结果文件地址"` + FileType string `p:"file_type" json:"file_type" dc:"结果文件类型"` + ErrorMsg string `json:"error_msg"` + BillingData []map[string]any `json:"billing_data"` } type VideoCallbackReq struct { @@ -225,6 +227,8 @@ type UpdateFlowExecutionReq struct { OutputParams []map[string]interface{} `json:"outputParams" description:"输出参数"` ErrorMessage string `json:"errorMessage" description:"错误信息"` TraceId string `json:"traceId" description:"跟踪ID"` + TotalTokens int `json:"totalTokens" description:"总token"` + TotalFee float64 `json:"totalFee" description:"总费用"` } type GetFlowExecutionReq struct { diff --git a/workflow/model/dto/node/node_execution_dto.go b/workflow/model/dto/node/node_execution_dto.go index e91ec1c..aa8c501 100644 --- a/workflow/model/dto/node/node_execution_dto.go +++ b/workflow/model/dto/node/node_execution_dto.go @@ -38,6 +38,7 @@ type UpdateNodeExecutionReq struct { PromptTokens int `json:"promptTokens"` CompletionTokens int `json:"completionTokens"` TotalTokens int `json:"totalTokens"` + TokenInfo []map[string]any `json:"tokenInfo"` Status node.NodeExecutionStatus `json:"status"` DurationMs int64 `json:"durationMs"` ErrorMessage string `json:"errorMessage"` @@ -60,6 +61,7 @@ type ListNodeExecutionByFlowReq struct { g.Meta `path:"/listByFlow" method:"get" tags:"节点执行记录" summary:"查询流程节点执行列表" dc:"查询指定流程执行下的所有节点执行记录"` Page *beans.Page `json:"page"` FlowExecutionId int64 `json:"flowExecutionId" v:"required#流程执行ID不能为空"` + NodeGroupId string `json:"nodeGroupId"` } // NodeExecutionResp 节点执行记录响应 diff --git a/workflow/model/entity/flow_execution.go b/workflow/model/entity/flow_execution.go index 6f90b0e..0bbdc3f 100644 --- a/workflow/model/entity/flow_execution.go +++ b/workflow/model/entity/flow_execution.go @@ -22,6 +22,7 @@ type FlowExecution struct { TraceId string `orm:"trace_id" json:"traceId" description:"跟踪ID"` SessionId string `orm:"session_id" json:"sessionId" description:"会话ID"` TotalTokens int `orm:"total_tokens" json:"totalTokens" description:"总token消耗"` + TotalFee int `orm:"total_fee" json:"totalFee" description:"总费用"` } type flowExecutionCol struct { @@ -39,6 +40,7 @@ type flowExecutionCol struct { TraceId string SessionId string TotalTokens string + TotalFee string } var FlowExecutionCol = flowExecutionCol{ @@ -56,4 +58,5 @@ var FlowExecutionCol = flowExecutionCol{ TraceId: "trace_id", SessionId: "session_id", TotalTokens: "total_tokens", + TotalFee: "total_fee", } diff --git a/workflow/model/entity/node_execution.go b/workflow/model/entity/node_execution.go index b8b5e08..d9d9f3a 100644 --- a/workflow/model/entity/node_execution.go +++ b/workflow/model/entity/node_execution.go @@ -22,6 +22,7 @@ type NodeExecution struct { PromptTokens int `orm:"prompt_tokens" json:"promptTokens" description:"提示词token消耗"` CompletionTokens int `orm:"completion_tokens" json:"completionTokens" description:"补全token消耗"` TotalTokens int `orm:"total_tokens" json:"totalTokens" description:"总token消耗"` + TokenInfo []map[string]interface{} `orm:"token_info" json:"tokenInfo" description:"token信息"` Status node.NodeExecutionStatus `orm:"status" json:"status" description:"执行状态:1-运行中,2-成功,3-失败,4-暂停,5-等待执行"` DurationMs int64 `orm:"duration_ms" json:"durationMs" description:"执行时长（毫秒）"` ErrorMessage string `orm:"error_message" json:"errorMessage" description:"错误信息"` @@ -40,6 +41,7 @@ type nodeExecutionCol struct { PromptTokens string CompletionTokens string TotalTokens string + TokenInfo string Status string DurationMs string ErrorMessage string @@ -58,6 +60,7 @@ var NodeExecutionCol = nodeExecutionCol{ PromptTokens: "prompt_tokens", CompletionTokens: "completion_tokens", TotalTokens: "total_tokens", + TokenInfo: "token_info", Status: "status", DurationMs: "duration_ms", ErrorMessage: "error_message", diff --git a/workflow/service/flow/lambda_node.go b/workflow/service/flow/lambda_node.go index 64b3115..1cc3684 100644 --- a/workflow/service/flow/lambda_node.go +++ b/workflow/service/flow/lambda_node.go @@ -6,12 +6,14 @@ import ( "ai-agent/workflow/consts/public" fileDao "ai-agent/workflow/dao/file" flowDao "ai-agent/workflow/dao/flow" + nodeDao "ai-agent/workflow/dao/node" "ai-agent/workflow/model/dto" fileDto "ai-agent/workflow/model/dto/file" flowDto "ai-agent/workflow/model/dto/flow" + nodeDto "ai-agent/workflow/model/dto/node" + "ai-agent/workflow/model/entity" "context" "fmt" - "path/filepath" "strconv" "strings" "sync" @@ -86,7 +88,7 @@ func JudgeLambda(ctx context.Context, input any) (string, error) { if err != nil { return "", err } - composeResult, err := GetComposeResult(ctx, 2, getIsChatModel.Model.ModelName, "", "", []map[string]any{{"prompt": strings.Join(branchIdNameLines, "\n")}}, []map[string]any{{"prompt": contextParts}}, nodeInput.Global.FileUrl, nodeInput.Global.SessionId, nodeInput.Config.Id, "判断节点") + composeResult, err := GetComposeResult(ctx, nodeInput.NodeExecutionId, 2, getIsChatModel.Model.ModelName, "", "", []map[string]any{{"prompt": strings.Join(branchIdNameLines, "\n")}}, []map[string]any{{"prompt": contextParts}}, nodeInput.Global.FileUrl, nodeInput.Global.SessionId, nodeInput.Config.Id, "判断节点") if err != nil { return "", err } @@ -299,7 +301,7 @@ func VideoModelLambda(ctx context.Context, input any) (any, error) { return nil, fmt.Errorf("下载图片失败: %w", err) } // 构造文件名 - fileName := fmt.Sprintf("ai_video_%d%s", time.Now().UnixMilli(), strings.ToLower(filepath.Ext(videoURL[0]))) + fileName := fmt.Sprintf("ai_video_%d%s", time.Now().UnixMilli(), GetUrlSuffix(videoURL[0], true)) // 上传到你的OSS（你项目已有的Upload方法） var upResp *dto.UploadFileBytesRes upResp, err = Upload(ctx, &dto.UploadFileBytesReq{ @@ -650,10 +652,28 @@ func SummaryLambda(ctx context.Context, input any) (any, error) { return err } + res, _, err := nodeDao.NodeExecutionDao.ListByFlowExecutionId(ctx, &nodeDto.ListNodeExecutionByFlowReq{ + NodeGroupId: execInput.Global.NodeGroupId, + }, entity.NodeExecutionCol.TokenInfo) + if err != nil { + return err + } + var totalTokens int + var totalFee float64 + for _, item := range res { + for _, itemToken := range item.TokenInfo { + m := gconv.Map(itemToken) + totalTokens += gconv.Int(m["total_tokens"]) + totalFee += gconv.Float64(m["total_fee"]) + } + } + executionReq := flowDto.UpdateFlowExecutionReq{ Id: execInput.Global.ExecutionId, Status: flow.FlowExecutionStatusSuccess.Code(), OutputParams: summaryResult, + TotalTokens: totalTokens, + TotalFee: totalFee, } _, err = flowDao.FlowExecutionDao.Update(ctx, &executionReq) diff --git a/workflow/service/flow/lambda_node_util.go b/workflow/service/flow/lambda_node_util.go index 6af2455..125277e 100644 --- a/workflow/service/flow/lambda_node_util.go +++ b/workflow/service/flow/lambda_node_util.go @@ -14,6 +14,7 @@ import ( "mime/multipart" "net/http" "net/url" + "path" "path/filepath" "regexp" "strconv" @@ -97,7 +98,7 @@ func GetModelInfo(ctx context.Context, req *flowDto.GetModelInfoReq) (res *flowD return } -func GetComposeResult(ctx context.Context, buildType int, modelName, promptContent, skillName string, form []map[string]any, userForm []map[string]any, fileUrl []string, sessionId, nodeId string, cause string) (res *flowDto.ComposeCallbackReq, err error) { +func GetComposeResult(ctx context.Context, nodeExecutionId int64, buildType int, modelName, promptContent, skillName string, form []map[string]any, userForm []map[string]any, fileUrl []string, sessionId, nodeId string, cause string) (res *flowDto.ComposeCallbackReq, err error) { var callbackUrl = utils.GetCallbackURL(ctx, "/flow/execution/composeCallBack") var consult = make([]flowDto.Consult, 0) var collectFileUrls func(val any) (fullyConsumed bool) @@ -206,18 +207,19 @@ func GetComposeResult(ctx context.Context, buildType int, modelName, promptConte if err = gconv.Struct(waitRes, msg); err != nil { return nil, err } + updateTokenCount(ctx, nodeExecutionId, msg.BillingData) if !g.IsEmpty(msg.ErrorMsg) { return nil, fmt.Errorf(msg.ErrorMsg) } return msg, nil } -func CreateGatewayTask(ctx context.Context, epicycleId int64, model string, content map[string]any) (map[string]any, error) { +func CreateGatewayTask(ctx context.Context, nodeExecutionId int64, epicycleId int64, model string, content map[string]any) (map[string]any, error) { taskId, err := createGatewayTaskOnly(ctx, epicycleId, model, content) if err != nil { return nil, err } - return waitGatewayResult(ctx, taskId) + return waitGatewayResult(ctx, nodeExecutionId, taskId) } // createGatewayTaskOnly creates a gateway task and returns the taskId only @@ -249,12 +251,11 @@ func createGatewayTaskOnly(ctx context.Context, epicycleId int64, model string, if g.IsEmpty(res.TaskId) { return "", fmt.Errorf("创建模型任务失败，taskId为空") } - return res.TaskId, nil } // waitGatewayResult waits for a created gateway task to complete and returns the result -func waitGatewayResult(ctx context.Context, taskId string) (map[string]any, error) { +func waitGatewayResult(ctx context.Context, nodeExecutionId int64, taskId string) (map[string]any, error) { waitRes, err := Wait(ctx, taskId) if err != nil { return nil, err @@ -264,6 +265,7 @@ func waitGatewayResult(ctx context.Context, taskId string) (map[string]any, erro if err = gconv.Struct(waitRes, task); err != nil { return nil, err } + updateTokenCount(ctx, nodeExecutionId, task.BillingData) if task.State == 3 || !g.IsEmpty(task.ErrorMsg) { return nil, fmt.Errorf("模型执行失败：%s", task.ErrorMsg) } @@ -279,14 +281,23 @@ func waitGatewayResult(ctx context.Context, taskId string) (map[string]any, erro } // updateTokenCount updates the token count in node execution -func updateTokenCount(ctx context.Context, nodeExecutionId int64, responseField string, result map[string]any) { - if responseField == "" { +func updateTokenCount(ctx context.Context, nodeExecutionId int64, tokenInfo []map[string]any) { + res, err := nodeDao.NodeExecutionDao.Get(ctx, &nodeDto.GetNodeExecutionReq{ + Id: nodeExecutionId, + }, entity.NodeExecutionCol.TokenInfo) + if err != nil { return } + var t []map[string]any + for _, item := range res.TokenInfo { + t = append(t, item) + } + for _, item := range tokenInfo { + t = append(t, item) + } _, _ = nodeDao.NodeExecutionDao.Update(ctx, &nodeDto.UpdateNodeExecutionReq{ - Id: nodeExecutionId, - CompletionTokens: gconv.Int(result[responseField]), - TotalTokens: gconv.Int(result[responseField]), + Id: nodeExecutionId, + TokenInfo: t, }) } @@ -315,7 +326,7 @@ func GetModelResult(ctx context.Context, sessionId string, nodeInput *flowDto.No } } } - composeResult, err := GetComposeResult(ctx, buildType, nodeInput.Config.ModelConfig.ModelName, nodeInput.Config.PromptContent, skillName, form, userForm, nodeInput.Global.FileUrl, sessionId, nodeInput.Config.Id, nodeInput.Config.Name) + composeResult, err := GetComposeResult(ctx, nodeInput.NodeExecutionId, buildType, nodeInput.Config.ModelConfig.ModelName, nodeInput.Config.PromptContent, skillName, form, userForm, nodeInput.Global.FileUrl, sessionId, nodeInput.Config.Id, nodeInput.Config.Name) if err != nil { return nil, err } @@ -344,7 +355,7 @@ func GetModelResult(ctx context.Context, sessionId string, nodeInput *flowDto.No } var taskResult map[string]any - taskResult, err = CreateGatewayTask(ctx, composeResult.EpicycleId, nodeInput.Config.ModelConfig.ModelName, item) + taskResult, err = CreateGatewayTask(ctx, nodeInput.NodeExecutionId, composeResult.EpicycleId, nodeInput.Config.ModelConfig.ModelName, item) if err != nil { return nil, err } @@ -364,7 +375,7 @@ func GetModelResult(ctx context.Context, sessionId string, nodeInput *flowDto.No } mapTaskResult[idx] = taskResult - updateTokenCount(ctx, nodeInput.NodeExecutionId, modelInfo.Model.ResponseTokenField, taskResult) + //updateTokenCount(ctx, nodeInput.NodeExecutionId, modelInfo.Model.ResponseTokenField, taskResult) } } else { taskIdList := make([]string, len(composeResult.Messages.Rounds)) @@ -393,7 +404,7 @@ func GetModelResult(ctx context.Context, sessionId string, nodeInput *flowDto.No go func(idx int, taskId string) { defer wg.Done() - taskResult, err := waitGatewayResult(subCtx, taskId) + taskResult, err := waitGatewayResult(subCtx, nodeInput.NodeExecutionId, taskId) if err != nil { errChan <- err globalCancel() // 全局取消，所有协程收到ctx取消信号快速退出 @@ -405,7 +416,7 @@ func GetModelResult(ctx context.Context, sessionId string, nodeInput *flowDto.No mapTaskResult[idx] = taskResult mu.Unlock() - updateTokenCount(ctx, nodeInput.NodeExecutionId, modelInfo.Model.ResponseTokenField, taskResult) + //updateTokenCount(ctx, nodeInput.NodeExecutionId, modelInfo.Model.ResponseTokenField, taskResult) }(idx, taskId) } @@ -426,7 +437,7 @@ func GetModelResult(ctx context.Context, sessionId string, nodeInput *flowDto.No } else { for idx, item := range composeResult.Messages.Rounds { mapTaskResult[idx] = item - updateTokenCount(ctx, nodeInput.NodeExecutionId, modelInfo.Model.ResponseTokenField, item) + //updateTokenCount(ctx, nodeInput.NodeExecutionId, modelInfo.Model.ResponseTokenField, item) } } @@ -622,6 +633,32 @@ func GetFileTypeByPath(filePath string) string { } } +// GetUrlSuffix 获取URL文件后缀 +// rawUrl: 原始链接 +// withDot: true 返回 .mp4 false 返回 mp4 +func GetUrlSuffix(rawUrl string, withDot bool) string { + // 解析URL，剥离查询参数 + u, err := url.Parse(rawUrl) + if err != nil { + return "" + } + + // 提取路径部分 + filePath := u.Path + // 获取文件名 + fileName := path.Base(filePath) + if fileName == "" || !strings.Contains(fileName, ".") { + return "" + } + + // 截取后缀 + suffix := path.Ext(fileName) + if !withDot { + suffix = strings.TrimPrefix(suffix, ".") + } + return suffix +} + func BuildText(text string) string { // 生成单条HTML var htmlBuilder strings.Builder