852 lines
27 KiB
Go
852 lines
27 KiB
Go
package flow
|
|
|
|
import (
|
|
"ai-agent/workflow/consts/flow"
|
|
"ai-agent/workflow/consts/node"
|
|
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"
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
"strconv"
|
|
"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/util/gconv"
|
|
"github.com/google/uuid"
|
|
"go.opentelemetry.io/otel/trace"
|
|
)
|
|
|
|
var FlowExecutionService = &flowExecutionService{}
|
|
|
|
type flowExecutionService struct{}
|
|
|
|
func (s *flowExecutionService) Get(ctx context.Context, req *flowDto.GetFlowExecutionReq) (res *flowDto.VOFlowExecution, err error) {
|
|
r, err := flowDao.FlowExecutionDao.Get(ctx, req)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
res = new(flowDto.VOFlowExecution)
|
|
res.ImgAddressPrefix, err = utils.GetFileAddressPrefix(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
err = gconv.Struct(r, &res)
|
|
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 = false
|
|
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.IsResult = true
|
|
list, _, err := flowDao.FlowExecutionDao.List(ctx, req)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// 过滤出有有效输出的执行记录
|
|
var validList []*entity.FlowExecution
|
|
for _, execution := range list {
|
|
if !g.IsEmpty(execution.OutputParams) {
|
|
validList = append(validList, execution)
|
|
}
|
|
}
|
|
|
|
// 1. 按日期归集，严格使用 Y-m-d 格式
|
|
dateMap := make(map[string][]flowDto.OutputItem)
|
|
for _, execution := range validList {
|
|
// 按要求使用 Y-m-d
|
|
createDate := execution.CreatedAt.Format("Y-m-d")
|
|
execID := execution.Id
|
|
outputParams := execution.OutputParams
|
|
|
|
var tempItems []flowDto.OutputItem
|
|
for _, paramMap := range outputParams {
|
|
for tsKey, value := range paramMap {
|
|
if _, err := strconv.ParseInt(tsKey, 10, 64); err != nil {
|
|
continue
|
|
}
|
|
tempItems = append(tempItems, flowDto.OutputItem{
|
|
Id: execID,
|
|
Timestamp: tsKey,
|
|
Content: gconv.String(value),
|
|
})
|
|
}
|
|
}
|
|
if len(tempItems) == 0 {
|
|
continue
|
|
}
|
|
|
|
// 单条执行内按时间戳正序
|
|
sort.Slice(tempItems, func(i, j int) bool {
|
|
t1, _ := strconv.ParseInt(tempItems[i].Timestamp, 10, 64)
|
|
t2, _ := strconv.ParseInt(tempItems[j].Timestamp, 10, 64)
|
|
return t1 < t2
|
|
})
|
|
|
|
dateMap[createDate] = append(dateMap[createDate], tempItems...)
|
|
}
|
|
|
|
// ========== 修复编号乱序核心逻辑 ==========
|
|
// 1. 取出所有日期并 倒序排序（和前端展示顺序一致）
|
|
var sortedDates []string
|
|
for d := range dateMap {
|
|
sortedDates = append(sortedDates, d)
|
|
}
|
|
// 日期字符串倒序
|
|
sort.Slice(sortedDates, func(i, j int) bool {
|
|
return sortedDates[i] > sortedDates[j]
|
|
})
|
|
|
|
// 2. 按【前端展示顺序】拼接所有条目，用于统计总数量
|
|
var allItems []flowDto.OutputItem
|
|
for _, d := range sortedDates {
|
|
allItems = append(allItems, dateMap[d]...)
|
|
}
|
|
|
|
// 3. 统计各类型总数
|
|
type totalCnt struct {
|
|
total int
|
|
idx int
|
|
}
|
|
typeTotal := make(map[string]*totalCnt)
|
|
for _, item := range allItems {
|
|
val := item.Content
|
|
suffix := "内容"
|
|
ext := GetFileTypeByPath(val)
|
|
|
|
switch ext {
|
|
case "image":
|
|
suffix = "图片"
|
|
case "video":
|
|
suffix = "视频"
|
|
case "audio":
|
|
suffix = "音频"
|
|
case "text":
|
|
suffix = "文案"
|
|
case "html":
|
|
suffix = "HTML"
|
|
}
|
|
if _, ok := typeTotal[suffix]; !ok {
|
|
typeTotal[suffix] = &totalCnt{}
|
|
}
|
|
typeTotal[suffix].total++
|
|
}
|
|
// 初始序号 = 总数，从最大值开始倒序
|
|
for _, v := range typeTotal {
|
|
v.idx = v.total
|
|
}
|
|
// ======================================
|
|
|
|
var tree []flowDto.DateNode
|
|
// 按有序日期遍历生成最终数据
|
|
for _, date := range sortedDates {
|
|
items := dateMap[date]
|
|
if len(items) == 0 {
|
|
continue
|
|
}
|
|
// 逐个生成倒序标签
|
|
for idx := range items {
|
|
item := &items[idx]
|
|
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,
|
|
Items: items,
|
|
})
|
|
}
|
|
|
|
imgPrefix, err := utils.GetFileAddressPrefix(ctx)
|
|
return &flowDto.ListFlowExecutionTreeRes{
|
|
Tree: tree,
|
|
ImgAddressPrefix: imgPrefix,
|
|
}, nil
|
|
}
|
|
|
|
// ComposeCallback 提示词回调接口
|
|
func (s *flowExecutionService) ComposeCallback(ctx context.Context, req *flowDto.ComposeCallbackReq) (err error) {
|
|
Notify(req.TaskId, req)
|
|
return nil
|
|
}
|
|
|
|
// ModelCallback 模型回调接口
|
|
func (s *flowExecutionService) ModelCallback(ctx context.Context, req *flowDto.ModelCallbackReq) (err error) {
|
|
// 唤醒等待的任务
|
|
Notify(req.TaskId, req)
|
|
return nil
|
|
}
|
|
|
|
// VideoCallback 视频拼接回调接口
|
|
func (s *flowExecutionService) VideoCallback(ctx context.Context, req *flowDto.VideoCallbackReq) (err error) {
|
|
// 唤醒等待的任务
|
|
Notify(req.TaskId, req)
|
|
return nil
|
|
}
|
|
|
|
// HttpNodeCallback http节点回调接口
|
|
func (s *flowExecutionService) HttpNodeCallback(ctx context.Context) (err error) {
|
|
r := g.RequestFromCtx(ctx)
|
|
taskId := r.Get("task_id").String()
|
|
Notify(taskId, r)
|
|
return nil
|
|
}
|
|
|
|
// ===================== 核心改造：替换为 sync.Map 存储取消上下文 =====================
|
|
var (
|
|
// cancelMap: traceID -> context.CancelFunc
|
|
cancelMap sync.Map
|
|
)
|
|
|
|
func (s *flowExecutionService) Cancel(ctx context.Context, req *flowDto.CancelReq) (err error) {
|
|
getRes, err := flowDao.FlowExecutionDao.Get(ctx, &flowDto.GetFlowExecutionReq{
|
|
SessionId: req.SessionId,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if g.IsEmpty(getRes) {
|
|
return fmt.Errorf("会话[%s] 不存在", req.SessionId)
|
|
}
|
|
// 从 sync.Map 获取取消函数
|
|
cancelVal, exist := cancelMap.Load(getRes.TraceId)
|
|
if !exist {
|
|
return fmt.Errorf("traceID[%s] 不存在或已执行完成", getRes.TraceId)
|
|
}
|
|
|
|
// 执行取消
|
|
cancel, ok := cancelVal.(context.CancelFunc)
|
|
if !ok {
|
|
return fmt.Errorf("traceID[%s] 对应的取消函数类型错误", getRes.TraceId)
|
|
}
|
|
cancel()
|
|
|
|
// 取消后清理（可选：也可以在流程结束时统一清理）
|
|
cancelMap.Delete(getRes.TraceId)
|
|
|
|
// 同步更新流程执行状态为已取消
|
|
_, err = flowDao.FlowExecutionDao.Update(ctx, &flowDto.UpdateFlowExecutionReq{
|
|
Id: getRes.Id,
|
|
Status: flow.FlowExecutionStatusCancel.Code(),
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("更新取消状态失败: %v", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *flowExecutionService) Execute(ctx context.Context, req *flowDto.ExecuteReq) (res *flowDto.ExecuteRes, err error) {
|
|
// ===================== 核心改造1：创建可取消的上下文 =====================
|
|
execCtx, cancel := context.WithCancel(ctx)
|
|
traceId := ""
|
|
defer func() {
|
|
// 流程结束（成功/失败）时清理 cancelMap
|
|
if traceId != "" {
|
|
cancelMap.Delete(traceId)
|
|
}
|
|
cancel()
|
|
}()
|
|
|
|
flowInfo, err := flowDao.FlowExecutionDao.Get(ctx, &flowDto.GetFlowExecutionReq{
|
|
SessionId: req.SessionId,
|
|
})
|
|
if err != nil {
|
|
return
|
|
}
|
|
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 {
|
|
var r = new(flowDto.CreateFlowExecutionReq)
|
|
r.FlowUserId = req.FlowId
|
|
r.FlowName = flowName
|
|
r.NodeGroupId = nodeGroupId
|
|
r.TriggerType = flow.FlowExecutionTriggerTypeManual.Code()
|
|
r.FlowContent = req.FlowContent
|
|
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()
|
|
traceId = r.TraceId
|
|
cancelMap.Store(traceId, cancel)
|
|
}
|
|
executionId, err = flowDao.FlowExecutionDao.Insert(ctx, r)
|
|
if err != nil {
|
|
return
|
|
}
|
|
} else {
|
|
executionId = flowInfo.Id
|
|
span := trace.SpanFromContext(ctx)
|
|
if span != nil && span.SpanContext().HasTraceID() {
|
|
traceId = span.SpanContext().TraceID().String()
|
|
cancelMap.Store(traceId, cancel)
|
|
}
|
|
executionReq := flowDto.UpdateFlowExecutionReq{
|
|
Id: executionId,
|
|
NodeGroupId: nodeGroupId,
|
|
FlowContent: req.FlowContent,
|
|
NodeInputParams: req.NodeInputParams,
|
|
Extension: map[string]any{
|
|
"templates": req.Templates,
|
|
},
|
|
Status: flow.FlowExecutionStatusRunning.Code(),
|
|
TraceId: traceId,
|
|
}
|
|
_, err = flowDao.FlowExecutionDao.Update(ctx, &executionReq)
|
|
if err != nil {
|
|
return
|
|
}
|
|
}
|
|
|
|
if !g.IsEmpty(req.FileUrl) {
|
|
createFileTempReq := make([]*fileDto.CreateFileTempReq, 0, len(req.FileUrl))
|
|
for _, fileUrl := range req.FileUrl {
|
|
var createReq = new(fileDto.CreateFileTempReq)
|
|
createReq.BusinessId = req.SessionId
|
|
createReq.FileUrl = fileUrl
|
|
createFileTempReq = append(createFileTempReq, createReq)
|
|
}
|
|
_, err = fileDao.FileTempDao.BatchInsert(ctx, createFileTempReq)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
//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
|
|
}
|
|
}
|
|
|
|
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 runGraph compose.Runnable[any, any]
|
|
runGraph, err = BuildGraphFromFlowContent(execCtx, req.FlowContent, judge2IntentNodeMap, summaryNodeID)
|
|
if err != nil {
|
|
executionReq := flowDto.UpdateFlowExecutionReq{
|
|
Id: executionId,
|
|
Status: flow.FlowExecutionStatusFailed.Code(),
|
|
ErrorMessage: err.Error(),
|
|
}
|
|
_, err1 := flowDao.FlowExecutionDao.Update(ctx, &executionReq)
|
|
if err1 != nil {
|
|
return
|
|
}
|
|
return nil, fmt.Errorf("执行工作流失败: %v", err)
|
|
}
|
|
|
|
// =========================================================================
|
|
// ✅【第3步】构建 ConfigMap
|
|
// =========================================================================
|
|
configMap := make(map[string]*entity.FlowNode)
|
|
for _, cfg := range req.NodeInputParams {
|
|
configMap[cfg.Id] = cfg
|
|
}
|
|
// 自动给意图节点复制配置
|
|
for judgeID, intentID := range judge2IntentNodeMap {
|
|
if cfg, ok := configMap[judgeID]; ok {
|
|
configMap[intentID] = cfg
|
|
}
|
|
}
|
|
// 初始化汇总节点配置
|
|
configMap[summaryNodeID] = &summaryNode
|
|
|
|
// =========================================================================
|
|
// ✅【第4步】构建全局执行入参（现在 schemaMap 是有值的！）
|
|
// =========================================================================
|
|
execInput := &flowDto.FlowExecutionInput{
|
|
NodeGroupId: nodeGroupId,
|
|
IsDialogue: isDialogue,
|
|
ExecutionId: executionId,
|
|
ConfigMap: configMap,
|
|
Templates: req.Templates,
|
|
SessionId: req.SessionId,
|
|
Desc: req.Desc,
|
|
SkillName: req.SkillName,
|
|
FileUrl: req.FileUrl,
|
|
}
|
|
// 执行工作流
|
|
_, err = runGraph.Invoke(execCtx, execInput)
|
|
if err != nil {
|
|
// 检测是否是取消导致的错误
|
|
if errors.Is(execCtx.Err(), context.Canceled) {
|
|
executionReq := flowDto.UpdateFlowExecutionReq{
|
|
Id: executionId,
|
|
Status: flow.FlowExecutionStatusCancel.Code(),
|
|
}
|
|
_, _ = flowDao.FlowExecutionDao.Update(ctx, &executionReq)
|
|
return nil, fmt.Errorf("工作流已被取消: %v", err)
|
|
}
|
|
executionReq := flowDto.UpdateFlowExecutionReq{
|
|
Id: executionId,
|
|
Status: flow.FlowExecutionStatusFailed.Code(),
|
|
ErrorMessage: err.Error(),
|
|
}
|
|
_, err1 := flowDao.FlowExecutionDao.Update(ctx, &executionReq)
|
|
if err1 != nil {
|
|
return
|
|
}
|
|
return nil, fmt.Errorf("执行工作流失败: %v", err)
|
|
}
|
|
return
|
|
}
|
|
|
|
// 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 引用类型，所有并行分支修改已经写入共享内存
|
|
// 直接返回第一个实例即可，所有修改都已经可见
|
|
compose.RegisterValuesMergeFunc(func(values []*flowDto.FlowExecutionInput) (*flowDto.FlowExecutionInput, error) {
|
|
if len(values) == 0 {
|
|
return nil, nil
|
|
}
|
|
// 返回第一个实例，ConfigMap 是指针，所有修改都已经写入共享数据结构
|
|
return values[0], nil
|
|
})
|
|
|
|
graph := compose.NewGraph[any, any]()
|
|
nodeMap := make(map[string]entity.FlowNode)
|
|
|
|
// 注册所有节点
|
|
for _, item := range flowContent.Nodes {
|
|
nodeMap[item.Id] = item
|
|
if item.NodeCode != node.NodeTypeJudge {
|
|
registerNodeToGraph(graph, item)
|
|
}
|
|
}
|
|
|
|
// 构建边关系
|
|
upstreamMap := make(map[string][]string)
|
|
edgeMap := make(map[string][]entity.FlowEdge)
|
|
for _, edge := range flowContent.Edges {
|
|
edgeMap[edge.From] = append(edgeMap[edge.From], edge)
|
|
upstreamMap[edge.To] = append(upstreamMap[edge.To], edge.From)
|
|
}
|
|
|
|
// 处理连线 & 分支
|
|
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
|
|
}
|
|
|
|
judgeLambda := func(ctx context.Context, input any) (string, error) {
|
|
execInput, ok := input.(*flowDto.FlowExecutionInput)
|
|
if !ok {
|
|
return "", fmt.Errorf("入参类型错误")
|
|
}
|
|
|
|
currentConfig := execInput.ConfigMap[fromNodeID]
|
|
if currentConfig == nil {
|
|
return "", fmt.Errorf("判断节点%s无配置", fromNodeID)
|
|
}
|
|
|
|
branchIdNameMap := make(map[string]string)
|
|
var branchIDs []string
|
|
for nodeID := range branchMap {
|
|
branchIDs = append(branchIDs, nodeID)
|
|
// 从configMap获取分支节点的名称
|
|
if branchNodeCfg, ok := execInput.ConfigMap[nodeID]; ok {
|
|
branchIdNameMap[nodeID] = branchNodeCfg.Name
|
|
} else {
|
|
branchIdNameMap[nodeID] = "未命名节点" // 兜底
|
|
}
|
|
}
|
|
|
|
// 把分支ID-名称映射塞进 ModelConfig，带给意图节点
|
|
m := make(map[string]interface{})
|
|
m["branch_ids"] = branchIDs
|
|
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, // 当前判断节点配置
|
|
Global: execInput, // 全局执行入参
|
|
}
|
|
return JudgeLambda(ctx, nodeExecInput) // 传入 NodeExecutionInput 类型
|
|
}
|
|
|
|
_ = 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)
|
|
}
|
|
}
|
|
|
|
// ==================== 第四步：处理开始/结束节点 ====================
|
|
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) {
|
|
// ✅ 【关键】全程入参类型永远不变
|
|
execInput, ok := input.(*flowDto.FlowExecutionInput)
|
|
if !ok {
|
|
return nil, fmt.Errorf("入参必须是 *FlowExecutionInput, 实际是 %T", input)
|
|
}
|
|
|
|
configMap := execInput.ConfigMap
|
|
currentConfig := configMap[nodeID]
|
|
if currentConfig == nil {
|
|
return nil, fmt.Errorf("节点%s无配置", nodeID)
|
|
}
|
|
|
|
// 获取入参 - 适配切片类型：遍历所有来源节点
|
|
realInput := new(flowDto.NodeExecutionInput)
|
|
if len(flowNode.InputSource) > 0 { // 改为判断切片长度
|
|
// 遍历所有指定的来源节点，聚合输出结果
|
|
for _, inputSource := range flowNode.InputSource { // 遍历切片
|
|
if sourceConfig, ok := configMap[inputSource.NodeId]; ok {
|
|
currentConfig.OutputResult = append(currentConfig.OutputResult, sourceConfig.OutputResult...)
|
|
}
|
|
}
|
|
}
|
|
|
|
// ✅ 封装节点执行入参（配置+表单架构）
|
|
realInput = &flowDto.NodeExecutionInput{
|
|
Config: currentConfig,
|
|
Global: execInput, // ✅ 把【全部节点】的对象直接塞进来
|
|
}
|
|
// ✅ 插入节点执行记录，初始状态为运行中
|
|
startTime := time.Now()
|
|
|
|
// 上传OSS（每条独立上传）
|
|
ossResult, err := Upload(ctx, &dto.UploadFileBytesReq{
|
|
FileBytes: gconv.Bytes(gconv.String(realInput)),
|
|
FileName: fmt.Sprintf("nodeInput:%v.txt", time.Now().UnixMilli()),
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
nodeExecutionId, err := nodeDao.NodeExecutionDao.Insert(ctx, &nodeDto.CreateNodeExecutionReq{
|
|
FlowExecutionId: execInput.ExecutionId,
|
|
NodeId: nodeID,
|
|
NodeName: flowNode.Name,
|
|
NodeGroupId: execInput.NodeGroupId,
|
|
InputParamsPath: ossResult.FileURL,
|
|
Status: node.NodeExecutionStatusRunning.Code(),
|
|
})
|
|
if err != nil {
|
|
// 记录失败到已执行列表
|
|
execInput.ExecutedNodes = append(execInput.ExecutedNodes, flowDto.ExecutedNode{
|
|
NodeId: nodeID,
|
|
Status: node.NodeExecutionStatusFailed.Code(),
|
|
})
|
|
return nil, err
|
|
}
|
|
realInput.NodeExecutionId = nodeExecutionId
|
|
// 执行节点
|
|
_, err = lambda(ctx, realInput)
|
|
durationMs := time.Since(startTime).Milliseconds()
|
|
updateReq := &nodeDto.UpdateNodeExecutionReq{
|
|
Id: nodeExecutionId,
|
|
DurationMs: durationMs,
|
|
}
|
|
if err != nil {
|
|
// 执行失败，更新状态
|
|
updateReq.Status = node.NodeExecutionStatusFailed.Code()
|
|
updateReq.ErrorMessage = err.Error()
|
|
_, _ = nodeDao.NodeExecutionDao.Update(ctx, updateReq)
|
|
// 记录失败到已执行列表
|
|
execInput.ExecutedNodes = append(execInput.ExecutedNodes, flowDto.ExecutedNode{
|
|
NodeId: nodeID,
|
|
Status: node.NodeExecutionStatusFailed.Code(),
|
|
})
|
|
return nil, err
|
|
}
|
|
// 上传OSS（每条独立上传）
|
|
ossResult1, err := Upload(ctx, &dto.UploadFileBytesReq{
|
|
FileBytes: gconv.Bytes(gconv.String(realInput)),
|
|
FileName: fmt.Sprintf("nodeInput:%v.txt", time.Now().UnixMilli()),
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
updateReq.OutputParamsPath = ossResult1.FileURL
|
|
// 执行成功，更新状态
|
|
updateReq.Status = node.NodeExecutionStatusSuccess.Code()
|
|
_, _ = nodeDao.NodeExecutionDao.Update(ctx, updateReq)
|
|
// 记录成功到已执行列表
|
|
execInput.ExecutedNodes = append(execInput.ExecutedNodes, flowDto.ExecutedNode{
|
|
NodeId: nodeID,
|
|
Status: node.NodeExecutionStatusSuccess.Code(),
|
|
})
|
|
|
|
// ✅ 关键：返回整个 execInput，让下一个节点继续用！
|
|
return execInput, nil
|
|
}
|
|
}
|
|
if nodeID == "summary_node" {
|
|
_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(SummaryLambda)))
|
|
return
|
|
}
|
|
switch code {
|
|
case "__start__":
|
|
_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(StartLambda)))
|
|
case node.NodeTypeTextModel:
|
|
_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(TextModelLambda)))
|
|
case node.NodeTypeImageModel:
|
|
_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(ImageModelLambda)))
|
|
case node.NodeTypeVideoModel:
|
|
_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(VideoModelLambda)))
|
|
case node.NodeTypeAudioModel:
|
|
_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(AudioModelLambda)))
|
|
case node.NodeTypeBatchModel:
|
|
_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(BatchModelLambda)))
|
|
case node.NodeTypeDataConversionModel:
|
|
_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(DataConversionLambda)))
|
|
//case node.NodeTypeSenseOptimizeModel:
|
|
// _ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(SenseOptimizeModelLambda)))
|
|
//case node.NodeTypeStoryOptimizeModel:
|
|
// _ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(StoryOptimizeModelLambda)))
|
|
//case node.NodeTypeScriptOptimizeModel:
|
|
// _ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(ScriptOptimizeModelLambda)))
|
|
case node.NodeTypeCustomNode:
|
|
_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(CustomLambda)))
|
|
case node.NodeTypeForm:
|
|
_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(FormLambda)))
|
|
case node.NodeTypeIntent:
|
|
_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(IntentLambda)))
|
|
case node.NodeTypeMerge:
|
|
_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(MergeLambda)))
|
|
case node.NodeTypeDataMerge:
|
|
_ = graph.AddLambdaNode(nodeID, compose.InvokableLambda(wrapLambda(DataMergeLambda)))
|
|
case node.NodeTypeHttp:
|
|
_ = 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{})
|
|
|
|
// 🚀 只从【开始节点】递归遍历（关键修复）
|
|
findLeafNodes(startNodeId, nextMap, endNodeSet)
|
|
|
|
// 转成数组返回
|
|
endNodes := make([]string, 0, len(endNodeSet))
|
|
for id := range endNodeSet {
|
|
endNodes = append(endNodes, id)
|
|
}
|
|
return endNodes
|
|
}
|
|
|
|
// --------------------------------------------------------------------
|
|
// ✅ 递归：查找以 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)
|
|
}
|
|
}
|