From 08251d9a7395f33ef30298af9cfdd40fe914f811 Mon Sep 17 00:00:00 2001 From: qhd <1766646056@qq.com> Date: Thu, 9 Jul 2026 13:44:13 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E5=B9=B6=E5=8F=91?= =?UTF-8?q?=E5=8F=96=E6=B6=88=E9=80=BB=E8=BE=91=E4=B8=8EHTTP=E8=AF=B7?= =?UTF-8?q?=E6=B1=82=E8=B6=85=E6=97=B6=E5=8F=8A=E5=93=8D=E5=BA=94=E8=A7=A3?= =?UTF-8?q?=E6=9E=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- workflow/service/flow/lambda_node.go | 35 ++++---- workflow/service/flow/lambda_node_util.go | 100 +++++++++++++++++++--- 2 files changed, 106 insertions(+), 29 deletions(-) diff --git a/workflow/service/flow/lambda_node.go b/workflow/service/flow/lambda_node.go index 570dae4..afd8e7c 100644 --- a/workflow/service/flow/lambda_node.go +++ b/workflow/service/flow/lambda_node.go @@ -161,9 +161,8 @@ func BatchModelLambda(ctx context.Context, input any) (any, error) { res := make([][]node.NodeFormField, len(reqMap)) var wg sync.WaitGroup - subCtx, cancel := context.WithCancel(ctx) - defer cancel() - + // 只创建基础上下文，不再主动批量 cancel + subCtx := context.WithoutCancel(ctx) // 缓冲1错误通道，仅接收第一个错误 errCh := make(chan error, 1) @@ -173,7 +172,7 @@ func BatchModelLambda(ctx context.Context, input any) (any, error) { go func(idx int, userItem map[string]any) { defer wg.Done() - // 上下文已取消则直接退出 + // 基础上下文仅响应上游原始 ctx 取消，内部任务失败不触发这里 select { case <-subCtx.Done(): return @@ -181,12 +180,12 @@ func BatchModelLambda(ctx context.Context, input any) (any, error) { } singleUserFrom := []map[string]any{userItem} + // 下游调用使用 subCtx，不会因为同批次其他任务报错而取消 output, err := TextNode(subCtx, nodeInput, skillName, from, singleUserFrom) if err != nil { - // 仅第一个错误写入通道 + // 只往错误通道塞第一个错误，不调用全局 cancel select { case errCh <- err: - cancel() // 触发全局取消，其他协程快速退出 default: } return @@ -195,26 +194,30 @@ func BatchModelLambda(ctx context.Context, input any) (any, error) { }(idx, item) } - // 任务全部结束后关闭错误通道 + // 所有协程跑完再关闭通道 go func() { wg.Wait() close(errCh) }() - // ========== 修正后的等待逻辑 ========== + // ========== 修复区域 start ========== var execErr error select { - // 优先捕获业务错误 case execErr = <-errCh: - if execErr != nil { - // 收到真实业务错误，等待剩余协程收尾后返回 - wg.Wait() - return nil, execErr - } - // execErr == nil 代表通道关闭、无任何错误，走到下方返回完整结果 + // 捕获第一个业务错误，等待剩余协程收尾 + wg.Wait() case <-subCtx.Done(): - // 上下文被取消，阻塞读完errCh，确认是否存在业务错误 + // 上游根上下文被终止，读取已存在的错误 execErr = <-errCh + wg.Wait() + if execErr != nil { + execErr = fmt.Errorf("global context canceled: %w", execErr) + } + } + + // 有错误直接返回，不再走结果拼接 + if execErr != nil { + return nil, execErr } // 拼接输出结果 diff --git a/workflow/service/flow/lambda_node_util.go b/workflow/service/flow/lambda_node_util.go index 1fefe23..1670ec9 100644 --- a/workflow/service/flow/lambda_node_util.go +++ b/workflow/service/flow/lambda_node_util.go @@ -9,6 +9,7 @@ import ( "ai-agent/workflow/model/entity" "bytes" "context" + "errors" "fmt" "io" "mime/multipart" @@ -20,11 +21,13 @@ import ( "strconv" "strings" "sync" + "time" commonHttp "gitea.redpowerfuture.com/red-future/common/http" "gitea.redpowerfuture.com/red-future/common/utils" "github.com/gogf/gf/v2/errors/gerror" "github.com/gogf/gf/v2/frame/g" + "github.com/gogf/gf/v2/net/ghttp" "github.com/gogf/gf/v2/util/gconv" "github.com/tidwall/sjson" ) @@ -140,11 +143,14 @@ func GetComposeResult(ctx context.Context, nodeExecutionId int64, buildType int, var newUserForm []map[string]any for _, m := range userForm { // 先替换字段 + if val, ok := m["audioDuration"]; ok { + delete(m, "audioDuration") + m["视频总时长"] = val + } if val, ok := m["videoDuration"]; ok { delete(m, "videoDuration") m["视频总时长"] = val } - // 收集待删除 key var delKeys []string for k, v := range m { @@ -183,19 +189,55 @@ func GetComposeResult(ctx context.Context, nodeExecutionId int64, buildType int, SessionId: sessionId, NodeId: nodeId, } - headers := make(map[string]string) + msgRes := new(flowDto.ComposeMessagesRes) + + // 1. 隔离上游取消（防止节点执行被中断时下游请求被 cancel）+ 设置独立超时 + baseCtx := context.WithoutCancel(ctx) + postCtx, cancel := context.WithTimeout(baseCtx, 30*time.Minute) + defer cancel() // 必须释放，防止上下文泄露 + + // 2. 克隆 commonHttp 客户端（保留 Consul 服务发现），显式设置超时和 ResponseHeaderTimeout + client := commonHttp.Httpclient.Clone() + client.SetTimeout(30 * time.Minute) + if tr, ok := client.Transport.(*http.Transport); ok { + tr.ResponseHeaderTimeout = 30 * time.Minute + } if r := g.RequestFromCtx(ctx); r != nil { for k, v := range r.Request.Header { if len(v) > 0 { - headers[k] = v[0] + client.SetHeader(k, v[0]) } } } - msgRes := new(flowDto.ComposeMessagesRes) - err = commonHttp.Post(ctx, "prompts-core/prompt/composeMessages", headers, msgRes, &msgReq) + resp, err := client.ContentJson().Post(postCtx, "prompts-core/prompt/composeMessages", &msgReq) if err != nil { return } + defer resp.Close() + result, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("读取composeMessages响应失败: %w", err) + } + // 统一处理内部API响应格式：{code:200,message:"",data:{...}} + resultStrut := &ghttp.DefaultHandlerResponse{} + + if err = gconv.Struct(result, &resultStrut); err != nil { // 修复：增加err检查 + return nil, fmt.Errorf("响应解析失败: " + err.Error()) + } + + // 添加调试日志：打印解析后的结构 + g.Log().Debugf(ctx, "[HTTP] 解析后结构: Code=%d, Message=%s, Data类型=%T, Data值=%+v", + resultStrut.Code, resultStrut.Message, resultStrut.Data, resultStrut.Data) + + if resultStrut.Code == 200 || resultStrut.Code == 0 { + if err = gconv.Struct(resultStrut.Data, &msgRes); err != nil { // 修复：增加err检查 + return nil, fmt.Errorf("数据解析失败: " + err.Error()) + } + // 添加调试日志：打印最终的target + g.Log().Debugf(ctx, "[HTTP] 最终target: %+v", &msgRes) + } else { + err = errors.New(resultStrut.Message) + } if g.IsEmpty(msgRes.TaskId) { return nil, fmt.Errorf("msg is empty") } @@ -234,20 +276,55 @@ func createGatewayTaskOnly(ctx context.Context, epicycleId int64, model string, EpicycleId: epicycleId, } - headers := make(map[string]string) + res := new(flowDto.ModelGatewayRes) + + // 1. 隔离上游取消（防止节点执行被中断时下游请求被 cancel）+ 设置独立超时 + baseCtx := context.WithoutCancel(ctx) + postCtx, cancel := context.WithTimeout(baseCtx, 30*time.Minute) + defer cancel() // 必须释放，防止上下文泄露 + + // 2. 克隆 commonHttp 客户端（保留 Consul 服务发现），显式设置超时和 ResponseHeaderTimeout + client := commonHttp.Httpclient.Clone() + client.SetTimeout(30 * time.Minute) + if tr, ok := client.Transport.(*http.Transport); ok { + tr.ResponseHeaderTimeout = 30 * time.Minute + } if r := g.RequestFromCtx(ctx); r != nil { for k, v := range r.Request.Header { if len(v) > 0 { - headers[k] = v[0] + client.SetHeader(k, v[0]) } } } - - res := new(flowDto.ModelGatewayRes) - err := commonHttp.Post(ctx, "model-gateway/task/createTask", headers, res, &req) + rpcResp, err := client.ContentJson().Post(postCtx, "model-gateway/task/createTask", &req) if err != nil { return "", err } + defer rpcResp.Close() + result, err := io.ReadAll(rpcResp.Body) + if err != nil { + return "", fmt.Errorf("读取createTask响应失败: %w", err) + } + // 统一处理内部API响应格式：{code:200,message:"",data:{...}} + resultStrut := &ghttp.DefaultHandlerResponse{} + + if err = gconv.Struct(result, &resultStrut); err != nil { // 修复：增加err检查 + return "", fmt.Errorf("响应解析失败: " + err.Error()) + } + + // 添加调试日志：打印解析后的结构 + g.Log().Debugf(ctx, "[HTTP] 解析后结构: Code=%d, Message=%s, Data类型=%T, Data值=%+v", + resultStrut.Code, resultStrut.Message, resultStrut.Data, resultStrut.Data) + + if resultStrut.Code == 200 || resultStrut.Code == 0 { + if err = gconv.Struct(resultStrut.Data, &res); err != nil { // 修复：增加err检查 + return "", fmt.Errorf("数据解析失败: " + err.Error()) + } + // 添加调试日志：打印最终的target + g.Log().Debugf(ctx, "[HTTP] 最终target: %+v", &res) + } else { + err = errors.New(resultStrut.Message) + } if g.IsEmpty(res.TaskId) { return "", fmt.Errorf("创建模型任务失败，taskId为空") } @@ -413,9 +490,7 @@ func GetModelResult(ctx context.Context, sessionId string, nodeInput *flowDto.No // 加锁写入map，解决并发竞态 mu.Lock() - fmt.Println("taskResult======================", idx, taskResult) mapTaskResult[idx] = taskResult - fmt.Println("mapTaskResult======================", mapTaskResult) mu.Unlock() //updateTokenCount(ctx, nodeInput.NodeExecutionId, modelInfo.Model.ResponseTokenField, taskResult) @@ -442,7 +517,6 @@ func GetModelResult(ctx context.Context, sessionId string, nodeInput *flowDto.No //updateTokenCount(ctx, nodeInput.NodeExecutionId, modelInfo.Model.ResponseTokenField, item) } } - fmt.Println("mapTaskResult--------------------------------------", mapTaskResult) return mapTaskResult, nil }