From 7c211719d27d120cf8cda1ec565b835084bbb161 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Thu, 2 Jul 2026 10:49:54 +0800 Subject: [PATCH] =?UTF-8?q?fix(dispatcher):=20=E7=86=94=E6=96=AD=E5=99=A8?= =?UTF-8?q?=E6=8E=A5=E5=9B=9E=20failover=20=E2=80=94=E2=80=94=20=E6=8C=82?= =?UTF-8?q?=E6=8E=89=E7=9A=84=E4=B8=BB=E6=A8=A1=E5=9E=8B=E8=B7=B3=E8=BF=87?= =?UTF-8?q?=E8=80=8C=E9=9D=9E=E6=AF=8F=E6=AC=A1=E7=99=BD=E8=AF=95=EF=BC=88?= =?UTF-8?q?T4.E=20=E6=94=B6=E5=AE=98=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- DEPTH_ROADMAP.md | 4 +- .../internal/harness/circuitbreaker.go | 21 +++- sundynix-dispatcher/internal/llm/failover.go | 112 ++++++++++++------ .../internal/llm/failover_test.go | 80 +++++++++++++ 4 files changed, 173 insertions(+), 44 deletions(-) diff --git a/DEPTH_ROADMAP.md b/DEPTH_ROADMAP.md index 197444f..fbb1374 100644 --- a/DEPTH_ROADMAP.md +++ b/DEPTH_ROADMAP.md @@ -183,13 +183,13 @@ RBAC 未做，暂以单管理员账号代理；概览口径必须是**系统级* - [ ] mcp-py 算法层去桩：文档解析接 PaddleOCR/magic-pdf（mineru.py:11 返回空块）| M - [ ] 报告原生 PDF（现仅 Word，office/unioffice.go）| M -### [~] T4.E 编排引擎边角 +### [x] T4.E 编排引擎边角 ✅（5/5 全清） - [x] Map 节点错误传播 + 汇总 ✅ —— writeSection 返回 (body,err) 仅真·LLM 失败计错(预算/无模型是主动降级不算)；writeSections 汇总失败数、失败项 Body 带可见标记；mapNode 全失败→置 b.fatalErr(判 failed 非静默 done-空)、部分失败→trace+流式告警。report handleReport 同步提示部分失败。3 单测(全失败/部分精确计数/mapNode 置 fatalErr)。 - [ ] 熔断器接回 failover（harness/circuitbreaker.go 与 llm/failover.go 未接合，可能长卡备用）| M - [x] coordinator 专家超时 ✅ —— specialistTool 加 timeout(默认 specialistTimeout=3min)，InvokableRun 包 WithTimeout；超时作为"观察"跳过该专家(err=nil，lead 据其余综合，不中断协调)。2 单测(超时~50ms 跳过 / 正常不受影响)。 - [x] Branch else 兜底 ✅ —— branchNode 识别 default/else 边，主选择(true/false)未命中时走它；trace 明确区分「走 default / 未匹配收口结束(END) / 正常选路」，杜绝静默落 END 被误当 bug。3 断言单测(default 命中/条件真走 true/无 default 仍空)。compose 层原有 len==0→END 收口保留。 - [x] DSL 拓扑校验 ✅ —— gateway ParseAndAssemble 加 validateTopology：拦重复/空 id + 悬挂边(source/target 指向不存在节点)；对空图/报告任务{topic}/非标准载荷宽松放过不误伤。单测(合法/重复id/空id/悬挂边×2/空端点/空图放过) + live(悬挂边→422，合法→202)。节点-工具映射校验需工具注册表，留后续。 -- [ ] 熔断器接回 failover（circuitbreaker.go 与 failover.go 未接合，可能长卡备用）| M ← T4.E 唯一剩项(最微妙/blast radius 最大) +- [x] 熔断器接回 failover ✅ —— failover 加每模型熔断器(阈值3/冷却20s)：主持续失败→熔断→后续请求跳过主直连备用(省重复白试)；冷却半开探测回主、成功自动恢复(无需外部通知)。WithTools 重包共享 breakers(状态不清零，关键坑)。harness 加 NewCircuitBreakerWith 参数化构造。3 新单测(熔断跳过/WithTools共享状态/冷却恢复)。**澄清**：原审计"卡在备用"不准——failover 每次都重试主，真问题是宕机时每次白试主，本项修的是这个。 ### [ ] T4.F 健壮性 / 安全 / 性能收口（多为 S，可穿插着做） - [ ] 关键 DB 写失败上浮 5xx（现 best-effort 返 200，前端无感，task_handler.go:69 等）| M diff --git a/sundynix-dispatcher/internal/harness/circuitbreaker.go b/sundynix-dispatcher/internal/harness/circuitbreaker.go index 6037ff9..6d9b715 100644 --- a/sundynix-dispatcher/internal/harness/circuitbreaker.go +++ b/sundynix-dispatcher/internal/harness/circuitbreaker.go @@ -50,11 +50,26 @@ type CircuitBreaker struct { } func NewCircuitBreaker() *CircuitBreaker { + return NewCircuitBreakerWith(defaultThreshold, defaultCooldown, defaultHalfOpenMax) +} + +// NewCircuitBreakerWith 用自定义参数建熔断器（非法入参回退默认）。 +// 供模型层 failover 用更紧的阈值/冷却（比编排层更快跳过挂掉的主模型）；也便于测试注短冷却。 +func NewCircuitBreakerWith(threshold int, cooldown time.Duration, halfOpenMax int) *CircuitBreaker { + if threshold <= 0 { + threshold = defaultThreshold + } + if cooldown <= 0 { + cooldown = defaultCooldown + } + if halfOpenMax <= 0 { + halfOpenMax = defaultHalfOpenMax + } return &CircuitBreaker{ state: Closed, - threshold: defaultThreshold, - cooldown: defaultCooldown, - halfOpenMax: defaultHalfOpenMax, + threshold: threshold, + cooldown: cooldown, + halfOpenMax: halfOpenMax, now: time.Now, } } diff --git a/sundynix-dispatcher/internal/llm/failover.go b/sundynix-dispatcher/internal/llm/failover.go index 4003d0a..1e1d0f9 100644 --- a/sundynix-dispatcher/internal/llm/failover.go +++ b/sundynix-dispatcher/internal/llm/failover.go @@ -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 } diff --git a/sundynix-dispatcher/internal/llm/failover_test.go b/sundynix-dispatcher/internal/llm/failover_test.go index 5bdb8f2..7c1b0a1 100644 --- a/sundynix-dispatcher/internal/llm/failover_test.go +++ b/sundynix-dispatcher/internal/llm/failover_test.go @@ -4,9 +4,12 @@ import ( "context" "fmt" "testing" + "time" "github.com/cloudwego/eino/components/model" "github.com/cloudwego/eino/schema" + + "github.com/sundynix/sundynix-dispatcher/internal/harness" ) // fakeModel 是 model.ToolCallingChatModel 测试替身：fail=true 则 Generate/Stream 报错。 @@ -128,6 +131,83 @@ func TestFailover_StreamFailover(t *testing.T) { } } +// 主模型持续失败达阈值 → 熔断 → 后续请求跳过主、直连备用（省掉每次白试主的失败往返）。 +func TestFailover_TrippedPrimarySkipped(t *testing.T) { + pc, fc := 0, 0 + p := &fakeModel{name: "p", fail: true, calls: &pc} + fb := &fakeModel{name: "fb", reply: "备", calls: &fc} + m := newFailoverModel([]model.ToolCallingChatModel{p, fb}, nil) + // 连打 threshold+2 次：前 threshold 次主被调并失败→切备；熔断后主被跳过。 + for i := 0; i < fbBreakerThreshold+2; i++ { + if ans, err := genText(t, context.Background(), m); err != nil || ans != "备" { + t.Fatalf("每次都应最终拿到备用回答, got %q err=%v", ans, err) + } + } + if pc != fbBreakerThreshold { + t.Fatalf("主应在失败 %d 次后被熔断跳过，实际被调 %d 次", fbBreakerThreshold, pc) + } + if fc != fbBreakerThreshold+2 { + t.Fatalf("备用应每次都被调: got %d", fc) + } +} + +// 关键：WithTools 重包必须共享熔断状态。主已熔断 → 新链仍跳过主（否则 rewrap 清零 = 熔断失效）。 +func TestFailover_WithToolsSharesBreakerState(t *testing.T) { + pc, fc := 0, 0 + p := &fakeModel{name: "p", fail: true, calls: &pc} + fb := &fakeModel{name: "fb", reply: "备", calls: &fc} + m := newFailoverModel([]model.ToolCallingChatModel{p, fb}, nil) + for i := 0; i < fbBreakerThreshold; i++ { // 打到主熔断 + _, _ = genText(t, context.Background(), m) + } + if pc != fbBreakerThreshold { + t.Fatalf("主应被调 %d 次, got %d", fbBreakerThreshold, pc) + } + bound, err := m.WithTools(nil) + if err != nil { + t.Fatal(err) + } + before := pc + if _, err := genText(t, context.Background(), bound); err != nil { + t.Fatal(err) + } + if pc != before { + t.Fatalf("重包后主不应再被调（熔断状态须共享），却多调了 %d 次", pc-before) + } +} + +// 熔断后冷却到点 → 半开探测打回主 → 主已恢复则成功切回主。 +func TestFailover_RecoversAfterCooldown(t *testing.T) { + pc, fc := 0, 0 + p := &fakeModel{name: "p", reply: "主回答", fail: true, calls: &pc} + fb := &fakeModel{name: "fb", reply: "备", calls: &fc} + m := &failoverModel{ + models: []model.ToolCallingChatModel{p, fb}, + breakers: []*harness.CircuitBreaker{ + harness.NewCircuitBreakerWith(fbBreakerThreshold, 30*time.Millisecond, 1), + harness.NewCircuitBreakerWith(fbBreakerThreshold, 30*time.Millisecond, 1), + }, + } + for i := 0; i < fbBreakerThreshold; i++ { // 打到主熔断 + _, _ = genText(t, context.Background(), m) + } + tripped := pc + p.fail = false // 主恢复健康 + // 冷却前：主仍被跳过。 + if _, _ = genText(t, context.Background(), m); pc != tripped { + t.Fatalf("冷却前主应仍被跳过，却被调用了") + } + // 冷却到点：半开探测打回主 → 成功恢复。 + time.Sleep(40 * time.Millisecond) + ans, err := genText(t, context.Background(), m) + if err != nil || ans != "主回答" { + t.Fatalf("冷却后应探测回主并成功, got %q err=%v", ans, err) + } + if pc <= tripped { + t.Fatalf("冷却后主应被再次探测调用") + } +} + // TestFailover_WithToolsBindsAll WithTools 给链上每个模型绑定，返回仍是 failover 链。 func TestFailover_WithToolsBindsAll(t *testing.T) { m := newFailoverModel([]model.ToolCallingChatModel{