diff --git a/server/task_llm.go b/server/task_llm.go index 9fbf5b4..48accbc 100644 --- a/server/task_llm.go +++ b/server/task_llm.go @@ -183,27 +183,46 @@ func streamTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest committed := false var pending []llm.StreamEvent var streamErr error - for event, err := range selection.provider.Stream(ctx, req) { - if err != nil { - streamErr = err - break + // 同 provider 安全窗口重试:committed 之前(还没向调用方交付任何输出) + // 的瞬时失败可以原样重放,不会重复模型输出或工具执行。committed 之后、 + // ctx 取消、或确定性/额度错误则跳出,交给下方原有的透传/故障转移逻辑。 + for attempt := 0; ; attempt++ { + committed = false + pending = nil + streamErr = nil + for event, err := range selection.provider.Stream(ctx, req) { + if err != nil { + streamErr = err + break + } + if !committed && !streamEventCommitsOutput(event) { + pending = append(pending, event) + continue + } + if !committed { + for _, buffered := range pending { + if !yield(buffered, nil) { + return + } + } + pending = nil + committed = true + } + if !yield(event, nil) { + return + } } - if !committed && !streamEventCommitsOutput(event) { - pending = append(pending, event) + if streamErr != nil && !committed && ctx.Err() == nil && + attempt < sameProviderStreamRetries && isRetryableStreamError(streamErr) { + backoff := sameProviderRetryBackoff(attempt) + log.Printf("[task-llm] task %s 提交前流失败,%v 后同 provider 重试 (%d/%d): %v", + taskID, backoff, attempt+1, sameProviderStreamRetries, streamErr) + if sleepCtx(ctx, backoff) { + break // 退避期间 ctx 取消 → 停止重试 + } continue } - if !committed { - for _, buffered := range pending { - if !yield(buffered, nil) { - return - } - } - pending = nil - committed = true - } - if !yield(event, nil) { - return - } + break } if streamErr == nil { for _, buffered := range pending { @@ -262,6 +281,48 @@ func streamEventCommitsOutput(event llm.StreamEvent) bool { } } +// 提交前安全窗口内、对同一 provider 的重试次数。SDK 的 doStream 只重试建连阶段 +// (拿到 200 之前);流一旦开始,中途断流 / overloaded / 流内 429 等瞬时故障会直接 +// 冒泡成 model_error,零重试。只要一个 token 都还没交给调用方(!committed),重放 +// 完全相同的请求就不会重复模型输出或工具副作用,因此这里补一层同 provider 退避重试, +// 把这类抖动挡在意图整体重跑之前。 +const sameProviderStreamRetries = 2 + +// sameProviderRetryBackoff 是第 attempt 次重试前的退避(0.5s、1s…,上限 4s), +// 与 SDK backoffSleep 同风格但封顶更小,避免拖住 worker 的收尾/取消响应。 +// 以变量形式暴露,便于测试将退避置零。 +var sameProviderRetryBackoff = func(attempt int) time.Duration { + return min(500*time.Millisecond*(1<