fix: 修正流程执行列表查询过滤逻辑

先将日期倒序排序并按展示顺序统计各类型总数，确保倒序编号与前端展示一致；同时修正日期格式为 Y-m-d。

.dockerignore

@@ -0,0 +1 @@
+.git

.gitea/workflows/dev-deploy.yaml

@@ -0,0 +1,37 @@
+# ============================================================
+# 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

Dockerfile

@@ -1,7 +1,8 @@
 # 阶段1: 构建
 FROM golang:alpine AS builder
 
-RUN apk add --no-cache git ca-certificates tzdata
+RUN sed -i 's/dl-cdn.alpinelinux.org/mirrors.aliyun.com/g' /etc/apk/repositories && \
+      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

config.yml

@@ -64,5 +64,8 @@ consul:
 jaeger:
   addr: 192.168.0.83:4318
 
-# 文件上传服务地址，与oss模块minio中的endpoint一致
-filePrefix: "http://192.168.0.83:9000"
+# 文件上传服务地址，cdn访问地址
+filePrefix: "http://cdn.redpowerfuture.com"
+
+# 文件上传服务地址，minio内网访问地址
+minioPrefix: "http://192.168.0.83:9000"

deploy/k8s/ai-agent.yaml

@@ -0,0 +1,46 @@
+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

go.mod

@@ -3,15 +3,13 @@ module ai-agent
 go 1.26.0
 
 require (
-	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
+	gitea.redpowerfuture.com/red-future/common v0.0.29
+	github.com/cloudwego/eino v0.9.12
 	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
@@ -23,7 +21,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.4 // indirect
+	github.com/bytedance/gopkg v0.1.3 // 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
@@ -31,7 +29,6 @@ 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
@@ -86,7 +83,6 @@ 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

go.sum

@@ -1,8 +1,6 @@
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
-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=
+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=
 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=
@@ -38,8 +36,6 @@ 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=
@@ -60,16 +56,12 @@ 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.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 v0.9.12 h1:mHAMo5k7GdvnVD8Lc2sLyfpkxEm0S/y3PkEMhsSYt78=
+github.com/cloudwego/eino v0.9.12/go.mod h1:OBD1mrkfkt/pJa4rkg1P0VnaMeOVl7l8IAdEqY//3IQ=
 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=

update.sql

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

workflow/consts/node/node_template.go

@@ -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,25 +48,26 @@ type NodeType string
 
 const (
 	// 组件
-	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"
+	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"
 	// 基础
-	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"
+	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"
 	// 自定义
 	NodeTypeCustomNode NodeType = "custom_node"
-	// 系统
-	NodeTypeSystemSum NodeType = "system_sum"
 )
 
 const (
@@ -105,6 +106,7 @@ 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"`

workflow/controller/flow/flow_execution_controller.go

@@ -38,3 +38,17 @@ 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
+}

workflow/dao/flow/flow_execution_dao.go

@@ -7,6 +7,7 @@ import (
 	"context"
 
 	"gitea.redpowerfuture.com/red-future/common/db/gfdb"
+	"github.com/gogf/gf/v2/frame/g"
 	"github.com/gogf/gf/v2/util/gconv"
 )
 
@@ -49,8 +50,17 @@ 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).OmitEmpty()
+	model := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameFlowExecution).Fields(fields)
 	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))

workflow/dao/node/node_execution_dao.go

