fix: 修复并发取消逻辑与HTTP请求超时及响应解析

2026-07-09 13:44:13 +08:00
parent cc29dd21e4
commit 08251d9a73
2 changed files with 106 additions and 29 deletions
@@ -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
 }