fix(dispatcher): 熔断器接回 failover —— 挂掉的主模型跳过而非每次白试（T4.E 收官）

- failoverModel 加每模型熔断器(阈值3/冷却20s，比编排层更紧)：主模型持续失败达阈值 → 熔断 → 后续请求直接跳过主、直连备用(省掉每次白试主的失败往返)；冷却到点半开放行探测打回主，成功即自动恢复走主(靠熔断器半开机制，无需外部通知) - 全部模型都熔断时强制试主兜底(编排层 o.breaker 兜"全挂") - WithTools 重包共享同一批 breakers(状态不清零)——否则每次 rewrap 熔断失效，关键坑 - harness 加 NewCircuitBreakerWith(threshold,cooldown,halfOpenMax) 参数化构造 - Generate/Stream 用泛型 runFailover 共用选路循环(去重) - 3 新单测：熔断跳过主/WithTools 共享熔断状态/冷却后半开恢复(全三态) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-02 10:49:54 +08:00
parent 65e939889e
commit 7c211719d2
4 changed files with 173 additions and 44 deletions
@@ -2,66 +2,107 @@ package llm

 import (
 	"context"
+	"errors"
 	"fmt"
+	"time"

 	"github.com/cloudwego/eino/components/model"
 	"github.com/cloudwego/eino/schema"
+
+	"github.com/sundynix/sundynix-dispatcher/internal/harness"
 )

+// 模型层 failover 的每模型熔断参数：比编排层更紧（更快跳过挂掉的主模型），冷却后半开探测恢复。
+const (
+	fbBreakerThreshold = 3                // 某模型连续失败达此数 → 熔断，后续请求跳过它
+	fbBreakerCooldown  = 20 * time.Second // 熔断后多久放行探测（回主模型）
+	fbBreakerHalfOpen  = 1
+)
+
+var errAllModelsFailed = errors.New("failover: 所有模型均不可用")
+
 // failoverModel 把多个 ToolCallingChatModel 串成主备链：按序调用，主模型遇错即切下一个，
 // 直到成功或全部失败。它本身就是个 model.ToolCallingChatModel，故 compose / ReAct / Chat
 // 全路径透明白嫖 failover —— 单 provider 抖动/挂掉时平台不整体宕。
 //