@@ -86,6 +86,7 @@ func (d *nodeExecutionDao) Get(ctx context.Context, req *nodeDto.GetNodeExecutio
 func (d *nodeExecutionDao) ListByFlowExecutionId(ctx context.Context, req *nodeDto.ListNodeExecutionByFlowReq, fields ...string) (res []*entity.NodeExecution, total int, err error) {
 	model := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameNodeExecution).NoTenantId(ctx).Fields(fields).OmitEmpty()
 	model.Where(entity.NodeExecutionCol.FlowExecutionId, req.FlowExecutionId)
+	model.Where(entity.NodeExecutionCol.NodeGroupId, req.NodeGroupId)
 	model.OrderAsc(entity.NodeExecutionCol.CreatedAt)
 	if req.Page != nil {
 		model.Page(int(req.Page.PageNum), int(req.Page.PageSize))

workflow/model/dto/flow/flow_execution_dto.go

@@ -30,6 +30,7 @@ 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"`
@@ -49,6 +50,7 @@ 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"`
@@ -74,6 +76,7 @@ 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"`
@@ -122,17 +125,19 @@ type ComposeCallbackReq struct {
 		TotalRounds int              `json:"total_rounds"` // 总轮数
 		Rounds      []map[string]any `json:"rounds"`       // 每轮详情（动态类型）
 	} `json:"messages,omitempty"`
-	EpicycleId int64  `json:"epicycleId"`
-	ErrorMsg   string `json:"errorMsg,omitempty"`
+	EpicycleId  int64            `json:"epicycleId"`
+	ErrorMsg    string           `json:"errorMsg,omitempty"`
+	BillingData []map[string]any `json:"billing_data"`
 }
 
 type ModelCallbackReq struct {
-	g.Meta   `path:"/modelCallback" method:"post" tags:"提示词处理" summary:"model-gateway 回调" dc:"model-gateway 成功后 GET 回调：callbackUrl/{bizName}"`
-	TaskId   string `p:"task_id" json:"task_id" v:"required#task_id不能为空" dc:"网关任务ID"`
-	State    int    `p:"state" json:"state" dc:"网关任务状态"`
-	OssFile  string `p:"oss_file" json:"oss_file" dc:"结果文件地址"`
-	FileType string `p:"file_type" json:"file_type" dc:"结果文件类型"`
-	ErrorMsg string `json:"error_msg"`
+	g.Meta      `path:"/modelCallback" method:"post" tags:"提示词处理" summary:"model-gateway 回调" dc:"model-gateway 成功后 GET 回调：callbackUrl/{bizName}"`
+	TaskId      string           `p:"task_id" json:"task_id" v:"required#task_id不能为空" dc:"网关任务ID"`
+	State       int              `p:"state" json:"state" dc:"网关任务状态"`
+	OssFile     string           `p:"oss_file" json:"oss_file" dc:"结果文件地址"`
+	FileType    string           `p:"file_type" json:"file_type" dc:"结果文件类型"`
+	ErrorMsg    string           `json:"error_msg"`
+	BillingData []map[string]any `json:"billing_data"`
 }
 
 type VideoCallbackReq struct {
@@ -180,6 +185,7 @@ 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"`
@@ -209,6 +215,7 @@ 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 {
@@ -216,13 +223,33 @@ 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-失败"`
-	OutputParams []map[string]interface{} `json:"outputParams" description:"输出参数"`
-	ErrorMessage string                   `json:"errorMessage" description:"错误信息"`
-	TraceId      string                   `json:"traceId" description:"跟踪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-失败"`
+	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不能为空"`
 }
 
 type GetFlowExecutionReq struct {
@@ -232,11 +259,20 @@ 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:"任务列表"`
 
-	Page    *beans.Page `json:"page"`
-	Creator string      `json:"creator"`
+	*beans.Page `json:"page"`
+	Creator     string `json:"creator"`
+	SessionDel  *bool  `json:"sessionDel"`
+	ResultDel   *bool  `json:"resultDel"`
+	IsResult    bool   `json:"isResult"`
 }
 
 type ListFlowExecutionRes struct {
@@ -255,6 +291,7 @@ 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:"创建时间"`
@@ -263,22 +300,17 @@ 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:"创建日期"`
-	Flows      []FlowNode `json:"flows" description:"流程列表"`
+	CreateDate string       `json:"createDate" description:"创建日期"`
+	Items      []OutputItem `json:"items" description:"直接是结果项列表"`
 }
 
 type ListFlowExecutionTreeRes struct {

workflow/model/dto/node/node_execution_dto.go

@@ -38,6 +38,7 @@ type UpdateNodeExecutionReq struct {
 	PromptTokens     int                      `json:"promptTokens"`
 	CompletionTokens int                      `json:"completionTokens"`
 	TotalTokens      int                      `json:"totalTokens"`
+	TokenInfo        []map[string]any         `json:"tokenInfo"`
 	Status           node.NodeExecutionStatus `json:"status"`
 	DurationMs       int64                    `json:"durationMs"`
 	ErrorMessage     string                   `json:"errorMessage"`
@@ -60,6 +61,7 @@ type ListNodeExecutionByFlowReq struct {
 	g.Meta          `path:"/listByFlow" method:"get" tags:"节点执行记录" summary:"查询流程节点执行列表" dc:"查询指定流程执行下的所有节点执行记录"`
 	Page            *beans.Page `json:"page"`
 	FlowExecutionId int64       `json:"flowExecutionId" v:"required#流程执行ID不能为空"`
+	NodeGroupId     string      `json:"nodeGroupId"`
 }
 
 // NodeExecutionResp 节点执行记录响应

workflow/model/entity/flow_execution.go

@@ -22,6 +22,10 @@ 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 {
@@ -39,6 +43,10 @@ type flowExecutionCol struct {
 	TraceId         string
 	SessionId       string
 	TotalTokens     string
+	TotalFee        string
+	SessionDel      string
+	ResultDel       string
+	Extension       string
 }
 
 var FlowExecutionCol = flowExecutionCol{
@@ -56,4 +64,8 @@ var FlowExecutionCol = flowExecutionCol{
 	TraceId:         "trace_id",
 	SessionId:       "session_id",
 	TotalTokens:     "total_tokens",
+	TotalFee:        "total_fee",
+	SessionDel:      "session_del",
+	ResultDel:       "result_del",
+	Extension:       "extension",
 }

workflow/model/entity/flow_user.go

@@ -15,39 +15,26 @@ 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"`
-	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:"节点输出结果"`
+	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:"节点输出结果"`
 }
 
 type FlowNodeInputSource struct {
-	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
+	NodeId      string   `json:"nodeId"`
+	QuoteOutput bool     `json:"quoteOutput"`
+	Field       []string `json:"field"`
 }
 
 type FlowEdge struct {

workflow/model/entity/node_execution.go

@@ -22,6 +22,7 @@ type NodeExecution struct {
 	PromptTokens     int                      `orm:"prompt_tokens" json:"promptTokens" description:"提示词token消耗"`
 	CompletionTokens int                      `orm:"completion_tokens" json:"completionTokens" description:"补全token消耗"`
 	TotalTokens      int                      `orm:"total_tokens" json:"totalTokens" description:"总token消耗"`
+	TokenInfo        []map[string]interface{} `orm:"token_info" json:"tokenInfo" description:"token信息"`
 	Status           node.NodeExecutionStatus `orm:"status" json:"status" description:"执行状态:1-运行中,2-成功,3-失败,4-暂停,5-等待执行"`
 	DurationMs       int64                    `orm:"duration_ms" json:"durationMs" description:"执行时长（毫秒）"`
 	ErrorMessage     string                   `orm:"error_message" json:"errorMessage" description:"错误信息"`
@@ -40,6 +41,7 @@ type nodeExecutionCol struct {
 	PromptTokens     string
 	CompletionTokens string
 	TotalTokens      string
+	TokenInfo        string
 	Status           string
 	DurationMs       string
 	ErrorMessage     string
@@ -58,6 +60,7 @@ var NodeExecutionCol = nodeExecutionCol{
 	PromptTokens:     "prompt_tokens",
 	CompletionTokens: "completion_tokens",
 	TotalTokens:      "total_tokens",
+	TokenInfo:        "token_info",
 	Status:           "status",
 	DurationMs:       "duration_ms",
 	ErrorMessage:     "error_message",

workflow/service/flow/flow_execution_service.go

@@ -16,14 +16,12 @@ 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"
@@ -47,54 +45,94 @@ 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
 	}
 
-	// ===================== 核心修复：只统计【有数据】的执行记录，空的直接跳过 =====================
-
-	executionNumber := make(map[int64]int) // executionId -> 倒序编号（最新=1）
-
-	// 第一次遍历：只处理【有输出参数】的记录，统计并分配编号
-	var validList []*entity.FlowExecution // 只存有效（非空）记录
+	// 过滤出有有效输出的执行记录
+	var validList []*entity.FlowExecution
 	for _, execution := range list {
-		if g.IsEmpty(execution.OutputParams) {
-			continue // 空数据直接过滤，不参与编号、不展示
+		if !g.IsEmpty(execution.OutputParams) {
+			validList = append(validList, execution)
 		}
-		validList = append(validList, execution)
 	}
 
-	// 给有效记录分配【时间倒序编号】（最新=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)
-
-	// 遍历【有效数据】构建结构
+	// 1. 按日期归集，严格使用 Y-m-d 格式
+	dateMap := make(map[string][]flowDto.OutputItem)
 	for _, execution := range validList {
+		// 按要求使用 Y-m-d
 		createDate := execution.CreatedAt.Format("Y-m-d")
-		flowName := execution.FlowName
+		execID := execution.Id
 		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 {
@@ -102,98 +140,116 @@ 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
 		})
 
-		// 标号：相同类型递增，不同重置
-		suffixCount := make(map[string]int)
-		for idx := range tempItems {
-			item := &tempItems[idx]
-			val := item.Content
-			suffix := "内容"
-			ext := ""
-			ext = GetFileTypeByPath(val)
-			if ext == "image" {
-				suffix = "图片"
-			}
-			if ext == "video" {
-				suffix = "视频"
-			}
-			if ext == "audio" {
-				suffix = "音频"
-			}
-			if ext == "text" {
-				suffix = "文案"
-			}
-			if ext == "html" {
-				suffix = "HTML"
-			}
-			suffixCount[suffix]++
-			item.Type = ext
-			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,
-		})
+		dateMap[createDate] = append(dateMap[createDate], tempItems...)
 	}
 
-	// 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)
-		})
+	// ========== 修复编号乱序核心逻辑 ==========
+	// 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]
+	})
 
