fix: 优化模型网关请求重试与业务字段前置写入

- 流式调用增加可重试错误码检测与指数退避重试
- 业务字段写入改为前置追加，不覆盖已有值
- 请求体按模板元数据递归合并，补充数组/对象字段处理
- 异步任务查询支持请求体映射与占位符替换
- 升级 common 与 gmq 依赖版本，移除本地 replace

config.yml

@@ -33,14 +33,14 @@ database:
       user: "postgres"
       pass: "Bjang09@686^*^"
       name: "model-gateway"
-      prefix: ""
+      prefix: "model_gateway_"
       role: "master"
       debug: true
       dryRun: false
       charset: "utf8"
       timezone: "Asia/Shanghai"
-      maxIdle: 5
-      maxOpen: 20
+      maxIdle: 15
+      maxOpen: 60
       maxLifetime: "30s"
       maxIdleConnTime: "30s"
       createdAt: "created_at"
@@ -59,6 +59,10 @@ consul:
 jaeger:
   addr: 192.168.0.83:4318
 
+nats:
+  addr: 192.168.0.83
+  port: 4222
+
 # 本地调试用：可选自动执行 worker/cleaner（默认关闭）
 asynch:
   queryPending:

go.mod

@@ -3,20 +3,15 @@ module model-gateway
 go 1.26.1
 
 require (
-	gitea.redpowerfuture.com/red-future/common v0.0.23
-	github.com/bjang03/gmq v0.0.1
+	gitea.redpowerfuture.com/red-future/common v0.0.31
+	github.com/bjang03/gmq v0.0.2
 	github.com/gogf/gf/contrib/drivers/pgsql/v2 v2.10.2
 	github.com/gogf/gf/contrib/nosql/redis/v2 v2.10.2
 	github.com/gogf/gf/v2 v2.10.2
 	github.com/google/uuid v1.6.0
 	github.com/tidwall/gjson v1.19.0
-	github.com/tidwall/sjson v1.2.5
 )
 
