19904408334
d699f7ce14
- 新增计费模块：执行开始建单、终态结算/取消/失败处理，支持按条/按秒/按token计费 - 新增执行生命周期跟踪：优雅关停时取消运行中执行并等待落库 - 新增异步任务等待/通知机制（Wait/Notify） - 重构执行记录落库与进度上报，统一失败分类与重试语义 - 重命名文件：async_task.go→async.go、flow_checkpoint_store.go→exec_checkpoint.go、flow_graph_util.go→exec_record.go - 更新 .gitignore 与数据库密码配置
255 lines
8.5 KiB
Go
255 lines
8.5 KiB
Go
package values
|
||
|
||
import (
|
||
"ai-agent/workflow/consts/node"
|
||
flowDto "ai-agent/workflow/model/dto/flow"
|
||
"ai-agent/workflow/model/entity"
|
||
"encoding/json"
|
||
"regexp"
|
||
"strconv"
|
||
"strings"
|
||
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
"github.com/gogf/gf/v2/util/gconv"
|
||
"github.com/tidwall/gjson"
|
||
)
|
||
|
||
var (
|
||
// 匹配 [数字]
|
||
regNumIndex = regexp.MustCompile(`\[\d+\]`)
|
||
// 匹配 .attrs
|
||
regAttrs = regexp.MustCompile(`\.attrs`)
|
||
// 匹配带捕获组的数组下标，转扁平点分路径用
|
||
arrayIndexPath = regexp.MustCompile(`\[(\d+)\]`)
|
||
)
|
||
|
||
// CleanFieldPath 清理字段路径：移除 .attrs、数字下标转为 .#（gjson 数组通配符）
|
||
// 示例：usage.attrs.total_tokens → usage.total_tokens
|
||
// 示例：choices.attrs[0].attrs.message.attrs.content → choices.#.message.content
|
||
func CleanFieldPath(path string) string {
|
||
index := CleanFieldPathReplaceNumIndex(path)
|
||
attrs := CleanFieldPathRemoveAttrs(index)
|
||
return attrs
|
||
}
|
||
|
||
func CleanFieldPathReplaceNumIndex(path string) string {
|
||
// 1. 替换 [数字] 为 [*]
|
||
s := regNumIndex.ReplaceAllString(path, `.#`)
|
||
return s
|
||
}
|
||
|
||
func CleanFieldPathRemoveAttrs(path string) string {
|
||
// 2. 移除所有 .attrs
|
||
s := regAttrs.ReplaceAllString(path, "")
|
||
return s
|
||
}
|
||
|
||
// ProcessValueSourceRecursive 递归遍历map，同级同时存在value和valueSource则把value设置为"AA"
|
||
func ProcessValueSourceRecursive(rawParams map[string]interface{}, globalParams *flowDto.FlowExecutionInput) {
|
||
walkMap(rawParams, globalParams)
|
||
}
|
||
|
||
// ResolveValueSource 解析 valueSource {nodeId, field} 引用的实际值。
|
||
// 返回 (value, refsName, ok)；ok=false 表示引用节点不存在或引用值仍为空。
|
||
// - 开始/表单节点：OutputConfig 平铺条目按 field == 引用字段匹配（前端约定以 field 为主，
|
||
// 不兼容 path），直接读 entry 的 value / refsName
|
||
// - scriptTranscribe 节点：OutputResult 是各段扁平请求参数，按段序收集字段为数组（段位留 nil）
|
||
// - 其他节点：读 OutputResult 中引用字段 field 路径对应的值
|
||
func ResolveValueSource(global *flowDto.FlowExecutionInput, nodeId, field string) (value any, refsName any, ok bool) {
|
||
if global == nil || global.ConfigMap == nil {
|
||
return nil, nil, false
|
||
}
|
||
nodeConfig := global.ConfigMap[nodeId]
|
||
if nodeConfig == nil {
|
||
return nil, nil, false
|
||
}
|
||
switch nodeConfig.NodeCode {
|
||
case node.NodeTypeStart, node.NodeTypeForm:
|
||
for _, output := range nodeConfig.OutputConfig {
|
||
if gconv.String(output["field"]) != field {
|
||
continue
|
||
}
|
||
if !g.IsEmpty(output["value"]) {
|
||
return output["value"], output["refsName"], true
|
||
}
|
||
}
|
||
case node.NodeTypeScriptTranscribe:
|
||
// 脚本转写节点 OutputResult 是各段扁平请求参数（split_shots_pipeline 产出，key 为字面量
|
||
// prompt/duration/seed 等），按段序读取 output[field] 收集为数组，供分段模型节点整体引用。
|
||
// 每段都占一位（字段缺失/为空留 nil），保证数组与段序对齐，供 split_segment 按段取值。
|
||
var list []any
|
||
for _, output := range nodeConfig.OutputResult {
|
||
list = append(list, output[field])
|
||
}
|
||
for _, v := range list {
|
||
if !g.IsEmpty(v) {
|
||
return list, "", true
|
||
}
|
||
}
|
||
default:
|
||
// templates 是模型节点在前端配置的静态输出模板，不在 OutputResult 中，需单独取
|
||
if field == "templates" {
|
||
if !g.IsEmpty(nodeConfig.Templates) {
|
||
return nodeConfig.Templates, "", true
|
||
}
|
||
return nil, nil, false
|
||
}
|
||
for _, output := range nodeConfig.OutputResult {
|
||
// 模型节点输出记录是单 key 的字面量扁平 key（如 "choices.attrs[0].attrs.delta.attrs.content"），
|
||
// gjson 会把 . 和 [0] 当结构路径解析，无法命中字面量 key，故先按字面量 key 直接取值；
|
||
// 未命中再回退 gjson 路径查询（兼容真正嵌套的输出结构）。
|
||
if v, has := output[field]; has {
|
||
value = v
|
||
} else {
|
||
value = gjson.Get(gconv.String(output), field).Value()
|
||
}
|
||
if !g.IsEmpty(value) {
|
||
return value, gjson.Get(gconv.String(output), CleanFieldPath("refsName")).Value(), true
|
||
}
|
||
}
|
||
}
|
||
return nil, nil, false
|
||
}
|
||
|
||
// walkMap 递归处理map/数组
|
||
func walkMap(data interface{}, globalParams *flowDto.FlowExecutionInput) {
|
||
switch v := data.(type) {
|
||
case map[string]interface{}:
|
||
// 有 valueSource：解析引用节点值
|
||
if valueSource, hasSource := v["valueSource"]; hasSource {
|
||
sources := new([]entity.ValueSource)
|
||
gconv.Structs(valueSource, sources)
|
||
|
||
// 多个引用源：把各源解析出的值拼成 "label: value"（无 label 只拼值），逗号分隔
|
||
if len(*sources) > 1 {
|
||
text, refsName, ok := joinValueSources(globalParams, *sources, schemaValueEmpty,
|
||
func(src entity.ValueSource, value any) any {
|
||
// 引用非模型节点（开始/表单/HTTP/脚本转写等）时，值按当前字段声明的 type 做类型化转换；
|
||
// 模型节点值由模型网关处理，复制时不需要转换
|
||
if !isModelSourceNode(globalParams, src.NodeId) {
|
||
return assignBySchemaType(v, value)
|
||
}
|
||
return value
|
||
})
|
||
if ok {
|
||
v["value"] = text
|
||
if !g.IsEmpty(refsName) {
|
||
v["refsName"] = refsName
|
||
}
|
||
return
|
||
}
|
||
} else if len(*sources) == 1 {
|
||
// 单个引用源：保持旧行为，值按原样赋值（非模型节点按声明 type 转换）
|
||
src := (*sources)[0]
|
||
value, refsName, ok := ResolveValueSource(globalParams, src.NodeId, src.Field)
|
||
if ok && !isModelSourceNode(globalParams, src.NodeId) {
|
||
value = assignBySchemaType(v, value)
|
||
}
|
||
if ok && !schemaValueEmpty(value) {
|
||
v["value"] = value
|
||
if !g.IsEmpty(refsName) {
|
||
v["refsName"] = refsName
|
||
}
|
||
return
|
||
}
|
||
}
|
||
}
|
||
// 统一兜底：无 valueSource（或解析失败/值为空）时，value 为空或 0 则取 defaultValue
|
||
if defaultValue, hasDefault := v["defaultValue"]; hasDefault && isEmptyForFallback(v["value"]) && !schemaValueEmpty(defaultValue) {
|
||
v["value"] = defaultValue
|
||
}
|
||
// 递归遍历所有子元素
|
||
for _, child := range v {
|
||
walkMap(child, globalParams)
|
||
}
|
||
case []interface{}:
|
||
// 数组遍历
|
||
for _, item := range v {
|
||
walkMap(item, globalParams)
|
||
}
|
||
}
|
||
}
|
||
|
||
// isEmptyForFallback 兜底场景判空：除 schemaValueEmpty 规则外，数字 0 也视为未填写，
|
||
// 便于配置了 defaultValue 的字段在值为 0 时用默认值兜底。
|
||
// 覆盖 json.Number（gconv 反序列化数字的运行时类型）与字符串 "0"/"0.0"。
|
||
func isEmptyForFallback(v interface{}) bool {
|
||
if schemaValueEmpty(v) {
|
||
return true
|
||
}
|
||
switch val := v.(type) {
|
||
case float32:
|
||
return val == 0
|
||
case float64:
|
||
return val == 0
|
||
case int:
|
||
return val == 0
|
||
case int8:
|
||
return val == 0
|
||
case int16:
|
||
return val == 0
|
||
case int32:
|
||
return val == 0
|
||
case int64:
|
||
return val == 0
|
||
case uint:
|
||
return val == 0
|
||
case uint8:
|
||
return val == 0
|
||
case uint16:
|
||
return val == 0
|
||
case uint32:
|
||
return val == 0
|
||
case uint64:
|
||
return val == 0
|
||
case json.Number:
|
||
if f, err := val.Float64(); err == nil {
|
||
return f == 0
|
||
}
|
||
case string:
|
||
if f, err := strconv.ParseFloat(val, 64); err == nil {
|
||
return f == 0
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
// isModelSourceNode 判断引用源节点是否为模型节点（值由模型网关处理，复制时不转换）
|
||
func isModelSourceNode(global *flowDto.FlowExecutionInput, nodeId string) bool {
|
||
if global == nil || global.ConfigMap == nil {
|
||
return false
|
||
}
|
||
nodeConfig := global.ConfigMap[nodeId]
|
||
return nodeConfig != nil && nodeConfig.NodeCode == node.NodeTypeModel
|
||
}
|
||
|
||
// joinValueSources 拼接多个 valueSource 的解析值为 "label: value"（无 label 只拼值），逗号分隔。
|
||
// 返回拼接文本与第一个非空 refsName；所有源都为空/解析失败时 ok=false。
|
||
// isEmpty 为各场景的空值判断（walkMap 用 schemaValueEmpty，模型请求解析用 isParamEmpty）；
|
||
// transform 对每个非空解析值做转换（walkMap 按声明 type 类型化、模型源不转换；模型请求场景传 nil）。
|
||
func joinValueSources(global *flowDto.FlowExecutionInput, sources []entity.ValueSource,
|
||
isEmpty func(any) bool, transform func(src entity.ValueSource, value any) any) (text string, refsName any, ok bool) {
|
||
var parts []string
|
||
for _, src := range sources {
|
||
value, rn, ok := ResolveValueSource(global, src.NodeId, src.Field)
|
||
if !ok || isEmpty(value) {
|
||
continue
|
||
}
|
||
if transform != nil {
|
||
value = transform(src, value)
|
||
}
|
||
s := toPlainString(value)
|
||
if src.Label != "" {
|
||
s = src.Label + ": " + s
|
||
}
|
||
parts = append(parts, s)
|
||
if !g.IsEmpty(rn) && g.IsEmpty(refsName) {
|
||
refsName = rn
|
||
}
|
||
}
|
||
if len(parts) == 0 {
|
||
return "", nil, false
|
||
}
|
||
return strings.Join(parts, ", "), refsName, true
|
||
}
|