-		var flowNodes []flowDto.FlowNode
-		for _, w := range *wraps {
-			flowNodes = append(flowNodes, w.flowNode)
+	// 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
+	}
+	// ======================================
 
-		// ===================== 修复2：日期下没有流程，也过滤掉 =====================
-		if len(flowNodes) == 0 {
+	var tree []flowDto.DateNode
+	// 按有序日期遍历生成最终数据
+	for _, date := range sortedDates {
+		items := dateMap[date]
+		if len(items) == 0 {
 			continue
 		}
+		// 逐个生成倒序标签
+		for idx := range items {
+			item := &items[idx]
+			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"
+			}
+
+			cnt := typeTotal[suffix]
+			item.Type = ext
+			item.Label = fmt.Sprintf("%s_%d", suffix, cnt.idx)
+			cnt.idx--
+		}
 
 		tree = append(tree, flowDto.DateNode{
 			CreateDate: date,
-			Flows:      flowNodes,
+			Items:      items,
 		})
 	}
 
-	// 第一层日期倒序
-	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,
@@ -285,13 +341,6 @@ 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,
 	})
@@ -301,17 +350,24 @@ 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 = req.FlowName
+		r.FlowName = flowName
 		r.NodeGroupId = nodeGroupId
 		r.TriggerType = flow.FlowExecutionTriggerTypeManual.Code()
 		r.FlowContent = req.FlowContent
