From f7957761f0935b3c5f337bb5e96f702a47e4d7cc Mon Sep 17 00:00:00 2001 From: qhd <1766646056@qq.com> Date: Thu, 3 Sep 2026 16:50:18 +0800 Subject: [PATCH] =?UTF-8?q?fix(workflow):=20=E4=BF=AE=E5=A4=8D=E5=AD=90?= =?UTF-8?q?=E6=B5=81=E7=A8=8B=E6=89=B9=E9=87=8F=E6=89=A7=E8=A1=8C=E7=BC=93?= =?UTF-8?q?=E5=AD=98=E9=94=AE=E5=86=B2=E7=AA=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 递归识别子工作流中的视频模型以支持 per_second 计费，并为子流程批量子执行添加 SubFlowScope 隔离 async/segment 缓存键。 --- workflow/model/dto/flow/flow_execution_dto.go | 1 + workflow/service/flow/billing.go | 42 ++++++++++++++++--- workflow/service/flow/lambda_core.go | 21 ++++++++-- workflow/service/flow/lambda_subflow.go | 16 ++++--- 4 files changed, 65 insertions(+), 15 deletions(-) diff --git a/workflow/model/dto/flow/flow_execution_dto.go b/workflow/model/dto/flow/flow_execution_dto.go index 5330cb0..3ba1763 100644 --- a/workflow/model/dto/flow/flow_execution_dto.go +++ b/workflow/model/dto/flow/flow_execution_dto.go @@ -34,6 +34,7 @@ type FlowExecutionInput struct { SessionId string `json:"sessionId" dc:"会话ID"` ExecutedNodes []ExecutedNode `json:"executedNodes"` // 已执行节点列表，包含执行状态 ForceNewRun bool `json:"forceNewRun" dc:"是否全新执行（false=断点续跑，视频节点段级复用已成功段）"` + SubFlowScope string `json:"subFlowScope" dc:"子流程批量子执行缓存作用域（拼进 async/segment 缓存键隔离各份；顶层为空）"` } // ExecutedNode 已执行节点记录，包含节点ID和执行状态 diff --git a/workflow/service/flow/billing.go b/workflow/service/flow/billing.go index 1564108..4aa8055 100644 --- a/workflow/service/flow/billing.go +++ b/workflow/service/flow/billing.go @@ -14,6 +14,7 @@ import ( nodeDao "ai-agent/workflow/dao/node" sessionDao "ai-agent/workflow/dao/session" + flowDto "ai-agent/workflow/model/dto/flow" nodeDto "ai-agent/workflow/model/dto/node" "ai-agent/workflow/model/entity" ) @@ -134,22 +135,51 @@ func openBillingOrder(ctx context.Context, execId int64, flowContent *entity.Flo } // flowHasVideoModel 工作流是否包含视频模型节点（per_second 计费前置条件）。 -// 遍历节点，按 modelId 去重后经 isVideoModel 查模型类型，任一为视频模型即 true； -// flowContent 缺失/查模型失败按无视频模型处理（per_second 被拦截，fail-closed）。 +// 遍历节点，按 modelId 去重后经 isVideoModel 查模型类型；sub_flow 节点按 subConfig.workflowId +// 递归展开其子工作流（与 SubFlowLambda 运行期解析一致），任一可达层级的视频模型即 true—— +// 视频模型放在子工作流内同样满足 per_second 前置条件。 +// flowContent/子工作流失联、取子工作流失败按该支无视频模型处理（fail-closed：per_second 被拦截）。 +// 递归以被展开的子流程 workflowId 去重防环、跨层共享 modelId 去重，避免重复取子流程/重复查模型类型。 func flowHasVideoModel(ctx context.Context, flowContent *entity.FlowInfo) bool { if flowContent == nil { return false } - seen := make(map[int64]struct{}) + return hasVideoModelRecursive(ctx, flowContent, + make(map[int64]struct{}), make(map[int64]struct{})) +} + +// hasVideoModelRecursive 递归扫描单层 flow 是否含视频模型节点。 +// 先查节点自身模型 id，再展开 sub_flow 子工作流递归（sub_flow 节点自身无模型，ModelId 恒 0）。 +// expandedSubFlows 记录已展开的子流程 workflowId（展开前先标记，防 A→B→A 无限递归并去重重复引用）； +// seenModels 记录已查过的模型 id，跨层共享避免同一模型重复调 isVideoModel。 +func hasVideoModelRecursive(ctx context.Context, flowContent *entity.FlowInfo, + expandedSubFlows, seenModels map[int64]struct{}) bool { for i := range flowContent.Nodes { - modelId := flowContent.Nodes[i].ModelConfig.ModelId + n := &flowContent.Nodes[i] + // sub_flow 节点：取子工作流继续递归；取不到/无法解析按该支无视频模型，继续扫其余节点 + if sc := n.SubConfig; sc != nil && sc.WorkflowId > 0 { + if _, ok := expandedSubFlows[sc.WorkflowId]; ok { + continue + } + expandedSubFlows[sc.WorkflowId] = struct{}{} + subRes, err := FlowUserService.Get(ctx, &flowDto.GetFlowUserReq{Id: sc.WorkflowId}) + if err != nil || subRes == nil || subRes.FlowContent == nil { + continue + } + if hasVideoModelRecursive(ctx, subRes.FlowContent, expandedSubFlows, seenModels) { + return true + } + continue + } + // 普通模型节点：按 modelId 去重后查类型 + modelId := n.ModelConfig.ModelId if modelId <= 0 { continue } - if _, ok := seen[modelId]; ok { + if _, ok := seenModels[modelId]; ok { continue } - seen[modelId] = struct{}{} + seenModels[modelId] = struct{}{} if isVideoModel(ctx, modelId) { return true } diff --git a/workflow/service/flow/lambda_core.go b/workflow/service/flow/lambda_core.go index a32566c..0e37027 100644 --- a/workflow/service/flow/lambda_core.go +++ b/workflow/service/flow/lambda_core.go @@ -34,6 +34,17 @@ func FormLambda(ctx context.Context, input any) (any, error) { return nodeInput, nil } +// cacheNodeId 缓存键节点 id：scope 非空（子流程批量子执行）时拼上作用域后缀，使各份子执行的 +// async/segment 缓存行互相隔离（同一 exec 下多份内层节点 id 相同，不隔离会互相命中/覆盖 done 结果， +// 见《工作流子流程批量缓存隔离设计.md》）；顶层 scope 为空 → 原样返回，行为不变。 +// 仅用于缓存读写键，节点的 Config.Id / node_execution 记录 / 输出引用一律不受影响。 +func cacheNodeId(global *flowDto.FlowExecutionInput, nodeId string) string { + if global == nil || global.SubFlowScope == "" { + return nodeId + } + return nodeId + global.SubFlowScope +} + // ModelLambda 模型调用节点 func ModelLambda(ctx context.Context, input any) (any, error) { nodeInput, ok := input.(*flowDto.NodeExecutionInput) @@ -45,6 +56,8 @@ func ModelLambda(ctx context.Context, input any) (any, error) { if err != nil { return nil, err } + // async/segment 缓存键节点 id（子流程批量子执行带 scope，顶层即 Config.Id） + cNodeId := cacheNodeId(nodeInput.Global, nodeInput.Config.Id) // 2. 前置工具：决定模型调用入参（单次/多次） // 入参统一为扁平模型请求体（BuildModelRequestBody 输出，key 为点分路径）。 @@ -79,7 +92,7 @@ func ModelLambda(ctx context.Context, input any) (any, error) { // 续跑（!ForceNewRun）时读取该节点已成功段；全新执行不查（BuildExecution 已清旧段），saved 为 nil → 全量重生成 var saved map[int]entity.SegmentRef if !nodeInput.Global.ForceNewRun && segVideo { - saved, err = flowDao.FlowSegmentResultDao.ListByNode(ctx, nodeInput.Global.ExecutionId, nodeInput.Config.Id) + saved, err = flowDao.FlowSegmentResultDao.ListByNode(ctx, nodeInput.Global.ExecutionId, cNodeId) if err != nil { return nil, err } @@ -102,7 +115,7 @@ func ModelLambda(ctx context.Context, input any) (any, error) { defer wg.Done() // 每段单次调用，不原地重试：段失败即走节点失败收口（HandleFailedNodeExecution → Interrupt）， // 下次 reExecute 由 planSegmentResume 复用已成功段、仅重生成失败段 - results[i], tokenRes[i], isInference[i], errs[i] = ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params, nodeInput.Config.Prompt, nodeInput.Global.ExecutionId, nodeInput.Config.Id, idxList[i]) + results[i], tokenRes[i], isInference[i], errs[i] = ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params, nodeInput.Config.Prompt, nodeInput.Global.ExecutionId, cNodeId, idxList[i]) // 每段成功立即落库：该段刚成功即持久化，其他段仍在跑时已成功段也不丢； // 后续段失败或进程崩溃（panic/OOM/kill）时，已完成段已在库中，reExecute 可直接复用 if segVideo && errs[i] == nil { @@ -112,7 +125,7 @@ func ModelLambda(ctx context.Context, input any) (any, error) { if key == "" || url == "" { continue } - if err := flowDao.FlowSegmentResultDao.Save(ctx, nodeInput.Global.ExecutionId, nodeInput.Config.Id, idxList[i], key, url); err != nil { + if err := flowDao.FlowSegmentResultDao.Save(ctx, nodeInput.Global.ExecutionId, cNodeId, idxList[i], key, url); err != nil { saveErrs[i] = err } } @@ -160,7 +173,7 @@ func ModelLambda(ctx context.Context, input any) (any, error) { } } else { for _, params := range paramsList { - res, modelRes, _, err := ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params, nodeInput.Config.Prompt, nodeInput.Global.ExecutionId, nodeInput.Config.Id, flowDao.FlowAsyncSegSentinel) + res, modelRes, _, err := ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params, nodeInput.Config.Prompt, nodeInput.Global.ExecutionId, cNodeId, flowDao.FlowAsyncSegSentinel) if err != nil { return nil, err } diff --git a/workflow/service/flow/lambda_subflow.go b/workflow/service/flow/lambda_subflow.go index bd638c4..fa11310 100644 --- a/workflow/service/flow/lambda_subflow.go +++ b/workflow/service/flow/lambda_subflow.go @@ -61,12 +61,18 @@ func SubFlowLambda(ctx context.Context, input any) (any, error) { configMap := buildConfigMap(getRes.FlowContent, nodeList) batchInputs := make([]*flowDto.FlowExecutionInput, 0, batchCount) for j := 0; j < batchCount; j++ { + // 缓存作用域：拼本(外层)子流程节点 Id + 份号 j，供 async/segment 缓存键隔离各份 + // （同一 exec 下 N 份内层节点 id 相同，无作用域会互相命中/覆盖 done 结果，见《工作流子流程批量缓存隔离设计.md》）。 + // 嵌套子流程继承父 scope（Global.SubFlowScope）递归叠加，保证跨 exec/嵌套路径唯一； + // 跨 launch 续跑按同序重放 batch → 各份 scope 稳定，仍能命中自己那份结果。 + scope := nodeExecInput.Global.SubFlowScope + fmt.Sprintf("[%s#%d]", nodeExecInput.Config.Id, j) batchInputs = append(batchInputs, &flowDto.FlowExecutionInput{ - NodeGroupId: nodeExecInput.Global.NodeGroupId, - ExecutionId: nodeExecInput.Global.ExecutionId, - FlowId: nodeExecInput.Global.FlowId, - ConfigMap: cloneConfigMap(configMap), - SessionId: nodeExecInput.Global.SessionId, + NodeGroupId: nodeExecInput.Global.NodeGroupId, + ExecutionId: nodeExecInput.Global.ExecutionId, + FlowId: nodeExecInput.Global.FlowId, + ConfigMap: cloneConfigMap(configMap), + SessionId: nodeExecInput.Global.SessionId, + SubFlowScope: scope, }) } // 7. 执行批量子流程