19904408334andClaude
928682f121
- 删 CheckTenantBalance/DeductBalance/GetTenantSurplus（admin-go 租户余额退出 model-gateway） - 新增 pricing_client.go：门禁 config/get + calcModelCost 调 shop-user-trade /calc 算费 - sync/stream/async 三处 Cost 改为按用量调 /calc（媒体类型映射 text/audio/video/image） - 删 price.go 本地 PriceConfig 换算体系与 entity/dto 的 price_config 字段 Co-Authored-By: Claude <noreply@anthropic.com>
356 lines
15 KiB
Go
356 lines
15 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"model-gateway/dao"
|
||
"model-gateway/model/domain"
|
||
"model-gateway/model/dto"
|
||
"model-gateway/service/httpclient"
|
||
modelUtils "model-gateway/service/utils"
|
||
"net/http"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
"github.com/gogf/gf/v2/util/gconv"
|
||
)
|
||
|
||
// CreateSessionStreamOnce 流式调用上游模型 → 缓冲全量后一次返回（走 gf 框架正常返回）。
|
||
// 与同步请求一致：上游返回可重试错误码（限流/5xx）时按指数退避重试（最多 modelCallMaxRetries 次）。
|
||
func (s *modelSessionService) CreateSessionStreamOnce(ctx context.Context, req *dto.CallModelSessionReq) (docMsg *dto.ModelCallRes, err error) {
|
||
startTime := time.Now()
|
||
|
||
id := req.Id
|
||
modelInfo := req.ModelInfo
|
||
newRequestParams := req.RequestParams
|
||
|
||
attempt := 0
|
||
LOOP:
|
||
// 获取上游流式 reader（stream=false → w 不会被使用，传 nil）。
|
||
// 非 2xx 状态/网络错误在此返回；错误含可重试错误码（限流/5xx）时按指数退避重试，与同步请求一致。
|
||
streamReader, err := httpclient.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)
|
||
if waitErr := retryWait(ctx, attempt); waitErr != nil {
|
||
recordSessionError(context.WithoutCancel(ctx), id, startTime, "调用取消: "+waitErr.Error())
|
||
return nil, waitErr
|
||
}
|
||
goto LOOP
|
||
}
|
||
// 非重试错误/重试耗尽：请求失败即返回，需把失败信息写入模型会话记录，避免留半截无错误信息记录
|
||
recordSessionError(ctx, id, startTime, err.Error())
|
||
return nil, err
|
||
}
|
||
|
||
docMsg = new(dto.ModelCallRes)
|
||
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[k] = modelUtils.CleanFieldPath(k)
|
||
}
|
||
totalTokenPath := modelUtils.CleanFieldPath(modelInfo.TokenMapping.TotalTokens)
|
||
promptTokenPath := modelUtils.CleanFieldPath(modelInfo.TokenMapping.PromptTokens)
|
||
completionTokenPath := modelUtils.CleanFieldPath(modelInfo.TokenMapping.CompletionTokens)
|
||
|
||
httpclient.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 {
|
||
if realText := extractChunkText(ctx, modelUtils.GetByPathValue(chunk, jsonPath)); realText != "" {
|
||
contentBuf.WriteString(realText)
|
||
}
|
||
}
|
||
|
||
// Token 累加
|
||
docMsg.TotalTokens += gconv.Int64(modelUtils.GetByPathValue(chunk, totalTokenPath))
|
||
docMsg.PromptTokens += gconv.Int64(modelUtils.GetByPathValue(chunk, promptTokenPath))
|
||
docMsg.CompletionTokens += gconv.Int64(modelUtils.GetByPathValue(chunk, completionTokenPath))
|
||
return nil
|
||
})
|
||
|
||
// 流内返回可重试错误码：丢弃本次部分内容，指数退避后重新请求
|
||
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)
|
||
if waitErr := retryWait(ctx, attempt); waitErr != nil {
|
||
recordSessionError(context.WithoutCancel(ctx), id, startTime, "调用取消: "+waitErr.Error())
|
||
return nil, waitErr
|
||
}
|
||
goto LOOP
|
||
}
|
||
// 与同步一致：不可重试的错误码记录到 ErrorMsg 后正常走组装返回，不中断流程
|
||
docMsg.ErrorMsg = streamErrMsg
|
||
}
|
||
|
||
// 流结束后组装（流内出错时内容为空，与同步一致不再组装/上传空内容）
|
||
if streamErrCode == "" {
|
||
for k, _ := range modelInfo.ResponseBodyMapping {
|
||
docMsg.Content = map[string]any{k: contentBuf.String()}
|
||
}
|
||
}
|
||
|
||
// 补充更新会话记录
|
||
updateModelSessionReq := dto.UpdateModelSessionReq{
|
||
Id: id,
|
||
DurationSeconds: int64(time.Since(startTime).Seconds()),
|
||
TotalTokens: docMsg.TotalTokens,
|
||
PromptTokens: docMsg.PromptTokens,
|
||
CompletionTokens: docMsg.CompletionTokens,
|
||
ErrorMsg: docMsg.ErrorMsg,
|
||
}
|
||
if !g.IsEmpty(docMsg.Content) {
|
||
uploadNewResp, uploadErr := Upload(ctx, &dto.UploadFileBytesReq{
|
||
FileBytes: gconv.Bytes(gconv.String(docMsg.Content)),
|
||
FileName: fmt.Sprintf("modelNewRespParams:%v.json", time.Now().UnixMilli()),
|
||
})
|
||
if uploadErr != nil {
|
||
return nil, fmt.Errorf("上传模型返回参数文件失败：%v", uploadErr)
|
||
}
|
||
updateModelSessionReq.ResponsePath = uploadNewResp.FileURL
|
||
}
|
||
updateModelSessionReq.DurationSeconds = int64(time.Since(startTime).Seconds())
|
||
// 调 shop-user-trade 按用量算费（不本地换算；调用前门禁已保证配置存在，失败→0 不阻塞）
|
||
mediaType := DetectMediaType(modelInfo.RequestBusinessFieldMapping, newRequestParams)
|
||
docMsg.Cost = calcModelCost(ctx, modelInfo.Id,
|
||
buildModelUsage(docMsg.PromptTokens, docMsg.CompletionTokens, 0, mediaType, time.Since(startTime).Seconds()))
|
||
updateModelSessionReq.TotalCost = docMsg.Cost
|
||
if _, updateErr := dao.ModelSession.Update(ctx, &updateModelSessionReq); updateErr != nil {
|
||
g.Log().Errorf(ctx, "更新流式会话信息失败: %v", updateErr)
|
||
}
|
||
|
||
return docMsg, nil
|
||
}
|
||
|
||
// CreateSessionStream 流式调用上游模型 → SSE 逐分片推送给前端；流结束返回本次调用的 token/费用（docMsg.Cost）供调用方扣减。
|
||
func (s *modelSessionService) CreateSessionStream(ctx context.Context, w http.ResponseWriter, req *dto.CallModelSessionReq) (*dto.ModelCallRes, error) {
|
||
startTime := time.Now()
|
||
|
||
id := req.Id
|
||
modelInfo := req.ModelInfo
|
||
newRequestParams := req.RequestParams
|
||
|
||
// 获取上游流式 reader 并设置 SSE 响应头
|
||
streamReader, err := httpclient.ModelHttpStreamRequest(ctx, w, modelInfo.BaseURL, modelInfo.RequestHeadMapping, modelInfo.HttpMethod, newRequestParams)
|
||
if err != nil {
|
||
// 请求建立前失败：把错误写入模型会话记录，避免留半截无错误信息记录
|
||
recordSessionError(ctx, id, startTime, err.Error())
|
||
return nil, err
|
||
}
|
||
|
||
flusher := w.(http.Flusher)
|
||
|
||
docMsg := new(dto.ModelCallRes)
|
||
var contentBuf strings.Builder
|
||
|
||
// tool_calls 按 index 累加（OpenAI 兼容 delta），流末随 done 事件一次性返回
|
||
toolAcc := make(map[int]*streamToolCallAcc)
|
||
|
||
// 路径预处理
|
||
respMapping := make(map[string]string, len(modelInfo.ResponseBodyMapping))
|
||
for k, _ := range modelInfo.ResponseBodyMapping {
|
||
respMapping[modelUtils.CleanFieldPath(k)] = modelUtils.CleanFieldPath(k)
|
||
}
|
||
totalTokenPath := modelUtils.CleanFieldPath(modelInfo.TokenMapping.TotalTokens)
|
||
promptTokenPath := modelUtils.CleanFieldPath(modelInfo.TokenMapping.PromptTokens)
|
||
completionTokenPath := modelUtils.CleanFieldPath(modelInfo.TokenMapping.CompletionTokens)
|
||
|
||
// 解析 ResponseBusinessFieldMapping 字段
|
||
businessFieldRes := new(domain.ChatFieldsRes)
|
||
err = gconv.Struct(modelInfo.ResponseBusinessFieldMapping, businessFieldRes)
|
||
if err != nil {
|
||
g.Log().Errorf(ctx, "解析 ResponseBusinessFieldMapping 失败: %v", err)
|
||
}
|
||
// tool_calls 读取路径：优先走 ResponseBusinessFieldMapping 的 tools 配置，缺省兜底 OpenAI 兼容路径
|
||
//toolsPath := "choices[0].delta.tool_calls"
|
||
toolsPath := businessFieldRes.Tools
|
||
// reasoning_content 读取路径：走 ResponseBusinessFieldMapping 的 reasoning_content 配置，未配置则不返回思考内容
|
||
reasoningPath := modelUtils.CleanFieldPath(businessFieldRes.ReasoningContent)
|
||
|
||
httpclient.ParseSSEStream(ctx, streamReader, func(ctx context.Context, chunk map[string]any) error {
|
||
// 基于统一字段路径（GetByPath）在分片对象上取值，取首个数组元素文本
|
||
content := make(map[string]any, len(respMapping))
|
||
for bizKey, jsonPath := range respMapping {
|
||
if realText := extractChunkText(ctx, modelUtils.GetByPathValue(chunk, jsonPath)); realText != "" {
|
||
content[bizKey] = realText
|
||
contentBuf.WriteString(realText)
|
||
}
|
||
}
|
||
|
||
// Token 累加（记录增量：usage 常在无文本/思考的末分片出现，需据此放行推送）
|
||
prevTotal, prevPrompt, prevCompletion := docMsg.TotalTokens, docMsg.PromptTokens, docMsg.CompletionTokens
|
||
docMsg.TotalTokens += gconv.Int64(modelUtils.GetByPathValue(chunk, totalTokenPath))
|
||
docMsg.PromptTokens += gconv.Int64(modelUtils.GetByPathValue(chunk, promptTokenPath))
|
||
docMsg.CompletionTokens += gconv.Int64(modelUtils.GetByPathValue(chunk, completionTokenPath))
|
||
tokenDelta := docMsg.TotalTokens != prevTotal || docMsg.PromptTokens != prevPrompt || docMsg.CompletionTokens != prevCompletion
|
||
|
||
accumulateStreamToolCallsByPath(chunk, toolsPath, toolAcc)
|
||
|
||
// 思考内容提取（独立业务字段，不进回答全文）
|
||
var reasoningContent string
|
||
if reasoningPath != "" {
|
||
if v := modelUtils.GetByPathValue(chunk, reasoningPath); v != nil && !g.IsEmpty(v) {
|
||
reasoningContent = firstText(v)
|
||
}
|
||
}
|
||
|
||
// 纯 token 分片（无文本/思考）也放行：否则末分片 usage 被过滤，调用方拿不到 token 值
|
||
if len(content) == 0 && reasoningContent == "" && !tokenDelta {
|
||
return nil
|
||
}
|
||
|
||
// 逐 chunk SSE 推送给前端（字段名由 ModelCallStreamEvent 统一管理）
|
||
event := &dto.ModelCallStreamEvent{
|
||
Content: content,
|
||
ReasoningContent: reasoningContent,
|
||
TotalTokens: docMsg.TotalTokens,
|
||
PromptTokens: docMsg.PromptTokens,
|
||
CompletionTokens: docMsg.CompletionTokens,
|
||
}
|
||
outBytes, err := json.Marshal(event)
|
||
if err != nil {
|
||
g.Log().Errorf(ctx, "[SSE] marshal response failed: %v", err)
|
||
return nil
|
||
}
|
||
_, err = fmt.Fprintf(w, "data: %s\n\n", outBytes)
|
||
if err != nil {
|
||
g.Log().Errorf(ctx, "[SSE] write client failed: %v", err)
|
||
return err
|
||
}
|
||
flusher.Flush()
|
||
return nil
|
||
})
|
||
|
||
// 流结束：调 shop-user-trade 按用量算费（不本地换算；调用前门禁已保证配置存在，失败→0 不阻塞）
|
||
mediaType := DetectMediaType(modelInfo.RequestBusinessFieldMapping, newRequestParams)
|
||
docMsg.Cost = calcModelCost(ctx, modelInfo.Id,
|
||
buildModelUsage(docMsg.PromptTokens, docMsg.CompletionTokens, 0, mediaType, time.Since(startTime).Seconds()))
|
||
|
||
// 流末 done 事件：携带该步最终 token 与费用。工具调用时附带完整 tool_calls，
|
||
// 纯文本流同样补发，使调用方拿到最终费用与 token；不识别 type=done 的消费方忽略该事件。
|
||
event := &dto.ModelCallStreamEvent{
|
||
Type: "done",
|
||
TotalTokens: docMsg.TotalTokens,
|
||
PromptTokens: docMsg.PromptTokens,
|
||
CompletionTokens: docMsg.CompletionTokens,
|
||
Cost: docMsg.Cost,
|
||
}
|
||
if tools := finalizeStreamToolCalls(toolAcc); len(tools) > 0 {
|
||
var toolModels []dto.ModelTool
|
||
if err := gconv.Structs(tools, &toolModels); err != nil {
|
||
g.Log().Errorf(ctx, "[SSE] convert tools failed: %v", err)
|
||
} else {
|
||
event.Tools = toolModels
|
||
}
|
||
}
|
||
outBytes, err := json.Marshal(event)
|
||
if err == nil {
|
||
_, _ = fmt.Fprintf(w, "data: %s\n\n", outBytes)
|
||
flusher.Flush()
|
||
}
|
||
|
||
// 流结束后补充更新会话记录
|
||
updateModelSessionReq := dto.UpdateModelSessionReq{
|
||
Id: id,
|
||
DurationSeconds: int64(time.Since(startTime).Seconds()),
|
||
TotalTokens: docMsg.TotalTokens,
|
||
PromptTokens: docMsg.PromptTokens,
|
||
CompletionTokens: docMsg.CompletionTokens,
|
||
TotalCost: docMsg.Cost,
|
||
}
|
||
if !g.IsEmpty(contentBuf.String()) {
|
||
uploadNewResp, uploadErr := Upload(ctx, &dto.UploadFileBytesReq{
|
||
FileBytes: gconv.Bytes(gconv.String(map[string]any{"respBody": contentBuf.String()})),
|
||
FileName: fmt.Sprintf("modelNewRespParams:%v.json", time.Now().UnixMilli()),
|
||
})
|
||
if uploadErr != nil {
|
||
return nil, fmt.Errorf("上传模型返回参数文件失败：%v", uploadErr)
|
||
}
|
||
updateModelSessionReq.ResponsePath = uploadNewResp.FileURL
|
||
}
|
||
updateModelSessionReq.DurationSeconds = int64(time.Since(startTime).Seconds())
|
||
if _, updateErr := dao.ModelSession.Update(ctx, &updateModelSessionReq); updateErr != nil {
|
||
return nil, fmt.Errorf("更新会话信息失败: %v", updateErr)
|
||
}
|
||
|
||
return docMsg, nil
|
||
}
|
||
|
||
// recordSessionError 请求建立前失败（上游不可达/非 2xx 且非重试/重试耗尽/调用取消）时，
|
||
// 把错误与耗时写入模型会话记录，避免流式调用留半截无错误信息记录。
|
||
// 仅写 ErrorMsg/DurationSeconds（OmitEmpty 不会影响已落库字段）；ctx 已取消时须传 WithoutCancel(ctx)。
|
||
func recordSessionError(ctx context.Context, id int64, startTime time.Time, errMsg string) {
|
||
if _, updateErr := dao.ModelSession.Update(ctx, &dto.UpdateModelSessionReq{
|
||
Id: id,
|
||
DurationSeconds: int64(time.Since(startTime).Seconds()),
|
||
ErrorMsg: errMsg,
|
||
}); updateErr != nil {
|
||
g.Log().Errorf(ctx, "更新模型会话错误信息失败: %v", updateErr)
|
||
}
|
||
}
|
||
|
||
// 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 时 httpclient.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 ""
|
||
}
|