feat: 支持工作流断点续跑并拆分错误信息存储

- 新增同会话+同工作流最近执行失败且参数一致时断点续跑逻辑
- exec_workflow/exec_chat 新增 error 字段存储原始错误,error_message 仅存友好提示
- 新增 UpdateExecChatReq 与 exec_chat_dao Update 方法
- 新增 GetLatestBySessionAndFlow 查询最近执行记录
- 修正 ListDates 分组与排序 SQL 表达式
- 新增 pipeline 配置结构,删除旧设计文档
This commit is contained in:
2026-08-21 09:54:06 +08:00
parent a939c45508
commit cef837a35c
34 changed files with 1519 additions and 515 deletions
@@ -1,10 +1,13 @@
// Package split_batch 工作流前置处理器:按各字段 constraint.uploadTotalMaxCount 拆分模型请求参数。
// Package split_batch 工作流前置处理器:按默认上限(uploadTotalMaxCount 默认 15拆分模型请求参数为多批
// 入参为已构建好的扁平模型请求体(BuildModelRequestBody 输出,key 为点分路径,value 已解析填充),
// 集合字段按上限分批、元素对象取 url 为值;返回结构同入参(扁平 map 数组),未超量返回单份。
// 处理器实现自包含(算法随处理器走,不依赖业务包),通过 init 注册进 processor 注册表。
package split_batch
import (
"context"
"fmt"
"reflect"
"sort"
"github.com/gogf/gf/v2/util/gconv"
@@ -12,16 +15,23 @@ import (
"ai-agent/workflow/service/flow/processor"
)
// ProcessorName 处理器注册名,供模型节点前置工具分发按名判定入参形态(扁平请求体)。
const ProcessorName = "split_batch_model_params"
// defaultMaxCount 每批最大元素数(constraint.uploadTotalMaxCount 当前默认值,后续动态传递)。
const defaultMaxCount = 15
func init() {
processor.Register(SplitBatchModelParamsProcessor())
}
// SplitBatchModelParamsProcessor 将模型请求参数按各字段 constraint.uploadTotalMaxCount 分批的前置处理器。
// 入参 args 即模型请求参数本体(需已解析好 valueSourcevalue 已填充)。
// SplitBatchModelParamsProcessor 将模型请求参数按默认上限分批的前置处理器。
// 入参 args 即扁平模型请求体(valueSource 已解析、value 已填充)。
func SplitBatchModelParamsProcessor() *processor.Processor {
return &processor.Processor{
Name: "split_batch_model_params",
Description: "按 constraint.uploadTotalMaxCount 分批模型请求数",
Name: ProcessorName,
Description: "按最大约束构建分批模型请求数",
IsShow: true,
Func: func(ctx context.Context, args map[string]any) (any, error) {
if args == nil {
return nil, fmt.Errorf("缺少模型请求参数")
@@ -31,37 +41,41 @@ func SplitBatchModelParamsProcessor() *processor.Processor {
}
}
// splitBatchField 描述一个需要按 constraint.uploadTotalMaxCount 分批的字段
type splitBatchField struct {
path []any // 从参数根节点到该字段 value 的路径(map 用 key,数组用下标)
items []any // 该字段 value 拆出的待分批元素列表
maxCount int // 每批最大元素数(constraint.uploadTotalMaxCount
// batchField 描述一个需要分批的扁平字段
type batchField struct {
key string // 扁平点分 key
items []any // 分批元素(元素为带 url 字段的对象时已取 url 为值)
}
// SplitBatchModelParams 把模型请求参数拆成多批,供分批请求模型使用。
// SplitBatchModelParams 把扁平模型请求拆成多批,供分批请求模型使用。
// 处理流程:
// 1. 深拷贝入参,避免污染调用方数据;
// 2. 递归解析 valueSource,把引用节点输出的字段值写入对应字段的 value;
// 3. 找出所有带 constraint.uploadTotalMaxCount 且 value 为集合(map/slice)的字段,
// map 按 key 排序取 value 列表作为元素;总批数 = 各字段 (元素数/上限) 向上取整的最大值;
// 4. 每批 = 整份参数深拷贝 + 各分批字段 value 替换为对应切片。
// 2. 遍历扁平字段,值为集合(slice/array/map)的按元素数 / defaultMaxCount 向上取整,
// map 按 key 排序取 value 列表作为元素;元素为带 url 字段的对象时取 url 为值;
// 3. 总批数 = 各字段批数的最大值;每批 = 整份参数深拷贝 + 各分批字段替换为该批切片,
// 元素已耗尽的分批字段替换为切片。
//
// 未超量时返回单份value 已被解析填充)参数。调用方可遍历返回值逐个请求模型。
// 未超量时返回单份参数。调用方可遍历返回值逐个请求模型。
func SplitBatchModelParams(rawParams map[string]any) []map[string]any {
params, ok := deepCopyAny(rawParams).(map[string]any)
if !ok {
params = make(map[string]any)
}
fields := make([]splitBatchField, 0)
collectBatchFields(params, nil, &fields)
fields := make([]batchField, 0)
batchCount := 1
for _, f := range fields {
n := (len(f.items) + f.maxCount - 1) / f.maxCount
for key, v := range params {
items, has := toItems(v)
if !has {
continue
}
n := (len(items) + defaultMaxCount - 1) / defaultMaxCount
if n > batchCount {
batchCount = n
}
if n > 1 {
fields = append(fields, batchField{key: key, items: items})
}
}
if batchCount <= 1 {
return []map[string]any{params}
@@ -71,67 +85,46 @@ func SplitBatchModelParams(rawParams map[string]any) []map[string]any {
for i := 0; i < batchCount; i++ {
batch, _ := deepCopyAny(params).(map[string]any)
for _, f := range fields {
start := i * f.maxCount
start := i * defaultMaxCount
if start >= len(f.items) {
setValueAtPath(batch, f.path, []any{})
batch[f.key] = []any{}
continue
}
end := start + f.maxCount
end := start + defaultMaxCount
if end > len(f.items) {
end = len(f.items)
}
setValueAtPath(batch, f.path, f.items[start:end])
batch[f.key] = f.items[start:end]
}
batches = append(batches, batch)
}
return batches
}
// collectBatchFields 递归收集所有带 constraint.uploadTotalMaxCount 且 value 为集合的分批字段
func collectBatchFields(node any, path []any, out *[]splitBatchField) {
switch v := node.(type) {
case map[string]any:
if maxCount, ok := uploadTotalMaxCountOf(v); ok {
if items, has := toItems(v["value"]); has {
*out = append(*out, splitBatchField{
path: append(append([]any{}, path...), "value"),
items: items,
maxCount: maxCount,
})
}
}
for key, child := range v {
collectBatchFields(child, append(append([]any{}, path...), key), out)
}
case []any:
for i, child := range v {
collectBatchFields(child, append(append([]any{}, path...), i), out)
// itemValue 取集合元素作为分批粒度时的值:元素为带 url 字段的对象时取 url"取url为值"),
// 其余元素原样保留。
func itemValue(e any) any {
if m := gconv.Map(e); m != nil {
if u, ok := m["url"]; ok && u != nil {
return u
}
}
return e
}
// uploadTotalMaxCountOf 读取 schema 节点 constraint.uploadTotalMaxCount
func uploadTotalMaxCountOf(m map[string]any) (int, bool) {
c, ok := m["constraint"]
if !ok {
return 0, false
}
cm := gconv.Map(c)
if cm == nil {
return 0, false
}
n := gconv.Int(cm["uploadTotalMaxCount"])
if n <= 0 {
return 0, false
}
return n, true
}
// toItems 把集合 value 转成元素列表:切片原样返回;map 按 key 排序取 value,保证分批顺序稳定
// toItems 把集合 value 转成元素列表:切片逐元素转换(对象取 url 为值);map 按 key 排序取 value
// 类型化切片([]string 等)经反射逐元素转换。
func toItems(v any) ([]any, bool) {
switch val := v.(type) {
case []any:
return val, len(val) > 0
if len(val) == 0 {
return nil, false
}
items := make([]any, 0, len(val))
for _, e := range val {
items = append(items, itemValue(e))
}
return items, true
case map[string]any:
if len(val) == 0 {
return nil, false
@@ -143,50 +136,26 @@ func toItems(v any) ([]any, bool) {
sort.Strings(keys)
items := make([]any, 0, len(keys))
for _, k := range keys {
items = append(items, val[k])
items = append(items, itemValue(val[k]))
}
return items, true
}
rv := reflect.ValueOf(v)
if rv.IsValid() && (rv.Kind() == reflect.Slice || rv.Kind() == reflect.Array) {
n := rv.Len()
if n == 0 {
return nil, false
}
items := make([]any, 0, n)
for i := 0; i < n; i++ {
items = append(items, itemValue(rv.Index(i).Interface()))
}
return items, true
}
return nil, false
}
// setValueAtPath 沿 path 逐级导航(map 用 key、数组用下标),在最后一级写入 value
func setValueAtPath(root map[string]any, path []any, value any) {
if len(path) == 0 {
return
}
var cur any = root
for i := 0; i < len(path)-1; i++ {
switch step := path[i].(type) {
case string:
m, ok := cur.(map[string]any)
if !ok {
return
}
cur = m[step]
case int:
s, ok := cur.([]any)
if !ok || step < 0 || step >= len(s) {
return
}
cur = s[step]
default:
return
}
}
switch last := path[len(path)-1].(type) {
case string:
if m, ok := cur.(map[string]any); ok {
m[last] = value
}
case int:
if s, ok := cur.([]any); ok && last >= 0 && last < len(s) {
s[last] = value
}
}
}
// deepCopyAny 深拷贝 map[string]any / []any 嵌套结构,避免批次之间互相影响
// deepCopyAny 深拷贝 map[string]any / []any / 类型化切片嵌套结构,避免批次之间互相影响
func deepCopyAny(v any) any {
switch val := v.(type) {
case map[string]any:
@@ -201,7 +170,18 @@ func deepCopyAny(v any) any {
res[i] = deepCopyAny(child)
}
return res
default:
return val
}
rv := reflect.ValueOf(v)
if rv.IsValid() && (rv.Kind() == reflect.Slice || rv.Kind() == reflect.Array) {
n := rv.Len()
out := reflect.MakeSlice(rv.Type(), n, n)
for i := 0; i < n; i++ {
d := deepCopyAny(rv.Index(i).Interface())
if d != nil {
out.Index(i).Set(reflect.ValueOf(d))
}
}
return out.Interface()
}
return v
}