-		//r.NodeInputParams = nodeInputParams
+		r.NodeInputParams = req.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()
@@ -323,7 +379,6 @@ 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() {
@@ -331,10 +386,15 @@ func (s *flowExecutionService) Execute(ctx context.Context, req *flowDto.Execute
 			cancelMap.Store(traceId, cancel)
 		}
 		executionReq := flowDto.UpdateFlowExecutionReq{
-			Id:          executionId,
-			NodeGroupId: nodeGroupId,
-			Status:      flow.FlowExecutionStatusRunning.Code(),
-			TraceId:     traceId,
+			Id:              executionId,
+			NodeGroupId:     nodeGroupId,
+			FlowContent:     req.FlowContent,
+			NodeInputParams: req.NodeInputParams,
+			Extension: map[string]any{
+				"templates": req.Templates,
+			},
+			Status:  flow.FlowExecutionStatusRunning.Code(),
+			TraceId: traceId,
 		}
 		_, err = flowDao.FlowExecutionDao.Update(ctx, &executionReq)
 		if err != nil {
@@ -356,26 +416,63 @@ 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
+	//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
 		}
-		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]
-	nodeList, runGraph, err = BuildGraphFromFlowContent(execCtx, req.FlowContent)
+	runGraph, err = BuildGraphFromFlowContent(execCtx, req.FlowContent, judge2IntentNodeMap, summaryNodeID)
 	if err != nil {
 		executionReq := flowDto.UpdateFlowExecutionReq{
 			Id:           executionId,
@@ -388,16 +485,23 @@ 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 nodeInputParams {
+	for _, cfg := range req.NodeInputParams {
 		configMap[cfg.Id] = cfg
 	}
-	for _, i := range nodeList {
-		configMap[i.Id] = &i
+	// 自动给意图节点复制配置
+	for judgeID, intentID := range judge2IntentNodeMap {
+		if cfg, ok := configMap[judgeID]; ok {
+			configMap[intentID] = cfg
+		}
 	}
+	// 初始化汇总节点配置
+	configMap[summaryNodeID] = &summaryNode
+
 	// =========================================================================
 	// ✅【第4步】构建全局执行入参（现在 schemaMap 是有值的！）
 	// =========================================================================
@@ -406,6 +510,7 @@ 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,
@@ -437,7 +542,8 @@ func (s *flowExecutionService) Execute(ctx context.Context, req *flowDto.Execute
 	return
 }
 
-func BuildGraph(ctx context.Context, flowContent *entity.FlowInfo) ([]entity.FlowNode, *compose.Graph[any, any]) {
+// BuildGraphFromFlowContent 根据前端保存的工作流JSON，自动构建执行图
+func BuildGraphFromFlowContent(ctx context.Context, flowContent *entity.FlowInfo, judge2IntentNodeMap map[string]string, summaryNodeID string) (compose.Runnable[any, any], error) {
 	// 注册自定义合并函数：处理 *flowDto.FlowExecutionInput 类型合并
 	// 由于 ConfigMap 是 map 引用类型，所有并行分支修改已经写入共享内存
 	// 直接返回第一个实例即可，所有修改都已经可见
@@ -450,26 +556,9 @@ func BuildGraph(ctx context.Context, flowContent *entity.FlowInfo) ([]entity.Flo
 	})
 
 	graph := compose.NewGraph[any, any]()
-
-	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)
 
 	// 注册所有节点
-	nodeMap := make(map[string]entity.FlowNode)
 	for _, item := range flowContent.Nodes {
 		nodeMap[item.Id] = item
 		if item.NodeCode != node.NodeTypeJudge {
@@ -477,16 +566,6 @@ func BuildGraph(ctx context.Context, flowContent *entity.FlowInfo) ([]entity.Flo
 		}
 	}
 
-	// 注册所有边
-	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)
@@ -499,8 +578,15 @@ func BuildGraph(ctx context.Context, flowContent *entity.FlowInfo) ([]entity.Flo
 	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
@@ -535,6 +621,11 @@ func BuildGraph(ctx context.Context, flowContent *entity.FlowInfo) ([]entity.Flo
 				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, // 当前判断节点配置
@@ -543,31 +634,41 @@ func BuildGraph(ctx context.Context, flowContent *entity.FlowInfo) ([]entity.Flo
 				return JudgeLambda(ctx, nodeExecInput) // 传入 NodeExecutionInput 类型
 			}
 
-			_ = graph.AddBranch(upstreamMap[fromNodeID][0], compose.NewGraphBranch(judgeLambda, branchMap))
+			_ = graph.AddBranch(intentNodeID, 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
-}
 
-// 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
+	// ==================== 第四步：处理开始/结束节点 ====================
+	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))
 }
 
 // -------------------------- 节点自动注册器（核心分发） --------------------------
 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) {
@@ -578,9 +679,9 @@ func registerNodeToGraph(graph *compose.Graph[any, any], flowNode entity.FlowNod
 			}
 
 			configMap := execInput.ConfigMap
-			currentConfig := configMap[flowNode.Id]
+			currentConfig := configMap[nodeID]
 			if currentConfig == nil {
-				return nil, fmt.Errorf("节点%s无配置", flowNode.Id)
+				return nil, fmt.Errorf("节点%s无配置", nodeID)
 			}
 
 			// 获取入参 - 适配切片类型：遍历所有来源节点
@@ -613,7 +714,7 @@ func registerNodeToGraph(graph *compose.Graph[any, any], flowNode entity.FlowNod
 
 			nodeExecutionId, err := nodeDao.NodeExecutionDao.Insert(ctx, &nodeDto.CreateNodeExecutionReq{
 				FlowExecutionId: execInput.ExecutionId,
-				NodeId:          flowNode.Id,
+				NodeId:          nodeID,
 				NodeName:        flowNode.Name,
 				NodeGroupId:     execInput.NodeGroupId,
 				InputParamsPath: ossResult.FileURL,
@@ -622,7 +723,7 @@ func registerNodeToGraph(graph *compose.Graph[any, any], flowNode entity.FlowNod
 			if err != nil {
 				// 记录失败到已执行列表
 				execInput.ExecutedNodes = append(execInput.ExecutedNodes, flowDto.ExecutedNode{
-					NodeId: flowNode.Id,
+					NodeId: nodeID,
 					Status: node.NodeExecutionStatusFailed.Code(),
 				})
 				return nil, err
@@ -642,7 +743,7 @@ func registerNodeToGraph(graph *compose.Graph[any, any], flowNode entity.FlowNod
 				_, _ = nodeDao.NodeExecutionDao.Update(ctx, updateReq)
 				// 记录失败到已执行列表
 				execInput.ExecutedNodes = append(execInput.ExecutedNodes, flowDto.ExecutedNode{
-					NodeId: flowNode.Id,
+					NodeId: nodeID,
 					Status: node.NodeExecutionStatusFailed.Code(),
 				})
 				return nil, err
@@ -661,7 +762,7 @@ func registerNodeToGraph(graph *compose.Graph[any, any], flowNode entity.FlowNod
 			_, _ = nodeDao.NodeExecutionDao.Update(ctx, updateReq)
 			// 记录成功到已执行列表
 			execInput.ExecutedNodes = append(execInput.ExecutedNodes, flowDto.ExecutedNode{
-				NodeId: flowNode.Id,
+				NodeId: nodeID,
 				Status: node.NodeExecutionStatusSuccess.Code(),
 			})
 
@@ -669,70 +770,83 @@ func registerNodeToGraph(graph *compose.Graph[any, any], flowNode entity.FlowNod
 			return execInput, nil
 		}
 	}
-	switch flowNode.NodeCode {
+	if nodeID == "summary_node" {
+		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(SummaryLambda)))
+		return
+	}
+	switch code {
 	case "__start__":
-		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(StartLambda)))
-	case node.NodeTypeSystemSum:
-		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(SummaryLambda)))
+		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(StartLambda)))
 	case node.NodeTypeTextModel:
-		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(TextModelLambda)))
+		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(TextModelLambda)))
 	case node.NodeTypeImageModel:
