feat: 添加子流程节点与重构工作流图构建

- 新增子流程节点(SubFlow)及SubFlowConfig配置结构
- 拆分BuildGraph与BuildGraphFromFlowContent，支持独立构建图与编译
- 移除ModelCallbackReq中多余字段，移除Intent节点注册
- 优化节点执行逻辑：调整OSS上传时机及变量引用
- 更新本地开发配置及依赖版本

workflow/service/flow/lambda_node.go

@@ -9,6 +9,7 @@ import (
 	"ai-agent/workflow/model/dto"
 	fileDto "ai-agent/workflow/model/dto/file"
 	flowDto "ai-agent/workflow/model/dto/flow"
+	"ai-agent/workflow/model/entity"
 	"context"
 	"fmt"
 	"strconv"
@@ -18,6 +19,8 @@ import (
 
 	"gitea.redpowerfuture.com/red-future/common/db/gfdb"
 	"gitea.redpowerfuture.com/red-future/common/utils"
+	"github.com/cloudwego/eino-examples/compose/batch/batch"
+	"github.com/cloudwego/eino/compose"
 	"github.com/gogf/gf/v2/database/gdb"
 	"github.com/gogf/gf/v2/frame/g"
 	"github.com/gogf/gf/v2/util/gconv"
@@ -31,8 +34,78 @@ func FormLambda(ctx context.Context, input any) (any, error) {
 	return input, nil
 }
 
-func IntentLambda(ctx context.Context, input any) (any, error) {
-	return input, nil
+func SubFlowLambda(ctx context.Context, input any) (any, error) {
+	// 1. 类型断言（和其他节点保持一致的入参结构）
+	nodeExecInput, ok := input.(*flowDto.NodeExecutionInput)
+	if !ok {
+		return nil, fmt.Errorf("子流程节点入参类型错误，期望*flowDto.NodeExecutionInput，实际%T", input)
+	}
+	// 2. 解析子流程配置
+	subFlowConfig := nodeExecInput.Config.SubConfig
+	if subFlowConfig == nil {
+		return nil, fmt.Errorf("子流程节点缺少配置")
+	}
+	getRes, err := FlowUserService.Get(ctx, &flowDto.GetFlowUserReq{
+		Id: subFlowConfig.FlowId,
+	})
+	if err != nil {
+		return nil, err
+	}
+	// 3. 编译子流程Graph（复用现有 BuildGraphFromFlowContent 逻辑）
+	nodeList, subGraph := BuildGraph(ctx, getRes.FlowContent)
+	// 4. 构建子流程Workflow（绑定START/END，和示例对齐）
+	innerWorkflow := compose.NewWorkflow[*flowDto.FlowExecutionInput, *flowDto.FlowExecutionInput]()
+	// 挂载子图节点并绑定全局START
+	innerWorkflow.AddGraphNode("sub_flow_graph", subGraph).AddInput(compose.START)
+	// 绑定子图输出到全局END
+	innerWorkflow.End().AddInput("sub_flow_graph")
+	// 5. 构建BatchNode（批量执行子流程，复用示例逻辑）
+	batchNode := batch.NewBatchNode(&batch.NodeConfig[*flowDto.FlowExecutionInput, *flowDto.FlowExecutionInput]{
+		Name:           fmt.Sprintf("sub_flow_batch_%s", nodeExecInput.Config.Id),
+		InnerTask:      innerWorkflow,
+		MaxConcurrency: subFlowConfig.MaxConcurrency,
+	})
+
+	skillName, from, userFrom := BuildParam(nodeExecInput)
+	fmt.Printf("skillName: %s, from: %s, userFrom: %s\n", skillName, from, userFrom)
+
+	// 6. 提取批量输入（从全局入参中获取）
+	batchInputs := make([]*flowDto.FlowExecutionInput, 0)
+
+	nodeInputParams := ExtractFlowNodeFrom(getRes.FlowContent)
+	configMap := make(map[string]*entity.FlowNode)
+	for _, cfg := range nodeInputParams {
+		configMap[cfg.Id] = cfg
+	}
+	for _, i := range nodeList {
+		configMap[i.Id] = &i
+	}
+	// =========================================================================
+	// ✅【第4步】构建全局执行入参（现在 schemaMap 是有值的！）
+	// =========================================================================
+	execInput := &flowDto.FlowExecutionInput{
+		NodeGroupId: nodeExecInput.Global.NodeGroupId,
+		IsDialogue:  nodeExecInput.Global.IsDialogue,
+		ExecutionId: nodeExecInput.Global.ExecutionId,
+		ConfigMap:   configMap,
+		SessionId:   nodeExecInput.Global.SessionId,
+		Desc:        nodeExecInput.Global.Desc,
+		SkillName:   nodeExecInput.Global.SkillName,
+		FileUrl:     nodeExecInput.Global.FileUrl,
+	}
+	batchInputs = append(batchInputs, execInput)
+
+	// 7. 执行批量子流程
+	batchOutput, err := batchNode.Invoke(ctx, batchInputs)
+	if err != nil {
+		return nil, fmt.Errorf("执行子流程BatchNode失败: %v", err)
+	}
+	for idx, singleSubResult := range batchOutput {
+		fmt.Printf("【批量任务%d 最终消息条数】: %v\n", idx+1, singleSubResult)
+	}
+	// 8. 保存子流程执行结果到当前节点输出
+	//nodeExecInput.Config.OutputResult = append(nodeExecInput.Config.OutputResult, batchOutput)
+	return nodeExecInput, nil
 }
 
 // JudgeLambda 分支判断核心：读取IntentLambda的输出 → 返回目标节点ID做路由
@@ -139,62 +212,75 @@ func BatchModelLambda(ctx context.Context, input any) (any, error) {
 			}
 		}
 	}
-	// 结果按索引存放，保证顺序
+	// 结果按索引存放，切片不同下标并发写无竞争，不用锁
 	res := make([][]node.NodeFormField, len(reqMap))
 	var wg sync.WaitGroup
-	// 用一个通道标记是否完成
-	done := make(chan struct{})
-	// 错误只存一个
-	var execErr error
 
-	// 并发执行
+	subCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	// 缓冲1错误通道，仅接收第一个错误
+	errCh := make(chan error, 1)
+
+	// 并发执行任务
 	for idx, item := range reqMap {
 		wg.Add(1)
 		go func(idx int, userItem map[string]any) {
 			defer wg.Done()
 
+			// 上下文已取消则直接退出
+			select {
+			case <-subCtx.Done():
+				return
+			default:
+			}
+
 			singleUserFrom := []map[string]any{userItem}
-			output, err := TextNode(ctx, nodeInput, skillName, from, singleUserFrom)
+			output, err := TextNode(subCtx, nodeInput, skillName, from, singleUserFrom)
 			if err != nil {
-				// 并发安全赋值错误
-				if execErr == nil {
-					execErr = err
+				// 仅第一个错误写入通道
+				select {
+				case errCh <- err:
+					cancel() // 触发全局取消，其他协程快速退出
+				default:
 				}
 				return
 			}
-
-			// 直接按原索引写，顺序绝对正确
 			res[idx] = output
 		}(idx, item)
 	}
 
-	// 后台等待所有协程完成，然后关闭 done 通道
+	// 任务全部结束后关闭错误通道
 	go func() {
 		wg.Wait()
-		close(done)
+		close(errCh)
 	}()
 
-	// 等待全部完成
-	<-done
-
-	// 如果有错误，直接返回
-	if execErr != nil {
-		return nil, execErr
+	// ========== 修正后的等待逻辑 ==========
+	var execErr error
+	select {
+	// 优先捕获业务错误
+	case execErr = <-errCh:
+		if execErr != nil {
+			// 收到真实业务错误，等待剩余协程收尾后返回
+			wg.Wait()
+			return nil, execErr
+		}
+		// execErr == nil 代表通道关闭、无任何错误，走到下方返回完整结果
+	case <-subCtx.Done():
+		// 上下文被取消，阻塞读完errCh，确认是否存在业务错误
+		execErr = <-errCh
 	}
 
-	// 全局自增 i
+	// 拼接输出结果
 	var globalIndex int
 	var outputRes []node.NodeFormField
 	for _, items := range res {
 		for _, item := range items {
-			// 1. 拿到原来的 Field：例如 "text_content:2:0"
 			oldField := item.Field
-			// 2. 找到最后一个 : 的位置
 			if idx := strings.LastIndex(oldField, ":"); idx != -1 {
-				// 3. 截断前面部分，拼接上新的 globalIndex
 				item.Field = oldField[:idx+1] + fmt.Sprint(globalIndex)
 			}
-			// Label 同理
 			oldLabel := item.Label
 			if idx := strings.LastIndex(oldLabel, ":"); idx != -1 {
 				item.Label = oldLabel[:idx+1] + fmt.Sprint(globalIndex)
@@ -290,6 +376,7 @@ func VideoModelLambda(ctx context.Context, input any) (any, error) {
 	if err != nil {
 		return nil, err
 	}
+	newS := strings.ReplaceAll(urlPrefix, g.Cfg().MustGet(ctx, "filePrefix").String(), g.Cfg().MustGet(ctx, "minioPrefix").String())
 
 	outputRes := make([]node.NodeFormField, 0)
 	if nodeInput.Config.IsSaveFile {
@@ -302,7 +389,7 @@ func VideoModelLambda(ctx context.Context, input any) (any, error) {
 	}
 	outputRes = append(outputRes, node.NodeFormField{
 		Field: fmt.Sprintf("concat_video_url:content:%d", 0),
-		Value: urlPrefix + msg.FileURL,
+		Value: newS + msg.FileURL,
 		Label: fmt.Sprintf("视频内容:content:%d", 0),
 		Type:  "string",
 	})