-replace gitea.redpowerfuture.com/red-future/common v0.0.23 => ../common
-
-replace github.com/bjang03/gmq v0.0.1 => ../gmq
-
 require (
 	github.com/BurntSushi/toml v1.5.0 // indirect
 	github.com/armon/go-metrics v0.4.1 // indirect
@@ -81,7 +76,8 @@ require (
 	github.com/rivo/uniseg v0.4.7 // indirect
 	github.com/spf13/cast v1.10.0 // indirect
 	github.com/tidwall/match v1.1.1 // indirect
-	github.com/tidwall/pretty v1.2.0 // indirect
+	github.com/tidwall/pretty v1.2.1 // indirect
+	github.com/tidwall/sjson v1.2.5 // indirect
 	github.com/tiger1103/gfast-token v1.0.10 // indirect
 	github.com/vcaesar/cedar v0.30.0 // indirect
 	github.com/vmihailenco/msgpack v4.0.4+incompatible // indirect

go.sum

@@ -1,4 +1,6 @@
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
+gitea.redpowerfuture.com/red-future/common v0.0.31 h1:9H8nL5Drazcv7Hs9d4j+cXhaB+7uOllIUqEOyZy1Eao=
+gitea.redpowerfuture.com/red-future/common v0.0.31/go.mod h1:xPU7aaMxn8rtNnWc2LDUXZL+IkaUkpQeLgflqw9FvdU=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/toml v1.5.0 h1:W5quZX/G/csjUnuI8SUYlsHs9M38FC7znL0lIO+DvMg=
 github.com/BurntSushi/toml v1.5.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho=
@@ -19,6 +21,8 @@ github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6r
 github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs=
 github.com/bitly/go-simplejson v0.5.1 h1:xgwPbetQScXt1gh9BmoJ6j9JMr3TElvuIyjR8pgdoow=
 github.com/bitly/go-simplejson v0.5.1/go.mod h1:YOPVLzCfwK14b4Sff3oP1AmGhI9T9Vsg84etUnlyp+Q=
+github.com/bjang03/gmq v0.0.2 h1:3CcVorDXYoRIN65bbzwRuUxzkBCkEpHWmKHOkfXzUo0=
+github.com/bjang03/gmq v0.0.2/go.mod h1:Y7TwWGuV4Cw97WUDaM7x+NC4kyFx1z44WAvNwJV3HV8=
 github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
 github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
 github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
@@ -323,6 +327,8 @@ github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
 github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
 github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
 github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
+github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4=
+github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
 github.com/tidwall/sjson v1.2.5 h1:kLy8mja+1c9jlljvWTlSazM7cKDRfJuR/bOJhcY5NcY=
 github.com/tidwall/sjson v1.2.5/go.mod h1:Fvgq9kS/6ociJEDnK0Fk1cpYF4FIW6ZF7LAe+6jwd28=
 github.com/tiger1103/gfast-token v1.0.10 h1:fNiBE/Dq5iTHvTGlCx3DmXa2o4hr0NtumFpffZ39k6s=

model/domain/schema_mapping.go

@@ -29,6 +29,7 @@ type ChatFieldsRes struct {
 // VideoFields 视频模型业务字段映射
 // 适用于 视频模型(600) 及其子类型
 type VideoFields struct {
+	SystemPrompt           string `json:"system_prompt" dc:"系统提示词,用于描述系统的需求"`
 	MinDuration            string `json:"min_duration" dc:"模型支持的最小视频时长（秒）"`
 	MaxDuration            string `json:"max_duration" dc:"模型支持的最大视频时长（秒）"`
 	FirstFrame             string `json:"first_frame" dc:"视频的首帧/初始画面，传入一张图片作为视频第一帧画面"`

model/dto/model_call_dto.go

@@ -69,6 +69,11 @@ var ModelErrorResp struct {
 	} `json:"error"`
 }
 
+var ModelError1Resp struct {
+	Code    int    `json:"code"`
+	Message string `json:"message"`
+}
+
 type ModelMsg struct {
 	TaskID           int64          `json:"id" dc:"任务ID"`
 	State            task.Status    `json:"state" dc:"状态"`

model/entity/model_manage.go

@@ -90,6 +90,7 @@ type ModelManage struct {
 	MinDuration                  int                `orm:"min_duration" json:"minDuration" description:"最小时长（秒）"`
 	MaxDuration                  int                `orm:"max_duration" json:"maxDuration" description:"最大时长（秒）"`
 	LastFrame                    string             `orm:"last_frame" json:"lastFrame" description:"视频的尾帧图像"`
+	ErrorMessageMapping          map[string]any     `orm:"error_message_mapping" json:"errorMessageMapping" description:"错误消息映射"`
 }
 
 type TokenMapping struct {
@@ -101,6 +102,7 @@ type TokenMapping struct {
 type AsyncTaskMapping struct {
 	Url                string            `json:"url" dc:"url"`
 	HttpMethod         string            `json:"httpMethod" dc:"http方法" d:"POST"`
+	RequestBodyMapping map[string]any    `json:"requestBodyMapping" description:"请求体映射"`
 	RequestHeadMapping map[string]string `json:"requestHeadMapping" description:"请求头映射"`
 	ResponseMapping    map[string]any    `json:"responseMapping" description:"响应映射"`
 	TaskId             string            `json:"taskId" dc:"任务id"`

service/model_call_service.go

@@ -378,20 +378,39 @@ LOOP:
 //  3. businessParams 只装业务字段，按业务字段名（RequestBusinessFieldMapping 的 key）传值，
 //     TakeBusinessFields 解析为写入路径，构建完成后由 WriteBusinessFields 按路径写入最终请求体
 func buildChatRequestParams(modelInfo *entity.ModelManage, requestParams, businessParams map[string]any) (map[string]any, error) {
+
+	// requestParams 可能混有扁平路径 key（messages.enumValues...）与已是对象/数组的值（stream）。
+	// IsFlatMap 遇 map/slice 值即整体返回 false 会跳过 unflatten；sjson.Set 能处理任意值类型作为子树，
+	// 带点 key 按路径展开、无点 key 直接赋值，故始终 unflatten
+	var err error
+	requestParams, err = utils.UnFlatBySjson(requestParams)
+	if err != nil {
+		return nil, err
+	}
+
 	rest := make(map[string]any, len(requestParams))
 	for k, v := range requestParams {
 		rest[k] = v
 	}
 	// 请求结构源：模板字段兜底（模板 value/defaultValue 生效）+ requestParams 覆盖同名；
 	// 保证模板声明的结构字段（如 stream_options 对象）即使 requestParams 未传也进请求体
+	// 用户已传字段按模板类型元数据递归合并（补 type 包装、补默认字段），数组字段仅在用户提供时才合并
 	src := rest
 	for k, v := range modelInfo.RequestBodyMapping {
-		mapV := gconv.Map(v)
-		if mapV["type"] == "array" {
-			continue
+		tmplMap, _ := v.(map[string]any)
+		if tmplMap != nil {
+			if t, _ := tmplMap["type"].(string); t == "array" {
+				if _, has := src[k]; !has {
+					continue
+				}
+				src[k] = modelUtils.MergeNode(src[k], v)
+				continue
+			}
 		}
-		if _, has := src[k]; !has {
-			src[k] = v
+		if userVal, has := src[k]; has {
+			src[k] = modelUtils.MergeNode(userVal, v)
+		} else {
+			src[k] = modelUtils.DeepCopyNode(v)
 		}
 	}
 	if len(requestParams) > 0 {

service/model_session_service.go

@@ -22,7 +22,7 @@ var ModelSession = &modelSessionService{}
 type modelSessionService struct{}
 
 // modelCallMaxRetries 上游调用最大重试次数
-const modelCallMaxRetries = 3
+const modelCallMaxRetries = 15
 
 // CreateSession 创建会话
 func (s *modelSessionService) CreateSession(ctx context.Context, req *dto.CallModelSessionReq) (res *dto.ModelCallRes, err error) {
@@ -80,7 +80,7 @@ LOOP:
 		if *model.ResponseTypeSync.Code() == *modelInfo.ResponseType {
 			respBodyMap := make(map[string]string, len(modelInfo.ResponseBodyMapping))
 			for k, _ := range modelInfo.ResponseBodyMapping {
-				respBodyMap[modelUtils.CleanFieldPath(k)] = modelUtils.CleanFieldPath(k)
+				respBodyMap[k] = modelUtils.CleanFieldPath(k)
 			}
 			// 基于统一字段路径（GetByPath）按映射取值组装结果
 			var respObj map[string]any
@@ -89,7 +89,7 @@ LOOP:
 			}
 			content := make(map[string]any, len(respBodyMap))
 			for bizKey, jsonPath := range respBodyMap {
-				content[bizKey] = modelUtils.GetByPathValue(respObj, jsonPath)
+				content[bizKey] = uploadTempURLToOSS(ctx, modelUtils.GetByPathValue(respObj, jsonPath))
 			}
 
 			businessField := make(map[string]any, len(modelInfo.ResponseBusinessFieldMapping))
@@ -139,7 +139,8 @@ LOOP:
 	return docMsg, nil
 }
 
-// CreateSessionStreamOnce 流式调用上游模型 → 缓冲全量后一次返回（走 gf 框架正常返回）
+// CreateSessionStreamOnce 流式调用上游模型 → 缓冲全量后一次返回（走 gf 框架正常返回）。
+// 与同步请求一致：上游返回可重试错误码（限流/5xx）时按指数退避重试（最多 modelCallMaxRetries 次）。
 func (s *modelSessionService) CreateSessionStreamOnce(ctx context.Context, req *dto.CallModelSessionReq) (docMsg *dto.ModelCallRes, err error) {
 	startTime := time.Now()
 
@@ -147,9 +148,23 @@ func (s *modelSessionService) CreateSessionStreamOnce(ctx context.Context, req *
 	modelInfo := req.ModelInfo
 	newRequestParams := req.RequestParams
 
-	// 获取上游流式 reader（stream=false → w 不会被使用，传 nil）
+	attempt := 0
+LOOP:
+	// 获取上游流式 reader（stream=false → w 不会被使用，传 nil）。
+	// 非 2xx 状态/网络错误在此返回；错误含可重试错误码（限流/5xx）时按指数退避重试，与同步请求一致。
 	streamReader, err := ModelHttpStreamRequest(ctx, nil, modelInfo.BaseURL, modelInfo.RequestHeadMapping, modelInfo.HttpMethod, newRequestParams)
 	if err != nil {
+		if retryCode := streamRetryCodeOfError(err); retryCode != "" && attempt < modelCallMaxRetries {
+			attempt++
+			wait := time.Duration(1<<attempt) * time.Second
+			g.Log().Warningf(ctx, "模型流式请求异常，第 %d 次重试(等待 %v): code=%s err=%v", attempt+1, wait, retryCode, err)
+			select {
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			case <-time.After(wait):
+			}
+			goto LOOP
+		}
 		return nil, err
 	}
 
@@ -157,22 +172,31 @@ func (s *modelSessionService) CreateSessionStreamOnce(ctx context.Context, req *
 	docMsg.TaskId = id
 	var contentBuf strings.Builder
 
+	// 记录流内 error 事件（OpenAI 兼容 error 分片），供流结束后统一判定重试/报错
+	var streamErrCode, streamErrMsg string
+
 	// 路径预处理
 	respMapping := make(map[string]string, len(modelInfo.ResponseBodyMapping))
 	for k, _ := range modelInfo.ResponseBodyMapping {
-		respMapping[modelUtils.CleanFieldPath(k)] = modelUtils.CleanFieldPath(k)
+		respMapping[k] = modelUtils.CleanFieldPath(k)
 	}
 	totalTokenPath := modelUtils.CleanFieldPath(modelInfo.TokenMapping.TotalTokens)
 	promptTokenPath := modelUtils.CleanFieldPath(modelInfo.TokenMapping.PromptTokens)
 	completionTokenPath := modelUtils.CleanFieldPath(modelInfo.TokenMapping.CompletionTokens)
 
 	ParseSSEStream(ctx, streamReader, func(ctx context.Context, chunk map[string]any) error {
+		// 流内错误事件（OpenAI 兼容 error 分片）：暂存错误码/消息，不做内容累加，由流结束后统一判定
+		if code, msg := streamErrorOfChunk(chunk); code != "" {
+			streamErrCode, streamErrMsg = code, msg
+			return nil
+		}
 		// 基于统一字段路径（GetByPath）在分片对象上取值，取首个数组元素文本累加
 		for _, jsonPath := range respMapping {
 			v := modelUtils.GetByPathValue(chunk, jsonPath)
 			if v == nil || g.IsEmpty(v) {
 				continue
 			}
+			v = uploadTempURLToOSS(ctx, v)
 			var realText string
 			if arr, ok := v.([]any); ok && len(arr) > 0 {
 				realText = gconv.String(arr[0])
@@ -194,8 +218,29 @@ func (s *modelSessionService) CreateSessionStreamOnce(ctx context.Context, req *
 		return nil
 	})
 
-	// 流结束后组装
-	docMsg.Content = map[string]any{"respBody": contentBuf.String()}
+	// 流内返回可重试错误码：丢弃本次部分内容，指数退避后重新请求
+	if streamErrCode != "" {
+		if attempt < modelCallMaxRetries && isRetryableErrorCode(streamErrCode) {
+			attempt++
+			wait := time.Duration(1<<attempt) * time.Second
+			g.Log().Warningf(ctx, "模型流式调用异常，第 %d 次重试(等待 %v): code=%s msg=%s", attempt+1, wait, streamErrCode, streamErrMsg)
+			select {
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			case <-time.After(wait):
+			}
+			goto LOOP
+		}
+		// 与同步一致：不可重试的错误码记录到 ErrorMsg 后正常走组装返回，不中断流程
+		docMsg.ErrorMsg = streamErrMsg
+	}
+
+	// 流结束后组装（流内出错时内容为空，与同步一致不再组装/上传空内容）
+	if streamErrCode == "" {
+		for k, _ := range modelInfo.ResponseBodyMapping {
+			docMsg.Content = map[string]any{k: contentBuf.String()}
+		}
+	}
 
 	// 补充更新会话记录
 	updateModelSessionReq := dto.UpdateModelSessionReq{
@@ -204,6 +249,7 @@ func (s *modelSessionService) CreateSessionStreamOnce(ctx context.Context, req *
 		TotalTokens:      docMsg.TotalTokens,
 		PromptTokens:     docMsg.PromptTokens,
 		CompletionTokens: docMsg.CompletionTokens,
+		ErrorMsg:         docMsg.ErrorMsg,
 	}
 	if !g.IsEmpty(docMsg.Content) {
 		uploadNewResp, uploadErr := Upload(ctx, &dto.UploadFileBytesReq{
@@ -277,6 +323,7 @@ func (s *modelSessionService) CreateSessionStream(ctx context.Context, w http.Re
 			if v == nil || g.IsEmpty(v) {
 				continue
 			}
+			v = uploadTempURLToOSS(ctx, v)
 			var realText string
 			if arr, ok := v.([]any); ok && len(arr) > 0 {
 				realText = gconv.String(arr[0])
@@ -398,8 +445,57 @@ func (s *modelSessionService) CreateSessionStream(ctx context.Context, w http.Re
 // ModelHttpNormalRequest 不返回 HTTP status，只能按响应体 error.code 字符串判定。
 func isRetryableErrorCode(code string) bool {
 	switch code {
-	case "429", "500", "501", "502", "503", "limit_requests", "limit_tokens", "rate_limit_exceeded":
+	case "429", "500", "501", "502", "503", "InvalidParameter", "limit_requests", "limit_tokens", "rate_limit_exceeded":
 		return true
 	}
 	return false
 }
+
+// streamErrorOfChunk 从流式分片提取错误码与消息：优先 OpenAI 兼容 error 事件，顶层 code 兜底。
+func streamErrorOfChunk(chunk map[string]any) (code, msg string) {
+	if errObj := gconv.Map(chunk["error"]); errObj != nil {
+		code = gconv.String(errObj["code"])
+		msg = gconv.String(errObj["message"])
+	}
+	if code == "" {
+		code = gconv.String(chunk["code"])
+	}
+	return
+}
+
+// streamRetryCodeOfError 从流式请求错误中提取可重试错误码：优先解析错误体 error.code/顶层 code，
+// 其次取非 2xx 的 HTTP 状态码；纯网络错误等无错误码场景返回空串（与同步请求一致，不重试）。
+func streamRetryCodeOfError(err error) string {
+	if err == nil {
+		return ""
+	}
+	msg := err.Error()
+	// 非 2xx 时 ModelHttpStreamRequest 返回 "[HTTP][Stream] 状态码异常: %d, body={...}"
+	if idx := strings.Index(msg, "body="); idx >= 0 {
+		body := msg[idx+len("body="):]
+		var errResp struct {
+			Error struct {
+				Code string `json:"code"`
+			} `json:"error"`
+			Code string `json:"code"`
+		}
+		if json.Unmarshal([]byte(body), &errResp) == nil {
+			if errResp.Error.Code != "" {
+				return errResp.Error.Code
+			}
+			if errResp.Code != "" {
+				return errResp.Code
+			}
+		}
+	}
+	if idx := strings.Index(msg, "状态码异常: "); idx >= 0 {
+		codeStr := strings.TrimSpace(msg[idx+len("状态码异常: "):])
+		if comma := strings.IndexByte(codeStr, ','); comma >= 0 {
+			codeStr = codeStr[:comma]
+		}
+		if isRetryableErrorCode(codeStr) {
+			return codeStr
+		}
+	}
+	return ""
+}

service/model_task_end_service.go

@@ -139,12 +139,14 @@ func (s *modelTaskEndService) handleSingleTask(ctx context.Context, item *entity
 	LOOP:
 		// 替换URL占位符
 		url := urlParamReg.ReplaceAllString(modelInfo.AsyncTaskMapping.Url, item.TaskId)
+		// 组装查询请求体：POST 查询接口需要 body（从 RequestBodyMapping 出发，替换 {…} 占位符为任务 ID）
+		reqBody := buildAsyncTaskBody(modelInfo.AsyncTaskMapping.RequestBodyMapping, item.TaskId)
 		// 发起HTTP请求
 		modelRespBody, err := ModelHttpNormalRequest(
 			asyncCtx,
 			url,
 			modelInfo.AsyncTaskMapping.RequestHeadMapping,
-			modelInfo.AsyncTaskMapping.HttpMethod, nil,
+			modelInfo.AsyncTaskMapping.HttpMethod, reqBody,
 		)
 		if err != nil {
 			g.Log().Errorf(asyncCtx, "模型请求失败: %v", err)
@@ -251,6 +253,42 @@ func (s *modelTaskEndService) handleSingleTask(ctx context.Context, item *entity
 	})
 }
 
+// buildAsyncTaskBody 组装异步任务查询请求体：从 AsyncTaskMapping.RequestBodyMapping 出发，
+// 把 {…} 占位符（如 {taskId}）替换为实际任务 ID，兼容 POST 查询接口需要请求体的场景。
+// 未配置映射时返回 nil（GET 查询/无需 body 的场景）。
+func buildAsyncTaskBody(mapping map[string]any, taskID string) map[string]any {
+	if len(mapping) == 0 {
+		return nil
+	}
+	out := make(map[string]any, len(mapping))
+	for k, v := range mapping {
+		out[k] = replaceTaskPlaceholder(v, taskID)
+	}
+	return out
+}
+
+// replaceTaskPlaceholder 递归替换结构体中的 {…} 占位符为任务 ID
+func replaceTaskPlaceholder(v any, taskID string) any {
+	switch val := v.(type) {
+	case string:
+		return urlParamReg.ReplaceAllString(val, taskID)
+	case map[string]any:
+		m := make(map[string]any, len(val))
+		for k, x := range val {
+			m[k] = replaceTaskPlaceholder(x, taskID)
+		}
+		return m
+	case []any:
+		arr := make([]any, len(val))
+		for i, x := range val {
+			arr[i] = replaceTaskPlaceholder(x, taskID)
+		}
+		return arr
+	default:
+		return v
+	}
+}
+
 // tempDownloadTimeout 临时路径下载超时
 const tempDownloadTimeout = 5 * time.Minute
 
@@ -304,7 +342,8 @@ func uploadSingleURL(ctx context.Context, rawURL string) (string, bool) {
 		g.Log().Errorf(ctx, "临时路径转存OSS失败: url=%s err=%v", rawURL, err)
 		return rawURL, false
 	}
-	return ossRes.FileAddressPrefix + ossRes.FileURL, true
+	//ossRes.FileAddressPrefix + ossRes.FileURL, true
+	return ossRes.FileURL, true
 }
 
 func isHTTPURL(s string) bool {

service/model_task_start_service.go

@@ -49,10 +49,23 @@ func (s *modelTaskStartService) CreateTask(ctx context.Context, req *dto.CallMod
 	if err != nil {
 		return nil, fmt.Errorf("模型返回参数解析失败：%v", err)
 	}
-	if errMsg.Error.Code != "" {
-		docMsg.ErrorMsg = errMsg.Error.Message
-		updateModelReq.ErrorMsg = docMsg.ErrorMsg
+	if g.IsEmpty(errMsg.Error.Code) {
+		errMsg1 := new(dto.ModelError1Resp)
+		err = gconv.Struct(modelRespBody, errMsg1)
+		if err != nil {
+			return nil, fmt.Errorf("模型返回参数解析失败：%v", err)
+		}
+		if !g.IsEmpty(errMsg1.Code) && errMsg1.Code != 20000000 {
+			docMsg.ErrorMsg = errMsg1.Message
+			updateModelReq.ErrorMsg = docMsg.ErrorMsg
+		}
 	} else {
+		if errMsg.Error.Code != "" {
+			docMsg.ErrorMsg = errMsg.Error.Message
+			updateModelReq.ErrorMsg = docMsg.ErrorMsg
+		}
+	}
+	if docMsg.ErrorMsg == "" {
 		taskIDPath := modelUtils.CleanFieldPath(modelInfo.AsyncTaskMapping.TaskId)
 		docMsg.Content = map[string]any{
 			"respBody": modelUtils.GetByPathValue(respObj, taskIDPath),

service/utils/business_fields.go

@@ -86,16 +86,16 @@ func isEmptyBizValue(v any) bool {
 //
 // SetByPath（写，构建请求体）与 GetByPath（读，解析响应）共用 parsePath；
 // 读方向语义：数组段非末段取第 0 个元素继续下钻，[*] 为末段返回整个数组，选择器定位匹配元素；
-// 写方向语义：数组段非末段作用于最后一个元素，末段追加，选择器 upsert（命中更新/未命中新建），
+// 写方向语义：数组段非末段作用于最后一个元素，末段前置追加（业务值在前），选择器 upsert（命中更新/未命中新建），
 // 多值（[]any）仅在叶子选择器展开为多个独立元素（多个参考图/视频等）。
 // ============================================================
 
 // SetByPath 按业务字段映射路径把值写入请求结构（请求侧构建）。
 // 路径语法与 BuildSchemaMapping 输出一致（干净形态，无需 attrs 剔除）。
 //
-// 写入语义：
-//   - 目标字段已存在且是数组 → 追加
-//   - 目标字段已存在且非数组 → 覆盖（普通叶子路径）
+// 写入语义（业务字段一律前置追加，不覆盖已有值）：
+//   - 目标字段已存在且是数组 → 业务值（或其元素）前置插入数组头部，原元素依次后移
+//   - 目标字段已存在且非数组（普通叶子路径）→ 业务值前置拼接（字符串拼接/数组包裹），原值保留在后
 //   - 目标字段不存在 → 新建
 //   - 数组段/选择器段目标字段不是数组（如字符串 content）→ 返回错误，不覆盖已有值
 //   - 中间路径遇到非对象字段 → 返回错误
@@ -103,7 +103,7 @@ func isEmptyBizValue(v any) bool {
 //   - 数组段带选择器 → 命中则更新目标字段，未命中则按选择器字段构造新元素追加；
 //     选择器段即使未标 [*] 也按数组处理（如 input.media?type=first_frame&url=#）
 //   - 多级选择器 → 递归：中间级选择器定位/新建容器元素并继续下钻，叶子选择器写值
-//   - 值本身是数组 → 叶子选择器逐值追加；普通数组段/点号路径按 appendValues 追加
+//   - 值本身是数组 → 叶子选择器逐值追加；普通数组段/点号路径前置追加（业务值在前，原值保留在后，不覆盖）
 func SetByPath(root map[string]any, path string, value any) error {
 	steps := parsePath(NormalizeFieldPath(path))
 	if len(steps) == 0 {
@@ -146,7 +146,7 @@ func setBySteps(cur map[string]any, steps []step, value any) error {
 		return err
 	}
 	if last {
-		cur[first.key] = appendValues(arr, value)
+		cur[first.key] = prependToArray(arr, value)
 		return nil
 	}
 	// 无选择器数组段：作用于最后一个元素（追加语义）
@@ -572,23 +572,46 @@ func setLeafPath(m map[string]any, path string, value any) error {
 	return nil
 }
 
-// setLeaf 叶子写入：目标已是数组 → 追加；非数组/不存在 → 覆盖/新建
+// setLeaf 叶子写入（业务字段语义：值前置追加而非覆盖）：
+// 字段不存在 → 直接写入；已有值 → 业务字段值前置到原有值前面（数组前插/字符串拼接），原值保留在后。
 func setLeaf(parent map[string]any, key string, value any) {
-	if existing, ok := parent[key]; ok {
-		if arr, isArr := existing.([]any); isArr {
-			parent[key] = appendValues(arr, value)
-			return
-		}
+	existing, ok := parent[key]
+	if !ok {
+		parent[key] = value
+		return
 	}
-	parent[key] = value
+	parent[key] = prependValue(value, existing)
 }
 
-// appendValues 追加值到数组；value 为数组时逐个追加
-func appendValues(arr []any, value any) []any {
-	if vals, ok := value.([]any); ok {
-		return append(arr, vals...)
+// prependValue 把 value 前置到 existing 前（业务值在前，原值保留在后，不覆盖）：
+//   - existing 是数组 → value（或其元素）前插到数组头部
+//   - value 是数组（existing 为标量）→ value 各元素在前，existing 作为末位元素
+//   - 其余标量 → 字符串拼接，业务值在前
+func prependValue(value, existing any) any {
+	if arr, isArr := existing.([]any); isArr {
+		return prependToArray(arr, value)
 	}
-	return append(arr, value)
+	if vals, isArr := value.([]any); isArr {
+		out := make([]any, 0, len(vals)+1)
+		out = append(out, vals...)
+		out = append(out, existing)
+		return out
+	}
+	return gconv.String(value) + gconv.String(existing)
+}
+
+// prependToArray 把 value（或其元素）插到数组头部，原元素依次后移
+func prependToArray(arr []any, value any) []any {
+	if vals, isArr := value.([]any); isArr {
+		out := make([]any, 0, len(vals)+len(arr))
+		out = append(out, vals...)
+		out = append(out, arr...)
+		return out
+	}
+	out := make([]any, 0, len(arr)+1)
+	out = append(out, value)
+	out = append(out, arr...)
+	return out
 }
 
 // ensureMap 确保键对应 map，不存在则新建；已存在但非对象 → 返回错误

service/utils/json_flatten.go

@@ -1,43 +0,0 @@
-package utils
-
-import (
-	"encoding/json"
-	"fmt"
-
-	"github.com/tidwall/sjson"
-)
-
-// IsFlatMap 递归判断 map 是否扁平化
-func IsFlatMap(m map[string]interface{}) bool {
-	for _, v := range m {
-		switch val := v.(type) {
-		case map[string]interface{}:
-			return false
-		case []interface{}:
-			for _, item := range val {
-				if _, ok := item.(map[string]interface{}); ok {
-					return false
-				}
-			}
-		}
-	}
-	return true
-}
-
-// UnFlatBySjson 将扁平路径映射还原为嵌套 JSON
-func UnFlatBySjson(flatMap map[string]interface{}) (map[string]interface{}, error) {
-	raw := "{}"
-	for path, val := range flatMap {
-		var err error
-		raw, err = sjson.Set(raw, path, val)
-		if err != nil {
-			return nil, fmt.Errorf("sjson set path %s failed: %w", path, err)
-		}
-	}
-
-	var result map[string]interface{}
-	if err := json.Unmarshal([]byte(raw), &result); err != nil {
-		return nil, fmt.Errorf("parse final json failed: %w", err)
-	}
-	return result, nil
-}

service/utils/template_merge.go

@@ -0,0 +1,260 @@
+package utils
+
+import "github.com/gogf/gf/v2/util/gconv"
+
+// MergeNode 按模板节点的元数据把用户数据节点合并为模板可解析的节点：
+//   - 用户节点无 type 键 → 套上模板类型包装（标量取用户值，空则回落模板 value/defaultValue）
+//   - 用户节点有 type 键 → 递归合并子节点（子节点缺 type 时同样补模板类型，避免模板包装泄漏）
+//   - 对象：模板 attrs 缺省字段补进用户容器（保留模板默认值）
+//   - 数组：用户 enumValues/attrs 元素逐个与模板元素原型合并（模板未提供的槽位克隆原型）
+func MergeNode(user, tmpl any) any {
+	tmplMap, ok := tmpl.(map[string]any)
+	if !ok {
+		return user
+	}
+	fieldType, _ := tmplMap["type"].(string)
+
+	// 用户节点已带 type：按用户类型递归合并子节点；标量叶子已完备，直接返回
+	if userMap, ok := user.(map[string]any); ok {
+		if ut, isTpl := userMap["type"]; isTpl {
+			switch gconv.String(ut) {
+			case TypeObject:
+				return mergeObject(userMap, tmplMap)
+			case TypeArray:
+				return mergeArray(userMap, tmplMap)
+			default:
+				return userMap
+			}
+		}
+	}
+
+	switch fieldType {
+	case TypeString, TypeBool, TypeNumber, TypeNumberInt, TypeNumberFloat:
+		return wrapScalarNode(tmplMap, user)
+	case TypeObject:
+		return mergeObject(user, tmplMap)
+	case TypeArray:
+		return mergeArray(user, tmplMap)
+	default:
+		return user
+	}
+}
+
+// wrapScalarNode 把用户标量值包装为 {type, value} 节点；用户值为空时回落模板 value/defaultValue
+func wrapScalarNode(tmplMap map[string]any, user any) map[string]any {
+	val := user
+	if m, ok := user.(map[string]any); ok {
+		if v, has := m["value"]; has {
+			val = v
+		} else {
+			val = nil
+		}
+	}
+	node := map[string]any{"type": normalizeScalarType(gconv.String(tmplMap["type"]))}
+	if hasUsableValue(val) {
+		node["value"] = val
+	} else if v, has := tmplMap["value"]; has {
+		node["value"] = v
+	} else if d, has := tmplMap["defaultValue"]; has {
+		node["value"] = d
+	}
+	return node
+}
+
+// normalizeScalarType integer/float 统一为 number（resolveField 只分发 string/boolean/number）
+func normalizeScalarType(t string) string {
+	if t == TypeNumberInt || t == TypeNumberFloat {
+		return TypeNumber
+	}
+	return t
+}
+
+// mergeObject 合并对象节点：模板 attrs 缺省字段补进用户容器（模板默认值保留）
+func mergeObject(user any, tmplMap map[string]any) map[string]any {
+	tmplAttrs, _ := tmplMap["attrs"].(map[string]any)
+	node := map[string]any{"type": TypeObject}
+	if tmplAttrs == nil {
+		node["attrs"] = user
+		return node
+	}
+	var container map[string]any
+	switch u := user.(type) {
+	case map[string]any:
+		if a, has := u["attrs"]; has {
+			if am, ok := a.(map[string]any); ok {
+				container = copyMap(am)
+			} else {
+				container = map[string]any{}
+			}
+		} else if v, has := u["value"]; has {
+			if vm, ok := v.(map[string]any); ok {
+				container = copyMap(vm)
+			} else {
+				container = map[string]any{}
+			}
+		} else {
+			container = copyMap(u)
+		}
+	default:
+		container = map[string]any{}
+	}
+	for k, subTmpl := range tmplAttrs {
+		if _, has := container[k]; !has {
+			container[k] = DeepCopyNode(subTmpl)
+		} else {
+			container[k] = MergeNode(container[k], subTmpl)
+		}
+	}
+	node["attrs"] = container
+	return node
+}
+
+// mergeArray 合并数组节点：用户 enumValues/attrs 元素逐个与模板元素原型合并。
+// 用户未填的槽位（sjson null 填充）用模板对应槽位原型补位，保留下标不塌缩。
+func mergeArray(user any, tmplMap map[string]any) map[string]any {
+	proto := arrayElementTemplate(tmplMap)
+	tmplSlots, _ := tmplMap["enumValues"].([]any) // 模板各槽位原型，按下标一一对应
+	node := map[string]any{"type": TypeArray}
+	slotProto := func(i int) any {
+		if i < len(tmplSlots) {
+			return tmplSlots[i]
+		}
+		return proto
+	}
+	switch u := user.(type) {
+	case map[string]any:
+		if evs, ok := u["enumValues"].([]any); ok {
+			out := make([]any, len(evs))
+			for i, ev := range evs {
+				if ev == nil {
+					// 保留 null 槽位以维持下标；解析阶段 resolveArray 会丢弃空元素，
+					// 避免用模板原型填充 null 而物化出"幽灵"元素（用户未填的数组槽位）
+					out[i] = nil
+					continue
+				}
+				out[i] = mergeArrayElement(ev, slotProto(i))
+			}
+			node["enumValues"] = out
+		} else if a, ok := u["attrs"].([]any); ok {
+			out := make([]any, len(a))
+			for i, item := range a {
+				out[i] = mergeArrayElement(item, proto)
+			}
+			node["attrs"] = out
+		} else if v, has := u["value"]; has {
+			node["value"] = v
+		} else if len(u) > 0 {
+			node["attrs"] = []any{mergeArrayElement(u, proto)}
+		}
+	case []any:
+		if isSchemaElementList(u) {
+			out := make([]any, len(u))
+			for i, item := range u {
+				out[i] = mergeArrayElement(item, proto)
+			}
+			node["attrs"] = out
+		} else {
+			node["value"] = u
+		}
+	default:
+		node["value"] = user
+	}
+	return node
+}
+
+// mergeArrayElement 合并单个数组元素：元素缺 type 时按槽位原型补默认字段。
+// slotProto 可为 nil（模板未提供槽位原型时）→ 原样返回用户元素
+func mergeArrayElement(ev any, slotProto any) any {
+	if ev == nil {
+		return ev
+	}
+	proto, _ := slotProto.(map[string]any)
+	if proto == nil {
+		return ev
+	}
+	evMap, ok := ev.(map[string]any)
+	if !ok {
+		return ev
+	}
+	if _, isTpl := evMap["type"]; isTpl {
+		return ev
+	}
+	// 元素原型自身是包装节点（嵌套对象/数组）→ 按模板类型合并
+	if t, has := proto["type"].(string); has && t != "" {
+		return MergeNode(ev, proto)
+	}
+	protoAttrs, _ := proto["attrs"].(map[string]any)
+	var container map[string]any
+	if a, has := evMap["attrs"]; has {
+		if am, ok := a.(map[string]any); ok {
+			container = copyMap(am)
+		} else {
+			container = map[string]any{}
+		}
+	} else {
+		container = copyMap(evMap)
+	}
+	for k, subTmpl := range protoAttrs {
+		if _, has := container[k]; !has {
+			container[k] = DeepCopyNode(subTmpl)
+		} else {
+			container[k] = MergeNode(container[k], subTmpl)
+		}
+	}
+	return map[string]any{"attrs": container}
+}
+
+// isSchemaElementList 判断 []any 是否为 schema 元素列表（元素均为 map，且带 type 或 attrs 包装）
+func isSchemaElementList(list []any) bool {
+	if len(list) == 0 {
+		return false
+	}
+	for _, item := range list {
+		m, ok := item.(map[string]any)
+		if !ok {
+			return false
+		}
+		if _, hasType := m["type"]; !hasType {
+			if _, hasAttrs := m["attrs"]; !hasAttrs {
+				return false
+			}
+		}
+	}
+	return true
+}
+
+// DeepCopyNode 深拷贝任意嵌套节点（模板节点整体拷贝进 src 时使用，避免共享引用被后续回填修改）
+func DeepCopyNode(v any) any {
+	switch val := v.(type) {
+	case map[string]any:
+		out := make(map[string]any, len(val))
+		for k, sub := range val {
+			out[k] = DeepCopyNode(sub)
+		}
+		return out
+	case []any:
+		out := make([]any, len(val))
+		for i, sub := range val {
+			out[i] = DeepCopyNode(sub)
+		}
+		return out
+	default:
+		return v
+	}
+}
+
+// hasUsableValue 值是否可写入节点 value；0/false 视为有效值
+func hasUsableValue(v any) bool {
+	switch val := v.(type) {
+	case nil:
+		return false
+	case string:
+		return val != ""
+	case []any:
+		return len(val) > 0
+	case map[string]any:
+		return len(val) > 0
+	default:
+		return true
+	}
+}

service/utils/template_parse.go

@@ -4,6 +4,7 @@ import (
 	"regexp"
 	"strings"
 
+	"gitea.redpowerfuture.com/red-future/common/utils"
 	"github.com/gogf/gf/v2/util/gconv"
 )
 
@@ -106,10 +107,10 @@ func CleanMapFieldPath(m map[string]string) map[string]string {
 //   - 数组展开: enumValue 内某叶子字段 value 为数组时，按数组元素展开为多个项
 func ParseConfigTemplate(cfg map[string]interface{}) map[string]interface{} {
 	var flattenJSON map[string]interface{}
-	flatMap := IsFlatMap(cfg)
+	flatMap := utils.IsFlatMap(cfg)
 	if flatMap {
 		var err error
-		flattenJSON, err = UnFlatBySjson(cfg)
+		flattenJSON, err = utils.UnFlatBySjson(cfg)
 		if err != nil {
 			return nil
 		}
@@ -152,13 +153,13 @@ func resolveScalar(field map[string]interface{}) interface{} {
 			if vv != "" {
 				return vv
 			}
-		case float64:
-			if vv != 0 {
-				return vv
-			}
 		case bool:
 			return vv
 		default:
+			// 数值零值（int/float 各类型）视为未提供，跳过；bool/string 已在上方处理
+			if isNumericZero(v) {
+				return nil
+			}
 			return vv
 		}
 	}
@@ -168,19 +169,50 @@ func resolveScalar(field map[string]interface{}) interface{} {
 			if dv != "" {
 				return dv
 			}
-		case float64:
-			if dv != 0 {
-				return dv
-			}
 		case bool:
 			return dv
 		default:
+			if isNumericZero(d) {
+				return nil
+			}
 			return dv
 		}
 	}
 	return nil
 }
 
+// isNumericZero 判断是否为数值零值（模板 value 常为 int 字面量，经 gconv 可能为 float64）
+func isNumericZero(v interface{}) bool {
+	switch vv := v.(type) {
+	case int:
+		return vv == 0
+	case int8:
+		return vv == 0
+	case int16:
+		return vv == 0
+	case int32:
+		return vv == 0
+	case int64:
+		return vv == 0
+	case uint:
+		return vv == 0
+	case uint8:
+		return vv == 0
+	case uint16:
+		return vv == 0
+	case uint32:
+		return vv == 0
+	case uint64:
+		return vv == 0
+	case float32:
+		return vv == 0
+	case float64:
+		return vv == 0
+	default:
+		return false
+	}
+}
+
 // resolveObject 解析对象字段，递归处理 attrs
 //
 // 特殊处理「参数定义」结构：当 attrs 含 default 字段时，说明该对象是一个

service/utils/template_validate.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"model-gateway/model/dto"
 
+	"gitea.redpowerfuture.com/red-future/common/utils"
 	"github.com/gogf/gf/v2/frame/g"
 	"github.com/gogf/gf/v2/util/gconv"
 	"github.com/gogf/gf/v2/util/gutil"
@@ -40,8 +41,8 @@ func CheckBody(body map[string]interface{}, templateParams map[string]interface{
 func checkParams(userParams map[string]interface{}, templateParams map[string]interface{}, strictUnknown, backfill bool) error {
 	// 兼容扁平路径入参：还原为嵌套结构
 	orig := userParams
-	if IsFlatMap(userParams) {
-		nested, err := UnFlatBySjson(userParams)
+	if utils.IsFlatMap(userParams) {
+		nested, err := utils.UnFlatBySjson(userParams)
 		if err != nil {
 			return fmt.Errorf("无法解析用户参数: %w", err)
 		}
@@ -143,21 +144,29 @@ func validateNode(parent map[string]interface{}, key string, tmplNode interface{
 		case TypeObject:
 			// 对象元素：以元素 attrs 为容器递归校验子字段（模板对象节点 {type:object,attrs:{...}} 的
 			// 子字段藏在 attrs 下；纯对象 map 直接以自身为容器）
-			if attrs, ok := protoTmpl.Attrs.(map[string]interface{}); ok {
-				for i, elem := range userArr {
-					elemMap, isMap := elem.(map[string]interface{})
-					if !isMap {
-						continue
-					}
-					container, has := userObjectValue(elemMap, true)
-					if !has {
-						continue
-					}
-					for subKey, subTmpl := range attrs {
-						subPath := fmt.Sprintf("%s[%d].%s", path, i, subKey)
-						if err := validateNode(container, subKey, subTmpl, subPath, strictUnknown, backfill); err != nil {
-							return err
-						}
+			for i, elem := range userArr {
+				elemMap, isMap := elem.(map[string]interface{})
+				if !isMap {
+					continue
+				}
+				container, has := userObjectValue(elemMap, true)
+				if !has {
+					continue
+				}
+				// 元素带 type 键（schema 包裹）：已由 validateEnumValues 按其自身 attrs 校验，跳过，
+				// 避免用枚举首原型（可能是必填字段模板）误报其他槽位元素缺失
+				if _, isWrapped := elemMap["type"]; isWrapped {
+					continue
+				}
+				// 纯对象元素（解析后数组下标可能塌缩）：按 attrs 键集结构匹配槽位原型，避免恒用首原型误报必填
+				subAttrs := matchArraySlotProto(&tmpl, container)
+				if subAttrs == nil {
+					continue
+				}
+				for subKey, subTmpl := range subAttrs {
+					subPath := fmt.Sprintf("%s[%d].%s", path, i, subKey)
+					if err := validateNode(container, subKey, subTmpl, subPath, strictUnknown, backfill); err != nil {
+						return err
 					}
 				}
 			}
@@ -369,6 +378,38 @@ func arrayElementPrototype(tmpl *dto.Template) map[string]interface{} {
 	return nil
 }
 
+// matchArraySlotProto 按元素 attrs 键集与各槽位原型 attrs 键集的重合度匹配最合适的槽位原型。
+// 解析后数组下标可能塌缩（resolveArray 丢弃空元素），不能按 index 对齐，故用结构匹配。
+// 键集完全无重合时返回 nil（跳过该校验，避免用错误原型误报必填）。
+func matchArraySlotProto(tmpl *dto.Template, container map[string]interface{}) map[string]interface{} {
+	var best map[string]interface{}
+	bestCount := -1
+	for _, ev := range tmpl.EnumValues {
+		evMap, ok := ev.(map[string]interface{})
+		if !ok {
+			continue
+		}
+		attrs, ok := evMap["attrs"].(map[string]interface{})
+		if !ok {
+			continue
+		}
+		count := 0
+		for k := range container {
+			if _, has := attrs[k]; has {
+				count++
+			}
+		}
+		if count > bestCount {
+			bestCount = count
+			best = attrs
+		}
+	}
+	if bestCount <= 0 {
+		return nil
+	}
+	return best
+}
+
 // isValueEmptyByType 按 tmpl.Type 判断是否为"业务空值"
 func isValueEmptyByType(tmpl *dto.Template) bool {
 	switch tmpl.Type {