-		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(ImageModelLambda)))
+		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(ImageModelLambda)))
 	case node.NodeTypeVideoModel:
-		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(VideoModelLambda)))
+		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(VideoModelLambda)))
 	case node.NodeTypeAudioModel:
-		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(AudioModelLambda)))
+		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(AudioModelLambda)))
 	case node.NodeTypeBatchModel:
-		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(BatchModelLambda)))
+		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(BatchModelLambda)))
 	case node.NodeTypeDataConversionModel:
-		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(DataConversionLambda)))
+		_ = 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)))
 	case node.NodeTypeCustomNode:
-		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(CustomLambda)))
+		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(CustomLambda)))
 	case node.NodeTypeForm:
-		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(FormLambda)))
-	//case node.NodeTypeIntent:
-	//	_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(IntentLambda)))
+		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(FormLambda)))
+	case node.NodeTypeIntent:
+		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(IntentLambda)))
 	case node.NodeTypeMerge:
-		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(MergeLambda)))
+		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(MergeLambda)))
 	case node.NodeTypeDataMerge:
-		_ = 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)))
+		_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(DataMergeLambda)))
 	case node.NodeTypeHttp:
-		_ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(HttpLambda)))
+		_ = graph.AddLambdaNode(nodeID, 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}
 
-	for len(queue) > 0 {
-		node := queue[0]
-		queue = queue[1:]
+	// 🚀 只从【开始节点】递归遍历（关键修复）
+	findLeafNodes(startNodeId, nextMap, endNodeSet)
 
-		if _, exist := visited[node]; exist {
-			continue
-		}
-		visited[node] = struct{}{}
+	// 转成数组返回
+	endNodes := make([]string, 0, len(endNodeSet))
+	for id := range endNodeSet {
+		endNodes = append(endNodes, id)
+	}
+	return endNodes
+}
 
-		nextList := nextMap[node]
-		if len(nextList) == 0 {
-			endNodeSet[node] = struct{}{}
-			continue
-		}
-		queue = append(queue, nextList...)
+// --------------------------------------------------------------------
+// ✅ 递归：查找以 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)
 	}
-
-	res := make([]string, 0, len(endNodeSet))
-	for k := range endNodeSet {
-		res = append(res, k)
-	}
-	return res
 }

workflow/service/flow/lambda_node.go

@@ -6,9 +6,11 @@ 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"
@@ -19,8 +21,6 @@ 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,81 +31,40 @@ 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 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
+func IntentLambda(ctx context.Context, input any) (any, error) {
+	return input, nil
 }
 
 // JudgeLambda 分支判断核心：读取IntentLambda的输出 → 返回目标节点ID做路由
@@ -114,20 +73,6 @@ 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
@@ -172,7 +117,7 @@ func JudgeLambda(ctx context.Context, input any) (string, error) {
 	if err != nil {
 		return "", err
 	}
-	composeResult, err := GetComposeResult(ctx, 2, getIsChatModel.Model.ModelName, "", "", []map[string]any{{"prompt": strings.Join(branchIdNameLines, "\n")}}, []map[string]any{{"prompt": contextParts}}, nodeInput.Global.FileUrl, nodeInput.Global.SessionId, nodeInput.Config.Id, "判断节点")
+	composeResult, err := GetComposeResult(ctx, nodeInput.NodeExecutionId, 2, getIsChatModel.Model.ModelName, "", "", []map[string]any{{"prompt": strings.Join(branchIdNameLines, "\n")}}, []map[string]any{{"prompt": contextParts}}, nodeInput.Global.FileUrl, nodeInput.Global.SessionId, nodeInput.Config.Id, "判断节点")
 	if err != nil {
 		return "", err
 	}
@@ -216,9 +161,8 @@ func BatchModelLambda(ctx context.Context, input any) (any, error) {
 	res := make([][]node.NodeFormField, len(reqMap))
 	var wg sync.WaitGroup
 
-	subCtx, cancel := context.WithCancel(ctx)
-	defer cancel()
-
+	// 只创建基础上下文，不再主动批量 cancel
+	subCtx := context.WithoutCancel(ctx)
 	// 缓冲1错误通道，仅接收第一个错误
 	errCh := make(chan error, 1)
 
@@ -228,7 +172,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
@@ -236,12 +180,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
@@ -250,26 +194,30 @@ 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:
-		if execErr != nil {
-			// 收到真实业务错误，等待剩余协程收尾后返回
-			wg.Wait()
-			return nil, execErr
-		}
-		// execErr == nil 代表通道关闭、无任何错误，走到下方返回完整结果
+		// 捕获第一个业务错误，等待剩余协程收尾
+		wg.Wait()
 	case <-subCtx.Done():
-		// 上下文被取消，阻塞读完errCh，确认是否存在业务错误
+		// 上游根上下文被终止，读取已存在的错误
 		execErr = <-errCh
+		wg.Wait()
+		if execErr != nil {
+			execErr = fmt.Errorf("global context canceled: %w", execErr)
+		}
+	}
+
+	// 有错误直接返回，不再走结果拼接
+	if execErr != nil {
+		return nil, execErr
 	}
 
 	// 拼接输出结果
@@ -306,8 +254,6 @@ func TextModelLambda(ctx context.Context, input any) (any, error) {
 		return nil, err
 	}
 	nodeInput.Config.OutputResult = outputRes
-	//}
-
 	return nodeInput, nil
 }
 