+// 每模型带一个熔断器：主模型持续失败达阈值即熔断，后续请求**直接跳过主、走备用**（省掉每次
+// 白试主模型的失败往返）；冷却到点半开放行一个探测回主，成功即恢复走主（自动，无需外部通知）。
+//
 // 局限(v1)：Stream 仅在「建流(Stream() 调用)同步报错」时切备；已开始回流 token 的中途失败不切
 // （输出已半出，无法干净重来）。连接级失败（拒连/立即 5xx）由 openai 客户端在 Stream() 同步返回，
 // 已覆盖"provider 整体挂"的主场景。
 type failoverModel struct {
 	models     []model.ToolCallingChatModel // 主模型在前，其余为按序备用
+	breakers   []*harness.CircuitBreaker    // 与 models 一一对应；WithTools 重包时共享（状态不清零）
 	onFailover func(idx int, err error)     // 切换回调（日志/观测；可空）
 }

 // newFailoverModel 建主备链。models 至少 1 个；只有 1 个时调用方应直接用该模型而非本包装。
 func newFailoverModel(models []model.ToolCallingChatModel, onFailover func(int, error)) model.ToolCallingChatModel {
-	return &failoverModel{models: models, onFailover: onFailover}
+	breakers := make([]*harness.CircuitBreaker, len(models))
+	for i := range breakers {
+		breakers[i] = harness.NewCircuitBreakerWith(fbBreakerThreshold, fbBreakerCooldown, fbBreakerHalfOpen)
+	}
+	return &failoverModel{models: models, breakers: breakers, onFailover: onFailover}
+}
+
+// runFailover 是 Generate/Stream 共用的选路循环：按序过链，熔断的模型跳过，成功即返回并上报，
+// 失败上报后切下一个；调用方取消(ctx)则不再切。全被熔断跳过时强制试主模型兜底。
+func runFailover[T any](f *failoverModel, ctx context.Context, call func(model.ToolCallingChatModel) (T, error)) (T, error) {
+	var zero T
+	var lastErr error
+	attempted := false
+	for i, m := range f.models {
+		if b := f.breakers[i]; b != nil && !b.Allow() {
+			continue // 该模型熔断中（Open 冷却）→ 跳过，直连下一个可用模型
+		}
+		attempted = true
+		out, err := call(m)
+		if err == nil {
+			f.report(i, true)
+			return out, nil
+		}
+		f.report(i, false)
+		lastErr = err
+		if ctx.Err() != nil { // 调用方主动取消/截止 → 不再切（切了也没用，且违背用户意图）
+			return zero, err
+		}
+		if f.onFailover != nil {
+			f.onFailover(i, err)
+		}
+	}
+	// 全部模型都熔断、一个没试 → 强制试主模型兜底（编排层 o.breaker 兜"全挂"）。
+	if !attempted && len(f.models) > 0 {
+		out, err := call(f.models[0])
+		f.report(0, err == nil)
+		return out, err
+	}
+	if lastErr == nil {
+		lastErr = errAllModelsFailed
+	}
+	return zero, lastErr
+}
+
+func (f *failoverModel) report(i int, success bool) {
+	if i >= 0 && i < len(f.breakers) && f.breakers[i] != nil {
+		f.breakers[i].Report(success)
+	}
 }

 func (f *failoverModel) Generate(ctx context.Context, input []*schema.Message, opts ...model.Option) (*schema.Message, error) {
-	var lastErr error
-	for i, m := range f.models {
-		out, err := m.Generate(ctx, input, opts...)
-		if err == nil {
-			return out, nil
-		}
-		lastErr = err
-		if !f.shouldFailover(ctx, i) {
-			return nil, err
-		}
-		if f.onFailover != nil {
-			f.onFailover(i, err)
-		}
-	}
-	return nil, lastErr
+	return runFailover(f, ctx, func(m model.ToolCallingChatModel) (*schema.Message, error) {
+		return m.Generate(ctx, input, opts...)
+	})
 }

 func (f *failoverModel) Stream(ctx context.Context, input []*schema.Message, opts ...model.Option) (*schema.StreamReader[*schema.Message], error) {
-	var lastErr error
-	for i, m := range f.models {
-		sr, err := m.Stream(ctx, input, opts...)
-		if err == nil {
-			return sr, nil
-		}
-		lastErr = err
-		if !f.shouldFailover(ctx, i) {
-			return nil, err
-		}
-		if f.onFailover != nil {
-			f.onFailover(i, err)
-		}
-	}
-	return nil, lastErr
+	return runFailover(f, ctx, func(m model.ToolCallingChatModel) (*schema.StreamReader[*schema.Message], error) {
+		return m.Stream(ctx, input, opts...)
+	})
 }

 // WithTools 给链上每个模型绑定工具，返回新的 failover 链（不可变，并发安全；ReAct 用）。
+// 关键：复用同一批 breakers（共享指针）——否则每次 rewrap 熔断状态清零，熔断形同虚设。
 func (f *failoverModel) WithTools(tools []*schema.ToolInfo) (model.ToolCallingChatModel, error) {
 	bound := make([]model.ToolCallingChatModel, len(f.models))
 	for i, m := range f.models {
@@ -71,12 +112,5 @@ func (f *failoverModel) WithTools(tools []*schema.ToolInfo) (model.ToolCallingCh
 		}
 		bound[i] = b
 	}
-	return &failoverModel{models: bound, onFailover: f.onFailover}, nil
-}
-
-// shouldFailover：还有备用模型 且 调用方未主动取消/截止 → 切。
-// 调用方取消(ctx.Err()!=nil) 不切——切了也没用，且违背用户意图（注：模型自身的请求超时走的是
-// 内部派生 ctx，不会污染父 ctx，故仍会正常 failover）。
-func (f *failoverModel) shouldFailover(ctx context.Context, idx int) bool {
-	return idx < len(f.models)-1 && ctx.Err() == nil
+	return &failoverModel{models: bound, breakers: f.breakers, onFailover: f.onFailover}, nil
 }