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

- 新增子流程节点(SubFlow)及SubFlowConfig配置结构 - 拆分BuildGraph与BuildGraphFromFlowContent，支持独立构建图与编译 - 移除ModelCallbackReq中多余字段，移除Intent节点注册 - 优化节点执行逻辑：调整OSS上传时机及变量引用 - 更新本地开发配置及依赖版本
Merge branch '2026-06-10-video' of http://116.204.74.41:3000/red-future/ai-agent into sub-workflow
2026-06-24 09:30:55 +08:00 · 2026-06-24 09:25:40 +08:00 · 2026-06-24 09:24:53 +08:00 · 2026-06-22 09:26:53 +08:00 · 2026-06-16 13:26:27 +08:00 · 2026-06-16 13:08:41 +08:00
22 changed files with 590 additions and 1156 deletions
@@ -1 +0,0 @@
-.git
@@ -1,37 +0,0 @@
-# ============================================================
-# ai-agent dev 分支自动发布
-# 触发：push dev 分支
-# 流程：构建镜像 -> push 本地仓库(127.0.0.1:5000) -> 部署到 dev 宿主机本地 k3s
-# 前置：runner 挂载 dev k3s kubeconfig 到 /root/.kube/k3s-dev.yaml
-# ============================================================
-name: Deploy ai-agent to dev k3s
-
-on:
-  push:
-    branches: [ 2026-06-16 ]
-
-jobs:
-  build-and-deploy:
-    runs-on: ubuntu-latest
-    steps:
-      # 不依赖 actions/checkout（从 github.com 拉 action 源码，国内网络不稳定），
-      # 直接 clone 本地 Gitea，用 Actions 注入的 GITHUB_TOKEN 认证
-      - name: 拉取代码
-        run: |
-          git clone --branch dev http://x-access-token:${GITHUB_TOKEN}@127.0.0.1:3000/red-future/ai-agent.git .
-          git checkout ${GITHUB_SHA}
-
-      - name: 构建并推送镜像
-        run: |
-          TAG=dev-${GITHUB_SHA::8}
-          docker build -t 127.0.0.1:5000/ai-agent:${TAG} .
-          docker push 127.0.0.1:5000/ai-agent:${TAG}
-
-      - name: 部署到 dev k3s
-        env:
-          KUBECONFIG: /root/.kube/k3s-dev.yaml
-        run: |
-          kubectl create namespace dev --dry-run=client -o yaml | kubectl apply -f -
-          kubectl apply -f deploy/k8s/ai-agent.yaml
-          kubectl set image deployment/ai-agent ai-agent=127.0.0.1:5000/ai-agent:dev-${GITHUB_SHA::8} -n dev
-          kubectl rollout status deployment/ai-agent -n dev --timeout=5m
@@ -1,8 +1,7 @@
 # 阶段1: 构建
 FROM golang:alpine AS builder

-RUN sed -i 's/dl-cdn.alpinelinux.org/mirrors.aliyun.com/g' /etc/apk/repositories && \
-      apk add --no-cache git ca-certificates tzdata
+RUN apk add --no-cache git ca-certificates tzdata

 ENV TZ=Asia/Shanghai
 RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
@@ -64,8 +64,5 @@ consul:
 jaeger:
  addr: 192.168.0.83:4318

