model-gateway/service/session_sync.go

package service

import (
	"context"
	"encoding/json"
	"fmt"
	"model-gateway/consts/model"
	"model-gateway/dao"
	"model-gateway/model/dto"
	"model-gateway/service/httpclient"
	modelUtils "model-gateway/service/utils"
	"time"

	"gitea.redpowerfuture.com/red-future/common/oss"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/util/gconv"
)

var ModelSession = &modelSessionService{}

type modelSessionService struct{}

// CreateSession 创建会话（同步调用，非流式）
func (s *modelSessionService) CreateSession(ctx context.Context, req *dto.CallModelSessionReq) (res *dto.ModelCallRes, err error) {
	startTime := time.Now()
	attempt := 0
	id := req.Id
	modelInfo := req.ModelInfo
	newRequestParams := req.RequestParams
LOOP:
	// 6) 模型请求
	modelRespBody, err := httpclient.ModelHttpNormalRequest(ctx, modelInfo.BaseURL, modelInfo.RequestHeadMapping, modelInfo.HttpMethod, newRequestParams)
	if err != nil {
		return nil, err
	}
	if modelRespBody == nil {
		return nil, fmt.Errorf("模型返回参数是空")
	}
	// 7) 上传模型返回参数文件
	uploadOriginalResp, err := Upload(ctx, &dto.UploadFileBytesReq{
		FileBytes: modelRespBody,
		FileName:  fmt.Sprintf("modelRespParams:%v.json", time.Now().UnixMilli()),
	})
	if err != nil {
		return nil, fmt.Errorf("上传模型返回参数文件失败：%v", err)
	}
	// 8) 更新模型会话信息
	updateModelSessionReq := dto.UpdateModelSessionReq{
		Id:                   id,
		OriginalResponsePath: uploadOriginalResp.FileURL,
	}
	docMsg := new(dto.ModelCallRes)
	docMsg.TaskId = id
	errCode, errMsg, err := parseModelError(modelRespBody, modelInfo.ErrorMessageMapping)
	if err != nil {
		return nil, fmt.Errorf("模型返回参数解析失败：%v", err)
	}
	if errCode != "" {
		if attempt < modelCallMaxRetries && shouldRetryWithMemory(ctx, modelInfo, errCode, errMsg, string(modelRespBody)) {
			attempt++
			wait := time.Duration(1<<attempt) * time.Second
			g.Log().Warningf(ctx, "模型上游调用异常，第 %d 次重试(等待 %v): code=%s err=%v", attempt+1, wait, errCode, errMsg)
			if waitErr := retryWait(ctx, attempt); waitErr != nil {
				return nil, waitErr
			}
			goto LOOP
		}
		docMsg.ErrorMsg = errMsg
		updateModelSessionReq.ErrorMsg = docMsg.ErrorMsg
	} else {
		if *model.ResponseTypeSync.Code() == *modelInfo.ResponseType {
			respBodyMap := make(map[string]string, len(modelInfo.ResponseBodyMapping))
			for k, _ := range modelInfo.ResponseBodyMapping {
				respBodyMap[k] = modelUtils.CleanFieldPath(k)
			}
			// 基于统一字段路径（GetByPath）按映射取值组装结果
			var respObj map[string]any
			if err = json.Unmarshal(modelRespBody, &respObj); err != nil {
				return nil, fmt.Errorf("模型返回参数解析失败：%v", err)
			}
			content := make(map[string]any, len(respBodyMap))
			for bizKey, jsonPath := range respBodyMap {
				content[bizKey] = oss.TempURLToOSS(ctx, modelUtils.GetByPathValue(respObj, jsonPath))
			}

			businessField := make(map[string]any, len(modelInfo.ResponseBusinessFieldMapping))
			for key, value := range modelInfo.ResponseBusinessFieldMapping {
				businessField[key] = modelUtils.GetByPathValue(respObj, modelUtils.CleanFieldPath(value))
			}
			err = gconv.Struct(businessField, docMsg)
			if err != nil {
				return nil, fmt.Errorf("模型返回参数解析失败：%v", err)
			}

			docMsg.Content = content
			docMsg.TotalTokens = gconv.Int64(modelUtils.GetByPathValue(respObj, modelUtils.CleanFieldPath(modelInfo.TokenMapping.TotalTokens)))
			docMsg.PromptTokens = gconv.Int64(modelUtils.GetByPathValue(respObj, modelUtils.CleanFieldPath(modelInfo.TokenMapping.PromptTokens)))
			docMsg.CompletionTokens = gconv.Int64(modelUtils.GetByPathValue(respObj, modelUtils.CleanFieldPath(modelInfo.TokenMapping.CompletionTokens)))

			updateModelSessionReq.PromptTokens = docMsg.PromptTokens
			updateModelSessionReq.CompletionTokens = docMsg.CompletionTokens
			updateModelSessionReq.TotalTokens = docMsg.TotalTokens
		} else {
			docMsg.Content = map[string]any{
				"respBody": modelRespBody,
			}
		}
	}
	if !g.IsEmpty(docMsg.Content) {
		// 9) 上传模型返回参数文件
		uploadNewResp, err := Upload(ctx, &dto.UploadFileBytesReq{
			FileBytes: gconv.Bytes(gconv.String(docMsg.Content)),
			FileName:  fmt.Sprintf("modelNewRespParams:%v.json", time.Now().UnixMilli()),
		})
		if err != nil {
			return nil, fmt.Errorf("上传模型返回参数文件失败：%v", err)
		}
		updateModelSessionReq.ResponsePath = uploadNewResp.FileURL
	}
	updateModelSessionReq.DurationSeconds = int64(time.Since(startTime).Seconds())
	// 9.5) 调 shop-user-trade 按用量算费（不本地换算；调用前门禁已保证配置存在，失败→0 不阻塞）
	mediaType := modelUtils.DetectMediaType(modelInfo.RequestBusinessFieldMapping, newRequestParams)
	docMsg.ModelId = modelInfo.Id // 引用行=系统模型 id，供 per_token 结算按系统模型计价
	docMsg.MediaType = mediaType
	docMsg.Cost = calcModelCost(ctx, modelInfo.Id,
		buildModelUsage(docMsg.PromptTokens, docMsg.CompletionTokens, 0, mediaType, 0))
	updateModelSessionReq.TotalCost = docMsg.Cost
	// 10) 更新模型会话信息
	_, err = dao.ModelSession.Update(ctx, &updateModelSessionReq)
	if err != nil {
		return nil, fmt.Errorf("更新模型会话信息失败: %v", err)
	}

	return docMsg, nil
}