red-future/ai-agent
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 添加对话式工作流节点执行与结果合并逻辑

Browse Source
This commit is contained in:
曲皓东
2026-05-15 18:30:44 +08:00
parent ffba1f30ec
commit d5206df131
4 changed files with 957 additions and 327 deletions
Download Patch File Download Diff File Expand all files Collapse all files

618
workflow/service/flow/flow_execution_service.go
View File

@@ -5,16 +5,20 @@ import (
"ai-agent/workflow/consts/node"
fileDao "ai-agent/workflow/dao/file"
flowDao "ai-agent/workflow/dao/flow"
"ai-agent/workflow/model/dto"
fileDto "ai-agent/workflow/model/dto/file"
flowDto "ai-agent/workflow/model/dto/flow"
"ai-agent/workflow/model/entity"
"context"
"errors"
"fmt"
"os"
"regexp"
"sort"
"strconv"
"strings"
"sync"
"time"
"gitea.com/red-future/common/utils"
"github.com/cloudwego/eino/compose"
@@ -351,134 +355,512 @@ func (s *flowExecutionService) Execute(ctx context.Context, req *flowDto.Execute
}
}
//_, err = flowDao.FlowUserDao.Update(ctx, &flowDto.UpdateFlowUserReq{
// Id: req.FlowId,
// FlowContent: req.FlowContent,
// NodeInputParams: req.NodeInputParams,
//})
//if err != nil {
// return nil, err
//}
//nodeInsert := make([]*nodeDto.CreateNodeExecutionReq, 0, len(flowInfo.NodeInputParams))
//for _, flowNode := range flowInfo.NodeInputParams {
// nodeInsert = append(nodeInsert, &nodeDto.CreateNodeExecutionReq{
// FlowExecutionId: executionId,
// NodeId: flowNode.Id,
// Status: node.NodeExecutionStatusWait.Code(),
// NodeInputParams: flowNode,
// TraceId: r.TraceId,
// })
//}
//_, err = nodeDao.NodeExecutionDao.BatchInsert(ctx, nodeInsert)
//if err != nil {
// return
//}
// =========================================================================
// ✅【第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, // ✅ 系统配置
if isDialogue && !g.IsEmpty(flowInfo) {
// 查询节点中是否包含结果合并节点
var htmlUrl []string
var textNodeId string
var textModelName string
var textModelResponse map[string]any
textResultFrom := make(map[string]any)
var imgNodeId string
var imgModelName string
var imgModelResponse map[string]any
imgResultFrom := make(map[string]any)
for _, item := range flowInfo.NodeInputParams {
if item.NodeCode == node.NodeTypeMerge {
for _, outputParamsItem := range flowInfo.OutputParams {
outputParamsMap := gconv.Map(outputParamsItem)
for _, mapItem := range outputParamsMap {
if strings.HasSuffix(gconv.String(mapItem), ".html") {
htmlUrl = append(htmlUrl, gconv.String(mapItem))
}
}
}
}
if item.NodeCode == node.NodeTypeTextModel {
textNodeId = item.Id
textModelName = item.ModelConfig.ModelName
textModelResponse = item.ModelConfig.ModelResponse
for key, modelFormItem := range item.ModelConfig.ModelForm {
textResultFrom[key] = map[string]any{
"value": modelFormItem,
}
}
}
if item.NodeCode == node.NodeTypeImageModel {
imgNodeId = item.Id
imgModelName = item.ModelConfig.ModelName
imgModelResponse = item.ModelConfig.ModelResponse
for key, modelFormItem := range item.ModelConfig.ModelForm {
imgResultFrom[key] = map[string]any{
"value": modelFormItem,
}
}
}
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步】构建执行图
// =========================================================================
runGraph, err := BuildGraphFromFlowContent(execCtx, req.FlowContent, judge2IntentNodeMap, summaryNodeID)
if err != nil {
executionReq := flowDto.UpdateFlowExecutionReq{
Id: executionId,
Status: flow.FlowExecutionStatusFailed.Code(),
ErrorMessage: err.Error(),
var url string
url, err = utils.GetFileAddressPrefix(ctx)
if err != nil {
return nil, err
}
_, err1 := flowDao.FlowExecutionDao.Update(ctx, &executionReq)
if err1 != nil {
return
}
return nil, fmt.Errorf("执行工作流失败: %v", err)
}
if strings.HasSuffix(gconv.String(req.ResultUrl), ".md") {
resultUserFrom := make(map[string]any)
resultUserFrom["desc"] = req.Desc
// =========================================================================
// ✅【第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
var textNode []node.NodeFormField
textNode, err = TextNode(ctx, textNodeId, req.SessionId, textModelName, req.SkillName, textResultFrom, resultUserFrom, textModelResponse, req.FileUrl)
if err != nil {
return nil, err
}
var textContent []string
var textUrl []string
for _, item := range textNode {
if strings.Contains(item.Field, "text_content") {
textContent = append(textContent, item.Value)
}
if strings.Contains(item.Field, "text_url") {
textUrl = append(textUrl, item.Value)
}
}
// =========================================================================
// ✅【第4步】构建全局执行入参（现在 schemaMap 是有值的！）
// =========================================================================
execInput := &flowDto.FlowExecutionInput{
IsDialogue: isDialogue,
ExecutionId: executionId,
ConfigMap: configMap,
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) {
}
content := ""
// 第二步 执行目标节点
if content == "text" {
resultUserFrom := make(map[string]any)
resultUserFrom["desc"] = req.Desc
var textNode []node.NodeFormField
textNode, err = TextNode(ctx, textNodeId, req.SessionId, textModelName, req.SkillName, textResultFrom, resultUserFrom, textModelResponse, req.FileUrl)
if err != nil {
return nil, err
}
var htmlTags []string
var textUrl []string
for _, item := range textNode {
if strings.Contains(item.Field, "text_content") {
htmlTags = append(htmlTags, item.Value)
}
if strings.Contains(item.Field, "text_url") {
textUrl = append(textUrl, item.Value)
}
}
var htmlContentUrl []string
if !g.IsEmpty(htmlUrl) {
for i, item := range htmlUrl {
// 获取当前要替换的文本内容
textContent := htmlTags[i]
// 1. 读取 HTML 文件内容
var htmlBytes []byte
htmlBytes, err = os.ReadFile(url + item)
if err != nil {
fmt.Printf("读取文件失败 %s: %v", url+item, err)
continue
}
htmlContent := string(htmlBytes)
// 2. 构建要替换成的新 div 标签
newTextTag := fmt.Sprintf(`<div class="text">%s</div>`, textContent)
re := regexp.MustCompile(`<text>.*?</text>`)
result := re.ReplaceAllString(htmlContent, newTextTag)
fmt.Printf("成功处理文件：%s", result)
// 上传OSS（每条独立上传）
fileName := fmt.Sprintf("item_%d_%d.html", i, time.Now().UnixMilli())
var ossResult *dto.UploadFileBytesRes
ossResult, err = Upload(ctx, &dto.UploadFileBytesReq{
FileBytes: []byte(result),
FileName: fileName,
})
if err != nil {
return nil, err
}
fmt.Printf("上传OSS成功：%s", ossResult.FileURL)
htmlContentUrl = append(htmlContentUrl, ossResult.FileURL)
}
}
var summaryResult []map[string]interface{}
if !g.IsEmpty(textUrl) {
for _, outputParamsItem := range flowInfo.OutputParams {
if !g.IsEmpty(htmlContentUrl) {
if strings.HasSuffix(gconv.String(outputParamsItem), ".html") {
continue
}
}
if strings.HasSuffix(gconv.String(outputParamsItem), ".txt") {
continue
}
timeKey := strconv.FormatInt(time.Now().UnixMilli(), 10)
item := make(map[string]interface{})
item[timeKey] = outputParamsItem
summaryResult = append(summaryResult, item)
}
for _, textItem := range textUrl {
// 生成 毫秒时间戳 作为 KEY
timeKey := strconv.FormatInt(time.Now().UnixMilli(), 10)
item := make(map[string]interface{})
item[timeKey] = textItem
summaryResult = append(summaryResult, item)
}
}
if !g.IsEmpty(htmlContentUrl) {
for _, outputParamsItem := range flowInfo.OutputParams {
if !g.IsEmpty(textUrl) {
if strings.HasSuffix(gconv.String(outputParamsItem), ".txt") {
continue
}
}
if strings.HasSuffix(gconv.String(outputParamsItem), ".html") {
continue
}
timeKey := strconv.FormatInt(time.Now().UnixMilli(), 10)
item := make(map[string]interface{})
item[timeKey] = outputParamsItem
summaryResult = append(summaryResult, item)
}
for _, textItem := range htmlContentUrl {
// 生成 毫秒时间戳 作为 KEY
timeKey := strconv.FormatInt(time.Now().UnixMilli(), 10)
item := make(map[string]interface{})
item[timeKey] = textItem
summaryResult = append(summaryResult, item)
}
}
if !g.IsEmpty(summaryResult) {
executionReq := flowDto.UpdateFlowExecutionReq{
Id: flowInfo.Id,
Status: flow.FlowExecutionStatusSuccess.Code(),
OutputParams: summaryResult,
}
_, err = flowDao.FlowExecutionDao.Update(ctx, &executionReq)
}
} else if content == "img" {
resultUserFrom := make(map[string]any)
resultUserFrom["desc"] = req.Desc
var imgNode []node.NodeFormField
imgNode, err = ImgNode(ctx, imgNodeId, req.SessionId, imgModelName, req.SkillName, imgResultFrom, resultUserFrom, imgModelResponse, req.FileUrl)
if err != nil {
return nil, err
}
var imgCount int
var imgUrl []string
var htmlBuilder strings.Builder
htmlBuilder.WriteString(`<div class="image-group">`)
for _, item := range imgNode {
if strings.Contains(item.Field, "img_url") {
imgCount = imgCount + 1
htmlBuilder.WriteString(fmt.Sprintf(`<img src="%s" alt="图片"/>`, item.Value))
imgUrl = append(imgUrl, item.Value)
}
}
htmlBuilder.WriteString(`</div>`)
var htmlContentUrl []string
if !g.IsEmpty(htmlUrl) && imgCount > 0 {
for i, item := range htmlUrl {
// 1. 读取 HTML 文件内容
var htmlBytes []byte
htmlBytes, err = os.ReadFile(url + item)
if err != nil {
fmt.Printf("读取文件失败 %s: %v", url+item, err)
continue
}
htmlContent := string(htmlBytes)
re := regexp.MustCompile(`<div class="image-group">[\s\S]*?</div>`)
result := re.ReplaceAllString(htmlContent, htmlBuilder.String())
fmt.Printf("成功处理文件：%s", result)
// 上传OSS（每条独立上传）
fileName := fmt.Sprintf("item_%d_%d.html", i, time.Now().UnixMilli())
var ossResult *dto.UploadFileBytesRes
ossResult, err = Upload(ctx, &dto.UploadFileBytesReq{
FileBytes: []byte(result),
FileName: fileName,
})
if err != nil {
return nil, err
}
fmt.Printf("上传OSS成功：%s", ossResult.FileURL)
htmlContentUrl = append(htmlContentUrl, ossResult.FileURL)
}
}
var summaryResult []map[string]interface{}
if !g.IsEmpty(imgCount) {
for _, outputParamsItem := range flowInfo.OutputParams {
if !g.IsEmpty(htmlContentUrl) {
if strings.HasSuffix(gconv.String(outputParamsItem), ".html") {
continue
}
}
if strings.HasSuffix(gconv.String(outputParamsItem), ".png") {
continue
}
timeKey := strconv.FormatInt(time.Now().UnixMilli(), 10)
item := make(map[string]interface{})
item[timeKey] = outputParamsItem
summaryResult = append(summaryResult, item)
}
for _, textItem := range imgUrl {
// 生成 毫秒时间戳 作为 KEY
timeKey := strconv.FormatInt(time.Now().UnixMilli(), 10)
item := make(map[string]interface{})
item[timeKey] = textItem
summaryResult = append(summaryResult, item)
}
}
if !g.IsEmpty(htmlContentUrl) {
for _, outputParamsItem := range flowInfo.OutputParams {
if !g.IsEmpty(imgUrl) {
if strings.HasSuffix(gconv.String(outputParamsItem), ".png") {
continue
}
}
if strings.HasSuffix(gconv.String(outputParamsItem), ".html") {
continue
}
timeKey := strconv.FormatInt(time.Now().UnixMilli(), 10)
item := make(map[string]interface{})
item[timeKey] = outputParamsItem
summaryResult = append(summaryResult, item)
}
for _, textItem := range htmlContentUrl {
// 生成 毫秒时间戳 作为 KEY
timeKey := strconv.FormatInt(time.Now().UnixMilli(), 10)
item := make(map[string]interface{})
item[timeKey] = textItem
summaryResult = append(summaryResult, item)
}
}
if !g.IsEmpty(summaryResult) {
executionReq := flowDto.UpdateFlowExecutionReq{
Id: flowInfo.Id,
Status: flow.FlowExecutionStatusSuccess.Code(),
OutputParams: summaryResult,
}
_, err = flowDao.FlowExecutionDao.Update(ctx, &executionReq)
}
} else if content == "text_img" {
//userFrom := make(map[string]any)
//userFrom["desc"] = req.Desc
//
//var textNode []node.NodeFormField
//textNode, err = TextNode(ctx, textNodeId, req.SessionId, textModelName, req.SkillName, textResultFrom, userFrom, textModelResponse, req.FileUrl)
//if err != nil {
// return nil, err
//}
//
//var htmlTags []string
//var textUrl []string
//for _, item := range textNode {
// if strings.Contains(item.Field, "text_content") {
// htmlTags = append(htmlTags, StripHtmlTags(item.Value, false))
// }
// if strings.Contains(item.Field, "text_url") {
// textUrl = append(textUrl, item.Value)
// }
//}
//
//userFrom["prompt"] = htmlTags
//var imgNode []node.NodeFormField
//imgNode, err = ImgNode(ctx, imgNodeId, req.SessionId, imgModelName, req.SkillName, imgResultFrom, userFrom, imgModelResponse, req.FileUrl)
//if err != nil {
// return nil, err
//}
//var imgCount int
//var imgUrl []string
//var htmlBuilder strings.Builder
//htmlBuilder.WriteString(`<div class="image-group">`)
//for _, item := range imgNode {
// if strings.Contains(item.Field, "img_url") {
// imgCount = imgCount + 1
// htmlBuilder.WriteString(fmt.Sprintf(`<img src="%s" alt="图片"/>`, item.Value))
// imgUrl = append(imgUrl, item.Value)
// }
//}
//htmlBuilder.WriteString(`</div>`)
//
//var htmlContentUrl []string
//if !g.IsEmpty(htmlUrl) && imgCount > 0 {
// for i, item := range htmlUrl {
// // 获取当前要替换的文本内容
// textContent := htmlTags[i]
// // 1. 读取 HTML 文件内容
// var htmlBytes []byte
// htmlBytes, err = os.ReadFile(url + item)
// if err != nil {
// fmt.Printf("读取文件失败 %s: %v", url+item, err)
// continue
// }
// htmlContent := string(htmlBytes)
//
// re := regexp.MustCompile(`<div class="image-group">[\s\S]*?</div>`)
// result := re.ReplaceAllString(htmlContent, htmlBuilder.String())
// // 2. 构建要替换成的新 div 标签
// newTextTag := fmt.Sprintf(`<div class="text">%s</div>`, textContent)
// ret := regexp.MustCompile(`<text>.*?</text>`)
// result = ret.ReplaceAllString(htmlContent, newTextTag)
// fmt.Printf("成功处理文件：%s", result)
//
// // 上传OSS（每条独立上传）
// fileName := fmt.Sprintf("item_%d_%d.html", i, time.Now().UnixMilli())
// var ossResult *dto.UploadFileBytesRes
// ossResult, err = Upload(ctx, &dto.UploadFileBytesReq{
// FileBytes: []byte(result),
// FileName: fileName,
// })
// if err != nil {
// return nil, err
// }
// fmt.Printf("上传OSS成功：%s", ossResult.FileURL)
// htmlContentUrl = append(htmlContentUrl, ossResult.FileURL)
// }
//}
//if !g.IsEmpty(htmlContentUrl) {
// for _, outputParamsItem := range flowInfo.OutputParams {
// if !g.IsEmpty(imgUrl) {
// if strings.HasSuffix(gconv.String(outputParamsItem), ".png") {
// continue
// }
// }
// if strings.HasSuffix(gconv.String(outputParamsItem), ".html") {
// continue
// }
// timeKey := strconv.FormatInt(time.Now().UnixMilli(), 10)
// item := make(map[string]interface{})
// item[timeKey] = outputParamsItem
// summaryResult = append(summaryResult, item)
// }
// for _, textItem := range htmlContentUrl {
// // 生成 毫秒时间戳 作为 KEY
// timeKey := strconv.FormatInt(time.Now().UnixMilli(), 10)
// item := make(map[string]interface{})
// item[timeKey] = textItem
// summaryResult = append(summaryResult, item)
// }
//}
//if !g.IsEmpty(summaryResult) {
// executionReq := flowDto.UpdateFlowExecutionReq{
// Id: flowInfo.Id,
// Status: flow.FlowExecutionStatusSuccess.Code(),
// OutputParams: summaryResult,
// }
// _, err = flowDao.FlowExecutionDao.Update(ctx, &executionReq)
//}
} else {
return nil, fmt.Errorf("意图识别失败")
}
} else {
// =========================================================================
// ✅【第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.FlowExecutionStatusCancel.Code(),
Id: executionId,
Status: flow.FlowExecutionStatusFailed.Code(),
ErrorMessage: err.Error(),
}
_, _ = flowDao.FlowExecutionDao.Update(ctx, &executionReq)
return nil, fmt.Errorf("工作流已被取消: %v", err)
_, err1 := flowDao.FlowExecutionDao.Update(ctx, &executionReq)
if err1 != nil {
return
}
return nil, fmt.Errorf("执行工作流失败: %v", err)
}
executionReq := flowDto.UpdateFlowExecutionReq{
Id: executionId,
Status: flow.FlowExecutionStatusFailed.Code(),
ErrorMessage: err.Error(),
// =========================================================================
// ✅【第3步】构建 ConfigMap
// =========================================================================
configMap := make(map[string]*entity.FlowNode)
for _, cfg := range req.NodeInputParams {
configMap[cfg.Id] = cfg
}
_, err1 := flowDao.FlowExecutionDao.Update(ctx, &executionReq)
if err1 != nil {
return
// 自动给意图节点复制配置
for judgeID, intentID := range judge2IntentNodeMap {
if cfg, ok := configMap[judgeID]; ok {
configMap[intentID] = cfg
}
}
// 初始化汇总节点配置
configMap[summaryNodeID] = &summaryNode
// =========================================================================
// ✅【第4步】构建全局执行入参（现在 schemaMap 是有值的！）
// =========================================================================
execInput := &flowDto.FlowExecutionInput{
IsDialogue: isDialogue,
ExecutionId: executionId,
ConfigMap: configMap,
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 nil, fmt.Errorf("执行工作流失败: %v", err)
}
return