-# 文件上传服务地址，cdn访问地址
-filePrefix: "http://cdn.redpowerfuture.com"
-
-# 文件上传服务地址，minio内网访问地址
-minioPrefix: "http://192.168.0.83:9000"
+# 文件上传服务地址，与oss模块minio中的endpoint一致
+filePrefix: "http://192.168.0.83:9000"
@@ -1,46 +0,0 @@
-apiVersion: apps/v1
-kind: Deployment
-metadata:
-  name: ai-agent
-  namespace: dev
-spec:
-  replicas: 1
-  selector:
-    matchLabels:
-      app: ai-agent
-  template:
-    metadata:
-      labels:
-        app: ai-agent
-    spec:
-      containers:
-        - name: ai-agent
-          image: 127.0.0.1:5000/ai-agent:dev-latest
-          imagePullPolicy: IfNotPresent
-          ports:
-            - containerPort: 3005
-          env:
-            - name: TZ
-              value: Asia/Shanghai
-          readinessProbe:
-            tcpSocket:
-              port: 3005
-            initialDelaySeconds: 10
-            periodSeconds: 10
-          livenessProbe:
-            tcpSocket:
-              port: 3005
-            initialDelaySeconds: 30
-            periodSeconds: 15
---
-apiVersion: v1
-kind: Service
-metadata:
-  name: ai-agent
-  namespace: dev
-spec:
-  selector:
-    app: ai-agent
-  ports:
-    - port: 3005
-      targetPort: 3005
@@ -3,13 +3,15 @@ module ai-agent
 go 1.26.0

 require (
-	gitea.redpowerfuture.com/red-future/common v0.0.29
-	github.com/cloudwego/eino v0.9.12
+	gitea.redpowerfuture.com/red-future/common v0.0.24
+	github.com/cloudwego/eino v0.9.5
+	github.com/cloudwego/eino-examples v0.0.0-20260611092511-bd64846fbc1d
 	github.com/cloudwego/eino-ext/components/model/qwen v0.1.9
 	github.com/gogf/gf/contrib/drivers/pgsql/v2 v2.10.2
 	github.com/gogf/gf/contrib/nosql/redis/v2 v2.10.2
 	github.com/gogf/gf/v2 v2.10.2
 	github.com/google/uuid v1.6.0
+	github.com/stretchr/testify v1.11.1
 	github.com/tidwall/gjson v1.19.0
 	github.com/tidwall/sjson v1.2.5
 	go.opentelemetry.io/otel/trace v1.44.0
@@ -21,7 +23,7 @@ require (
 	github.com/bahlo/generic-list-go v0.2.0 // indirect
 	github.com/buger/jsonparser v1.1.1 // indirect
 	github.com/bwmarrin/snowflake v0.3.0 // indirect
-	github.com/bytedance/gopkg v0.1.3 // indirect
+	github.com/bytedance/gopkg v0.1.4 // indirect
 	github.com/bytedance/sonic v1.15.0 // indirect
 	github.com/bytedance/sonic/loader v0.5.0 // indirect
 	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
@@ -29,6 +31,7 @@ require (
 	github.com/clbanning/mxj/v2 v2.7.0 // indirect
 	github.com/cloudwego/base64x v0.1.6 // indirect
 	github.com/cloudwego/eino-ext/libs/acl/openai v0.1.17 // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/dgraph-io/badger/v4 v4.2.0 // indirect
 	github.com/dgraph-io/ristretto v0.1.1 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
@@ -83,6 +86,7 @@ require (
 	github.com/olekukonko/tablewriter v1.1.0 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.4 // indirect
 	github.com/pkg/errors v0.9.2-0.20201214064552-5dd12d0cfe7f // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/r3labs/diff/v2 v2.15.1 // indirect
 	github.com/redis/go-redis/v9 v9.17.2 // indirect
 	github.com/rivo/uniseg v0.4.7 // indirect
@@ -1,6 +1,8 @@
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
-gitea.redpowerfuture.com/red-future/common v0.0.29 h1:5McaN5pSewvrLUHQzWMX6EaUvD+B5I5bMYoU+clHJk4=
-gitea.redpowerfuture.com/red-future/common v0.0.29/go.mod h1:50U1Xi+Ie56z09S5LQbZvaken0Mxv3OeS9LgR7U/ZRY=
+gitea.redpowerfuture.com/red-future/common v0.0.23 h1:xieoA00iKOCDm5SO9iXn+cSyMKBAlZwI0fuEVPWrHLg=
+gitea.redpowerfuture.com/red-future/common v0.0.23/go.mod h1:50U1Xi+Ie56z09S5LQbZvaken0Mxv3OeS9LgR7U/ZRY=
+gitea.redpowerfuture.com/red-future/common v0.0.24 h1:sXxhnmDmCgn+KwH/3gDnhAtAQ7FCmf/5AsMfvxRmri0=
+gitea.redpowerfuture.com/red-future/common v0.0.24/go.mod h1:50U1Xi+Ie56z09S5LQbZvaken0Mxv3OeS9LgR7U/ZRY=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/toml v1.5.0 h1:W5quZX/G/csjUnuI8SUYlsHs9M38FC7znL0lIO+DvMg=
 github.com/BurntSushi/toml v1.5.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho=
@@ -36,6 +38,8 @@ github.com/bwmarrin/snowflake v0.3.0 h1:xm67bEhkKh6ij1790JB83OujPR5CzNe8QuQqAgIS
 github.com/bwmarrin/snowflake v0.3.0/go.mod h1:NdZxfVWX+oR6y2K0o6qAYv6gIOP9rjG0/E9WsDpxqwE=
 github.com/bytedance/gopkg v0.1.3 h1:TPBSwH8RsouGCBcMBktLt1AymVo2TVsBVCY4b6TnZ/M=
 github.com/bytedance/gopkg v0.1.3/go.mod h1:576VvJ+eJgyCzdjS+c4+77QF3p7ubbtiKARP3TxducM=
+github.com/bytedance/gopkg v0.1.4 h1:oZnQwnX82KAIWb7033bEwtxvTqXcYMxDBaQxo5JJHWM=
+github.com/bytedance/gopkg v0.1.4/go.mod h1:v1zWfPm21Fb+OsyXN2VAHdL6TBb2L88anLQgdyje6R4=
 github.com/bytedance/mockey v1.3.0 h1:ONLRdvhqmCfr9rTasUB8ZKCfvbdD2tohOg4u+4Q/ed0=
 github.com/bytedance/mockey v1.3.0/go.mod h1:1BPHF9sol5R1ud/+0VEHGQq/+i2lN+GTsr3O2Q9IENY=
 github.com/bytedance/sonic v1.15.0 h1:/PXeWFaR5ElNcVE84U0dOHjiMHQOwNIx3K4ymzh/uSE=
@@ -56,12 +60,16 @@ github.com/clbanning/mxj/v2 v2.7.0/go.mod h1:hNiWqW14h+kc+MdF9C6/YoRfjEJoR3ou6tn
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/cloudwego/base64x v0.1.6 h1:t11wG9AECkCDk5fMSoxmufanudBtJ+/HemLstXDLI2M=
 github.com/cloudwego/base64x v0.1.6/go.mod h1:OFcloc187FXDaYHvrNIjxSe8ncn0OOM8gEHfghB2IPU=
-github.com/cloudwego/eino v0.9.12 h1:mHAMo5k7GdvnVD8Lc2sLyfpkxEm0S/y3PkEMhsSYt78=
-github.com/cloudwego/eino v0.9.12/go.mod h1:OBD1mrkfkt/pJa4rkg1P0VnaMeOVl7l8IAdEqY//3IQ=
+github.com/cloudwego/eino v0.9.5 h1:0Nftjx9gPek/2S/hzm38LVxSjk5/6mqRr3I9VKrKvm4=
+github.com/cloudwego/eino v0.9.5/go.mod h1:OBD1mrkfkt/pJa4rkg1P0VnaMeOVl7l8IAdEqY//3IQ=
+github.com/cloudwego/eino-examples v0.0.0-20260611092511-bd64846fbc1d h1:NrAxhU58S5SgK5YbrYnaOQrQFwAz3x4/0qg46JM8Eo4=
+github.com/cloudwego/eino-examples v0.0.0-20260611092511-bd64846fbc1d/go.mod h1:VVmcWGhnLIxLkrQAaCoOtQoEbcvOCrMQvbXgbo9O34Q=
 github.com/cloudwego/eino-ext/components/model/qwen v0.1.9 h1:xCz/mp43JeWqupjPR3zLRArmwC6P29/6lTwbwh1yzYM=
 github.com/cloudwego/eino-ext/components/model/qwen v0.1.9/go.mod h1:slTGTuhzkzhNavf+1UtUg1FvUSA31iNAF+rq1mT4SnI=
 github.com/cloudwego/eino-ext/libs/acl/openai v0.1.17 h1:EeVcR1TslRA2IdNW1h/2LaGbPlffwGhQm99jM3zWZiI=
 github.com/cloudwego/eino-ext/libs/acl/openai v0.1.17/go.mod h1:Zkcx6DPTR2NfWmtSXbhItswGw6hqUezNPhNcke0pOG8=
+github.com/cloudwego/hertz v0.10.5 h1:N4oBqAJShSjYQm2Jfr0ryTzzJ9fnY9qSvIvUBMkoFWg=
+github.com/cloudwego/hertz v0.10.5/go.mod h1:Im9u6rUa1v2mL2HiDKKJoof/CPQ3mPBBpT92v67Cetg=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -415,7 +415,7 @@ CREATE TABLE IF NOT EXISTS black_deacon_flow_execution (
    node_input_params JSONB DEFAULT '[]'::JSONB,
    output_params JSONB DEFAULT '[]'::JSONB,
    error_message TEXT DEFAULT '', -- 错误信息
-    trace_id VARCHAR(64) DEFAULT '' -- 跟踪ID
+    trace_id VARCHAR(64) DEFAULT '', -- 跟踪ID
    session_id VARCHAR(64) DEFAULT '' -- 会话ID
    );

@@ -15,17 +15,17 @@ const (
 	NodeNameVideoModel          = "生成视频"
 	NodeNameAudioModel          = "生成音频"
 	NodeNameBatchModel          = "批量处理一起返回"
-	NodeNameSenseOptimizeModel  = "语义优化"
-	NodeNameStoryOptimizeModel  = "分镜优化"
-	NodeNameScriptOptimizeModel = "剧本优化"
 	NodeNameDataConversionModel = "参数转换"
 	NodeNameModel               = "模型"
 	NodeNameMerge               = "结果合并"
 	NodeNameDataMerge           = "结果汇集"
 	NodeNameJudge               = "条件判断"
+	NodeNameLoop                = "循环"
 	NodeNameForm                = "表单"
+	NodeSubFlow                 = "子流程"
 	NodeNameHttp                = "HTTP(S)接口"
 	NodeNameCustomNode          = "自定义节点"
+	NodeNameSystemSum           = "系统-结果汇总"
 )

 // 表单字段 Label
@@ -48,26 +48,25 @@ type NodeType string

 const (
 	// 组件
-	NodeTypeTextModel  NodeType = "text_model"
-	NodeTypeImageModel NodeType = "image_model"
-	NodeTypeVideoModel NodeType = "video_model"
-	NodeTypeAudioModel NodeType = "audio_model"
-	NodeTypeBatchModel NodeType = "batch_model"
-
-	NodeTypeSenseOptimizeModel  NodeType = "sense_optimize_model"
-	NodeTypeStoryOptimizeModel  NodeType = "story_optimize_model"
-	NodeTypeScriptOptimizeModel NodeType = "script_optimize_model"
-	// 基础
+	NodeTypeTextModel           NodeType = "text_model"
+	NodeTypeImageModel          NodeType = "image_model"
+	NodeTypeVideoModel          NodeType = "video_model"
+	NodeTypeAudioModel          NodeType = "audio_model"
+	NodeTypeBatchModel          NodeType = "batch_model"
 	NodeTypeDataConversionModel NodeType = "data_conversion_model"
-	NodeTypeModel               NodeType = "model"
-	NodeTypeMerge               NodeType = "merge"
-	NodeTypeDataMerge           NodeType = "data_merge"
-	NodeTypeJudge               NodeType = "judge"
-	NodeTypeForm                NodeType = "form"
-	NodeTypeIntent              NodeType = "intent"
-	NodeTypeHttp                NodeType = "http"
+	// 基础
+	NodeTypeModel     NodeType = "model"
+	NodeTypeMerge     NodeType = "merge"
+	NodeTypeDataMerge NodeType = "data_merge"
+	NodeTypeJudge     NodeType = "judge"
+	NodeTypeForm      NodeType = "form"
+	NodeTypeIntent    NodeType = "intent"
+	NodeTypeSubFlow   NodeType = "sub_flow"
+	NodeTypeHttp      NodeType = "http"
 	// 自定义
 	NodeTypeCustomNode NodeType = "custom_node"
+	// 系统
+	NodeTypeSystemSum NodeType = "system_sum"
 )

 const (
@@ -106,7 +105,6 @@ type NodeItem struct {
 	NodeCode     NodeType        `json:"nodeCode"`
 	ModelType    int             `json:"modelType"`
 	NodeName     string          `json:"nodeName"` // 从常量来
-	PatchLayout  bool            `json:"patchLayout"`
 	SkillOption  bool            `json:"skillOption"`
 	PromptOption bool            `json:"promptOption"`
 	IsSaveFile   bool            `json:"isSaveFile"`
@@ -38,17 +38,3 @@ func (c *flowExecution) Get(ctx context.Context, req *flowDto.GetFlowExecutionRe
 func (c *flowExecution) List(ctx context.Context, req *flowDto.ListFlowExecutionReq) (res *flowDto.ListFlowExecutionTreeRes, err error) {
 	return flowService.FlowExecutionService.List(ctx, req)
 }
-
-func (c *flowExecution) GetSessionList(ctx context.Context, req *flowDto.GetSessionListReq) (res *flowDto.ListFlowExecutionRes, err error) {
-	return flowService.FlowExecutionService.GetSessionList(ctx, req)
-}
-
-func (c *flowExecution) DeleteResult(ctx context.Context, req *flowDto.DeleteResultReq) (res *beans.ResponseEmpty, err error) {
-	err = flowService.FlowExecutionService.DeleteResult(ctx, req)
-	return
-}
-
-func (c *flowExecution) DeleteSession(ctx context.Context, req *flowDto.DeleteSessionReq) (res *beans.ResponseEmpty, err error) {
-	err = flowService.FlowExecutionService.DeleteSession(ctx, req)
-	return
-}
@@ -7,7 +7,6 @@ import (
 	"context"

 	"gitea.redpowerfuture.com/red-future/common/db/gfdb"
-	"github.com/gogf/gf/v2/frame/g"
 	"github.com/gogf/gf/v2/util/gconv"
 )

@@ -50,17 +49,8 @@ func (d *flowExecutionDao) Get(ctx context.Context, req *flowDto.GetFlowExecutio
 }

 func (d *flowExecutionDao) List(ctx context.Context, req *flowDto.ListFlowExecutionReq, fields ...string) (res []*entity.FlowExecution, total int, err error) {
-	model := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameFlowExecution).Fields(fields)
+	model := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameFlowExecution).Fields(fields).OmitEmpty()
 	model.Where(entity.FlowExecutionCol.Creator, req.Creator)
-	if !g.IsEmpty(req.ResultDel) {
-		model.Where(entity.FlowExecutionCol.ResultDel, false)
-	}
-	if !g.IsEmpty(req.SessionDel) {
-		model.Where(entity.FlowExecutionCol.SessionDel, false)
-	}
-	if req.IsResult {
-		model.WhereNot(entity.FlowExecutionCol.OutputParams, "[]")
-	}
 	model.OrderDesc(entity.FlowExecutionCol.CreatedAt)
 	if req.Page != nil {
 		model.Page(int(req.Page.PageNum), int(req.Page.PageSize))
@@ -86,7 +86,6 @@ 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))
@@ -30,7 +30,6 @@ type FlowExecutionInput struct {
 	ExecutionId   int64                       `json:"executionId"`
 	ConfigMap     map[string]*entity.FlowNode `json:"configMap"`
 	SessionId     string                      `json:"sessionId" dc:"会话ID"`
-	Templates     []map[string]any            `json:"templates"`
 	Desc          string                      `json:"desc"`
 	SkillName     string                      `json:"skillName"`
 	FileUrl       []string                    `json:"fileUrl"`
@@ -50,7 +49,6 @@ type GetModelInfoReq struct {

 type GetModelInfoRes struct {
 	Model struct {
-		FirstFrame         string         `json:"firstFrame"`
 		LastFrame          string         `json:"lastFrame"`
 		ResponseTokenField string         `json:"responseTokenField"`
 		ResponseMapping    map[string]any `json:"responseMapping"`
@@ -76,7 +74,6 @@ type ComposeMessagesReq struct {
 	CallbackUrl string           `json:"callbackUrl"`
 	Form        []map[string]any `json:"form"`
 	UserForm    []map[string]any `json:"userForm"`
-	UserPrompt  string           `json:"userPrompt" dc:"用户提示词"`
 	Consult     []Consult        `json:"consult"`
 	SessionId   string           `json:"sessionId" dc:"会话ID"`
 	NodeId      string           `json:"nodeId"`
@@ -125,19 +122,17 @@ 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"`
-	BillingData []map[string]any `json:"billing_data"`
+	EpicycleId int64  `json:"epicycleId"`
+	ErrorMsg   string `json:"errorMsg,omitempty"`
 }

 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"`
-	BillingData []map[string]any `json:"billing_data"`
+	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"`
 }

 type VideoCallbackReq struct {
@@ -185,7 +180,6 @@ type ExecuteReq struct {
 	FlowContent     *entity.FlowInfo   `json:"flowContent" description:"流程内容"`
 	NodeInputParams []*entity.FlowNode `json:"nodeInputParams" description:"节点输入参数"`
 	SessionId       string             `json:"sessionId" dc:"会话ID"`
-	Templates       []map[string]any   `json:"templates"`
 	Desc            string             `json:"desc"`
 	SkillName       string             `json:"skillName"`
 	FileUrl         []string           `json:"fileUrl"`
@@ -215,7 +209,6 @@ type CreateFlowExecutionReq struct {
 	ErrorMessage    string                        `json:"errorMessage" description:"错误信息"`
 	TraceId         string                        `json:"traceId" description:"跟踪ID"`
 	SessionId       string                        `json:"sessionId" dc:"会话ID"`
-	Extension       map[string]interface{}        `json:"extension"`
 }

 type CreateFlowExecutionRes struct {
@@ -223,33 +216,13 @@ type CreateFlowExecutionRes struct {
 }

 type UpdateFlowExecutionReq struct {
-	Id              int64                    `json:"id" v:"required#ID不能为空"`
-	NodeGroupId     string                   `json:"nodeGroupId"`
-	DurationMs      int64                    `json:"durationMs" description:"执行时长（毫秒）"`
-	Status          flow.FlowExecutionStatus `json:"status" description:"状态:1-运行中,2-成功,3-失败"`
-	FlowContent     *entity.FlowInfo         `json:"flowContent" description:"流程内容"`
-	NodeInputParams []*entity.FlowNode       `json:"nodeInputParams" description:"节点输入参数"`
-	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:"总费用"`
-	SessionDel      bool                     `json:"sessionDel" description:"会话是否删除"`
-	ResultDel       bool                     `json:"resultDel" description:"结果是否删除"`
-	Extension       map[string]interface{}   `json:"extension"`
-}
-
-type DeleteResultReq struct {
-	g.Meta `path:"/deleteResult" method:"delete" tags:"任务管理" summary:"删除结果" dc:"删除结果"`
-
-	Id      int64  `json:"id" v:"required#ID不能为空"`
-	Content string `json:"content"`
-}
-
-type DeleteSessionReq struct {
-	g.Meta `path:"/deleteSession" method:"delete" tags:"任务管理" summary:"删除会话" dc:"删除会话"`
-
-	Id int64 `json:"id" v:"required#ID不能为空"`
+	Id           int64                    `json:"id" v:"required#ID不能为空"`
+	NodeGroupId  string                   `json:"nodeGroupId"`
+	DurationMs   int64                    `json:"durationMs" description:"执行时长（毫秒）"`
+	Status       flow.FlowExecutionStatus `json:"status" description:"状态:1-运行中,2-成功,3-失败"`
+	OutputParams []map[string]interface{} `json:"outputParams" description:"输出参数"`
+	ErrorMessage string                   `json:"errorMessage" description:"错误信息"`
+	TraceId      string                   `json:"traceId" description:"跟踪ID"`
 }

 type GetFlowExecutionReq struct {
@@ -259,20 +232,11 @@ type GetFlowExecutionReq struct {
 	SessionId string `json:"sessionId" dc:"会话ID"`
 }

-type GetSessionListReq struct {
-	g.Meta `path:"/sessionList" method:"get" tags:"任务管理" summary:"会话列表" dc:"会话列表"`
-
-	*beans.Page `json:"page"`
-}
-
 type ListFlowExecutionReq struct {
 	g.Meta `path:"/list" method:"get" tags:"任务管理" summary:"任务列表" dc:"任务列表"`

-	*beans.Page `json:"page"`
-	Creator     string `json:"creator"`
-	SessionDel  *bool  `json:"sessionDel"`
-	ResultDel   *bool  `json:"resultDel"`
-	IsResult    bool   `json:"isResult"`
+	Page    *beans.Page `json:"page"`
+	Creator string      `json:"creator"`
 }

 type ListFlowExecutionRes struct {
@@ -291,7 +255,6 @@ type VOFlowExecution struct {
 	NodeInputParams  []*entity.FlowNode            `json:"nodeInputParams" description:"节点输入参数"`
 	OutputParams     []map[string]interface{}      `json:"outputParams" description:"输出参数"`
 	ErrorMessage     string                        `json:"errorMessage" description:"错误信息"`
-	Extension        map[string]interface{}        `json:"extension"`
 	TraceId          string                        `json:"traceId" description:"跟踪ID"`
 	SessionId        string                        `json:"sessionId" dc:"会话ID"`
 	CreatedAt        *gtime.Time                   `json:"createdAt" dc:"创建时间"`
@@ -300,17 +263,22 @@ type VOFlowExecution struct {
 }

 // ========== 核心：构建树状结构 ==========
+// 定义树结构
 type OutputItem struct {
-	Id        int64  `json:"id,string" description:"ID"`
 	Timestamp string `json:"timestamp" description:"时间戳key"`
 	Content   string `json:"content" description:"内容值"`
 	Type      string `json:"type" description:"类型"`
 	Label     string `json:"label" description:"后缀+数字标号"`
 }
-
+type FlowNode struct {
+	FlowName  string       `json:"flowName" description:"流程名称"`
+	Id        int64        `json:"Id,string" description:"任务ID"`
+	SessionId string       `json:"sessionId" description:"会话ID"`
+	Items     []OutputItem `json:"items" description:"输出项列表"`
+}
 type DateNode struct {
-	CreateDate string       `json:"createDate" description:"创建日期"`
-	Items      []OutputItem `json:"items" description:"直接是结果项列表"`
+	CreateDate string     `json:"createDate" description:"创建日期"`
+	Flows      []FlowNode `json:"flows" description:"流程列表"`
 }

 type ListFlowExecutionTreeRes struct {
@@ -38,7 +38,6 @@ 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"`
@@ -61,7 +60,6 @@ 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 节点执行记录响应
@@ -22,10 +22,6 @@ 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:"总费用"`
-	SessionDel      bool                          `orm:"session_del" json:"sessionDel" description:"会话是否删除"`
-	ResultDel       bool                          `orm:"result_del" json:"resultDel" description:"结果是否删除"`
-	Extension       map[string]interface{}        `orm:"extension" json:"extension" description:"扩展字段"`
 }

 type flowExecutionCol struct {
@@ -43,10 +39,6 @@ type flowExecutionCol struct {
 	TraceId         string
 	SessionId       string
 	TotalTokens     string
-	TotalFee        string
-	SessionDel      string
-	ResultDel       string
-	Extension       string
 }

 var FlowExecutionCol = flowExecutionCol{
@@ -64,8 +56,4 @@ var FlowExecutionCol = flowExecutionCol{
 	TraceId:         "trace_id",
 	SessionId:       "session_id",
 	TotalTokens:     "total_tokens",
-	TotalFee:        "total_fee",
-	SessionDel:      "session_del",
-	ResultDel:       "result_del",
-	Extension:       "extension",
 }
@@ -15,26 +15,39 @@ type FlowInfo struct {
 }

 type FlowNode struct {
-	Id                string                 `json:"id"`
-	NodeCode          node.NodeType          `json:"nodeCode"`
-	Name              string                 `json:"name"`
-	Config            map[string]interface{} `json:"config"`
-	PatchLayout       bool                   `json:"patchLayout"`
-	SkillName         string                 `json:"skillName"`
-	PromptContent     string                 `json:"promptContent"`
-	IsSaveFile        bool                   `json:"isSaveFile"`
-	InputSource       []FlowNodeInputSource  `json:"inputSource"` // 前端指定：来源节点ID
-	FormConfig        []node.NodeFormField   `json:"formConfig"`
-	ModelConfig       node.ModelItem         `json:"modelConfig"`
-	ModelOutputFields []string               `json:"modelOutputFields"`
-	OutputConfig      []node.NodeFormField   `json:"outputConfig"`
-	OutputResult      []node.NodeFormField   `json:"outputResult" ds:"节点输出结果"`
+	Id            string                 `json:"id"`
+	NodeCode      node.NodeType          `json:"nodeCode"`
+	Name          string                 `json:"name"`
+	Config        map[string]interface{} `json:"config"`
+	SkillName     string                 `json:"skillName"`
+	PromptContent string                 `json:"promptContent"`
+	IsSaveFile    bool                   `json:"isSaveFile"`
+	InputSource   []FlowNodeInputSource  `json:"inputSource"` // 前端指定：来源节点ID
+	SubConfig     *SubFlowConfig         `json:"subConfig"`
+	FormConfig    []node.NodeFormField   `json:"formConfig"`
+	ModelConfig   node.ModelItem         `json:"modelConfig"`
+	OutputConfig  []node.NodeFormField   `json:"outputConfig"`
+	OutputResult  []node.NodeFormField   `json:"outputResult" ds:"节点输出结果"`
 }

 type FlowNodeInputSource struct {
-	NodeId      string   `json:"nodeId"`
-	QuoteOutput bool     `json:"quoteOutput"`
-	Field       []string `json:"field"`
+	NodeId      string      `json:"nodeId"`
+	QuoteOutput bool        `json:"quoteOutput"`
+	Field       []string    `json:"field"`
+	FieldMap    []FlowField `json:"fieldMap"`
+}
+
+type FlowField struct {
+	Key   string `json:"key"`
+	Value string `json:"value"`
+	Desc  string `json:"desc"`
+}
+
+// SubFlowConfig 子流程节点配置
+type SubFlowConfig struct {
+	FlowId         int64                 `json:"flowId"`
+	MaxConcurrency int                   `json:"maxConcurrency"` // 子流程并发数
+	InputSource    []FlowNodeInputSource `json:"inputSource"`    // 前端指定：来源节点ID
 }

 type FlowEdge struct {
@@ -22,7 +22,6 @@ 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:"错误信息"`
@@ -41,7 +40,6 @@ type nodeExecutionCol struct {
 	PromptTokens     string
 	CompletionTokens string
 	TotalTokens      string
-	TokenInfo        string
 	Status           string
 	DurationMs       string
 	ErrorMessage     string
@@ -60,7 +58,6 @@ var NodeExecutionCol = nodeExecutionCol{
 	PromptTokens:     "prompt_tokens",
 	CompletionTokens: "completion_tokens",
 	TotalTokens:      "total_tokens",
-	TokenInfo:        "token_info",
 	Status:           "status",
 	DurationMs:       "duration_ms",
 	ErrorMessage:     "error_message",
@@ -16,12 +16,14 @@ import (
 	"fmt"
 	"sort"
 	"strconv"
+	"strings"
 	"sync"
 	"time"

 	"gitea.redpowerfuture.com/red-future/common/utils"
 	"github.com/cloudwego/eino/compose"
 	"github.com/gogf/gf/v2/frame/g"
+	"github.com/gogf/gf/v2/os/gtime"
 	"github.com/gogf/gf/v2/util/gconv"
 	"github.com/google/uuid"
 	"go.opentelemetry.io/otel/trace"
@@ -45,94 +47,54 @@ func (s *flowExecutionService) Get(ctx context.Context, req *flowDto.GetFlowExec
 	return res, err
 }

-func (s *flowExecutionService) DeleteResult(ctx context.Context, req *flowDto.DeleteResultReq) (err error) {
-	r, err := flowDao.FlowExecutionDao.Get(ctx, &flowDto.GetFlowExecutionReq{Id: req.Id})
-	if err != nil {
-		return err
-	}
-	// 遍历并剔除值等于 req.Url 的数据
-	newOutputParams := make([]map[string]any, 0)
-	for _, paramMap := range r.OutputParams {
-		// 单个 paramMap 过滤
-		cleanMap := make(map[string]any)
-		for k, v := range paramMap {
-			// 转为字符串对比
-			if gconv.String(v) != req.Content {
-				cleanMap[k] = v
-			}
-		}
-		// 只保留非空 map，避免出现空层级
-		if len(cleanMap) > 0 {
-			newOutputParams = append(newOutputParams, cleanMap)
-		}
-	}
-	// 赋值回原数据
-	r.OutputParams = newOutputParams
-	// 执行更新：更新 OutputParams + 标记删除
-	flowUpdateReq := new(flowDto.UpdateFlowExecutionReq)
-	flowUpdateReq.Id = req.Id
-	flowUpdateReq.OutputParams = r.OutputParams
-	_, err = flowDao.FlowExecutionDao.Update(ctx, flowUpdateReq)
-	return
-}
-
-func (s *flowExecutionService) DeleteSession(ctx context.Context, req *flowDto.DeleteSessionReq) (err error) {
-	flowUpdateReq := new(flowDto.UpdateFlowExecutionReq)
-	flowUpdateReq.Id = req.Id
-	flowUpdateReq.SessionDel = true
-	_, err = flowDao.FlowExecutionDao.Update(ctx, flowUpdateReq)
-	return
-}
-
-func (s *flowExecutionService) GetSessionList(ctx context.Context, req *flowDto.GetSessionListReq) (res *flowDto.ListFlowExecutionRes, err error) {
-	user, err := utils.GetUserInfo(ctx)
-	if err != nil {
-		return
-	}
-	flowReq := new(flowDto.ListFlowExecutionReq)
-	flowReq.Page = req.Page
-	flowReq.Creator = user.UserName
-	flowReq.SessionDel = gconv.PtrBool(true)
-	list, total, err := flowDao.FlowExecutionDao.List(ctx, flowReq)
-	if err != nil {
-		return nil, err
-	}
-	res = &flowDto.ListFlowExecutionRes{
-		Total: total,
-	}
-	err = gconv.Struct(list, &res.List)
-	return res, err
-}
-
 func (s *flowExecutionService) List(ctx context.Context, req *flowDto.ListFlowExecutionReq) (res *flowDto.ListFlowExecutionTreeRes, err error) {
 	user, err := utils.GetUserInfo(ctx)
 	if err != nil {
 		return
 	}
 	req.Creator = user.UserName
-	req.ResultDel = gconv.PtrBool(true)
-	req.IsResult = true
 	list, _, err := flowDao.FlowExecutionDao.List(ctx, req)
 	if err != nil {
 		return nil, err
 	}

-	// 过滤出有有效输出的执行记录
-	var validList []*entity.FlowExecution
+	// ===================== 核心修复：只统计【有数据】的执行记录，空的直接跳过 =====================
+
+	executionNumber := make(map[int64]int) // executionId -> 倒序编号（最新=1）
+
+	// 第一次遍历：只处理【有输出参数】的记录，统计并分配编号
+	var validList []*entity.FlowExecution // 只存有效（非空）记录
 	for _, execution := range list {
-		if !g.IsEmpty(execution.OutputParams) {
-			validList = append(validList, execution)
+		if g.IsEmpty(execution.OutputParams) {
+			continue // 空数据直接过滤，不参与编号、不展示
 		}
+		validList = append(validList, execution)
 	}

-	// 1. 按日期归集，严格使用 Y-m-d 格式
-	dateMap := make(map[string][]flowDto.OutputItem)
+	// 给有效记录分配【时间倒序编号】（最新=1）
+	totalValid := len(validList)
+	for idx, execution := range validList {
+		executionNumber[execution.Id] = totalValid - idx
+	}
+
+	// 2. 分组映射：日期 -> 流程节点
+	type flowWrap struct {
+		flowNode  flowDto.FlowNode
+		createdAt *gtime.Time
+	}
+	dateMap := make(map[string]*[]flowWrap)
+
+	// 遍历【有效数据】构建结构
 	for _, execution := range validList {
-		// 按要求使用 Y-m-d
 		createDate := execution.CreatedAt.Format("Y-m-d")
-		execID := execution.Id
+		flowName := execution.FlowName
 		outputParams := execution.OutputParams

+		// 编号只算有效数据，不会把空的算进去
+		num := executionNumber[execution.Id]
+		displayFlowName := fmt.Sprintf("会话-%d(%s)", num, flowName)
+
+		// 3. 解析 outputParams
 		var tempItems []flowDto.OutputItem
 		for _, paramMap := range outputParams {
 			for tsKey, value := range paramMap {
@@ -140,116 +102,98 @@ func (s *flowExecutionService) List(ctx context.Context, req *flowDto.ListFlowEx
 					continue
 				}
 				tempItems = append(tempItems, flowDto.OutputItem{
-					Id:        execID,
 					Timestamp: tsKey,
 					Content:   gconv.String(value),
 				})
 			}
 		}
+
+		// ===================== 修复1：如果解析后依然为空，直接跳过，不生成第二层节点 =====================
 		if len(tempItems) == 0 {
 			continue
 		}

-		// 单条执行内按时间戳正序
+		// 时间戳正序
 		sort.Slice(tempItems, func(i, j int) bool {
 			t1, _ := strconv.ParseInt(tempItems[i].Timestamp, 10, 64)
 			t2, _ := strconv.ParseInt(tempItems[j].Timestamp, 10, 64)
 			return t1 < t2
 		})

-		dateMap[createDate] = append(dateMap[createDate], tempItems...)
-	}
-
-	// ========== 修复编号乱序核心逻辑 ==========
-	// 1. 取出所有日期并 倒序排序（和前端展示顺序一致）
-	var sortedDates []string
-	for d := range dateMap {
-		sortedDates = append(sortedDates, d)
-	}
-	// 日期字符串倒序
-	sort.Slice(sortedDates, func(i, j int) bool {
-		return sortedDates[i] > sortedDates[j]
-	})
-
-	// 2. 按【前端展示顺序】拼接所有条目，用于统计总数量
-	var allItems []flowDto.OutputItem
-	for _, d := range sortedDates {
-		allItems = append(allItems, dateMap[d]...)
-	}
-
-	// 3. 统计各类型总数
-	type totalCnt struct {
-		total int
-		idx   int
-	}
-	typeTotal := make(map[string]*totalCnt)
-	for _, item := range allItems {
-		val := item.Content
-		suffix := "内容"
-		ext := GetFileTypeByPath(val)
-
-		switch ext {
-		case "image":
-			suffix = "图片"
-		case "video":
-			suffix = "视频"
-		case "audio":
-			suffix = "音频"
-		case "text":
-			suffix = "文案"
-		case "html":
-			suffix = "HTML"
-		}
-		if _, ok := typeTotal[suffix]; !ok {
-			typeTotal[suffix] = &totalCnt{}
-		}
-		typeTotal[suffix].total++
-	}
-	// 初始序号 = 总数，从最大值开始倒序
-	for _, v := range typeTotal {
-		v.idx = v.total
-	}
-	// ======================================
-
-	var tree []flowDto.DateNode
-	// 按有序日期遍历生成最终数据
-	for _, date := range sortedDates {
-		items := dateMap[date]
-		if len(items) == 0 {
-			continue
-		}
-		// 逐个生成倒序标签
-		for idx := range items {
-			item := &items[idx]
+		// 标号：相同类型递增，不同重置
+		suffixCount := make(map[string]int)
+		for idx := range tempItems {
+			item := &tempItems[idx]
 			val := item.Content
 			suffix := "内容"
-			ext := GetFileTypeByPath(val)
-
-			switch ext {
-			case "image":
+			ext := ""
+			ext = GetFileTypeByPath(val)
+			if ext == "image" {
 				suffix = "图片"
-			case "video":
+			}
+			if ext == "video" {
 				suffix = "视频"
-			case "audio":
+			}
+			if ext == "audio" {
 				suffix = "音频"
-			case "text":
+			}
+			if ext == "text" {
 				suffix = "文案"
-			case "html":
+			}
+			if ext == "html" {
 				suffix = "HTML"
 			}
-
-			cnt := typeTotal[suffix]
+			suffixCount[suffix]++
 			item.Type = ext
-			item.Label = fmt.Sprintf("%s_%d", suffix, cnt.idx)
-			cnt.idx--
+			item.Label = fmt.Sprintf("%s_%d", suffix, suffixCount[suffix])
+		}
+
+		// 组装节点
+		flowNode := flowDto.FlowNode{
+			FlowName:  displayFlowName,
+			Id:        execution.Id,
+			SessionId: gconv.String(execution.SessionId),
+			Items:     tempItems,
+		}
+
+		if dateMap[createDate] == nil {
+			dateMap[createDate] = &[]flowWrap{}
+		}
+		*dateMap[createDate] = append(*dateMap[createDate], flowWrap{
+			flowNode:  flowNode,
+			createdAt: execution.CreatedAt,
+		})
+	}
+
+	// 6. 构建树 + 排序
+	var tree []flowDto.DateNode
+	for date, wraps := range dateMap {
+		// 第二层按创建时间倒序（最新在前）
+		sort.Slice(*wraps, func(i, j int) bool {
+			return (*wraps)[i].createdAt.After((*wraps)[j].createdAt)
+		})
+
+		var flowNodes []flowDto.FlowNode
+		for _, w := range *wraps {
+			flowNodes = append(flowNodes, w.flowNode)
+		}
+
+		// ===================== 修复2：日期下没有流程，也过滤掉 =====================
+		if len(flowNodes) == 0 {
+			continue
 		}

 		tree = append(tree, flowDto.DateNode{
 			CreateDate: date,
-			Items:      items,
+			Flows:      flowNodes,
 		})
 	}

+	// 第一层日期倒序
+	sort.Slice(tree, func(i, j int) bool {
+		return tree[i].CreateDate > tree[j].CreateDate
+	})
+
 	imgPrefix, err := utils.GetFileAddressPrefix(ctx)
 	return &flowDto.ListFlowExecutionTreeRes{
 		Tree:             tree,
@@ -341,6 +285,13 @@ func (s *flowExecutionService) Execute(ctx context.Context, req *flowDto.Execute
 		cancel()
 	}()

+	//getRes, err := FlowUserService.Get(ctx, &flowDto.GetFlowUserReq{
+	//	Id: req.FlowId,
+	//})
+	//if err != nil {
+	//	return nil, err
+	//}
+	nodeInputParams := ExtractFlowNodeFrom(req.FlowContent)
 	flowInfo, err := flowDao.FlowExecutionDao.Get(ctx, &flowDto.GetFlowExecutionReq{
 		SessionId: req.SessionId,
 	})
@@ -350,24 +301,17 @@ func (s *flowExecutionService) Execute(ctx context.Context, req *flowDto.Execute
 	var executionId int64
 	var isDialogue bool
 	var nodeGroupId = uuid.NewString()
-	flowName := req.FlowName
-	if !g.IsEmpty(req.Desc) {
-		flowName = req.Desc
-	}
-	isDialogue = false
 	if flowInfo == nil {
+		isDialogue = false
 		var r = new(flowDto.CreateFlowExecutionReq)
 		r.FlowUserId = req.FlowId
-		r.FlowName = flowName
+		r.FlowName = req.FlowName
 		r.NodeGroupId = nodeGroupId
 		r.TriggerType = flow.FlowExecutionTriggerTypeManual.Code()
 		r.FlowContent = req.FlowContent
-		r.NodeInputParams = req.NodeInputParams
+		//r.NodeInputParams = nodeInputParams
 		r.SessionId = req.SessionId
 		r.Status = flow.FlowExecutionStatusRunning.Code()
-		r.Extension = map[string]any{
-			"templates": req.Templates,
-		}
 		span := trace.SpanFromContext(ctx)
 		if span != nil && span.SpanContext().HasTraceID() {
 			r.TraceId = span.SpanContext().TraceID().String()
@@ -379,6 +323,7 @@ func (s *flowExecutionService) Execute(ctx context.Context, req *flowDto.Execute
 			return
 		}
 	} else {
+		isDialogue = true
 		executionId = flowInfo.Id
 		span := trace.SpanFromContext(ctx)
 		if span != nil && span.SpanContext().HasTraceID() {
@@ -386,15 +331,10 @@ func (s *flowExecutionService) Execute(ctx context.Context, req *flowDto.Execute
 			cancelMap.Store(traceId, cancel)
 		}
 		executionReq := flowDto.UpdateFlowExecutionReq{
-			Id:              executionId,
-			NodeGroupId:     nodeGroupId,
-			FlowContent:     req.FlowContent,
-			NodeInputParams: req.NodeInputParams,
-			Extension: map[string]any{
-				"templates": req.Templates,
-			},
-			Status:  flow.FlowExecutionStatusRunning.Code(),
-			TraceId: traceId,
+			Id:          executionId,
+			NodeGroupId: nodeGroupId,
+			Status:      flow.FlowExecutionStatusRunning.Code(),
+			TraceId:     traceId,
 		}
 		_, err = flowDao.FlowExecutionDao.Update(ctx, &executionReq)
 		if err != nil {
@@ -416,63 +356,26 @@ func (s *flowExecutionService) Execute(ctx context.Context, req *flowDto.Execute
 		}
 	}

-	//if isDialogue && !g.IsEmpty(flowInfo) && !g.IsEmpty(req.ResultUrl) {
-	//	req.NodeGroupId = nodeGroupId
-	//	if strings.HasSuffix(gconv.String(req.ResultUrl), ".inc") {
-	//		err = TextModelSingleLambda(ctx, req, flowInfo)
-	//		return
-	//	} else if strings.HasSuffix(gconv.String(req.ResultUrl), ".png") {
-	//		err = ImgModelSingleLambda(ctx, req, flowInfo)
-	//		return
-	//	} else if strings.HasSuffix(gconv.String(req.ResultUrl), ".html") {
-	//		err = TextImgModelSingleLambda(ctx, req, flowInfo)
-	//		return
-	//	}
-	//	return nil, errors.New("文件格式不支持")
-	//}
-
-	// =========================================================================
-	// ✅【第1步】给所有判断节点自动生成意图识别节点
-	// =========================================================================
-	judge2IntentNodeMap := make(map[string]string)
-	finalNodes := make([]entity.FlowNode, 0, len(req.FlowContent.Nodes)*2)
-	for _, item := range req.FlowContent.Nodes {
-		finalNodes = append(finalNodes, item)
-		// 判断节点自动加 intent 节点
-		if item.NodeCode == node.NodeTypeJudge {
-			intentNodeID := fmt.Sprintf("intent_%s", item.Id)
-			intentNode := entity.FlowNode{
-				Id:          intentNodeID,
-				NodeCode:    node.NodeTypeIntent,
-				Name:        fmt.Sprintf("意图识别-%s", item.Name),
-				InputSource: item.InputSource, // ✅ 正确赋值
-				FormConfig:  item.FormConfig,  // ✅ 用户配置
-				ModelConfig: item.ModelConfig, // ✅ 系统配置
-			}
-			finalNodes = append(finalNodes, intentNode)
-			judge2IntentNodeMap[item.Id] = intentNodeID
+	if isDialogue && !g.IsEmpty(flowInfo) && !g.IsEmpty(req.ResultUrl) {
+		req.NodeGroupId = nodeGroupId
+		if strings.HasSuffix(gconv.String(req.ResultUrl), ".inc") {
+			err = TextModelSingleLambda(ctx, req, flowInfo)
+			return
+		} else if strings.HasSuffix(gconv.String(req.ResultUrl), ".png") {
+			err = ImgModelSingleLambda(ctx, req, flowInfo)
+			return
+		} else if strings.HasSuffix(gconv.String(req.ResultUrl), ".html") {
+			err = TextImgModelSingleLambda(ctx, req, flowInfo)
+			return
 		}
+		return nil, errors.New("文件格式不支持")
 	}
-
-	summaryNodeID := "summary_node"
-	summaryNode := entity.FlowNode{
-		Id:          summaryNodeID,
-		NodeCode:    node.NodeTypeCustomNode, // 复用自定义节点类型，也可新增专属类型
-		Name:        "结果汇总节点",
-		InputSource: []entity.FlowNodeInputSource{}, // 后续自动聚合所有节点输出
-		FormConfig:  nil,
-		ModelConfig: node.ModelItem{},
-	}
-	finalNodes = append(finalNodes, summaryNode)
-
-	// 替换节点列表
-	req.FlowContent.Nodes = finalNodes
-
 	// =========================================================================
 	// ✅【第2步】构建执行图
 	// =========================================================================
+	var nodeList []entity.FlowNode
 	var runGraph compose.Runnable[any, any]
-	runGraph, err = BuildGraphFromFlowContent(execCtx, req.FlowContent, judge2IntentNodeMap, summaryNodeID)
+	nodeList, runGraph, err = BuildGraphFromFlowContent(execCtx, req.FlowContent)
 	if err != nil {
 		executionReq := flowDto.UpdateFlowExecutionReq{
 			Id:           executionId,
@@ -485,23 +388,16 @@ func (s *flowExecutionService) Execute(ctx context.Context, req *flowDto.Execute
 		}
 		return nil, fmt.Errorf("执行工作流失败: %v", err)
 	}
-
 	// =========================================================================
 	// ✅【第3步】构建 ConfigMap
 	// =========================================================================
 	configMap := make(map[string]*entity.FlowNode)
-	for _, cfg := range req.NodeInputParams {
+	for _, cfg := range nodeInputParams {
 		configMap[cfg.Id] = cfg
 	}
-	// 自动给意图节点复制配置
-	for judgeID, intentID := range judge2IntentNodeMap {
-		if cfg, ok := configMap[judgeID]; ok {
-			configMap[intentID] = cfg
-		}
+	for _, i := range nodeList {
+		configMap[i.Id] = &i
 	}
-	// 初始化汇总节点配置
-	configMap[summaryNodeID] = &summaryNode
-
 	// =========================================================================
 	// ✅【第4步】构建全局执行入参（现在 schemaMap 是有值的！）
 	// =========================================================================
@@ -510,7 +406,6 @@ func (s *flowExecutionService) Execute(ctx context.Context, req *flowDto.Execute
 		IsDialogue:  isDialogue,
 		ExecutionId: executionId,
 		ConfigMap:   configMap,
-		Templates:   req.Templates,
 		SessionId:   req.SessionId,
 		Desc:        req.Desc,
 		SkillName:   req.SkillName,
@@ -542,8 +437,7 @@ func (s *flowExecutionService) Execute(ctx context.Context, req *flowDto.Execute
 	return
 }

-// BuildGraphFromFlowContent 根据前端保存的工作流JSON，自动构建执行图
-func BuildGraphFromFlowContent(ctx context.Context, flowContent *entity.FlowInfo, judge2IntentNodeMap map[string]string, summaryNodeID string) (compose.Runnable[any, any], error) {
+func BuildGraph(ctx context.Context, flowContent *entity.FlowInfo) ([]entity.FlowNode, *compose.Graph[any, any]) {
 	// 注册自定义合并函数：处理 *flowDto.FlowExecutionInput 类型合并
 	// 由于 ConfigMap 是 map 引用类型，所有并行分支修改已经写入共享内存
 	// 直接返回第一个实例即可，所有修改都已经可见
@@ -556,9 +450,26 @@ func BuildGraphFromFlowContent(ctx context.Context, flowContent *entity.FlowInfo
 	})

 	graph := compose.NewGraph[any, any]()
-	nodeMap := make(map[string]entity.FlowNode)
+
+	var nodeList []entity.FlowNode
+	nodeId := uuid.NewString()
+	originalEndNodes := findEndNodes(flowContent.StartNodeId, flowContent.Edges)
+	for i := range originalEndNodes {
+		sprintf := fmt.Sprintf("%v_%d", nodeId, i)
+		summaryNode := entity.FlowNode{
+			Id:          sprintf,
+			NodeCode:    node.NodeTypeSystemSum,
+			Name:        node.NodeNameSystemSum,
+			InputSource: []entity.FlowNodeInputSource{}, // 后续自动聚合所有节点输出
+			FormConfig:  nil,
+			ModelConfig: node.ModelItem{},
+		}
+		nodeList = append(nodeList, summaryNode)
+		flowContent.Nodes = append(flowContent.Nodes, summaryNode)
+	}

 	// 注册所有节点
+	nodeMap := make(map[string]entity.FlowNode)
 	for _, item := range flowContent.Nodes {
 		nodeMap[item.Id] = item
 		if item.NodeCode != node.NodeTypeJudge {
@@ -566,6 +477,16 @@ func BuildGraphFromFlowContent(ctx context.Context, flowContent *entity.FlowInfo
 		}
 	}

+	// 注册所有边
+	if flowContent.StartNodeId != "" {
+		_ = graph.AddEdge(compose.START, flowContent.StartNodeId)
+	}
+	for i, endID := range originalEndNodes {
+		sprintf := fmt.Sprintf("%v_%d", nodeId, i)
+		_ = graph.AddEdge(endID, sprintf)
+		_ = graph.AddEdge(sprintf, compose.END)
+	}
+
 	// 构建边关系
 	upstreamMap := make(map[string][]string)
 	edgeMap := make(map[string][]entity.FlowEdge)
@@ -578,15 +499,8 @@ func BuildGraphFromFlowContent(ctx context.Context, flowContent *entity.FlowInfo
 	for fromNodeID, edges := range edgeMap {
 		fromNode := nodeMap[fromNodeID]

-		// --------------------------
 		// 判断节点 → 分支处理
-		// --------------------------
 		if fromNode.NodeCode == node.NodeTypeJudge {
-			intentNodeID, ok := judge2IntentNodeMap[fromNodeID]
-			if !ok {
-				return nil, fmt.Errorf("判断节点[%s]未生成意图节点", fromNodeID)
-			}
-
 			branchMap := make(map[string]bool)
 			for _, e := range edges {
 				branchMap[e.To] = true
@@ -621,11 +535,6 @@ func BuildGraphFromFlowContent(ctx context.Context, flowContent *entity.FlowInfo
 				m["branch_id_name_map"] = branchIdNameMap // 传递ID-名称映射
 				currentConfig.Config = m

-				// 从意图节点取输出
-				if intentCfg, ok := execInput.ConfigMap[intentNodeID]; ok {
-					currentConfig.OutputResult = intentCfg.OutputResult
-				}
-
 				// 关键修改：构造 NodeExecutionInput 传入 JudgeLambda
 				nodeExecInput := &flowDto.NodeExecutionInput{
 					Config: currentConfig, // 当前判断节点配置
@@ -634,41 +543,31 @@ func BuildGraphFromFlowContent(ctx context.Context, flowContent *entity.FlowInfo
 				return JudgeLambda(ctx, nodeExecInput) // 传入 NodeExecutionInput 类型
 			}

-			_ = graph.AddBranch(intentNodeID, compose.NewGraphBranch(judgeLambda, branchMap))
+			_ = graph.AddBranch(upstreamMap[fromNodeID][0], compose.NewGraphBranch(judgeLambda, branchMap))
 			continue
 		}

-		// --------------------------
 		// 普通节点连线
-		// --------------------------
 		for _, e := range edges {
 			toNode := nodeMap[e.To]
 			if toNode.NodeCode == node.NodeTypeJudge {
-				_ = graph.AddEdge(e.From, fmt.Sprintf("intent_%s", toNode.Id))
 				continue
 			}
 			_ = graph.AddEdge(e.From, e.To)
 		}
 	}
+	return nodeList, graph
+}

-	// ==================== 第四步：处理开始/结束节点 ====================
-	if flowContent.StartNodeId != "" {
-		_ = graph.AddEdge(compose.START, flowContent.StartNodeId)
-	}
-	originalEndNodes := findEndNodes(flowContent.StartNodeId, flowContent.Edges)
-	for _, endID := range originalEndNodes {
-		_ = graph.AddEdge(endID, summaryNodeID)
-	}
-	_ = graph.AddEdge(summaryNodeID, compose.END)
-
-	return graph.Compile(ctx, compose.WithGraphName("auto_build_workflow"), compose.WithNodeTriggerMode(compose.AllPredecessor))
+// BuildGraphFromFlowContent 根据前端保存的工作流JSON，自动构建执行图
+func BuildGraphFromFlowContent(ctx context.Context, flowContent *entity.FlowInfo) ([]entity.FlowNode, compose.Runnable[any, any], error) {
+	nodeList, graph := BuildGraph(ctx, flowContent)
+	compile, err := graph.Compile(ctx, compose.WithGraphName("auto_build_workflow"))
+	return nodeList, compile, err
 }

 // -------------------------- 节点自动注册器（核心分发） --------------------------
 func registerNodeToGraph(graph *compose.Graph[any, any], flowNode entity.FlowNode) {
-	nodeID := flowNode.Id
-	code := flowNode.NodeCode
-
 	// 通用包装：全程入参都是 *FlowExecutionInput
 	wrapLambda := func(lambda func(ctx context.Context, input any) (any, error)) func(ctx context.Context, input any) (any, error) {
 		return func(ctx context.Context, input any) (any, error) {
@@ -679,9 +578,9 @@ func registerNodeToGraph(graph *compose.Graph[any, any], flowNode entity.FlowNod
 			}

 			configMap := execInput.ConfigMap
-			currentConfig := configMap[nodeID]
+			currentConfig := configMap[flowNode.Id]
 			if currentConfig == nil {
-				return nil, fmt.Errorf("节点%s无配置", nodeID)
+				return nil, fmt.Errorf("节点%s无配置", flowNode.Id)
 			}

 			// 获取入参 - 适配切片类型：遍历所有来源节点
@@ -714,7 +613,7 @@ func registerNodeToGraph(graph *compose.Graph[any, any], flowNode entity.FlowNod

 			nodeExecutionId, err := nodeDao.NodeExecutionDao.Insert(ctx, &nodeDto.CreateNodeExecutionReq{
 				FlowExecutionId: execInput.ExecutionId,
-				NodeId:          nodeID,
+				NodeId:          flowNode.Id,
 				NodeName:        flowNode.Name,
 				NodeGroupId:     execInput.NodeGroupId,
 				InputParamsPath: ossResult.FileURL,
@@ -723,7 +622,7 @@ func registerNodeToGraph(graph *compose.Graph[any, any], flowNode entity.FlowNod
 			if err != nil {
 				// 记录失败到已执行列表
 				execInput.ExecutedNodes = append(execInput.ExecutedNodes, flowDto.ExecutedNode{
-					NodeId: nodeID,
+					NodeId: flowNode.Id,
 					Status: node.NodeExecutionStatusFailed.Code(),
 				})
 				return nil, err
@@ -743,7 +642,7 @@ func registerNodeToGraph(graph *compose.Graph[any, any], flowNode entity.FlowNod
 				_, _ = nodeDao.NodeExecutionDao.Update(ctx, updateReq)
 				// 记录失败到已执行列表
 				execInput.ExecutedNodes = append(execInput.ExecutedNodes, flowDto.ExecutedNode{
-					NodeId: nodeID,
+					NodeId: flowNode.Id,
 					Status: node.NodeExecutionStatusFailed.Code(),
 				})
 				return nil, err
@@ -762,7 +661,7 @@ func registerNodeToGraph(graph *compose.Graph[any, any], flowNode entity.FlowNod
 			_, _ = nodeDao.NodeExecutionDao.Update(ctx, updateReq)
 			// 记录成功到已执行列表
 			execInput.ExecutedNodes = append(execInput.ExecutedNodes, flowDto.ExecutedNode{
-				NodeId: nodeID,
+				NodeId: flowNode.Id,
 				Status: node.NodeExecutionStatusSuccess.Code(),
 			})

@@ -770,83 +669,70 @@ func registerNodeToGraph(graph *compose.Graph[any, any], flowNode entity.FlowNod
 			return execInput, nil
 		}
 	}
-	if nodeID == "summary_node" {
-		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(SummaryLambda)))
-		return
-	}
-	switch code {
+	switch flowNode.NodeCode {
 	case "__start__":
-		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(StartLambda)))
+		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(StartLambda)))
+	case node.NodeTypeSystemSum:
+		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(SummaryLambda)))
 	case node.NodeTypeTextModel:
-		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(TextModelLambda)))
+		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(TextModelLambda)))
 	case node.NodeTypeImageModel:
-		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(ImageModelLambda)))
+		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(ImageModelLambda)))
 	case node.NodeTypeVideoModel:
-		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(VideoModelLambda)))
+		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(VideoModelLambda)))
 	case node.NodeTypeAudioModel:
-		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(AudioModelLambda)))
+		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(AudioModelLambda)))
 	case node.NodeTypeBatchModel:
-		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(BatchModelLambda)))
+		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(BatchModelLambda)))
 	case node.NodeTypeDataConversionModel:
-		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(DataConversionLambda)))
-	//case node.NodeTypeSenseOptimizeModel:
-	//	_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(SenseOptimizeModelLambda)))
-	//case node.NodeTypeStoryOptimizeModel:
-	//	_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(StoryOptimizeModelLambda)))
-	//case node.NodeTypeScriptOptimizeModel:
-	//	_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(ScriptOptimizeModelLambda)))
+		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(DataConversionLambda)))
 	case node.NodeTypeCustomNode:
-		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(CustomLambda)))
+		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(CustomLambda)))
 	case node.NodeTypeForm:
-		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(FormLambda)))
-	case node.NodeTypeIntent:
-		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(IntentLambda)))
+		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(FormLambda)))
+	//case node.NodeTypeIntent:
+	//	_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(IntentLambda)))
 	case node.NodeTypeMerge:
-		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(MergeLambda)))
+		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(MergeLambda)))
 	case node.NodeTypeDataMerge:
-		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(DataMergeLambda)))
+		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(DataMergeLambda)), compose.WithGraphCompileOptions(compose.WithNodeTriggerMode(compose.AllPredecessor)))
+	case node.NodeTypeSubFlow:
+		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(SubFlowLambda)))
 	case node.NodeTypeHttp:
-		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(HttpLambda)))
+		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(HttpLambda)))
 	}
 }

-// --------------------------------------------------------------------
-// ✅【工具方法】找出所有没有出边的节点 → 作为结束节点连接 END
-// --------------------------------------------------------------------
 func findEndNodes(startNodeId string, edges []entity.FlowEdge) []string {
-	// 构建 节点 → 后续节点 的映射
 	nextMap := make(map[string][]string)
 	for _, e := range edges {
 		nextMap[e.From] = append(nextMap[e.From], e.To)
 	}

 	endNodeSet := make(map[string]struct{})
+	visited := make(map[string]struct{})
+	queue := []string{startNodeId}

-	// 🚀 只从【开始节点】递归遍历（关键修复）
-	findLeafNodes(startNodeId, nextMap, endNodeSet)
+	for len(queue) > 0 {
+		node := queue[0]
+		queue = queue[1:]

-	// 转成数组返回
-	endNodes := make([]string, 0, len(endNodeSet))
-	for id := range endNodeSet {
-		endNodes = append(endNodes, id)
-	}
-	return endNodes
-}
+		if _, exist := visited[node]; exist {
+			continue
+		}
+		visited[node] = struct{}{}

-// --------------------------------------------------------------------
-// ✅ 递归：查找以 nodeId 开头的所有叶子节点
-// --------------------------------------------------------------------
-func findLeafNodes(nodeId string, nextMap map[string][]string, endNodeSet map[string]struct{}) {
-	nextNodes := nextMap[nodeId]
-
-	// 🚩 没有下一个节点 = 真实结束节点
-	if len(nextNodes) == 0 {
-		endNodeSet[nodeId] = struct{}{}
-		return
-	}
-
-	// 递归继续找下一个
-	for _, nextId := range nextNodes {
-		findLeafNodes(nextId, nextMap, endNodeSet)
+		nextList := nextMap[node]
+		if len(nextList) == 0 {
+			endNodeSet[node] = struct{}{}
+			continue
+		}
+		queue = append(queue, nextList...)
 	}
+
+	res := make([]string, 0, len(endNodeSet))
+	for k := range endNodeSet {
+		res = append(res, k)
+	}
+	return res
 }
@@ -6,11 +6,9 @@ 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"
@@ -21,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,40 +31,81 @@ func StartLambda(ctx context.Context, input any) (any, error) {
 }

 func FormLambda(ctx context.Context, input any) (any, error) {
-	nodeInput, ok := input.(*flowDto.NodeExecutionInput)
-	if !ok {
-		return nil, fmt.Errorf("入参类型错误")
-	}
-	if !g.IsEmpty(nodeInput.Config.FormConfig) {
-		for _, v := range nodeInput.Config.FormConfig {
-			if strings.Contains(v.Field, "视频时长") {
-				if g.IsEmpty(v.Value) {
-					return nil, fmt.Errorf("视频时长不能为空")
-				}
-				if gconv.Int(v.Value) >= 16 {
-					return nil, fmt.Errorf("视频时长超过15秒了")
-				}
-				if gconv.Int(v.Value) <= 3 {
-					return nil, fmt.Errorf("视频时长不能小于4秒")
-				}
-			}
-			if strings.Contains(v.Field, "视频分辨率") {
-				if gconv.String(v.Value) != "480p" && gconv.String(v.Value) != "720p" {
-					return nil, fmt.Errorf("视频分辨率不合法, 可选值: 480p, 720p")
-				}
-			}
-			if strings.Contains(v.Field, "宽高比例") {
-				if gconv.String(v.Value) != "21:9" && gconv.String(v.Value) != "1:1" && gconv.String(v.Value) != "16:9" && gconv.String(v.Value) != "4:3" && gconv.String(v.Value) != "9:16" && gconv.String(v.Value) != "3:4" {
-					return nil, fmt.Errorf("宽高比例不合法, 可选值: 21:9, 1:1, 16:9, 4:3, 9:16, 3:4")
-				}
-			}
-		}
-	}
 	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做路由
@@ -73,6 +114,20 @@ func JudgeLambda(ctx context.Context, input any) (string, error) {
 	if !ok {
 		return "", fmt.Errorf("入参类型错误，期望 *flowDto.NodeExecutionInput，实际 %T", input)
 	}
+	//inputMap, outputMap, modelMap := GetNodeContextContent(nodeInput.Global, nodeInput.Config)
+	//fmt.Printf("JudgeLambda路由：输入=%s\n", gjson.MustEncode(inputMap))
+	//fmt.Printf("JudgeLambda路由：输出=%s\n", gjson.MustEncode(outputMap))
+	//fmt.Printf("JudgeLambda路由：模型=%s\n", gjson.MustEncode(modelMap))
+	//configMap := gconv.Map(nodeInput.Config.Config)
+	//ids := gconv.Strings(configMap["branch_ids"])
+	//fmt.Printf("JudgeLambda路由：目标节点ID=%s\n", gconv.String(ids))
+	//
+	//m := map[string]bool{
+	//	"80000a50-81e1-4c15-adae-aab6c0d781ad": true,
+	//	"59a6ffa2-3252-4535-b6ed-d3e49cdf6c55": true,
+	//}
+	//
+	//return m, nil
 	// 1. 直接用你原来的方法（返回两个 map）
 	inputMap, outputMap, modelMap := GetNodeContextContent(nodeInput.Global, nodeInput.Config)
 	var outputResult []node.NodeFormField
@@ -117,7 +172,7 @@ func JudgeLambda(ctx context.Context, input any) (string, error) {
 	if err != nil {
 		return "", err
 	}
-	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, "判断节点")
+	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, "判断节点")
 	if err != nil {
 		return "", err
 	}
@@ -161,8 +216,9 @@ func BatchModelLambda(ctx context.Context, input any) (any, error) {
 	res := make([][]node.NodeFormField, len(reqMap))
 	var wg sync.WaitGroup

-	// 只创建基础上下文，不再主动批量 cancel
-	subCtx := context.WithoutCancel(ctx)
+	subCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
 	// 缓冲1错误通道，仅接收第一个错误
 	errCh := make(chan error, 1)

@@ -172,7 +228,7 @@ func BatchModelLambda(ctx context.Context, input any) (any, error) {
 		go func(idx int, userItem map[string]any) {
 			defer wg.Done()

-			// 基础上下文仅响应上游原始 ctx 取消，内部任务失败不触发这里
+			// 上下文已取消则直接退出
 			select {
 			case <-subCtx.Done():
 				return
@@ -180,12 +236,12 @@ func BatchModelLambda(ctx context.Context, input any) (any, error) {
 			}

 			singleUserFrom := []map[string]any{userItem}
-			// 下游调用使用 subCtx，不会因为同批次其他任务报错而取消
 			output, err := TextNode(subCtx, nodeInput, skillName, from, singleUserFrom)
 			if err != nil {
-				// 只往错误通道塞第一个错误，不调用全局 cancel
+				// 仅第一个错误写入通道
 				select {
 				case errCh <- err:
+					cancel() // 触发全局取消，其他协程快速退出
 				default:
 				}
 				return
@@ -194,30 +250,26 @@ func BatchModelLambda(ctx context.Context, input any) (any, error) {
 		}(idx, item)
 	}

-	// 所有协程跑完再关闭通道
+	// 任务全部结束后关闭错误通道
 	go func() {
 		wg.Wait()
 		close(errCh)
 	}()

-	// ========== 修复区域 start ==========
+	// ========== 修正后的等待逻辑 ==========
 	var execErr error
 	select {
+	// 优先捕获业务错误
 	case execErr = <-errCh:
-		// 捕获第一个业务错误，等待剩余协程收尾
-		wg.Wait()
-	case <-subCtx.Done():
-		// 上游根上下文被终止，读取已存在的错误
-		execErr = <-errCh
-		wg.Wait()
 		if execErr != nil {
-			execErr = fmt.Errorf("global context canceled: %w", execErr)
+			// 收到真实业务错误，等待剩余协程收尾后返回
+			wg.Wait()
+			return nil, execErr
 		}
-	}
-
-	// 有错误直接返回，不再走结果拼接
-	if execErr != nil {
-		return nil, execErr
+		// execErr == nil 代表通道关闭、无任何错误，走到下方返回完整结果
+	case <-subCtx.Done():
+		// 上下文被取消，阻塞读完errCh，确认是否存在业务错误
+		execErr = <-errCh
 	}

 	// 拼接输出结果
@@ -254,6 +306,8 @@ func TextModelLambda(ctx context.Context, input any) (any, error) {
 		return nil, err
 	}
 	nodeInput.Config.OutputResult = outputRes
+	//}
+
 	return nodeInput, nil
 }

@@ -299,7 +353,7 @@ func VideoModelLambda(ctx context.Context, input any) (any, error) {
 	if err != nil {
 		return nil, err
 	}
-	videoUrl := ""
+
 	videoURL := make([]string, 0)
 	for _, v := range res {
 		if strings.Contains(v.Field, "content") {
@@ -309,56 +363,34 @@ func VideoModelLambda(ctx context.Context, input any) (any, error) {
 	if g.IsEmpty(videoURL) {
 		return nil, fmt.Errorf("视频合成失败:模型生成视频失败")
 	}
-	if len(videoURL) > 1 {
-		var waitRes any
-		waitRes, err = VideoConcat(ctx, videoURL)
-		if err != nil {
-			return nil, err
-		}
-		msg := new(flowDto.VideoCallbackReq)
-		if err = gconv.Struct(waitRes, msg); err != nil {
-			return nil, err
-		}
-		var urlPrefix string
-		urlPrefix, err = utils.GetFileAddressPrefix(ctx)
-		newS := strings.ReplaceAll(urlPrefix, g.Cfg().MustGet(ctx, "filePrefix").String(), g.Cfg().MustGet(ctx, "minioPrefix").String())
-		if err != nil {
-			return nil, err
-		}
-		videoUrl = newS + msg.FileURL
-	} else {
-		var bytes []byte
-		bytes, err = GetFileBytesFromURL(ctx, videoURL[0])
-		if err != nil {
-			return nil, fmt.Errorf("下载图片失败: %w", err)
-		}
-		// 构造文件名
-		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{
-			FileName:  fileName,
-			FileBytes: bytes,
-		})
-		if err != nil {
-			return nil, fmt.Errorf("上传OSS失败: %w", err)
-		}
-		videoUrl = upResp.FileURL
+	waitRes, err := VideoConcat(ctx, videoURL)
+	if err != nil {
+		return nil, err
 	}
+	msg := new(flowDto.VideoCallbackReq)
+	if err = gconv.Struct(waitRes, msg); err != nil {
+		return nil, err
+	}
+
+	urlPrefix, err := utils.GetFileAddressPrefix(ctx)
+	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 {
 		outputRes = append(outputRes, node.NodeFormField{
 			Field: fmt.Sprintf("video_oss_url:content:%d", 0),
-			Value: videoUrl,
+			Value: msg.FileURL,
 			Label: fmt.Sprintf("video_oss_url:content:%d", 0),
 			Type:  "string",
 		})
 	}
 	outputRes = append(outputRes, node.NodeFormField{
 		Field: fmt.Sprintf("concat_video_url:content:%d", 0),
-		Value: videoUrl,
-		Label: fmt.Sprintf("concat_video_url:content:%d", 0),
+		Value: newS + msg.FileURL,
+		Label: fmt.Sprintf("视频内容:content:%d", 0),
 		Type:  "string",
 	})
 	nodeInput.Config.OutputResult = outputRes
@@ -375,6 +407,64 @@ func HttpLambda(ctx context.Context, input any) (any, error) {
 	outputRes := make([]node.NodeFormField, 0)
 	var err error
 	outputRes, err = HttpNode(ctx, nodeInput)
+	//if nodeInput.Config.Name == "生成视频" {
+	//	outputRes, err = HttpNode(ctx, nodeInput)
+	//} else {
+	//	a := []map[string]any{
+	//		{
+	//			"timeline": "0.0-2.1",
+	//			"url":      "https://ark-auto-2127201628-cn-beijing-default.tos-cn-beijing.volces.com/%E8%A7%86%E9%A2%91/1.mp4",
+	//		},
+	//		{
+	//			"timeline": "2.1-4.5",
+	//			"url":      "https://ark-auto-2127201628-cn-beijing-default.tos-cn-beijing.volces.com/%E8%A7%86%E9%A2%91/2.mp4",
+	//		},
+	//		{
+	//			"timeline": "4.5-12.2",
+	//			"url":      "https://ark-auto-2127201628-cn-beijing-default.tos-cn-beijing.volces.com/%E8%A7%86%E9%A2%91/3.mp4",
+	//		},
+	//		{
+	//			"timeline": "12.2-13.6",
+	//			"url":      "https://ark-auto-2127201628-cn-beijing-default.tos-cn-beijing.volces.com/%E8%A7%86%E9%A2%91/4.mp4",
+	//		},
+	//		{
+	//			"timeline": "13.6-17.7",
+	//			"url":      "https://ark-auto-2127201628-cn-beijing-default.tos-cn-beijing.volces.com/%E8%A7%86%E9%A2%91/5.mp4model-gateway",
+	//		},
+	//		{
+	//			"timeline": "17.7-31.0",
+	//			"url":      "https://ark-auto-2127201628-cn-beijing-default.tos-cn-beijing.volces.com/%E8%A7%86%E9%A2%91/6.mp4",
+	//		},
+	//		{
+	//			"timeline": "31.0-33.2",
+	//			"url":      "https://ark-auto-2127201628-cn-beijing-default.tos-cn-beijing.volces.com/%E8%A7%86%E9%A2%91/7.mp4",
+	//		},
+	//		{
+	//			"timeline": "33.2-37.4",
+	//			"url":      "https://ark-auto-2127201628-cn-beijing-default.tos-cn-beijing.volces.com/%E8%A7%86%E9%A2%91/8.mp4",
+	//		},
+	//		{
+	//			"timeline": "37.4-38.9",
+	//			"url":      "https://ark-auto-2127201628-cn-beijing-default.tos-cn-beijing.volces.com/%E8%A7%86%E9%A2%91/9.mp4",
+	//		},
+	//		{
+	//			"timeline": "38.9-57.9",
+	//			"url":      "https://ark-auto-2127201628-cn-beijing-default.tos-cn-beijing.volces.com/%E8%A7%86%E9%A2%91/10.mp4",
+	//		},
+	//	}
+	//	outputRes = append(outputRes, node.NodeFormField{
+	//		Field: fmt.Sprintf("segments"),
+	//		Value: a,
+	//		Label: fmt.Sprintf("segments"),
+	//		Type:  "string",
+	//	})
+	//	outputRes = append(outputRes, node.NodeFormField{
+	//		Field: fmt.Sprintf("audioUrl"),
+	//		Value: "http://116.204.74.41:9000/tenantid-94/2026-06-11/9915351c-55b9-46d8-b783-3815126b.m4a",
+	//		Label: fmt.Sprintf("audioUrl"),
+	//		Type:  "string",
+	//	})
+	//}
 	if err != nil {
 		return nil, err
 	}
@@ -509,12 +599,6 @@ func MergeLambda(ctx context.Context, input any) (res any, err error) {
 	// 1. 把所有节点输出拍平成 字段名->内容 的map
 	dataMap := make(map[string]node.NodeFormField)
 	_, outputMap, _ := GetNodeContextContent(nodeInput.Global, nodeInput.Config)
-	//for _, valueAny := range outputMap {
-	//	field := node.NodeFormField{}
-	//	if field, ok = valueAny.(node.NodeFormField); ok {
-	//		dataMap[field.Field] = field
-	//	}
-	//}
 	for _, field := range outputMap {
 		dataMap[field.Field] = field
 	}
@@ -666,7 +750,7 @@ func SummaryLambda(ctx context.Context, input any) (any, error) {
 					// 生成 毫秒时间戳 作为 KEY
 					timeKey := strconv.FormatInt(time.Now().UnixMilli(), 10)
 					item := make(map[string]interface{})
-					item[timeKey] = ProcessPath(ctx, gconv.String(field.Value))
+					item[timeKey] = field.Value
 					summaryResult = append(summaryResult, item)
 				}
 			}
@@ -684,28 +768,10 @@ 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)

@@ -583,7 +583,7 @@ func ImgNode(ctx context.Context, nodeInput *flowDto.NodeExecutionInput, skillNa
 			outputRes = append(outputRes, node.NodeFormField{
 				Field: fmt.Sprintf("img_url:%v:%d", k, i),
 				Value: v,
-				Label: fmt.Sprintf("img_url%v:%d", k, i),
+				Label: fmt.Sprintf("图片内容%v:%d", k, i),
 				Type:  "string",
 			})
 		}
@@ -674,7 +674,7 @@ func AudioOptimizeNode(ctx context.Context, nodeInput *flowDto.NodeExecutionInpu
 			outputRes = append(outputRes, node.NodeFormField{
 				Field: fmt.Sprintf("audio_url:%v:%d", k, i),
 				Value: v,
-				Label: fmt.Sprintf("audio_url:%v:%d", k, i),
+				Label: fmt.Sprintf("音频内容:%v:%d", k, i),
 				Type:  "string",
 			})
 		}
@@ -683,29 +683,24 @@ func AudioOptimizeNode(ctx context.Context, nodeInput *flowDto.NodeExecutionInpu
 	return outputRes, nil
 }

-// splitTextByPunct 按中文标点分割句子，同时保留标点在分段内
-// 例如："这个叫高血压调理方，注意是根源调理不是临时缓解，"
-// 会变成：["这个叫高血压调理方，", "注意是根源调理不是临时缓解，"]
 func splitTextByPunct(raw string) []string {
-	// 匹配中文标点并保留在文本中，按标点位置切分
-	re := regexp.MustCompile(`[，。；！？]`)
-	// 先找到所有标点的位置
-	indexes := re.FindAllStringIndex(raw, -1)
-	if len(indexes) == 0 {
-		return []string{raw}
-	}
-
+	// 按标点切分+拼接标点
+	slice := regexp.MustCompile(`([，。；！？])`).Split(raw, -1)
 	var res []string
-	prev := 0
-	for _, idx := range indexes {
-		end := idx[1] // 标点的结束位置
-		seg := raw[prev:end]
-		res = append(res, seg)
-		prev = end
+	var builder strings.Builder
+	for idx, s := range slice {
+		if s == "" {
+			continue
+		}
+		builder.WriteString(s)
+		// 偶数位是分隔标点（split后规律：文本、标点、文本、标点...）
+		if idx%2 == 1 {
+			res = append(res, builder.String())
+			builder.Reset()
+		}
 	}
-	// 处理最后一段没有标点的文本
-	if prev < len(raw) {
-		res = append(res, raw[prev:])
+	if builder.Len() > 0 {
+		res = append(res, builder.String())
 	}
 	return res
 }
@@ -713,53 +708,46 @@ func splitTextByPunct(raw string) []string {
 // BuildSubtitles 核心工具：单个sentence生成多条subtitle
 func BuildSubtitles(sents *[]flowDto.Sentence) ([]flowDto.Subtitle, error) {
 	var subtitles []flowDto.Subtitle
-
 	for _, sent := range *sents {
-		// 1. 先按标点把文本拆成多个片段（保留标点）
 		segList := splitTextByPunct(sent.Text)
 		if len(segList) == 0 {
-			continue
+			return nil, nil
 		}

+		var subs []flowDto.Subtitle
 		wordIdx := 0
 		allWords := sent.Words
-		// 2. 遍历每个文本片段，匹配对应的Words
+
 		for _, seg := range segList {
-			// 去除文本片段的标点，方便和Word.Word拼接内容匹配
-			segClean := strings.ReplaceAll(seg, "，", "")
-			segClean = strings.ReplaceAll(segClean, "。", "")
-			segClean = strings.ReplaceAll(segClean, "；", "")
-			segClean = strings.ReplaceAll(segClean, "！", "")
-			segClean = strings.ReplaceAll(segClean, "？", "")
-
 			var collectWords []flowDto.Word
-			var currentText strings.Builder
-
-			// 收集Word直到拼接内容覆盖当前分段
-			for wordIdx < len(allWords) {
+			currentText := ""
+			// 循环取 word，直到拼接内容 包含/匹配 seg
+			for {
+				if wordIdx >= len(allWords) {
+					break
+				}
 				word := allWords[wordIdx]
-				currentText.WriteString(word.Word)
+				currentText += word.Word
 				collectWords = append(collectWords, word)
 				wordIdx++

-				// 当拼接的文本包含当前分段的纯文本时，停止收集
-				if strings.Contains(currentText.String(), segClean) {
+				// 只要包含分段文本，就认为匹配（无视末尾标点差异）
+				if strings.Contains(currentText, seg) {
 					break
 				}
 			}
-
 			if len(collectWords) == 0 {
 				continue
 			}
-
-			// 3. 生成字幕（时间戳取首尾Word的时间）
+			// 生成字幕
 			sub := flowDto.Subtitle{
 				Start: collectWords[0].StartTime,
 				End:   collectWords[len(collectWords)-1].EndTime,
-				Text:  segClean,
+				Text:  seg,
 			}
-			subtitles = append(subtitles, sub)
+			subs = append(subs, sub)
 		}
+		subtitles = append(subtitles, subs...)
 	}

 	return subtitles, nil
@@ -788,52 +776,15 @@ func VideoOptimizeNode(ctx context.Context, nodeInput *flowDto.NodeExecutionInpu
 }

 func DataConversionNode(ctx context.Context, nodeInput *flowDto.NodeExecutionInput, skillName string, form []map[string]any, userForm []map[string]any) ([]node.NodeFormField, error) {
-	if strings.Contains(nodeInput.Config.Name, "字幕") || strings.Contains(nodeInput.Config.Name, "视频") {
-		jsonStr := ``
-		outputRes := make([]node.NodeFormField, 0)
-		for _, field := range nodeInput.Config.OutputConfig {
-			if strings.Contains(nodeInput.Config.Name, "视频") {
-				for _, item := range nodeInput.Global.ExecutedNodes {
-					refNode, ok := nodeInput.Global.ConfigMap[item.NodeId]
-					if !ok {
-						continue
-					}
-					for _, v := range refNode.FormConfig {
-						if v.Field == field.Value {
-							jsonStr, _ = sjson.Set(jsonStr, field.Field, v.Value)
-						}
-					}
-				}
-			} else {
-				jsonStr, _ = sjson.Set(jsonStr, field.Field, field.Value)
-			}
-		}
-		if strings.Contains(nodeInput.Config.Name, "视频") {
-			outputRes = append(outputRes, node.NodeFormField{
-				Field: fmt.Sprintf("data_content"),
-				Value: gconv.Map(jsonStr),
-				Label: fmt.Sprintf("data_content"),
-				Type:  "string",
-			})
-		} else {
-			outputRes = append(outputRes, node.NodeFormField{
-				Field: fmt.Sprintf("data_conversion"),
-				Value: gconv.Map(jsonStr),
-				Label: fmt.Sprintf("data_conversion"),
-				Type:  "string",
-			})
-		}
-		return outputRes, nil
-	}
-
 	jsonStr := ``
-	jsonVal := ""
+	jsonVal := "输出字段规范："
 	for _, field := range nodeInput.Config.OutputConfig {
 		jsonStr, _ = sjson.Set(jsonStr, field.Field, "")
-		//jsonVal += fmt.Sprintf("%s:%s;", field.Field, field.Value)
+		jsonVal += fmt.Sprintf("%s:%s;", field.Field, field.Value)
 	}
-	jsonVal += fmt.Sprintf("输出字段规范：%v", jsonStr)
+	jsonVal += fmt.Sprintf("输出模板结构,仅修改每个字段对应数值：%s", jsonStr)
 	nodeInput.Config.PromptContent = fmt.Sprintf("%s;%s", nodeInput.Config.PromptContent, jsonVal)
+
 	mapTaskResult, err := GetModelResult(ctx, "", nodeInput, skillName, form, userForm)
 	if err != nil {
 		return nil, err
@@ -898,7 +849,7 @@ func HttpNode(ctx context.Context, nodeInput *flowDto.NodeExecutionInput) ([]nod
 	}

 	// 构建请求参数
-	newBody := BuildNestedJson(body, nodeInput.Global)
+	newBody := BuildNestedJson(body, nodeInput.Global.ConfigMap)
 	// 1. 自己生成唯一 taskId（不用前端给）
 	taskId := "my_task_" + uuid.New().String() // 自己生成唯一ID
 	if responseType == "callback" {
@@ -924,8 +875,6 @@ func HttpNode(ctx context.Context, nodeInput *flowDto.NodeExecutionInput) ([]nod
 		return nil, err
 	}

-	var e = ""
-
 	finalResult := make(map[string]any)
 	if responseType == "sync" {
 		httpResultJson := gconv.String(rawHttpResult)
@@ -935,7 +884,6 @@ func HttpNode(ctx context.Context, nodeInput *flowDto.NodeExecutionInput) ([]nod
 				finalResult[key] = gjson.Get(httpResultJson, path).Value()
 			}
 		}
-		e = fmt.Sprintf("%v", httpResultJson)
 	}
 	if responseType == "callback" {
 		var waitResult any
@@ -962,16 +910,11 @@ func HttpNode(ctx context.Context, nodeInput *flowDto.NodeExecutionInput) ([]nod
 				}
 			}
 		}
-		e = fmt.Sprintf("%v", bodyStr)
 	}
 	if responseType == "pull" {

 	}

-	if g.IsEmpty(finalResult) {
-		return nil, fmt.Errorf("http请求异常，返回结果为空：%v", e)
-	}
-
 	outputRes := make([]node.NodeFormField, 0)
 	for i, item := range finalResult {
 		if nodeInput.Config.IsSaveFile {
@@ -1058,7 +1001,7 @@ func GetNodeContextContent(execInput *flowDto.FlowExecutionInput, nodeEntity *en
 				// 取指定字段
 				for _, f := range source.Field {
 					for _, v := range refNode.FormConfig {
-						if v.Label == f {
+						if strings.Contains(v.Label, f) {
 							input = append(input, v)
 						}
 					}
@@ -1081,146 +1024,3 @@ func GetNodeContextContent(execInput *flowDto.FlowExecutionInput, nodeEntity *en
 	}
 	return input, output, model
 }
-
-//func BuildParam(nodeInput *flowDto.NodeExecutionInput) (skillName string, resultFrom []map[string]any, resultUserFrom []map[string]any) {
-//	inputMap, outputMap, modelMap := GetNodeContextContent(nodeInput.Global, nodeInput.Config)
-//	var outputResult []node.NodeFormField
-//	for _, valueAny := range inputMap {
-//		if field, ok := valueAny.(node.NodeFormField); ok {
-//			outputResult = append(outputResult, field)
-//		}
-//	}
-//
-//	resultUserFrom = []map[string]any{}
-//	for _, valueAny := range outputMap {
-//		if field, ok := valueAny.(node.NodeFormField); ok {
-//			if !strings.Contains(field.Field, "text_url") && !strings.Contains(field.Field, "img_url") {
-//				if strings.Contains(field.Field, "text_content") {
-//					field.Value = StripHtmlTags(gconv.String(field.Value))
-//				}
-//				resultUserFrom = append(resultUserFrom, map[string]any{
-//					field.Label: field.Value,
-//				})
-//			}
-//		}
-//	}
-//	for _, valueAny := range modelMap {
-//		if field, ok := valueAny.(node.NodeFormField); ok {
-//			outputResult = append(outputResult, field)
-//		}
-//	}
-//	//if !nodeInput.Global.IsDialogue {
-//	for _, item := range outputResult {
-//		resultUserFrom = append(resultUserFrom, map[string]any{
-//			item.Label: item.Value,
-//		})
-//	}
-//	for _, item := range nodeInput.Config.FormConfig {
-//		resultUserFrom = append(resultUserFrom, map[string]any{
-//			item.Label: item.Value,
-//		})
-//	}
-//	//}
-//	if !g.IsEmpty(nodeInput.Global.Desc) {
-//		resultUserFrom = append(resultUserFrom, map[string]any{
-//			"desc": nodeInput.Global.Desc,
-//		})
-//	}
-//
-//	resultFrom = []map[string]any{}
-//	for _, item := range nodeInput.Config.ModelConfig.ModelForm {
-//		if g.IsEmpty(item.Value) {
-//			continue
-//		}
-//		resultFrom = append(resultFrom, map[string]any{
-//			item.Label: item.Value,
-//		})
-//	}
-//	skillName = nodeInput.Config.SkillName
-//	if g.IsEmpty(nodeInput.Config.SkillName) {
-//		skillName = nodeInput.Global.SkillName
-//	}
-//
-//	return skillName, resultFrom, resultUserFrom
-//}
-//
-//func GetNodeContextContent(execInput *flowDto.FlowExecutionInput, nodeEntity *entity.FlowNode) (map[string]any, map[string]any, map[string]any) {
-//	input := make(map[string]any)
-//	output := make(map[string]any)
-//	model := make(map[string]any)
-//	// 1. 有引用 → 取引用节点的字段值
-//	if len(nodeEntity.InputSource) > 0 {
-//		for _, source := range nodeEntity.InputSource {
-//			refNodeID := source.NodeId
-//			fields := source.Field
-//
-//			refNode, ok := execInput.ConfigMap[refNodeID]
-//			if !ok {
-//				continue
-//			}
-//
-//			inputMap := buildInputMap(refNode)
-//			outputMap := mergeOutput(refNode.OutputResult)
-//			modelMap := mergeModel(refNode.ModelConfig)
-//			if len(fields) > 0 {
-//				// 取指定字段
-//				for _, f := range fields {
-//					if v, ok := inputMap[f]; ok {
-//						input[f] = v
-//					}
-//					if v, ok := modelMap[f]; ok {
-//						model[f] = v
-//					}
-//					for k, v := range outputMap {
-//						if strings.Contains(k, f) {
-//							model[k] = v
-//						}
-//					}
-//				}
-//			} else {
-//				// 取全部
-//				if refNode.NodeCode != node.NodeTypeHttp {
-//					for k, v := range inputMap {
-//						input[k] = v
-//					}
-//				}
-//				for k, v := range modelMap {
-//					model[k] = v
-//				}
-//			}
-//		}
-//	}
-//	return input, output, model
-//}
-//
-//// buildInputMap 从 FormConfig 构造输入map
-//func buildInputMap(node *entity.FlowNode) map[string]any {
-//	m := make(map[string]any)
-//	for _, item := range node.FormConfig {
-//		m[item.Label] = item
-//	}
-//	return m
-//}
-//
-//// mergeOutput 合并节点输出 []map → 单map
-//func mergeOutput(output []node.NodeFormField) map[string]any {
-//	m := make(map[string]any)
-//	for _, item := range output {
-//		m[item.Label] = item
-//	}
-//	return m
-//}
-//
-//// mergeOutput 合并节点输出 []map → 单map
-//func mergeModel(output node.ModelItem) map[string]any {
-//	m := make(map[string]any)
-//	// 遍历 output.ModelForm 里的每一个 key 和原始值
-//	for _, rawValue := range output.ModelForm {
-//		if g.IsEmpty(rawValue.Value) {
-//			continue
-//		}
-//		// 包装成 { "value": 原始值 }
-//		m[rawValue.Label] = rawValue.Value
-//	}
-//	return m
-//}
@@ -9,25 +9,21 @@ import (
 	"ai-agent/workflow/model/entity"
 	"bytes"
 	"context"
-	"errors"
 	"fmt"
 	"io"
 	"mime/multipart"
 	"net/http"
 	"net/url"
-	"path"
 	"path/filepath"
 	"regexp"
 	"strconv"
 	"strings"
 	"sync"
-	"time"

 	commonHttp "gitea.redpowerfuture.com/red-future/common/http"
 	"gitea.redpowerfuture.com/red-future/common/utils"
 	"github.com/gogf/gf/v2/errors/gerror"
 	"github.com/gogf/gf/v2/frame/g"
-	"github.com/gogf/gf/v2/net/ghttp"
 	"github.com/gogf/gf/v2/util/gconv"
 	"github.com/tidwall/sjson"
 )
@@ -101,7 +97,12 @@ func GetModelInfo(ctx context.Context, req *flowDto.GetModelInfoReq) (res *flowD
 	return
 }

-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) {
+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) {
+	if !g.IsEmpty(promptContent) {
+		userForm = append(userForm, map[string]any{
+			"prompt": promptContent,
+		})
+	}
 	var callbackUrl = utils.GetCallbackURL(ctx, "/flow/execution/composeCallBack")
 	var consult = make([]flowDto.Consult, 0)
 	var collectFileUrls func(val any) (fullyConsumed bool)
@@ -142,27 +143,11 @@ func GetComposeResult(ctx context.Context, nodeExecutionId int64, buildType int,
 	}
 	var newUserForm []map[string]any
 	for _, m := range userForm {
-		// 先替换字段
-		if val, ok := m["audioDuration"]; ok {
-			delete(m, "audioDuration")
-			m["视频总时长"] = val
-		}
-		if val, ok := m["videoDuration"]; ok {
-			delete(m, "videoDuration")
-			m["视频总时长"] = val
-		}
-		// 收集待删除 key
-		var delKeys []string
 		for k, v := range m {
 			if collectFileUrls(v) {
-				delKeys = append(delKeys, k)
+				delete(m, k)
 			}
 		}
-		// 统一删除
-		for _, k := range delKeys {
-			delete(m, k)
-		}
-
 		if len(m) > 0 {
 			newUserForm = append(newUserForm, m)
 		}
@@ -184,60 +169,23 @@ func GetComposeResult(ctx context.Context, nodeExecutionId int64, buildType int,
 		Cause:       cause,
 		Form:        form,
 		UserForm:    newUserForm,
-		UserPrompt:  promptContent,
 		Consult:     consult,
 		SessionId:   sessionId,
 		NodeId:      nodeId,
 	}
-	msgRes := new(flowDto.ComposeMessagesRes)
-
-	// 1. 隔离上游取消（防止节点执行被中断时下游请求被 cancel）+ 设置独立超时
-	baseCtx := context.WithoutCancel(ctx)
-	postCtx, cancel := context.WithTimeout(baseCtx, 30*time.Minute)
-	defer cancel() // 必须释放，防止上下文泄露
-
-	// 2. 克隆 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
-	}
+	headers := make(map[string]string)
 	if r := g.RequestFromCtx(ctx); r != nil {
 		for k, v := range r.Request.Header {
 			if len(v) > 0 {
-				client.SetHeader(k, v[0])
+				headers[k] = v[0]
 			}
 		}
 	}
-	resp, err := client.ContentJson().Post(postCtx, "prompts-core/prompt/composeMessages", &msgReq)
+	msgRes := new(flowDto.ComposeMessagesRes)
+	err = commonHttp.Post(ctx, "prompts-core/prompt/composeMessages", headers, msgRes, &msgReq)
 	if err != nil {
 		return
 	}
-	defer resp.Close()
-	result, err := io.ReadAll(resp.Body)
-	if err != nil {
-		return nil, fmt.Errorf("读取composeMessages响应失败: %w", err)
-	}
-	// 统一处理内部API响应格式：{code:200,message:"",data:{...}}
-	resultStrut := &ghttp.DefaultHandlerResponse{}
-
-	if err = gconv.Struct(result, &resultStrut); err != nil { // 修复：增加err检查
-		return nil, fmt.Errorf("响应解析失败: " + err.Error())
-	}
-
-	// 添加调试日志：打印解析后的结构
-	g.Log().Debugf(ctx, "[HTTP] 解析后结构: Code=%d, Message=%s, Data类型=%T, Data值=%+v",
-		resultStrut.Code, resultStrut.Message, resultStrut.Data, resultStrut.Data)
-
-	if resultStrut.Code == 200 || resultStrut.Code == 0 {
-		if err = gconv.Struct(resultStrut.Data, &msgRes); err != nil { // 修复：增加err检查
-			return nil, fmt.Errorf("数据解析失败: " + err.Error())
-		}
-		// 添加调试日志：打印最终的target
-		g.Log().Debugf(ctx, "[HTTP] 最终target: %+v", &msgRes)
-	} else {
-		err = errors.New(resultStrut.Message)
-	}
 	if g.IsEmpty(msgRes.TaskId) {
 		return nil, fmt.Errorf("msg is empty")
 	}
@@ -249,19 +197,18 @@ func GetComposeResult(ctx context.Context, nodeExecutionId int64, buildType int,
 	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, nodeExecutionId int64, epicycleId int64, model string, content map[string]any) (map[string]any, error) {
+func CreateGatewayTask(ctx context.Context, 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, nodeExecutionId, taskId)
+	return waitGatewayResult(ctx, taskId)
 }

 // createGatewayTaskOnly creates a gateway task and returns the taskId only
@@ -276,63 +223,29 @@ func createGatewayTaskOnly(ctx context.Context, epicycleId int64, model string,
 		EpicycleId:     epicycleId,
 	}

-	res := new(flowDto.ModelGatewayRes)
-
-	// 1. 隔离上游取消（防止节点执行被中断时下游请求被 cancel）+ 设置独立超时
-	baseCtx := context.WithoutCancel(ctx)
-	postCtx, cancel := context.WithTimeout(baseCtx, 30*time.Minute)
-	defer cancel() // 必须释放，防止上下文泄露
-
-	// 2. 克隆 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
-	}
+	headers := make(map[string]string)
 	if r := g.RequestFromCtx(ctx); r != nil {
 		for k, v := range r.Request.Header {
 			if len(v) > 0 {
-				client.SetHeader(k, v[0])
+				headers[k] = v[0]
 			}
 		}
 	}
-	rpcResp, err := client.ContentJson().Post(postCtx, "model-gateway/task/createTask", &req)
+
+	res := new(flowDto.ModelGatewayRes)
+	err := commonHttp.Post(ctx, "model-gateway/task/createTask", headers, res, &req)
 	if err != nil {
 		return "", err
 	}
-	defer rpcResp.Close()
-	result, err := io.ReadAll(rpcResp.Body)
-	if err != nil {
-		return "", fmt.Errorf("读取createTask响应失败: %w", err)
-	}
-	// 统一处理内部API响应格式：{code:200,message:"",data:{...}}
-	resultStrut := &ghttp.DefaultHandlerResponse{}
-
-	if err = gconv.Struct(result, &resultStrut); err != nil { // 修复：增加err检查
-		return "", fmt.Errorf("响应解析失败: " + err.Error())
-	}
-
-	// 添加调试日志：打印解析后的结构
-	g.Log().Debugf(ctx, "[HTTP] 解析后结构: Code=%d, Message=%s, Data类型=%T, Data值=%+v",
-		resultStrut.Code, resultStrut.Message, resultStrut.Data, resultStrut.Data)
-
-	if resultStrut.Code == 200 || resultStrut.Code == 0 {
-		if err = gconv.Struct(resultStrut.Data, &res); err != nil { // 修复：增加err检查
-			return "", fmt.Errorf("数据解析失败: " + err.Error())
-		}
-		// 添加调试日志：打印最终的target
-		g.Log().Debugf(ctx, "[HTTP] 最终target: %+v", &res)
-	} else {
-		err = errors.New(resultStrut.Message)
-	}
 	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, nodeExecutionId int64, taskId string) (map[string]any, error) {
+func waitGatewayResult(ctx context.Context, taskId string) (map[string]any, error) {
 	waitRes, err := Wait(ctx, taskId)
 	if err != nil {
 		return nil, err
@@ -342,7 +255,6 @@ func waitGatewayResult(ctx context.Context, nodeExecutionId int64, taskId string
 	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)
 	}
@@ -358,23 +270,14 @@ func waitGatewayResult(ctx context.Context, nodeExecutionId int64, taskId string
 }

 // updateTokenCount updates the token count in node execution
-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 {
+func updateTokenCount(ctx context.Context, nodeExecutionId int64, responseField string, result map[string]any) {
+	if responseField == "" {
 		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,
-		TokenInfo: t,
+		Id:               nodeExecutionId,
+		CompletionTokens: gconv.Int(result[responseField]),
+		TotalTokens:      gconv.Int(result[responseField]),
 	})
 }

@@ -387,23 +290,8 @@ func GetModelResult(ctx context.Context, sessionId string, nodeInput *flowDto.No
 	if !nodeInput.Global.IsDialogue {
 		sessionId = ""
 	}
-	needSequential := false
-	for _, item := range userForm {
-		if g.NewVar(item).IsMap() {
-			valMap := gconv.Map(item)
-			for _, v := range valMap {
-				if g.NewVar(v).IsMap() {
-					vv := gconv.Map(v)
-					for kk, vvv := range vv {
-						if kk == "return_last_frame" {
-							needSequential = vvv.(bool)
-						}
-					}
-				}
-			}
-		}
-	}
-	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)
+
+	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)
 	if err != nil {
 		return nil, err
 	}
@@ -419,12 +307,13 @@ func GetModelResult(ctx context.Context, sessionId string, nodeInput *flowDto.No
 	mapTaskResult = make([]map[string]any, len(composeResult.Messages.Rounds))
 	var taskResultMap map[string]any

+	needSequential := false
 	if buildType == 1 {
 		if needSequential {
 			for idx, item := range composeResult.Messages.Rounds {
 				if !g.IsEmpty(taskResultMap) {
 					var set string
-					set, err = sjson.Set(gconv.String(item), modelInfo.Model.FirstFrame, gconv.String(taskResultMap["content"]))
+					set, err = sjson.Set(gconv.String(item), modelInfo.Model.LastFrame, gconv.String(taskResultMap[modelInfo.Model.ResponseBody]))
 					if err != nil {
 						return nil, err
 					}
@@ -432,7 +321,7 @@ func GetModelResult(ctx context.Context, sessionId string, nodeInput *flowDto.No
 				}

 				var taskResult map[string]any
-				taskResult, err = CreateGatewayTask(ctx, nodeInput.NodeExecutionId, composeResult.EpicycleId, nodeInput.Config.ModelConfig.ModelName, item)
+				taskResult, err = CreateGatewayTask(ctx, composeResult.EpicycleId, nodeInput.Config.ModelConfig.ModelName, item)
 				if err != nil {
 					return nil, err
 				}
@@ -441,7 +330,7 @@ func GetModelResult(ctx context.Context, sessionId string, nodeInput *flowDto.No
 				}

 				if nodeInput.Config.NodeCode == node.NodeTypeVideoModel {
-					ext := GetFileTypeByPath(gconv.String(taskResult["content"]))
+					ext := GetFileTypeByPath(gconv.String(taskResult[modelInfo.Model.ResponseBody]))
 					if ext == "image" {
 						taskResultMap = taskResult
 					} else {
@@ -452,34 +341,36 @@ 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))
+
+			for idx, item := range composeResult.Messages.Rounds {
+				taskId, err := createGatewayTaskOnly(ctx, composeResult.EpicycleId, nodeInput.Config.ModelConfig.ModelName, item)
+				if err != nil {
+					return nil, err
+				}
+				taskIdList[idx] = taskId
+			}
+
 			// 全局共享子上下文，实现一处报错全部终止
 			subCtx, globalCancel := context.WithCancel(ctx)
 			defer globalCancel() // 函数退出兜底释放

 			var wg sync.WaitGroup
-			errChan := make(chan error, len(composeResult.Messages.Rounds))
+			errChan := make(chan error, len(taskIdList))

 			// 加互斥锁保护结果map
 			var mu sync.Mutex

-			// 每个任务创建后立即启动等待协程：把「回调 vs Wait 注册」的竞争窗口从整个创建循环
-			// 压缩到微秒级，避免创建期间完成的回调被 Notify 静默丢弃导致 Wait 永久阻塞
-			for idx, item := range composeResult.Messages.Rounds {
-				taskId, err := createGatewayTaskOnly(ctx, composeResult.EpicycleId, nodeInput.Config.ModelConfig.ModelName, item)
-				if err != nil {
-					globalCancel() // 取消已启动的等待协程，避免泄漏
-					return nil, err
-				}
-
+			for idx, taskId := range taskIdList {
 				wg.Add(1)

 				go func(idx int, taskId string) {
 					defer wg.Done()

-					taskResult, err := waitGatewayResult(subCtx, nodeInput.NodeExecutionId, taskId)
+					taskResult, err := waitGatewayResult(subCtx, taskId)
 					if err != nil {
 						errChan <- err
 						globalCancel() // 全局取消，所有协程收到ctx取消信号快速退出
@@ -491,7 +382,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)
 			}

@@ -512,19 +403,16 @@ 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)
 		}
 	}
+
 	return mapTaskResult, nil
 }

-func BuildNestedJson(body g.Map, global *flowDto.FlowExecutionInput) g.Map {
+func BuildNestedJson(body g.Map, mockConfigMap map[string]*entity.FlowNode) g.Map {
 	jsonStr := "{}"
 	for originKey, originItem := range body {
-		if originKey == "templates" && !g.IsEmpty(global.Templates) {
-			jsonStr, _ = sjson.Set(jsonStr, originKey, global.Templates)
-			continue
-		}
 		bodyItemMap := gconv.Map(originItem)
 		val := bodyItemMap["value"]
 		if v, ok := bodyItemMap["value"]; ok {
@@ -535,7 +423,7 @@ func BuildNestedJson(body g.Map, global *flowDto.FlowExecutionInput) g.Map {
 			valMap := gconv.Map(val)
 			nodeId := gconv.String(valMap["nodeId"])
 			fieldName := gconv.String(valMap["field"])
-			if configValue, ok := global.ConfigMap[nodeId]; ok {
+			if configValue, ok := mockConfigMap[nodeId]; ok {
 				if !g.IsEmpty(configValue.OutputResult) {
 					for _, v := range configValue.OutputResult {
 						if strings.Contains(v.Field, fieldName) {
@@ -711,56 +599,6 @@ 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
-}
-
-// ProcessPath 处理请求路径
-// 1. 判断是否为合法完整请求路径(以/开头)
-// 2. 包含 tenantId-1 则截断其及前面所有内容
-func ProcessPath(ctx context.Context, path string) string {
-	// 判断是否是完整请求路径：以 / 开头
-	if strings.HasPrefix(path, "/") {
-		return path
-	}
-
-	target, err := utils.GetBucketName(ctx)
-	if err != nil {
-		return path
-	}
-	idx := strings.Index(path, target)
-	if idx == -1 {
-		// 不包含目标字符串，原样返回
-		return path
-	}
-
-	// 截取 tenantId-1 后面的内容
-	newPath := path[idx+len(target):]
-	return newPath
-}
-
 func BuildText(text string) string {
 	// 生成单条HTML
 	var htmlBuilder strings.Builder
@@ -67,7 +67,6 @@ func (s *nodeLibraryService) GetNodeLibrary(ctx context.Context, req *nodeDto.Wo
 					NodeCode:     node.NodeTypeVideoModel,
 					NodeName:     node.NodeNameVideoModel,
 					ModelType:    node.ModelTypeVideo,
-					PatchLayout:  true,
 					SkillOption:  false,
 					PromptOption: true,
 					IsSaveFile:   true,
@@ -94,36 +93,23 @@ func (s *nodeLibraryService) GetNodeLibrary(ctx context.Context, req *nodeDto.Wo
 					FormConfig:   []node.NodeFormField{},
 					ModelConfig:  []node.ModelItem{},
 				},
-				//{
-				//	NodeCode:    node.NodeTypeSenseOptimizeModel,
-				//	NodeName:    node.NodeNameSenseOptimizeModel,
-				//	ModelType:   node.ModelTypeText,
-				//	SkillOption: false,
-				//	FormConfig:  []node.NodeFormField{},
-				//	ModelConfig: []node.ModelItem{},
-				//},
-				//{
-				//	NodeCode:    node.NodeTypeStoryOptimizeModel,
-				//	NodeName:    node.NodeNameStoryOptimizeModel,
-				//	ModelType:   node.ModelTypeText,
-				//	SkillOption: false,
-				//	FormConfig:  []node.NodeFormField{},
-				//	ModelConfig: []node.ModelItem{},
-				//},
-				//{
-				//	NodeCode:    node.NodeTypeScriptOptimizeModel,
-				//	NodeName:    node.NodeNameScriptOptimizeModel,
-				//	ModelType:   node.ModelTypeText,
-				//	SkillOption: false,
-				//	FormConfig:  []node.NodeFormField{},
-				//	ModelConfig: []node.ModelItem{},
-				//},
 			},
 		},
 		{
 			Group: node.NodeGroupBase,
 			Label: node.NodeGroupNameBase,
 			Items: []node.NodeItem{
+				{
+					NodeCode:     node.NodeTypeSubFlow,
+					NodeName:     node.NodeSubFlow,
+					SkillOption:  false,
+					PromptOption: false,
+					IsSaveFile:   false,
+					FormConfig: []node.NodeFormField{
+						{Field: "maxConcurrency", Label: "最大并发数", Type: "input", Required: true},
+					},
+					ModelConfig: []node.ModelItem{},
+				},
 				{
 					NodeCode:     node.NodeTypeDataConversionModel,
 					NodeName:     node.NodeNameDataConversionModel,
@@ -360,10 +346,7 @@ func (s *nodeLibraryService) GetNodeLibrary(ctx context.Context, req *nodeDto.Wo
 				item.NodeCode == node.NodeTypeVideoModel ||
 				item.NodeCode == node.NodeTypeAudioModel ||
 				item.NodeCode == node.NodeTypeBatchModel ||
-				item.NodeCode == node.NodeTypeDataConversionModel ||
-				item.NodeCode == node.NodeTypeSenseOptimizeModel ||
-				item.NodeCode == node.NodeTypeStoryOptimizeModel ||
-				item.NodeCode == node.NodeTypeScriptOptimizeModel {
+				item.NodeCode == node.NodeTypeDataConversionModel {
 				item.ModelConfig = append(item.ModelConfig, node.ModelItem{
 					ModelName: "自定义",
 				})
Author	SHA1	Message	Date
19904408334	bbd62c082d	feat: 添加子流程节点与重构工作流图构建 - 新增子流程节点(SubFlow)及SubFlowConfig配置结构 - 拆分BuildGraph与BuildGraphFromFlowContent，支持独立构建图与编译 - 移除ModelCallbackReq中多余字段，移除Intent节点注册 - 优化节点执行逻辑：调整OSS上传时机及变量引用 - 更新本地开发配置及依赖版本	2026-06-24 09:30:55 +08:00
19904408334	cfad80b359	Merge branch '2026-06-10-video' of http://116.204.74.41:3000/red-future/ai-agent into sub-workflow # Conflicts: # config.yml	2026-06-24 09:25:40 +08:00
19904408334	f28ca0a50a	feat: 添加子流程节点与重构工作流图构建 - 新增子流程节点(SubFlow)及SubFlowConfig配置结构 - 拆分BuildGraph与BuildGraphFromFlowContent，支持独立构建图与编译 - 移除ModelCallbackReq中多余字段，移除Intent节点注册 - 优化节点执行逻辑：调整OSS上传时机及变量引用 - 更新本地开发配置及依赖版本	2026-06-24 09:24:53 +08:00
admin	db694c298c	封版	2026-06-22 09:26:53 +08:00
19904408334	695c00aed5	refactor: 重构工作流执行图构建与节点上下文处理	2026-06-16 13:26:27 +08:00
19904408334	6a249af7fa	fix: 移除优化类节点多余的自定义模型配置	2026-06-16 13:08:41 +08:00
19904408334	a6e98914e8	chore: 更新依赖包导入路径	2026-06-16 13:05:40 +08:00