red-future/model-gateway
4
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor: 三路接入LLM+记忆重试判定

Browse Source
This commit is contained in:
19904408334
2026-09-01 10:49:44 +08:00
parent ea949a4b66
commit 94d408074a
4 changed files with 54 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
service/model_task_start_service.go
+18 -1
View File
@@ -21,14 +21,23 @@ type modelTaskStartService struct{}
// CreateTask 创建任务
func (s *modelTaskStartService) CreateTask(ctx context.Context, req *dto.CallModelTaskStartReq) (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 {
if attempt < modelCallMaxRetries && shouldRetryWithMemory(ctx, modelInfo, "", err.Error(), "") {
attempt++
if waitErr := retryWait(ctx, attempt); waitErr != nil {
return nil, waitErr
}
goto LOOP
}
return nil, fmt.Errorf("模型请求失败: %v", err)
}
if modelRespBody == nil {
@@ -46,10 +55,18 @@ func (s *modelTaskStartService) CreateTask(ctx context.Context, req *dto.CallMod
docMsg := new(dto.ModelCallRes)
docMsg.TaskId = id
// 按模型 ErrorMessageMapping 解析错误响应，无错误返回空串
if _, docMsg.ErrorMsg, err = parseModelError(modelRespBody, modelInfo.ErrorMessageMapping); err != nil {
var errCode string
if errCode, docMsg.ErrorMsg, err = parseModelError(modelRespBody, modelInfo.ErrorMessageMapping); err != nil {
return nil, fmt.Errorf("模型返回参数解析失败：%v", err)
}
if docMsg.ErrorMsg != "" {
if attempt < modelCallMaxRetries && shouldRetryWithMemory(ctx, modelInfo, errCode, docMsg.ErrorMsg, string(modelRespBody)) {
attempt++
if waitErr := retryWait(ctx, attempt); waitErr != nil {
return nil, waitErr
}
goto LOOP
}
updateModelReq.ErrorMsg = docMsg.ErrorMsg
}
if docMsg.ErrorMsg == "" {
service/retry.go
-10
View File
@@ -25,16 +25,6 @@ func retryWait(ctx context.Context, attempt int) error {
}
}
// isRetryableErrorCode 判定上游返回的错误码是否可重试：限流(429/limit_requests/limit_tokens/rate_limit_exceeded)与 5xx(500-503)。
// httpclient.ModelHttpNormalRequest 不返回 HTTP status，只能按响应体 error.code 字符串判定。
func isRetryableErrorCode(code string) bool {
switch code {
case "429", "500", "501", "502", "503", "InvalidParameter", "limit_requests", "limit_tokens", "rate_limit_exceeded":
return true
}
return false
}
// firstText 取任意值首位文本：数组取首个元素，其余原样转字符串
func firstText(v any) string {
if arr, ok := v.([]any); ok && len(arr) > 0 {
service/session_stream.go
+35 -32
View File
@@ -18,7 +18,7 @@ import (
)
// CreateSessionStreamOnce 流式调用上游模型 → 缓冲全量后一次返回（走 gf 框架正常返回）。
// 与同步请求一致：上游返回可重试错误码（限流/5xx）时按指数退避重试（最多 modelCallMaxRetries 次）。
// 与同步请求一致：上游返回错误时按 shouldRetryWithMemory 判定是否指数退避重试（最多 modelCallMaxRetries 次）。
func (s *modelSessionService) CreateSessionStreamOnce(ctx context.Context, req *dto.CallModelSessionReq) (docMsg *dto.ModelCallRes, err error) {
startTime := time.Now()
@@ -29,20 +29,22 @@ func (s *modelSessionService) CreateSessionStreamOnce(ctx context.Context, req *
attempt := 0
LOOP:
// 获取上游流式 reader（stream=false → w 不会被使用，传 nil）。
// 非 2xx 状态/网络错误在此返回；错误含可重试错误码（限流/5xx）时按指数退避重试，与同步请求一致。
// 非 2xx 状态/网络错误在此返回；按 shouldRetryWithMemory 判定是否指数退避重试，与同步请求一致。
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
if attempt < modelCallMaxRetries {
if code, msg := streamErrorInfoOfError(err); shouldRetryWithMemory(ctx, modelInfo, code, msg, "") {
attempt++
wait := time.Duration(1<<attempt) * time.Second
g.Log().Warningf(ctx, "模型流式请求异常，第 %d 次重试(等待 %v): code=%s err=%v", attempt+1, wait, code, err)
if waitErr := retryWait(ctx, attempt); waitErr != nil {
recordSessionError(context.WithoutCancel(ctx), id, startTime, "调用取消: "+waitErr.Error())
return nil, waitErr
}
goto LOOP
}
goto LOOP
}
// 非重试错误/重试耗尽：请求失败即返回，需把失败信息写入模型会话记录，避免留半截无错误信息记录
// 非重试错误/重试耗尽：请求失败即返回，把失败信息写入模型会话记录
recordSessionError(ctx, id, startTime, err.Error())
return nil, err
}
@@ -52,7 +54,7 @@ LOOP:
var contentBuf strings.Builder
// 记录流内 error 事件（OpenAI 兼容 error 分片），供流结束后统一判定重试/报错
var streamErrCode, streamErrMsg string
var streamErrCode, streamErrMsg, streamErrBody string
// 路径预处理
respMapping := make(map[string]string, len(modelInfo.ResponseBodyMapping))
@@ -66,7 +68,7 @@ LOOP:
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
streamErrCode, streamErrMsg, streamErrBody = code, msg, gconv.String(chunk["error"])
return nil
}
// 基于统一字段路径（GetByPath）在分片对象上取值，取首个数组元素文本累加
@@ -83,9 +85,9 @@ LOOP:
return nil
})
// 流内返回可重试错误码：丢弃本次部分内容，指数退避后重新请求
// 流内返回错误：丢弃本次部分内容，指数退避后重新请求（判定交给 shouldRetryWithMemory）
if streamErrCode != "" {
if attempt < modelCallMaxRetries && isRetryableErrorCode(streamErrCode) {
if attempt < modelCallMaxRetries && shouldRetryWithMemory(ctx, modelInfo, streamErrCode, streamErrMsg, streamErrBody) {
attempt++
wait := time.Duration(1<<attempt) * time.Second
g.Log().Warningf(ctx, "模型流式调用异常，第 %d 次重试(等待 %v): code=%s msg=%s", attempt+1, wait, streamErrCode, streamErrMsg)
@@ -317,39 +319,40 @@ func streamErrorOfChunk(chunk map[string]any) (code, msg string) {
return
}
// streamRetryCodeOfError 从流式请求错误中提取可重试错误码：优先解析错误体 error.code/顶层 code，
// 其次取非 2xx 的 HTTP 状态码；纯网络错误等无错误码场景返回空串（与同步请求一致，不重试）。
func streamRetryCodeOfError(err error) string {
// streamErrorInfoOfError 从流式请求错误中提取错误码与消息（不做固定清单过滤，交由 shouldRetryWithMemory 判定）：
// 优先解析错误体 error.code/顶层 code 与 message，其次取非 2xx 的 HTTP 状态码。
// 纯网络错误等无错误码场景返回空串。
func streamErrorInfoOfError(err error) (code, msg string) {
if err == nil {
return ""
return "", ""
}
msg := err.Error()
e := err.Error()
// 非 2xx 时 httpclient.ModelHttpStreamRequest 返回 "[HTTP][Stream] 状态码异常: %d, body={...}"
if idx := strings.Index(msg, "body="); idx >= 0 {
body := msg[idx+len("body="):]
if idx := strings.Index(e, "body="); idx >= 0 {
body := e[idx+len("body="):]
var errResp struct {
Error struct {
Code string `json:"code"`
Code string `json:"code"`
Message string `json:"message"`
} `json:"error"`
Code string `json:"code"`
Code string `json:"code"`
Message string `json:"message"`
}
if json.Unmarshal([]byte(body), &errResp) == nil {
if errResp.Error.Code != "" {
return errResp.Error.Code
return errResp.Error.Code, errResp.Error.Message
}
if errResp.Code != "" {
return errResp.Code
return errResp.Code, errResp.Message
}
}
}
if idx := strings.Index(msg, "状态码异常: "); idx >= 0 {
codeStr := strings.TrimSpace(msg[idx+len("状态码异常: "):])
if idx := strings.Index(e, "状态码异常: "); idx >= 0 {
codeStr := strings.TrimSpace(e[idx+len("状态码异常: "):])
if comma := strings.IndexByte(codeStr, ','); comma >= 0 {
codeStr = codeStr[:comma]
}
if isRetryableErrorCode(codeStr) {
return codeStr
}
return codeStr, ""
}
return ""
return "", ""
}
service/session_sync.go
+1 -3
View File
@@ -56,8 +56,7 @@ LOOP:
return nil, fmt.Errorf("模型返回参数解析失败：%v", err)
}
if errCode != "" {
if attempt < modelCallMaxRetries && isRetryableErrorCode(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)
@@ -66,7 +65,6 @@ LOOP:
}
goto LOOP
}
docMsg.ErrorMsg = errMsg
updateModelSessionReq.ErrorMsg = docMsg.ErrorMsg
} else {