69 lines
2.1 KiB
Go
69 lines
2.1 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
|
|
"model-gateway/dao"
|
|
"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.ErrorMemory.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.ErrorMemory{
|
|
MemoryKey: key,
|
|
Upstream: modelInfo.BaseURL,
|
|
ErrorCode: code,
|
|
MsgFingerprint: msgFingerprint(msg),
|
|
Retryable: r,
|
|
Reason: reason,
|
|
AnalyzedBy: model.ModelName,
|
|
}
|
|
if err := dao.ErrorMemory.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)
|
|
}
|