@@ -353,7 +299,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") {
@@ -363,34 +309,56 @@ func VideoModelLambda(ctx context.Context, input any) (any, error) {
 	if g.IsEmpty(videoURL) {
 		return nil, fmt.Errorf("视频合成失败:模型生成视频失败")
 	}
-	waitRes, err := VideoConcat(ctx, videoURL)
-	if err != nil {
-		return nil, err
+	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
 	}
-	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: msg.FileURL,
+			Value: videoUrl,
 			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: newS + msg.FileURL,
-		Label: fmt.Sprintf("视频内容:content:%d", 0),
+		Value: videoUrl,
+		Label: fmt.Sprintf("concat_video_url:content:%d", 0),
 		Type:  "string",
 	})
 	nodeInput.Config.OutputResult = outputRes
@@ -407,64 +375,6 @@ 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
 	}
@@ -599,6 +509,12 @@ 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
 	}
@@ -750,7 +666,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] = field.Value
+					item[timeKey] = ProcessPath(ctx, gconv.String(field.Value))
 					summaryResult = append(summaryResult, item)
 				}
 			}
@@ -768,10 +684,28 @@ func SummaryLambda(ctx context.Context, input any) (any, error) {
 			return err
 		}
 
+		res, _, err := nodeDao.NodeExecutionDao.ListByFlowExecutionId(ctx, &nodeDto.ListNodeExecutionByFlowReq{
+			NodeGroupId: execInput.Global.NodeGroupId,
+		}, entity.NodeExecutionCol.TokenInfo)
+		if err != nil {
+			return err
+		}
+		var totalTokens int
+		var totalFee float64
+		for _, item := range res {
+			for _, itemToken := range item.TokenInfo {
+				m := gconv.Map(itemToken)
+				totalTokens += gconv.Int(m["total_tokens"])
+				totalFee += gconv.Float64(m["total_fee"])
+			}
+		}
+
 		executionReq := flowDto.UpdateFlowExecutionReq{
 			Id:           execInput.Global.ExecutionId,
 			Status:       flow.FlowExecutionStatusSuccess.Code(),
 			OutputParams: summaryResult,
+			TotalTokens:  totalTokens,
+			TotalFee:     totalFee,
 		}
 		_, err = flowDao.FlowExecutionDao.Update(ctx, &executionReq)
 

workflow/service/flow/lambda_node_imp.go

@@ -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("图片内容%v:%d", k, i),
+				Label: fmt.Sprintf("img_url%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("音频内容:%v:%d", k, i),
+				Label: fmt.Sprintf("audio_url:%v:%d", k, i),
 				Type:  "string",
 			})
 		}
@@ -683,24 +683,29 @@ func AudioOptimizeNode(ctx context.Context, nodeInput *flowDto.NodeExecutionInpu
 	return outputRes, nil
 }
 
+// splitTextByPunct 按中文标点分割句子，同时保留标点在分段内
+// 例如："这个叫高血压调理方，注意是根源调理不是临时缓解，"
+// 会变成：["这个叫高血压调理方，", "注意是根源调理不是临时缓解，"]
 func splitTextByPunct(raw string) []string {
-	// 按标点切分+拼接标点
-	slice := regexp.MustCompile(`([，。；！？])`).Split(raw, -1)
-	var res []string
-	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()
-		}
+	// 匹配中文标点并保留在文本中，按标点位置切分
+	re := regexp.MustCompile(`[，。；！？]`)
+	// 先找到所有标点的位置
+	indexes := re.FindAllStringIndex(raw, -1)
+	if len(indexes) == 0 {
+		return []string{raw}
 	}
-	if builder.Len() > 0 {
-		res = append(res, builder.String())
+
+	var res []string
+	prev := 0
+	for _, idx := range indexes {
+		end := idx[1] // 标点的结束位置
+		seg := raw[prev:end]
+		res = append(res, seg)
+		prev = end
+	}
+	// 处理最后一段没有标点的文本
+	if prev < len(raw) {
+		res = append(res, raw[prev:])
 	}
 	return res
 }
@@ -708,46 +713,53 @@ 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 {
-			return nil, nil
+			continue
 		}
 
-		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
-			currentText := ""
-			// 循环取 word，直到拼接内容 包含/匹配 seg
-			for {
-				if wordIdx >= len(allWords) {
-					break
-				}
+			var currentText strings.Builder
+
+			// 收集Word直到拼接内容覆盖当前分段
+			for wordIdx < len(allWords) {
 				word := allWords[wordIdx]
-				currentText += word.Word
+				currentText.WriteString(word.Word)
 				collectWords = append(collectWords, word)
 				wordIdx++
 
-				// 只要包含分段文本，就认为匹配（无视末尾标点差异）
-				if strings.Contains(currentText, seg) {
+				// 当拼接的文本包含当前分段的纯文本时，停止收集
+				if strings.Contains(currentText.String(), segClean) {
 					break
 				}
 			}
+
 			if len(collectWords) == 0 {
 				continue
 			}
-			// 生成字幕
+
+			// 3. 生成字幕（时间戳取首尾Word的时间）
 			sub := flowDto.Subtitle{
 				Start: collectWords[0].StartTime,
 				End:   collectWords[len(collectWords)-1].EndTime,
-				Text:  seg,
+				Text:  segClean,
 			}
-			subs = append(subs, sub)
+			subtitles = append(subtitles, sub)
 		}
-		subtitles = append(subtitles, subs...)
 	}
 
 	return subtitles, nil
