From ecf4a80466e38ad8fe93b723689a26bc26356c4b Mon Sep 17 00:00:00 2001 From: Blizzard Date: Tue, 30 Jun 2026 09:26:18 +0800 Subject: [PATCH] =?UTF-8?q?feat(llm):=20T2.1=20=E6=A8=A1=E5=9E=8B=E8=B7=AF?= =?UTF-8?q?=E7=94=B1=20+=20Fallback=20=E2=80=94=E2=80=94=20=E5=8D=95=20pro?= =?UTF-8?q?vider=20=E6=8A=96=E5=8A=A8=E4=B8=8D=E5=86=8D=E6=95=B4=E4=BD=93?= =?UTF-8?q?=E5=AE=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 现状：单 provider，一家 API 抖动/挂掉全平台不可用。加主备 failover：主模型调用失败 自动按序切备用，compose/ReAct/Chat 全路径透明白嫖。 - llm/failover.go: failoverModel 把多个 ToolCallingChatModel 串成主备链，按序调用、 遇错切下一个；它本身是 model.ToolCallingChatModel 故全路径透明。调用方主动取消 (ctx.Err()!=nil) 不切；模型自身请求超时走内部 ctx、不污染父 ctx 故仍 failover。 局限(v1)：Stream 仅建流同步报错时切（已回流 token 的中途失败不切）。 - llm/pool.go: SetConfig 用激活配置(含 Fallbacks)重建——主+可用备用串成 failover 链， 无备用则直接用主；备用单个构建失败跳过不影响主链。 - contract.ModelConfig: 加 Fallbacks 字段（骑在主配置里下发，不改任何 bus/ServeConfig 签名）。 - gateway store.ActiveConfig: chat 把"其它已登记 chat 模型"按序填进 Fallbacks； provide(main) + broadcast(admin) 共用 → 注册多个 chat 模型即自动成主备。 - bus.decryptConfig: 备用模型的 api_key（密文）一并解密。 测试：failover 单测（主可用不调备/主挂切备/全挂报错/取消不切/Stream 切备/WithTools 链）。 live 验证：active=死 ollama 主 + deepseek 备 → 任务连主拒连→自动切 deepseek→4s 完成。 DEPTH_ROADMAP T2.1（admin 注册多模型即主备，无需新 UI）。 Co-Authored-By: Claude Opus 4.8 --- sundynix-dispatcher/internal/llm/failover.go | 82 ++++++++++ .../internal/llm/failover_test.go | 145 ++++++++++++++++++ sundynix-dispatcher/internal/llm/pool.go | 48 +++++- sundynix-gateway/cmd/server/main.go | 6 +- sundynix-gateway/internal/handler/admin.go | 9 +- sundynix-gateway/internal/store/model.go | 24 +++ sundynix-shared/bus/bus.go | 17 +- sundynix-shared/contract/task.go | 3 + 8 files changed, 312 insertions(+), 22 deletions(-) create mode 100644 sundynix-dispatcher/internal/llm/failover.go create mode 100644 sundynix-dispatcher/internal/llm/failover_test.go diff --git a/sundynix-dispatcher/internal/llm/failover.go b/sundynix-dispatcher/internal/llm/failover.go new file mode 100644 index 0000000..4003d0a --- /dev/null +++ b/sundynix-dispatcher/internal/llm/failover.go @@ -0,0 +1,82 @@ +package llm + +import ( + "context" + "fmt" + + "github.com/cloudwego/eino/components/model" + "github.com/cloudwego/eino/schema" +) + +// failoverModel 把多个 ToolCallingChatModel 串成主备链：按序调用，主模型遇错即切下一个， +// 直到成功或全部失败。它本身就是个 model.ToolCallingChatModel，故 compose / ReAct / Chat +// 全路径透明白嫖 failover —— 单 provider 抖动/挂掉时平台不整体宕。 +// +// 局限(v1)：Stream 仅在「建流(Stream() 调用)同步报错」时切备；已开始回流 token 的中途失败不切 +// （输出已半出，无法干净重来）。连接级失败（拒连/立即 5xx）由 openai 客户端在 Stream() 同步返回， +// 已覆盖"provider 整体挂"的主场景。 +type failoverModel struct { + models []model.ToolCallingChatModel // 主模型在前，其余为按序备用 + 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} +} + +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 +} + +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 +} + +// WithTools 给链上每个模型绑定工具，返回新的 failover 链（不可变，并发安全；ReAct 用）。 +func (f *failoverModel) WithTools(tools []*schema.ToolInfo) (model.ToolCallingChatModel, error) { + bound := make([]model.ToolCallingChatModel, len(f.models)) + for i, m := range f.models { + b, err := m.WithTools(tools) + if err != nil { + return nil, fmt.Errorf("failover WithTools[%d]: %w", i, err) + } + 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 +} diff --git a/sundynix-dispatcher/internal/llm/failover_test.go b/sundynix-dispatcher/internal/llm/failover_test.go new file mode 100644 index 0000000..5bdb8f2 --- /dev/null +++ b/sundynix-dispatcher/internal/llm/failover_test.go @@ -0,0 +1,145 @@ +package llm + +import ( + "context" + "fmt" + "testing" + + "github.com/cloudwego/eino/components/model" + "github.com/cloudwego/eino/schema" +) + +// fakeModel 是 model.ToolCallingChatModel 测试替身：fail=true 则 Generate/Stream 报错。 +type fakeModel struct { + name string + reply string + fail bool + calls *int // 共享计数器，记录被调次数（验证 failover 没多调/早停） +} + +func (f *fakeModel) Generate(_ context.Context, _ []*schema.Message, _ ...model.Option) (*schema.Message, error) { + if f.calls != nil { + *f.calls++ + } + if f.fail { + return nil, fmt.Errorf("%s down", f.name) + } + return schema.AssistantMessage(f.reply, nil), nil +} + +func (f *fakeModel) Stream(_ context.Context, _ []*schema.Message, _ ...model.Option) (*schema.StreamReader[*schema.Message], error) { + if f.calls != nil { + *f.calls++ + } + if f.fail { + return nil, fmt.Errorf("%s down", f.name) + } + sr, sw := schema.Pipe[*schema.Message](1) + go func() { sw.Send(schema.AssistantMessage(f.reply, nil), nil); sw.Close() }() + return sr, nil +} + +func (f *fakeModel) WithTools(_ []*schema.ToolInfo) (model.ToolCallingChatModel, error) { return f, nil } + +func genText(t *testing.T, ctx context.Context, m model.ToolCallingChatModel) (string, error) { + t.Helper() + out, err := m.Generate(ctx, []*schema.Message{schema.UserMessage("hi")}) + if err != nil { + return "", err + } + return out.Content, nil +} + +// TestFailover_PrimaryWorks 主可用 → 直接用主，备用不被调用。 +func TestFailover_PrimaryWorks(t *testing.T) { + var pc, fc int + m := newFailoverModel([]model.ToolCallingChatModel{ + &fakeModel{name: "primary", reply: "主回答", calls: &pc}, + &fakeModel{name: "fb", reply: "备回答", calls: &fc}, + }, nil) + ans, err := genText(t, context.Background(), m) + if err != nil || ans != "主回答" { + t.Fatalf("应返回主回答, got %q err=%v", ans, err) + } + if pc != 1 || fc != 0 { + t.Fatalf("主可用时不应调备用: primary=%d fb=%d", pc, fc) + } +} + +// TestFailover_PrimaryDownUseFallback 主挂 → 切备用并成功。 +func TestFailover_PrimaryDownUseFallback(t *testing.T) { + var pc, fc int + switched := false + m := newFailoverModel([]model.ToolCallingChatModel{ + &fakeModel{name: "primary", fail: true, calls: &pc}, + &fakeModel{name: "fb", reply: "备回答", calls: &fc}, + }, func(int, error) { switched = true }) + ans, err := genText(t, context.Background(), m) + if err != nil || ans != "备回答" { + t.Fatalf("主挂应切备用, got %q err=%v", ans, err) + } + if pc != 1 || fc != 1 || !switched { + t.Fatalf("主备各调一次且触发回调: primary=%d fb=%d switched=%v", pc, fc, switched) + } +} + +// TestFailover_AllDown 全挂 → 返回最后一个错误。 +func TestFailover_AllDown(t *testing.T) { + m := newFailoverModel([]model.ToolCallingChatModel{ + &fakeModel{name: "p", fail: true}, + &fakeModel{name: "fb", fail: true}, + }, nil) + if _, err := genText(t, context.Background(), m); err == nil { + t.Fatal("全挂应返回错误") + } +} + +// TestFailover_CtxCanceledNoSwitch 调用方已取消 → 不切备用（切了也没用，且违背意图）。 +func TestFailover_CtxCanceledNoSwitch(t *testing.T) { + var pc, fc int + ctx, cancel := context.WithCancel(context.Background()) + cancel() // 立即取消 + m := newFailoverModel([]model.ToolCallingChatModel{ + &fakeModel{name: "p", fail: true, calls: &pc}, + &fakeModel{name: "fb", reply: "备", calls: &fc}, + }, nil) + if _, err := genText(t, ctx, m); err == nil { + t.Fatal("已取消仍应返回主错误") + } + if fc != 0 { + t.Fatalf("取消时不应切备用: fb 被调 %d 次", fc) + } +} + +// TestFailover_StreamFailover Stream 建流失败也切备用。 +func TestFailover_StreamFailover(t *testing.T) { + m := newFailoverModel([]model.ToolCallingChatModel{ + &fakeModel{name: "p", fail: true}, + &fakeModel{name: "fb", reply: "流备"}, + }, nil) + sr, err := m.Stream(context.Background(), []*schema.Message{schema.UserMessage("hi")}) + if err != nil { + t.Fatalf("Stream 应切到备用: %v", err) + } + defer sr.Close() + chunk, _ := sr.Recv() + if chunk == nil || chunk.Content != "流备" { + t.Fatalf("应收到备用的流, got %+v", chunk) + } +} + +// TestFailover_WithToolsBindsAll WithTools 给链上每个模型绑定，返回仍是 failover 链。 +func TestFailover_WithToolsBindsAll(t *testing.T) { + m := newFailoverModel([]model.ToolCallingChatModel{ + &fakeModel{name: "p", fail: true}, + &fakeModel{name: "fb", reply: "带工具的备"}, + }, nil) + bound, err := m.WithTools(nil) + if err != nil { + t.Fatal(err) + } + ans, err := genText(t, context.Background(), bound) + if err != nil || ans != "带工具的备" { + t.Fatalf("绑定工具后 failover 仍生效, got %q err=%v", ans, err) + } +} diff --git a/sundynix-dispatcher/internal/llm/pool.go b/sundynix-dispatcher/internal/llm/pool.go index 9bfb031..c9a1880 100644 --- a/sundynix-dispatcher/internal/llm/pool.go +++ b/sundynix-dispatcher/internal/llm/pool.go @@ -45,16 +45,11 @@ func NewPool() *Pool { return &Pool{} } // 绕开真实 LLM 推理与计费，只压全链路 plumbing（网关→NATS→调度→工具RTT→回流）。 func forceStub() bool { return os.Getenv("LLM_FORCE_STUB") == "1" } -// SetConfig 热更新后端配置：重建 ChatModel 实例（控制面变更时调用）。 +// SetConfig 热更新后端配置：用激活配置（含备用模型）重建 ChatModel（控制面变更时调用）。 func (p *Pool) SetConfig(cfg *contract.ModelConfig) { var cm model.BaseChatModel if cfg != nil && cfg.Ready() && !forceStub() { - built, err := buildChatModel(cfg) - if err != nil { - fmt.Printf("[llm] 构建 ChatModel 失败（降级桩运行）: %v\n", err) - } else { - cm = built - } + cm = buildWithFallbacks(cfg) } p.mu.Lock() p.cfg = cfg @@ -62,10 +57,47 @@ func (p *Pool) SetConfig(cfg *contract.ModelConfig) { p.mu.Unlock() if cfg != nil { // 不打印 api_key。 - fmt.Printf("[llm] model config set: provider=%s base=%s model=%s\n", cfg.Provider, normalizeBaseURL(cfg), cfg.Model) + fmt.Printf("[llm] model config set: provider=%s base=%s model=%s fallbacks=%d\n", + cfg.Provider, normalizeBaseURL(cfg), cfg.Model, len(cfg.Fallbacks)) } } +// buildWithFallbacks 构建主模型，并把可用的备用模型串成 failover 链（无备用则直接返回主模型）。 +// 主模型构建失败 → 返回 nil（降级桩）；备用单个失败 → 跳过该备用，不影响主链。 +func buildWithFallbacks(cfg *contract.ModelConfig) model.BaseChatModel { + primary, err := buildChatModel(cfg) + if err != nil { + fmt.Printf("[llm] 构建主 ChatModel 失败（降级桩运行）: %v\n", err) + return nil + } + ptcm, ok := primary.(model.ToolCallingChatModel) + if !ok || len(cfg.Fallbacks) == 0 { + return primary // 不支持 WithTools 或无备用 → 直接用主模型 + } + models := []model.ToolCallingChatModel{ptcm} + for i := range cfg.Fallbacks { + fb := cfg.Fallbacks[i] + if !fb.Ready() { + continue + } + fbm, ferr := buildChatModel(&fb) + if ferr != nil { + fmt.Printf("[llm] 构建备用模型 %s 失败，跳过: %v\n", fb.Model, ferr) + continue + } + if t, ok := fbm.(model.ToolCallingChatModel); ok { + models = append(models, t) + } + } + if len(models) == 1 { + return primary // 无有效备用 + } + fmt.Printf("[llm] 启用模型 failover：主 %s + %d 个备用\n", cfg.Model, len(models)-1) + return newFailoverModel(models, func(idx int, ferr error) { + fmt.Printf("[llm] 模型 failover：第 %d 个模型失败(%v)，切下一个\n", idx, ferr) + }) +} + // buildChatModel 据 provider 归一化连接参数后构建 OpenAI 兼容 ChatModel。 // vLLM 与 Ollama 都暴露 OpenAI 兼容 API（底层 go-openai 请求 {base}/chat/completions）， // 故统一走 openai 客户端，仅差在 BaseURL 是否带 /v1 与是否需要占位 key： diff --git a/sundynix-gateway/cmd/server/main.go b/sundynix-gateway/cmd/server/main.go index fee4d2d..e07143e 100644 --- a/sundynix-gateway/cmd/server/main.go +++ b/sundynix-gateway/cmd/server/main.go @@ -53,11 +53,7 @@ func main() { for _, kind := range []string{contract.ConfigKindChat, contract.ConfigKindEmbedding} { k := kind if _, err := bus.ServeConfig(k, func() *contract.ModelConfig { - row, _ := db.GetActiveModel(context.Background(), k) - if row == nil { - return nil - } - return &contract.ModelConfig{Provider: row.Provider, BaseURL: row.BaseURL, APIKey: row.APIKey, Model: row.Model} + return db.ActiveConfig(context.Background(), k) // chat 含 Fallbacks（其它模型作备用） }); err != nil { log.Printf("[gateway] serve %s config: %v", k, err) } diff --git a/sundynix-gateway/internal/handler/admin.go b/sundynix-gateway/internal/handler/admin.go index 1eb3647..fdc98da 100644 --- a/sundynix-gateway/internal/handler/admin.go +++ b/sundynix-gateway/internal/handler/admin.go @@ -202,15 +202,12 @@ func (h *Handler) TestModel(c *gin.Context) { } // broadcastActive 重新广播各 kind 当前激活配置，触发对应消费方热更新。 +// chat 配置带 Fallbacks（其它已登记 chat 模型作备用），dispatcher 据此重建 failover 链。 func (h *Handler) broadcastActive(ctx context.Context) { for _, kind := range []string{contract.ConfigKindChat, contract.ConfigKindEmbedding} { - row, _ := h.db.GetActiveModel(ctx, kind) - if row == nil { - continue + if cfg := h.db.ActiveConfig(ctx, kind); cfg != nil { + _ = h.bus.PublishConfigUpdated(kind, cfg) } - _ = h.bus.PublishConfigUpdated(kind, &contract.ModelConfig{ - Provider: row.Provider, BaseURL: row.BaseURL, APIKey: row.APIKey, Model: row.Model, - }) } } diff --git a/sundynix-gateway/internal/store/model.go b/sundynix-gateway/internal/store/model.go index 291c118..f8bf280 100644 --- a/sundynix-gateway/internal/store/model.go +++ b/sundynix-gateway/internal/store/model.go @@ -7,6 +7,8 @@ import ( "gorm.io/gorm" "gorm.io/gorm/clause" + + "github.com/sundynix/sundynix-shared/contract" ) // KB 是一个知识库（按 owner 隔离 + 按 kind 组织：文件夹/项目/案件/通用）。 @@ -321,6 +323,28 @@ func (p *Postgres) GetActiveModel(ctx context.Context, kind string) (*LLMModel, return &m, nil } +// ActiveConfig 取某 kind 的激活模型配置；chat 还把"其它已登记 chat 模型"按序填进 Fallbacks， +// 供 dispatcher 串成 failover 链（主 provider 抖动/挂掉时自动切备，平台不整体宕）。 +func (p *Postgres) ActiveConfig(ctx context.Context, kind string) *contract.ModelConfig { + row, _ := p.GetActiveModel(ctx, kind) + if row == nil { + return nil + } + cfg := &contract.ModelConfig{Provider: row.Provider, BaseURL: row.BaseURL, APIKey: row.APIKey, Model: row.Model} + if kind == contract.ConfigKindChat { + all, _ := p.ListModels(ctx, kind) + for _, m := range all { + if m.ID == row.ID { + continue // 跳过激活模型（它已是主） + } + cfg.Fallbacks = append(cfg.Fallbacks, contract.ModelConfig{ + Provider: m.Provider, BaseURL: m.BaseURL, APIKey: m.APIKey, Model: m.Model, + }) + } + } + return cfg +} + // DeleteModel 删除一条模型配置。 func (p *Postgres) DeleteModel(ctx context.Context, id string) error { if p.db == nil { diff --git a/sundynix-shared/bus/bus.go b/sundynix-shared/bus/bus.go index ddd5251..9c22aaf 100644 --- a/sundynix-shared/bus/bus.go +++ b/sundynix-shared/bus/bus.go @@ -24,12 +24,23 @@ import ( // decryptConfig 在消费侧把配置里的 api_key 从密文还原为明文（控制面以密文过线缆，见 secrets 包）。 // 失败（密钥不匹配 / 密文损坏）时清空 api_key 并不再降级阻断——调用方据 Ready() 判定。 +// 备用模型(Fallbacks)的 api_key 同样是密文，一并解密。 func decryptConfig(cfg *contract.ModelConfig) { - if cfg == nil || cfg.APIKey == "" { + if cfg == nil { return } - if plain, err := secrets.Decrypt(cfg.APIKey); err == nil { - cfg.APIKey = plain + if cfg.APIKey != "" { + if plain, err := secrets.Decrypt(cfg.APIKey); err == nil { + cfg.APIKey = plain + } + } + for i := range cfg.Fallbacks { + if cfg.Fallbacks[i].APIKey == "" { + continue + } + if plain, err := secrets.Decrypt(cfg.Fallbacks[i].APIKey); err == nil { + cfg.Fallbacks[i].APIKey = plain + } } } diff --git a/sundynix-shared/contract/task.go b/sundynix-shared/contract/task.go index cbd0b0c..ceb45cc 100644 --- a/sundynix-shared/contract/task.go +++ b/sundynix-shared/contract/task.go @@ -201,6 +201,9 @@ type ModelConfig struct { BaseURL string `json:"base_url"` // 如 https://api.deepseek.com APIKey string `json:"api_key,omitempty"` Model string `json:"model"` // 如 deepseek-chat / text-embedding-v3 + // Fallbacks 是主模型调用失败/超时时按序切换的备用模型（仅 chat 用，骑在主配置里一并下发）。 + // 网关把"其它 enabled chat 模型"填进来；dispatcher 据此把模型包成 failover 链。 + Fallbacks []ModelConfig `json:"fallbacks,omitempty"` } // Ready 报告该配置是否足以发起真实推理。