diff --git a/gateway/model.go b/gateway/model.go index 4f75125..e1391c9 100644 --- a/gateway/model.go +++ b/gateway/model.go @@ -138,7 +138,7 @@ func ModelCallResult(ctx context.Context, modelId int64, responseType model.Resp // 异步模型必须绑定消息主题接收结果:自动生成唯一主题, // 带业务标识(bizName/modelId/sessionId)便于排查,每次调用唯一避免并发串结果 msgTopic := "" - if *responseType == *model.ResponseTypeAsync.Code() { + if responseType != nil && *responseType == *model.ResponseTypeAsync.Code() { msgTopic = modelCallTopic(g.Cfg().MustGet(ctx, "server.name").String(), modelId, sessionId) } @@ -166,7 +166,7 @@ func ModelCallResult(ctx context.Context, modelId int64, responseType model.Resp return nil, fmt.Errorf("创建模型任务失败:%v", res.ErrorMsg) } // 3. 订阅模型结果(异步模型) - if *responseType == *model.ResponseTypeAsync.Code() { + if responseType != nil && *responseType == *model.ResponseTypeAsync.Code() { resultCh := make(chan *ModelCallRes, 1) errCh := make(chan error, 1) diff --git a/workflow/service/flow/flow_graph_builder.go b/workflow/service/flow/flow_graph_builder.go index 9910901..651fbed 100644 --- a/workflow/service/flow/flow_graph_builder.go +++ b/workflow/service/flow/flow_graph_builder.go @@ -17,14 +17,47 @@ import ( // BuildGraph 根据 FlowInfo 构建完整的 Eino Graph 拓扑 func BuildGraph(ctx context.Context, flowContent *entity.FlowInfo) ([]entity.FlowNode, *compose.Graph[any, any]) { // 注册自定义合并函数:处理 *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 + // 首次运行所有并行分支共享同一个 ConfigMap 指针,直接返回 values[0] 即可。 + // 但续跑(ReExecute)时各分支从 checkpoint 反序列化出独立的 ConfigMap 副本, + // 只返回 values[0] 会丢失其他分支写入的 OutputResult(用户实测:node-8 成功的结果 + // 在汇合节点 node-7 变 null)。以第一个为基底,把其余分支中缺失的节点输出合并进来。 + base := values[0] + for _, v := range values[1:] { + if v == nil { + continue + } + for nodeId, cfg := range v.ConfigMap { + if cfg == nil { + continue + } + baseCfg, ok := base.ConfigMap[nodeId] + if !ok || baseCfg == nil { + base.ConfigMap[nodeId] = cfg + continue + } + if len(baseCfg.OutputResult) == 0 && len(cfg.OutputResult) > 0 { + baseCfg.OutputResult = cfg.OutputResult + } + } + // 合并已执行节点列表(按 NodeId 去重),续跑时被恢复分支的进度不丢失 + for _, en := range v.ExecutedNodes { + dup := false + for _, b := range base.ExecutedNodes { + if b.NodeId == en.NodeId { + dup = true + break + } + } + if !dup { + base.ExecutedNodes = append(base.ExecutedNodes, en) + } + } + } + return base, nil }) graph := compose.NewGraph[any, any]( diff --git a/workflow/service/flow/lambda_node_util.go b/workflow/service/flow/lambda_node_util.go index 7ad2f6d..3dd5e4a 100644 --- a/workflow/service/flow/lambda_node_util.go +++ b/workflow/service/flow/lambda_node_util.go @@ -69,6 +69,11 @@ func ModelCallResultLambda(ctx context.Context, modelId int64, sessionId string, if err != nil { return nil, nil, false, fmt.Errorf("获取模型配置失败: %w", err) } + // model-gateway 对不存在的模型返回 modelManage=null(HTTP 仍 200),零值 struct 里 ResponseType 是 nil 指针, + // 直接传入 ModelCallResult 会在 *responseType 处 panic,这里提前报业务错误 + if g.IsEmpty(modelInfo.ModelManage.Id) { + return nil, nil, false, fmt.Errorf("模型配置不存在: modelId=%d", modelId) + } businessParams := make(map[string]any) if !g.IsEmpty(prompt) { if modelInfo.ModelManage.ModelType != nil && *modelInfo.ModelManage.ModelType == model.TypeVideo {