feat(llm): 五层重试的次数与间隔可配
一次模型调用的失败由内到外经过五层重试:建连(SDK)、空响应(SDK)、同 provider 安全窗口、轮询熔断、意图重跑。此前五层的次数与间隔全是写死的常量,遇到脾气各异 的中转站只能改代码。现在都能在 LLM 页配置,不配就是原来的默认值。 配置语义五层统一(db.RetryRule):次数 0=内置默认 / -1=关掉这层 / >0=该值; 间隔 0=原本的指数退避 / >0=改用固定间隔。0 被"未配置"占用,故关闭用 -1。 写库前夹紧到 次数 [-1,20]、间隔 [0,1h],DB 另有 CHECK 兜底。 粒度按层的性质分:前三层跟着端点走,可在每个模型配置里逐字段覆盖全局默认(只钉 间隔的仍继承全局次数);熔断与意图重跑是进程级语义,只有全局一份。 生效路径:①②是 provider 构建参数,保存后重建缓存;③随每次 hooks.current() 解析,故障转移换配置时重试节奏跟着换;④在失败路径上每次都要读,由 applyRetryPolicy 推给 llmpool.Registry;⑤每条意图收尾读一次。全部热生效,无需重启。 连接测试刻意不带这些参数——它有 30s 硬超时,叠上用户配的重试只会把本来能用的 端点测成超时失败。 norma 升 v0.3.5(llm.Config 新增 RetryInterval / EmptyResponseRetries / EmptyResponseInterval)。旧库升级后行为逐字节不变:新列默认 0,settings 键不 存在即全默认。设计见 docs/LLM重试设计.md。
This commit is contained in:
Autumn-27 committed
1 parent
377ccc7e5a
commit
a50bf02428
21 files changed
+1216
-146
No files matched your search
+50
-28
@@ -91,6 +91,9 @@ type taskLLMSelection struct {
|
||||
profileID int64
|
||||
revision int64
|
||||
provider llm.Provider
|
||||
// retry 是这次选中的配置解析出来的重试参数(profile 覆盖 → 全局策略 → 内置默认)。
|
||||
// 同 provider 安全窗口重试按它走,所以换了 profile 就换一套重试节奏。
|
||||
retry agent.RetryConfig
|
||||
}
|
||||
|
||||
type taskLLMStreamHooks struct {
|
||||
@@ -105,38 +108,43 @@ type taskLLMStreamHooks struct {
|
||||
// 构建失败时才降级到任务链，任务链为空时再降级到全局配置。
|
||||
// 返回的 profile id 只有走任务链时才非零 —— streamTaskLLM 以此判断额度错误是否
|
||||
// 应该推进任务的故障转移状态（绑定/全局路径不改任务链状态，沿用既有语义）。
|
||||
func (r *taskLLMRuntime) current() (*Task, int64, int64, llm.Provider, error) {
|
||||
func (r *taskLLMRuntime) current() (taskLLMSelection, error) {
|
||||
taskNum, err := parseTaskID(r.taskID)
|
||||
if err != nil {
|
||||
return nil, 0, 0, nil, err
|
||||
return taskLLMSelection{}, err
|
||||
}
|
||||
pt, err := r.s.m.pg.GetTask(taskNum)
|
||||
if err != nil {
|
||||
return nil, 0, 0, nil, err
|
||||
return taskLLMSelection{}, err
|
||||
}
|
||||
if pt == nil {
|
||||
return nil, 0, 0, nil, fmt.Errorf("task %s not found", r.taskID)
|
||||
return taskLLMSelection{}, fmt.Errorf("task %s not found", r.taskID)
|
||||
}
|
||||
r.s.syncTaskLLMState(pt)
|
||||
t, _ := r.s.m.Task(r.taskID)
|
||||
if prov, _, ok := r.s.agentBindingProvider(r.agentKey); ok {
|
||||
return t, 0, pt.LLMChainRevision, prov, nil
|
||||
sel := taskLLMSelection{task: t, revision: pt.LLMChainRevision}
|
||||
if prov, cfg, ok := r.s.agentBindingProvider(r.agentKey); ok {
|
||||
sel.provider, sel.retry = prov, cfg.Retry
|
||||
return sel, nil
|
||||
}
|
||||
if len(pt.LLMProfileIDs) > 0 {
|
||||
if pt.ActiveLLMProfileID == nil {
|
||||
return t, 0, pt.LLMChainRevision, nil, &taskLLMError{taskID: r.taskID, chainExhausted: true, cause: errors.New("all selected profiles are quota exhausted")}
|
||||
return sel, &taskLLMError{taskID: r.taskID, chainExhausted: true, cause: errors.New("all selected profiles are quota exhausted")}
|
||||
}
|
||||
prov, _, ok := r.s.providerForProfile(*pt.ActiveLLMProfileID)
|
||||
sel.profileID = *pt.ActiveLLMProfileID
|
||||
prov, cfg, ok := r.s.providerForProfile(sel.profileID)
|
||||
if !ok {
|
||||
return t, *pt.ActiveLLMProfileID, pt.LLMChainRevision, nil, fmt.Errorf("LLM profile #%d is missing or invalid", *pt.ActiveLLMProfileID)
|
||||
return sel, fmt.Errorf("LLM profile #%d is missing or invalid", sel.profileID)
|
||||
}
|
||||
return t, *pt.ActiveLLMProfileID, pt.LLMChainRevision, prov, nil
|
||||
sel.provider, sel.retry = prov, cfg.Retry
|
||||
return sel, nil
|
||||
}
|
||||
prov, _, ok := r.s.globalProvider()
|
||||
prov, cfg, ok := r.s.globalProvider()
|
||||
if !ok {
|
||||
return t, 0, pt.LLMChainRevision, nil, fmt.Errorf("task %s has no available fallback LLM provider", r.taskID)
|
||||
return sel, fmt.Errorf("task %s has no available fallback LLM provider", r.taskID)
|
||||
}
|
||||
return t, 0, pt.LLMChainRevision, prov, nil
|
||||
sel.provider, sel.retry = prov, cfg.Retry
|
||||
return sel, nil
|
||||
}
|
||||
|
||||
// activeCfg resolves the task's currently-active LLM config, mirroring current()'s
|
||||
@@ -194,10 +202,7 @@ func parseTaskID(id string) (int64, error) {
|
||||
// exhausted, and how to emit a failover transition.
|
||||
func (r *taskLLMRuntime) streamHooks() taskLLMStreamHooks {
|
||||
return taskLLMStreamHooks{
|
||||
current: func() (taskLLMSelection, error) {
|
||||
task, profileID, revision, provider, err := r.current()
|
||||
return taskLLMSelection{task: task, profileID: profileID, revision: revision, provider: provider}, err
|
||||
},
|
||||
current: r.current,
|
||||
exhaust: func(selection taskLLMSelection, cause error) (db.TaskLLMTransition, error) {
|
||||
taskNum, _ := parseTaskID(r.taskID)
|
||||
transition, err := r.s.m.pg.MarkTaskLLMProfileQuotaExhaustedAtRevision(taskNum, selection.profileID, selection.revision, cause.Error())
|
||||
@@ -244,13 +249,14 @@ func completeTaskLLM(ctx context.Context, taskID string, req llm.CompletionReque
|
||||
)
|
||||
// 同 provider 安全窗口重试:非流式调用要么整体成功、要么整体失败,没有
|
||||
// 中途已交付输出的问题,所以任何瞬时失败都可原样重试。
|
||||
retries, backoffOf := sameProviderRetryPolicy(selection.retry)
|
||||
for attempt := 0; ; attempt++ {
|
||||
msg, sr, usage, callErr = selection.provider.Complete(ctx, req)
|
||||
if callErr != nil && ctx.Err() == nil &&
|
||||
attempt < sameProviderStreamRetries && isRetryableStreamError(callErr) {
|
||||
backoff := sameProviderRetryBackoff(attempt)
|
||||
attempt < retries && isRetryableStreamError(callErr) {
|
||||
backoff := backoffOf(attempt)
|
||||
log.Printf("[task-llm] task %s 非流式调用失败,%v 后同 provider 重试 (%d/%d): %v",
|
||||
taskID, backoff, attempt+1, sameProviderStreamRetries, callErr)
|
||||
taskID, backoff, attempt+1, retries, callErr)
|
||||
if sleepCtx(ctx, backoff) {
|
||||
break // 退避期间 ctx 取消 → 停止重试
|
||||
}
|
||||
@@ -290,6 +296,7 @@ func streamTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest
|
||||
committed := false
|
||||
var pending []llm.StreamEvent
|
||||
var streamErr error
|
||||
retries, backoffOf := sameProviderRetryPolicy(selection.retry)
|
||||
// 同 provider 安全窗口重试:committed 之前(还没向调用方交付任何输出)
|
||||
// 的瞬时失败可以原样重放,不会重复模型输出或工具执行。committed 之后、
|
||||
// ctx 取消、或确定性/额度错误则跳出,交给下方原有的透传/故障转移逻辑。
|
||||
@@ -320,10 +327,10 @@ func streamTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest
|
||||
}
|
||||
}
|
||||
if streamErr != nil && !committed && ctx.Err() == nil &&
|
||||
attempt < sameProviderStreamRetries && isRetryableStreamError(streamErr) {
|
||||
backoff := sameProviderRetryBackoff(attempt)
|
||||
attempt < retries && isRetryableStreamError(streamErr) {
|
||||
backoff := backoffOf(attempt)
|
||||
log.Printf("[task-llm] task %s 提交前流失败,%v 后同 provider 重试 (%d/%d): %v",
|
||||
taskID, backoff, attempt+1, sameProviderStreamRetries, streamErr)
|
||||
taskID, backoff, attempt+1, retries, streamErr)
|
||||
if sleepCtx(ctx, backoff) {
|
||||
break // 退避期间 ctx 取消 → 停止重试
|
||||
}
|
||||
@@ -388,20 +395,35 @@ func streamEventCommitsOutput(event llm.StreamEvent) bool {
|
||||
}
|
||||
}
|
||||
|
||||
// 提交前安全窗口内、对同一 provider 的重试次数。SDK 的 doStream 只重试建连阶段
|
||||
// (拿到 200 之前);流一旦开始,中途断流 / overloaded / 流内 429 等瞬时故障会直接
|
||||
// 提交前安全窗口内、对同一 provider 的【默认】重试次数。SDK 的 doStream 只重试建连
|
||||
// 阶段(拿到 200 之前);流一旦开始,中途断流 / overloaded / 流内 429 等瞬时故障会直接
|
||||
// 冒泡成 model_error,零重试。只要一个 token 都还没交给调用方(!committed),重放
|
||||
// 完全相同的请求就不会重复模型输出或工具副作用,因此这里补一层同 provider 退避重试,
|
||||
// 把这类抖动挡在意图整体重跑之前。
|
||||
// 把这类抖动挡在意图整体重跑之前。可被 LLM 配置的重试覆盖/全局重试策略改写。
|
||||
const sameProviderStreamRetries = 2
|
||||
|
||||
// sameProviderRetryBackoff 是第 attempt 次重试前的退避(0.5s、1s…,上限 4s),
|
||||
// 与 SDK backoffSleep 同风格但封顶更小,避免拖住 worker 的收尾/取消响应。
|
||||
// sameProviderRetryBackoff 是第 attempt 次重试前的【默认】退避(0.5s、1s…,上限 4s),
|
||||
// 与 SDK 的指数梯度同风格但封顶更小,避免拖住 worker 的收尾/取消响应。
|
||||
// 以变量形式暴露,便于测试将退避置零。
|
||||
var sameProviderRetryBackoff = func(attempt int) time.Duration {
|
||||
return min(500*time.Millisecond*(1<<attempt), 4*time.Second)
|
||||
}
|
||||
|
||||
// sameProviderRetryPolicy 解析这次调用用哪套同 provider 重试参数:配置里设了次数就
|
||||
// 用配置的(负数 = 关掉这层重试),设了间隔就把指数退避换成固定间隔,两者都没设时
|
||||
// 与改可配之前逐字节一致。
|
||||
func sameProviderRetryPolicy(r agent.RetryConfig) (retries int, backoff func(int) time.Duration) {
|
||||
retries, backoff = sameProviderStreamRetries, sameProviderRetryBackoff
|
||||
if r.StreamAttempts != 0 {
|
||||
retries = max(r.StreamAttempts, 0)
|
||||
}
|
||||
if r.StreamInterval > 0 {
|
||||
d := r.StreamInterval
|
||||
backoff = func(int) time.Duration { return d }
|
||||
}
|
||||
return retries, backoff
|
||||
}
|
||||
|
||||
// isRetryableStreamError 判断「提交前的流失败」是否值得在同一 provider 上重放。
|
||||
// 瞬时的传输中断 / 供应商过载 / 限流会自行恢复,可安全重试;而以下三类不重试:
|
||||
// - 额度耗尽:交给 profile 故障转移处理,别在这里白烧重试
|
||||
|
||||
Reference in new issue
Block a user