feat: 添加执行费用统计及token信息记录

在流程执行和节点执行中新增 TotalFee 和 TokenInfo 字段，用于记录每次模型调用的详细计费数据；流程执行完成时汇总所有节点的 token 消耗和费用；重构 token 更新逻辑，从响应字段提取改为使用回调返回的 billingData；优化 URL 后缀提取方法以正确处理带查询参数的链接。

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))

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 {

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 节点执行记录响应

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",
 }

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",

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)
 

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