@@ -776,15 +788,52 @@ 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("输出模板结构,仅修改每个字段对应数值：%s", jsonStr)
+	jsonVal += fmt.Sprintf("输出字段规范：%v", 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
@@ -849,7 +898,7 @@ func HttpNode(ctx context.Context, nodeInput *flowDto.NodeExecutionInput) ([]nod
 	}
 
 	// 构建请求参数
-	newBody := BuildNestedJson(body, nodeInput.Global.ConfigMap)
+	newBody := BuildNestedJson(body, nodeInput.Global)
 	// 1. 自己生成唯一 taskId（不用前端给）
 	taskId := "my_task_" + uuid.New().String() // 自己生成唯一ID
 	if responseType == "callback" {
@@ -875,6 +924,8 @@ 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)
@@ -884,6 +935,7 @@ 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
@@ -910,11 +962,16 @@ 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 {
@@ -1001,7 +1058,7 @@ func GetNodeContextContent(execInput *flowDto.FlowExecutionInput, nodeEntity *en
 				// 取指定字段
 				for _, f := range source.Field {
 					for _, v := range refNode.FormConfig {
-						if strings.Contains(v.Label, f) {
+						if v.Label == f {
 							input = append(input, v)
 						}
 					}
@@ -1024,3 +1081,146 @@ 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
+//}

workflow/service/flow/lambda_node_util.go

@@ -9,21 +9,25 @@ 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"
 )
@@ -97,12 +101,7 @@ func GetModelInfo(ctx context.Context, req *flowDto.GetModelInfoReq) (res *flowD
 	return
 }
 
-func GetComposeResult(ctx context.Context, buildType int, modelName, promptContent, skillName string, form []map[string]any, userForm []map[string]any, fileUrl []string, sessionId, nodeId string, cause string) (res *flowDto.ComposeCallbackReq, err error) {
-	if !g.IsEmpty(promptContent) {
-		userForm = append(userForm, map[string]any{
-			"prompt": promptContent,
-		})
-	}
+func GetComposeResult(ctx context.Context, nodeExecutionId int64, buildType int, modelName, promptContent, skillName string, form []map[string]any, userForm []map[string]any, fileUrl []string, sessionId, nodeId string, cause string) (res *flowDto.ComposeCallbackReq, err error) {
 	var callbackUrl = utils.GetCallbackURL(ctx, "/flow/execution/composeCallBack")
 	var consult = make([]flowDto.Consult, 0)
 	var collectFileUrls func(val any) (fullyConsumed bool)
@@ -143,11 +142,27 @@ func GetComposeResult(ctx context.Context, buildType int, modelName, promptConte
 	}
 	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) {
-				delete(m, k)
+				delKeys = append(delKeys, k)
 			}
 		}
+		// 统一删除
+		for _, k := range delKeys {
+			delete(m, k)
+		}
+
 		if len(m) > 0 {
 			newUserForm = append(newUserForm, m)
 		}
@@ -169,23 +184,60 @@ func GetComposeResult(ctx context.Context, buildType int, modelName, promptConte
 		Cause:       cause,
 		Form:        form,
 		UserForm:    newUserForm,
+		UserPrompt:  promptContent,
 		Consult:     consult,
 		SessionId:   sessionId,
 		NodeId:      nodeId,
 	}
-	headers := make(map[string]string)
+	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
+	}
 	if r := g.RequestFromCtx(ctx); r != nil {
 		for k, v := range r.Request.Header {
 			if len(v) > 0 {
-				headers[k] = v[0]
+				client.SetHeader(k, v[0])
 			}
 		}
 	}
-	msgRes := new(flowDto.ComposeMessagesRes)
-	err = commonHttp.Post(ctx, "prompts-core/prompt/composeMessages", headers, msgRes, &msgReq)
+	resp, err := client.ContentJson().Post(postCtx, "prompts-core/prompt/composeMessages", &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")
 	}
@@ -197,18 +249,19 @@ func GetComposeResult(ctx context.Context, buildType int, modelName, promptConte
 	if err = gconv.Struct(waitRes, msg); err != nil {
 		return nil, err
 	}
+	updateTokenCount(ctx, nodeExecutionId, msg.BillingData)
 	if !g.IsEmpty(msg.ErrorMsg) {
 		return nil, fmt.Errorf(msg.ErrorMsg)
 	}
 	return msg, nil
 }
 
-func CreateGatewayTask(ctx context.Context, epicycleId int64, model string, content map[string]any) (map[string]any, error) {
+func CreateGatewayTask(ctx context.Context, nodeExecutionId int64, epicycleId int64, model string, content map[string]any) (map[string]any, error) {
 	taskId, err := createGatewayTaskOnly(ctx, epicycleId, model, content)
 	if err != nil {
 		return nil, err
 	}
-	return waitGatewayResult(ctx, taskId)
+	return waitGatewayResult(ctx, nodeExecutionId, taskId)
 }
 
 // createGatewayTaskOnly creates a gateway task and returns the taskId only
@@ -223,29 +276,63 @@ func createGatewayTaskOnly(ctx context.Context, epicycleId int64, model string,
 		EpicycleId:     epicycleId,
 	}
 
-	headers := make(map[string]string)
+	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
+	}
 	if r := g.RequestFromCtx(ctx); r != nil {
 		for k, v := range r.Request.Header {
 			if len(v) > 0 {
-				headers[k] = v[0]
+				client.SetHeader(k, v[0])
 			}
 		}
 	}
-
-	res := new(flowDto.ModelGatewayRes)
-	err := commonHttp.Post(ctx, "model-gateway/task/createTask", headers, res, &req)
+	rpcResp, err := client.ContentJson().Post(postCtx, "model-gateway/task/createTask", &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, taskId string) (map[string]any, error) {
+func waitGatewayResult(ctx context.Context, nodeExecutionId int64, taskId string) (map[string]any, error) {
 	waitRes, err := Wait(ctx, taskId)
 	if err != nil {
 		return nil, err
@@ -255,6 +342,7 @@ func waitGatewayResult(ctx context.Context, taskId string) (map[string]any, erro
 	if err = gconv.Struct(waitRes, task); err != nil {
 		return nil, err
 	}
+	updateTokenCount(ctx, nodeExecutionId, task.BillingData)
 	if task.State == 3 || !g.IsEmpty(task.ErrorMsg) {
 		return nil, fmt.Errorf("模型执行失败：%s", task.ErrorMsg)
 	}
@@ -270,14 +358,23 @@ func waitGatewayResult(ctx context.Context, taskId string) (map[string]any, erro
 }
 
 // updateTokenCount updates the token count in node execution
-func updateTokenCount(ctx context.Context, nodeExecutionId int64, responseField string, result map[string]any) {
-	if responseField == "" {
+func updateTokenCount(ctx context.Context, nodeExecutionId int64, tokenInfo []map[string]any) {
+	res, err := nodeDao.NodeExecutionDao.Get(ctx, &nodeDto.GetNodeExecutionReq{
+		Id: nodeExecutionId,
+	}, entity.NodeExecutionCol.TokenInfo)
+	if err != nil {
 		return
 	}
+	var t []map[string]any
+	for _, item := range res.TokenInfo {
+		t = append(t, item)
+	}
+	for _, item := range tokenInfo {
+		t = append(t, item)
+	}
 	_, _ = nodeDao.NodeExecutionDao.Update(ctx, &nodeDto.UpdateNodeExecutionReq{
-		Id:               nodeExecutionId,
-		CompletionTokens: gconv.Int(result[responseField]),
-		TotalTokens:      gconv.Int(result[responseField]),
+		Id:        nodeExecutionId,
+		TokenInfo: t,
 	})
 }
 
@@ -290,8 +387,23 @@ func GetModelResult(ctx context.Context, sessionId string, nodeInput *flowDto.No
 	if !nodeInput.Global.IsDialogue {
 		sessionId = ""
 	}
-
-	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)
+	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)
 	if err != nil {
 		return nil, err
 	}
@@ -307,13 +419,12 @@ 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.LastFrame, gconv.String(taskResultMap[modelInfo.Model.ResponseBody]))
+					set, err = sjson.Set(gconv.String(item), modelInfo.Model.FirstFrame, gconv.String(taskResultMap["content"]))
 					if err != nil {
 						return nil, err
 					}
@@ -321,7 +432,7 @@ func GetModelResult(ctx context.Context, sessionId string, nodeInput *flowDto.No
 				}
 
 				var taskResult map[string]any
-				taskResult, err = CreateGatewayTask(ctx, composeResult.EpicycleId, nodeInput.Config.ModelConfig.ModelName, item)
+				taskResult, err = CreateGatewayTask(ctx, nodeInput.NodeExecutionId, composeResult.EpicycleId, nodeInput.Config.ModelConfig.ModelName, item)
 				if err != nil {
 					return nil, err
 				}
@@ -330,7 +441,7 @@ func GetModelResult(ctx context.Context, sessionId string, nodeInput *flowDto.No
 				}
 
 				if nodeInput.Config.NodeCode == node.NodeTypeVideoModel {
-					ext := GetFileTypeByPath(gconv.String(taskResult[modelInfo.Model.ResponseBody]))
+					ext := GetFileTypeByPath(gconv.String(taskResult["content"]))
 					if ext == "image" {
 						taskResultMap = taskResult
 					} else {
@@ -341,36 +452,34 @@ 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(taskIdList))
+			errChan := make(chan error, len(composeResult.Messages.Rounds))
 
 			// 加互斥锁保护结果map
 			var mu sync.Mutex
 
-			for idx, taskId := range taskIdList {
+			// 每个任务创建后立即启动等待协程：把「回调 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
+				}
+
 				wg.Add(1)
 
 				go func(idx int, taskId string) {
 					defer wg.Done()
 
-					taskResult, err := waitGatewayResult(subCtx, taskId)
+					taskResult, err := waitGatewayResult(subCtx, nodeInput.NodeExecutionId, taskId)
 					if err != nil {
 						errChan <- err
 						globalCancel() // 全局取消，所有协程收到ctx取消信号快速退出
@@ -382,7 +491,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)
 			}
 
@@ -403,16 +512,19 @@ 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, mockConfigMap map[string]*entity.FlowNode) g.Map {
+func BuildNestedJson(body g.Map, global *flowDto.FlowExecutionInput) 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 {
@@ -423,7 +535,7 @@ func BuildNestedJson(body g.Map, mockConfigMap map[string]*entity.FlowNode) g.Ma
 			valMap := gconv.Map(val)
 			nodeId := gconv.String(valMap["nodeId"])
 			fieldName := gconv.String(valMap["field"])
-			if configValue, ok := mockConfigMap[nodeId]; ok {
+			if configValue, ok := global.ConfigMap[nodeId]; ok {
 				if !g.IsEmpty(configValue.OutputResult) {
 					for _, v := range configValue.OutputResult {
 						if strings.Contains(v.Field, fieldName) {
@@ -599,6 +711,56 @@ 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

workflow/service/node/node_library_service.go

@@ -67,6 +67,7 @@ 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,
@@ -93,23 +94,36 @@ 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,
@@ -346,7 +360,10 @@ 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.NodeTypeDataConversionModel ||
+				item.NodeCode == node.NodeTypeSenseOptimizeModel ||
+				item.NodeCode == node.NodeTypeStoryOptimizeModel ||
+				item.NodeCode == node.NodeTypeScriptOptimizeModel {
 				item.ModelConfig = append(item.ModelConfig, node.ModelItem{
 					ModelName: "自定义",
 				})