108 lines
3.3 KiB
Go
108 lines
3.3 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
|
|
"model-gateway/dao"
|
|
"model-gateway/model/dto"
|
|
"model-gateway/model/entity"
|
|
|
|
"github.com/gogf/gf/v2/frame/g"
|
|
"golang.org/x/sync/singleflight"
|
|
)
|
|
|
|
var analysisGroup singleflight.Group
|
|
|
|
// shouldRetryWithMemory 统一重试判定：查持久记忆，未命中则调分析模型并落库。
|
|
// 记忆/分析/DB 任一环节失败均 fail-closed 不重试。
|
|
func shouldRetryWithMemory(ctx context.Context, modelInfo *entity.ModelManage, code, msg, rawBody string) (retry bool) {
|
|
if modelInfo == nil || (code == "" && msg == "") {
|
|
return false
|
|
}
|
|
key := buildMemoryKey(modelInfo.BaseURL, code, msg)
|
|
if row, err := dao.ModelErrorMemory.GetByKey(ctx, key); err != nil {
|
|
g.Log().Errorf(ctx, "查询错误重试记忆失败: %v", err)
|
|
return false
|
|
} else if row != nil {
|
|
return row.Retryable
|
|
}
|
|
retry, _ = analyzeOnce(key, func() (bool, string) {
|
|
model, ok := resolveAnalysisModel(ctx, modelInfo)
|
|
if !ok {
|
|
g.Log().Warningf(ctx, "无可用分析模型(对话模型)，错误不重试: code=%s", code)
|
|
return false, ""
|
|
}
|
|
r, reason, err := callAnalysisLLM(ctx, model, code, msg, rawBody)
|
|
if err != nil {
|
|
g.Log().Warningf(ctx, "错误分析失败，fail-closed不重试: %v", err)
|
|
return false, ""
|
|
}
|
|
row := &entity.ModelErrorMemory{
|
|
MemoryKey: key,
|
|
Upstream: modelInfo.BaseURL,
|
|
ErrorCode: code,
|
|
MsgFingerprint: msgFingerprint(msg),
|
|
Retryable: r,
|
|
Reason: reason,
|
|
AnalyzedBy: model.ModelName,
|
|
}
|
|
if err := dao.ModelErrorMemory.Upsert(ctx, row); err != nil {
|
|
g.Log().Errorf(ctx, "错误重试记忆落库失败: %v", err)
|
|
}
|
|
return r, reason
|
|
})
|
|
return retry
|
|
}
|
|
|
|
// analyzeOnce 按记忆键合并并发分析请求（singleflight）。
|
|
// 注：fn 失败时结果在本次突发内共享（后续新错误会重新分析）。
|
|
func analyzeOnce(key string, fn func() (bool, string)) (bool, string) {
|
|
v, err, _ := analysisGroup.Do(key, func() (any, error) {
|
|
retry, reason := fn()
|
|
return []any{retry, reason}, nil
|
|
})
|
|
if err != nil {
|
|
return false, ""
|
|
}
|
|
vals := v.([]any)
|
|
return vals[0].(bool), vals[1].(string)
|
|
}
|
|
|
|
var ModelErrorMemory = &modelErrorMemoryService{}
|
|
|
|
type modelErrorMemoryService struct{}
|
|
|
|
// List 错误重试记忆列表
|
|
func (s *modelErrorMemoryService) List(ctx context.Context, req *dto.GetErrorMemoryListReq) (res *dto.GetErrorMemoryListRes, err error) {
|
|
page, size := 1, 20
|
|
if req.Page != nil && req.Page.PageNum > 0 {
|
|
page = int(req.Page.PageNum)
|
|
}
|
|
if req.Page != nil && req.Page.PageSize > 0 {
|
|
size = int(req.Page.PageSize)
|
|
}
|
|
list, total, err := dao.ModelErrorMemory.List(ctx, page, size)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
res = &dto.GetErrorMemoryListRes{Total: total, List: make([]dto.ErrorMemoryItem, 0, len(list))}
|
|
for _, m := range list {
|
|
res.List = append(res.List, dto.ErrorMemoryItem{
|
|
Id: m.Id,
|
|
MemoryKey: m.MemoryKey,
|
|
Upstream: m.Upstream,
|
|
ErrorCode: m.ErrorCode,
|
|
MsgFingerprint: m.MsgFingerprint,
|
|
Retryable: m.Retryable,
|
|
Reason: m.Reason,
|
|
AnalyzedBy: m.AnalyzedBy,
|
|
})
|
|
}
|
|
return res, nil
|
|
}
|
|
|
|
// Delete 删除错误重试记忆（手动纠错永久记忆）
|
|
func (s *modelErrorMemoryService) Delete(ctx context.Context, req *dto.DeleteErrorMemoryReq) (err error) {
|
|
return dao.ModelErrorMemory.Delete(ctx, req.Id)
|
|
}
|