diff --git a/agent/goals.go b/agent/goals.go index 1ce68fa..dbbae9f 100644 --- a/agent/goals.go +++ b/agent/goals.go @@ -7,6 +7,7 @@ import ( "github.com/Autumn-27/artex/db" "github.com/Autumn-27/norma/agentcore" + "github.com/Autumn-27/norma/llm" acperm "github.com/Autumn-27/norma/permission" actool "github.com/Autumn-27/norma/tool" ) @@ -63,6 +64,11 @@ type GoalSpec struct { // no provider is configured or the call yields nothing — the caller then falls // back to a rule-based split so goal nodes always exist. // +// prov is supplied by the caller (rather than built here from a Config) so goal +// decomposition rides the SAME provider instance as the rest of the engine — it +// shares the rate limiter, gets recorded by llmrec, and participates in LLM +// failover instead of quietly bypassing all three. +// // desc is the task's free-text description (背景:靶标范围/flag 数量/交战说明等). // It is fed alongside the goal so the decomposer no longer splits blind — the // prompt still forbids inventing anything the two texts don't state. @@ -77,9 +83,8 @@ type GoalSpec struct { // straight into it (the same managed tool the main agent uses to add goals at // runtime). The returned specs are read back from the store so callers can emit // per-goal activity and detect the "LLM produced nothing" case for their fallback. -func DecomposeGoals(ctx context.Context, c Config, dataDir, goalText, desc string, as *db.AssetStore, ts *db.ExplorationStore, taskID int64, emit func(db.Activity)) []GoalSpec { - prov, err := c.NewProvider() - if err != nil { +func DecomposeGoals(ctx context.Context, prov llm.Provider, dataDir, goalText, desc string, as *db.AssetStore, ts *db.ExplorationStore, taskID int64, emit func(db.Activity)) []GoalSpec { + if prov == nil { return nil } // worker="goals" tags the goal nodes' provenance; ts/taskID let set_goals link diff --git a/db/config.go b/db/config.go index df31ddf..67d9edc 100644 --- a/db/config.go +++ b/db/config.go @@ -27,10 +27,29 @@ type LLMProfile struct { // 开启思考并设强度. Mapped to the provider request in agent.Config.NewProvider. ReasoningEffort string `json:"reasoning_effort"` IsDefault bool `json:"is_default"` + // Priority orders the failover chain: higher goes first. The ACTIVE profile + // (IsDefault) always heads the chain regardless of this value. + Priority int `json:"priority"` + // PoolExclude=true keeps this profile out of the failover chain — it stays + // usable when an agent/task binds it explicitly, it just never gets picked up + // as a fallback target. + PoolExclude bool `json:"pool_exclude"` +} + +// profileCols is the read column list (hint variant, no api key) shared by the +// list query; profileColsKey is the same with api_key for the single-row loads. +const profileCols = `id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key_hint,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default,priority,pool_exclude` +const profileColsKey = `id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default,priority,pool_exclude` + +// scanProfile reads one row in the profileCols / profileColsKey column order. The +// 7th column lands in APIKeyHint or APIKey depending on which list the caller used. +func scanProfile(sc interface{ Scan(...any) error }, into *string, p *LLMProfile) error { + return sc.Scan(&p.ID, &p.Name, &p.Format, &p.BaseURL, &p.Proxy, &p.Model, into, + &p.RatePerSecond, &p.RatePerMinute, &p.ContextWindowK, &p.ReasoningEffort, &p.IsDefault, &p.Priority, &p.PoolExclude) } func (d *DB) ListProfiles() ([]*LLMProfile, error) { - rows, err := d.Query(`SELECT id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key_hint,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default FROM llm_profiles ORDER BY id`) + rows, err := d.Query(`SELECT ` + profileCols + ` FROM llm_profiles ORDER BY id`) if err != nil { return nil, err } @@ -38,7 +57,7 @@ func (d *DB) ListProfiles() ([]*LLMProfile, error) { var out []*LLMProfile for rows.Next() { var p LLMProfile - if err := rows.Scan(&p.ID, &p.Name, &p.Format, &p.BaseURL, &p.Proxy, &p.Model, &p.APIKeyHint, &p.RatePerSecond, &p.RatePerMinute, &p.ContextWindowK, &p.ReasoningEffort, &p.IsDefault); err != nil { + if err := scanProfile(rows, &p.APIKeyHint, &p); err != nil { return nil, err } out = append(out, &p) @@ -49,8 +68,7 @@ func (d *DB) ListProfiles() ([]*LLMProfile, error) { // ActiveProfile returns the default (active) profile with its api key, or nil. func (d *DB) ActiveProfile() (*LLMProfile, error) { var p LLMProfile - err := d.QueryRow(`SELECT id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default FROM llm_profiles WHERE is_default LIMIT 1`). - Scan(&p.ID, &p.Name, &p.Format, &p.BaseURL, &p.Proxy, &p.Model, &p.APIKey, &p.RatePerSecond, &p.RatePerMinute, &p.ContextWindowK, &p.ReasoningEffort, &p.IsDefault) + err := scanProfile(d.QueryRow(`SELECT `+profileColsKey+` FROM llm_profiles WHERE is_default LIMIT 1`), &p.APIKey, &p) if err == sql.ErrNoRows { return nil, nil } @@ -61,14 +79,37 @@ func (d *DB) ActiveProfile() (*LLMProfile, error) { // Used to run a task on a specific (non-default) LLM profile. func (d *DB) ProfileByID(id int64) (*LLMProfile, error) { var p LLMProfile - err := d.QueryRow(`SELECT id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default FROM llm_profiles WHERE id=$1`, id). - Scan(&p.ID, &p.Name, &p.Format, &p.BaseURL, &p.Proxy, &p.Model, &p.APIKey, &p.RatePerSecond, &p.RatePerMinute, &p.ContextWindowK, &p.ReasoningEffort, &p.IsDefault) + err := scanProfile(d.QueryRow(`SELECT `+profileColsKey+` FROM llm_profiles WHERE id=$1`, id), &p.APIKey, &p) if err == sql.ErrNoRows { return nil, nil } return &p, err } +// PoolProfiles returns the failover chain in run order, api keys included: the +// active profile first, then every other keyed profile that isn't excluded, by +// priority DESC (id ASC to stay stable). Profiles without an api key can't serve +// a request, so they never enter the chain. The ordering IS the policy — callers +// walk the slice front to back. +func (d *DB) PoolProfiles() ([]*LLMProfile, error) { + rows, err := d.Query(`SELECT ` + profileColsKey + ` FROM llm_profiles +WHERE COALESCE(api_key,'') <> '' AND (is_default OR NOT pool_exclude) +ORDER BY is_default DESC, priority DESC, id ASC`) + if err != nil { + return nil, err + } + defer rows.Close() + var out []*LLMProfile + for rows.Next() { + var p LLMProfile + if err := scanProfile(rows, &p.APIKey, &p); err != nil { + return nil, err + } + out = append(out, &p) + } + return out, rows.Err() +} + // SaveProfile inserts (id==0) or updates a profile. Empty apiKey on update keeps existing. func (d *DB) SaveProfile(p *LLMProfile) (int64, error) { hint := p.APIKeyHint @@ -77,18 +118,18 @@ func (d *DB) SaveProfile(p *LLMProfile) (int64, error) { } if p.ID == 0 { var id int64 - err := d.QueryRow(`INSERT INTO llm_profiles(name,format,base_url,proxy,model,api_key,api_key_hint,rate_per_second,rate_per_minute,context_window_k,reasoning_effort) -VALUES ($1,$2,NULLIF($3,''),NULLIF($4,''),$5,NULLIF($6,''),NULLIF($7,''),$8,$9,$10,$11) RETURNING id`, - p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.APIKey, hint, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort).Scan(&id) + err := d.QueryRow(`INSERT INTO llm_profiles(name,format,base_url,proxy,model,api_key,api_key_hint,rate_per_second,rate_per_minute,context_window_k,reasoning_effort,priority,pool_exclude) +VALUES ($1,$2,NULLIF($3,''),NULLIF($4,''),$5,NULLIF($6,''),NULLIF($7,''),$8,$9,$10,$11,$12,$13) RETURNING id`, + p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.APIKey, hint, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude).Scan(&id) return id, err } if p.APIKey == "" { - _, err := d.Exec(`UPDATE llm_profiles SET name=$1,format=$2,base_url=NULLIF($3,''),proxy=NULLIF($4,''),model=$5,rate_per_second=$6,rate_per_minute=$7,context_window_k=$8,reasoning_effort=$9 WHERE id=$10`, - p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.ID) + _, err := d.Exec(`UPDATE llm_profiles SET name=$1,format=$2,base_url=NULLIF($3,''),proxy=NULLIF($4,''),model=$5,rate_per_second=$6,rate_per_minute=$7,context_window_k=$8,reasoning_effort=$9,priority=$10,pool_exclude=$11 WHERE id=$12`, + p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ID) return p.ID, err } - _, err := d.Exec(`UPDATE llm_profiles SET name=$1,format=$2,base_url=NULLIF($3,''),proxy=NULLIF($4,''),model=$5,api_key=$6,api_key_hint=$7,rate_per_second=$8,rate_per_minute=$9,context_window_k=$10,reasoning_effort=$11 WHERE id=$12`, - p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.APIKey, hint, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.ID) + _, err := d.Exec(`UPDATE llm_profiles SET name=$1,format=$2,base_url=NULLIF($3,''),proxy=NULLIF($4,''),model=$5,api_key=$6,api_key_hint=$7,rate_per_second=$8,rate_per_minute=$9,context_window_k=$10,reasoning_effort=$11,priority=$12,pool_exclude=$13 WHERE id=$14`, + p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.APIKey, hint, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ID) return p.ID, err } diff --git a/db/config_test.go b/db/config_test.go index cd4c642..6f1743e 100644 --- a/db/config_test.go +++ b/db/config_test.go @@ -5,6 +5,54 @@ import ( "testing" ) +// TestPoolProfilesOrder pins the failover chain query: keyless profiles can't +// serve a request and excluded ones aren't fallback targets, so neither belongs +// in the chain; the rest come back by priority, highest first. +// Deliberately does NOT touch is_default — flipping the active profile would be a +// side effect on the shared dev database. +func TestPoolProfilesOrder(t *testing.T) { + d, err := Open(testDSN(t)) + if err != nil { + t.Skipf("postgres unavailable (%v) — skipping", err) + } + defer d.Close() + + mk := func(name string, priority int, exclude bool, key string) int64 { + id, err := d.SaveProfile(&LLMProfile{ + Name: name, Format: "openai", Model: "m", APIKey: key, + Priority: priority, PoolExclude: exclude, + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { d.Exec(`DELETE FROM llm_profiles WHERE id=$1`, id) }) + return id + } + lo := mk("t-pool-lo", 1, false, "k1") + hi := mk("t-pool-hi", 9, false, "k2") + mk("t-pool-excluded", 99, true, "k3") // excluded despite the top priority + mk("t-pool-nokey", 50, false, "") // no key → cannot serve anything + + chain, err := d.PoolProfiles() + if err != nil { + t.Fatal(err) + } + var got []int64 + for _, p := range chain { + switch p.Name { + case "t-pool-lo", "t-pool-hi": + got = append(got, p.ID) + case "t-pool-excluded": + t.Fatal("pool_exclude profile entered the failover chain") + case "t-pool-nokey": + t.Fatal("keyless profile entered the failover chain") + } + } + if len(got) != 2 || got[0] != hi || got[1] != lo { + t.Fatalf("chain order = %v, want [hi=%d lo=%d]", got, hi, lo) + } +} + func TestConfigStores(t *testing.T) { d, err := Open(testDSN(t)) if err != nil { diff --git a/db/llmhealth.go b/db/llmhealth.go new file mode 100644 index 0000000..21f9f02 --- /dev/null +++ b/db/llmhealth.go @@ -0,0 +1,57 @@ +package db + +import "time" + +// LLM failover health: one row per profile tracking its circuit-breaker state. +// The authoritative copy lives in memory (llmpool.Registry); this table only +// survives a restart so a cooling-off window isn't silently reset by one. + +// LLMHealth is one profile's circuit-breaker state. +type LLMHealth struct { + ProfileID int64 `json:"profile_id"` + Fails int `json:"fails"` // consecutive failures; cleared on success + Trips int `json:"trips"` // total trips, drives the backoff ladder + OpenUntil *time.Time `json:"open_until"` // nil/past = closed (healthy) + LastError string `json:"last_error"` + LastAt time.Time `json:"last_at"` +} + +// LoadLLMHealth returns the profiles still in an UNEXPIRED cooling-off window. +// Expired rows are deliberately skipped: after a restart a profile that has +// finished cooling should be treated as healthy again and re-probed on its next +// call, not resurrected as broken. +func (d *DB) LoadLLMHealth() ([]LLMHealth, error) { + rows, err := d.Query(`SELECT profile_id,fails,trips,open_until,COALESCE(last_error,''),last_at +FROM llm_profile_health WHERE open_until IS NOT NULL AND open_until > now()`) + if err != nil { + return nil, err + } + defer rows.Close() + var out []LLMHealth + for rows.Next() { + var h LLMHealth + if err := rows.Scan(&h.ProfileID, &h.Fails, &h.Trips, &h.OpenUntil, &h.LastError, &h.LastAt); err != nil { + return nil, err + } + out = append(out, h) + } + return out, rows.Err() +} + +// SaveLLMHealth upserts one profile's circuit-breaker state. +func (d *DB) SaveLLMHealth(h LLMHealth) error { + _, err := d.Exec(` +INSERT INTO llm_profile_health(profile_id,fails,trips,open_until,last_error,last_at) +VALUES ($1,$2,$3,$4,$5,now()) +ON CONFLICT (profile_id) DO UPDATE SET + fails=EXCLUDED.fails, trips=EXCLUDED.trips, open_until=EXCLUDED.open_until, + last_error=EXCLUDED.last_error, last_at=now()`, + h.ProfileID, h.Fails, h.Trips, h.OpenUntil, h.LastError) + return err +} + +// ClearLLMHealth drops one profile's state (manual "recover now" from the UI). +func (d *DB) ClearLLMHealth(profileID int64) error { + _, err := d.Exec(`DELETE FROM llm_profile_health WHERE profile_id=$1`, profileID) + return err +} diff --git a/db/schema.sql b/db/schema.sql index 053cfe5..32ba6f9 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -214,6 +214,11 @@ CREATE TABLE IF NOT EXISTS llm_profiles ( context_window_k INTEGER NOT NULL DEFAULT 0, reasoning_effort TEXT NOT NULL DEFAULT '', is_default BOOLEAN NOT NULL DEFAULT false, + -- 轮询(故障转移)参数,见 docs/LLM轮询设计.md: + -- priority 顺位,越大越先被选中;激活配置(is_default)永远排链首,与本值无关。 + -- pool_exclude true=不作为故障转移目标(仍可被 agent/任务显式绑定使用)。 + priority INTEGER NOT NULL DEFAULT 0, + pool_exclude BOOLEAN NOT NULL DEFAULT false, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); @@ -221,6 +226,21 @@ CREATE UNIQUE INDEX IF NOT EXISTS uq_llm_one_default ON llm_profiles(is_default) DROP TRIGGER IF EXISTS trg_llm_upd ON llm_profiles; CREATE TRIGGER trg_llm_upd BEFORE UPDATE ON llm_profiles FOR EACH ROW EXECUTE FUNCTION set_updated_at(); +-- 轮询顺位/排除标记;补旧库。默认 0 / false = 全部配置都参与轮询。 +ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS priority INTEGER NOT NULL DEFAULT 0; +ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS pool_exclude BOOLEAN NOT NULL DEFAULT false; + +-- LLM 轮询熔断状态:某个配置连续失败(余额不足/key 失效/限流)后进入冷却,冷却期内 +-- 轮询直接跳过它。内存态为准,这里落库只为重启后不丢冷却窗口——加载时只取尚未 +-- 到期的行(open_until > now),已到期的自然回到"正常",等下一次调用半开试探。 +CREATE TABLE IF NOT EXISTS llm_profile_health ( + profile_id BIGINT PRIMARY KEY REFERENCES llm_profiles(id) ON DELETE CASCADE, + fails INTEGER NOT NULL DEFAULT 0, -- 当前连续失败次数(成功即清零) + trips INTEGER NOT NULL DEFAULT 0, -- 累计熔断次数,用于冷却时间指数退避 + open_until TIMESTAMPTZ, -- 冷却截止;NULL/过期 = 未熔断 + last_error TEXT NOT NULL DEFAULT '', + last_at TIMESTAMPTZ NOT NULL DEFAULT now() +); -- ===================================================================== -- D. 任务层 diff --git a/llmpool/health.go b/llmpool/health.go new file mode 100644 index 0000000..80ce1f4 --- /dev/null +++ b/llmpool/health.go @@ -0,0 +1,149 @@ +// Package llmpool implements LLM failover ("轮询"): a Provider decorator that +// walks an ordered chain of LLM profiles and moves to the next one when the +// current one can't serve the request — out of credit, revoked key, rate-limited +// past the SDK's own retries, or down. +// +// The chain is built by the server from llm_profiles (active profile first, then +// by priority), so the whole process shares ONE chain and ONE circuit-breaker +// registry: when a task discovers a profile is out of credit, every other task +// skips it immediately. +package llmpool + +import ( + "sync" + "time" +) + +// backoff is the cooling-off ladder, indexed by trip count: the 1st trip cools +// for a minute, the 2nd for five, everything after that for half an hour. A +// profile that is merely rate-limited recovers quickly; one that keeps failing +// stops being probed every round. +var backoff = []time.Duration{time.Minute, 5 * time.Minute, 30 * time.Minute} + +// State is one profile's circuit-breaker state. +type State struct { + Fails int // consecutive failures; reset by a success + Trips int // total trips, indexes the backoff ladder + OpenUntil time.Time // zero / past = closed (usable) + LastError string + LastAt time.Time +} + +// Open reports whether the breaker is currently open (profile should be skipped). +func (s State) Open() bool { return time.Now().Before(s.OpenUntil) } + +// softTripAfter is how many consecutive TRANSIENT failures (429 / 5xx / network) +// trip the breaker. Deterministic failures (no credit, bad key) trip on the first. +const softTripAfter = 3 + +// Registry holds the circuit-breaker state of every profile, keyed by profile id. +// It is process-wide and outlives any single Pool instance, so rebuilding the pool +// (saving an unrelated profile, toggling a setting) never clears what we've learned +// about which backends are broken. +type Registry struct { + mu sync.Mutex + m map[int64]*State + + // persist / forget mirror state to the DB so a cooling-off window survives a + // restart. Both may be nil (no DB); both are called off the hot path. + persist func(id int64, st State) + forget func(id int64) +} + +// NewRegistry builds an empty registry. persist/forget may be nil. +func NewRegistry(persist func(id int64, st State), forget func(id int64)) *Registry { + return &Registry{m: map[int64]*State{}, persist: persist, forget: forget} +} + +// Restore seeds state loaded from the DB at startup (bypasses persistence). +func (r *Registry) Restore(id int64, st State) { + r.mu.Lock() + defer r.mu.Unlock() + cp := st + r.m[id] = &cp +} + +// Get returns a copy of one profile's state. +func (r *Registry) Get(id int64) State { + r.mu.Lock() + defer r.mu.Unlock() + if st := r.m[id]; st != nil { + return *st + } + return State{} +} + +// IsOpen reports whether a profile is in its cooling-off window. +func (r *Registry) IsOpen(id int64) bool { return r.Get(id).Open() } + +// Pass records a successful call: clears the failure counters so a profile that +// recovers is fully trusted again (and drops the persisted row). +func (r *Registry) Pass(id int64) { + r.mu.Lock() + st := r.m[id] + if st == nil || (st.Fails == 0 && st.Trips == 0 && st.OpenUntil.IsZero()) { + r.mu.Unlock() + return // already clean — nothing to write + } + delete(r.m, id) + forget := r.forget + r.mu.Unlock() + if forget != nil { + forget(id) + } +} + +// Trip records a failed call. hard=true marks a deterministic failure (no credit, +// invalid key, missing model) which opens the breaker immediately; hard=false is a +// transient one (429 / 5xx / network) that needs softTripAfter in a row. Returns +// true when this call is what opened the breaker, so the caller can log it once. +func (r *Registry) Trip(id int64, errMsg string, hard bool) (tripped bool) { + r.mu.Lock() + st := r.m[id] + if st == nil { + st = &State{} + r.m[id] = st + } + st.Fails++ + st.LastError = errMsg + st.LastAt = time.Now() + if hard || st.Fails >= softTripAfter { + st.Trips++ + d := backoff[len(backoff)-1] + if st.Trips-1 < len(backoff) { + d = backoff[st.Trips-1] + } + st.OpenUntil = time.Now().Add(d) + st.Fails = 0 // counted into this trip; start fresh for the half-open probe + tripped = true + } + snap := *st + persist := r.persist + r.mu.Unlock() + if persist != nil { + persist(id, snap) + } + return tripped +} + +// Reset clears one profile's state — the UI's "立即恢复" action. +func (r *Registry) Reset(id int64) { + r.mu.Lock() + delete(r.m, id) + forget := r.forget + r.mu.Unlock() + if forget != nil { + forget(id) + } +} + +// Snapshot returns a copy of all tracked state, for the status API. +func (r *Registry) Snapshot() map[int64]State { + r.mu.Lock() + defer r.mu.Unlock() + out := make(map[int64]State, len(r.m)) + for id, st := range r.m { + out[id] = *st + } + return out +} diff --git a/llmpool/pool.go b/llmpool/pool.go new file mode 100644 index 0000000..30b3665 --- /dev/null +++ b/llmpool/pool.go @@ -0,0 +1,261 @@ +package llmpool + +import ( + "context" + "errors" + "fmt" + "iter" + "log" + "regexp" + "strconv" + "strings" + "sync/atomic" + + "github.com/Autumn-27/norma/llm" +) + +// Member is one LLM profile in the chain, already built into a provider (wrapped +// with the recorder by the caller, so a failed attempt is still recorded under +// its own profile name). +type Member struct { + ID int64 // llm_profiles.id + Name string // profile name, for logs / UI + Model string + Format string // "anthropic" | "openai" + Priority int // the profile's configured priority (display only) + Active bool // is_default (display only) + // Rank is the ordering key the caller assigned: higher goes first, and members + // sharing a Rank take turns leading (load-spreading across duplicate keys). + // The caller encodes "active profile heads the chain" as a Rank above every + // user-settable priority, so this type needs no policy of its own. + Rank int + // WindowTokens is the profile's context window in tokens. A member whose + // window can't hold the request is skipped rather than made to fail on it. + WindowTokens int + Prov llm.Provider +} + +// RankActive is the Rank the caller gives the chain head (the active profile, or +// an explicitly bound one) so it always outranks any configured priority. +const RankActive = int(^uint(0)>>1) - 1 + +// ErrExhausted is returned when every member of the chain failed. +var ErrExhausted = errors.New("LLM 轮询:所有配置均不可用") + +// Pool is an llm.Provider that fails over across an ordered chain of members. +// It is safe for concurrent use: members are immutable after construction and +// all mutable state lives in the shared Registry. +type Pool struct { + members []*Member // in chain order (active first, then priority DESC) + health *Registry + rr atomic.Uint64 // rotates the starting point within an equal-priority group +} + +// New builds a Pool over members (already in chain order). Returns nil when the +// chain is empty. A single-member chain is still a valid Pool — it just behaves +// exactly like the bare provider. +func New(members []*Member, health *Registry) *Pool { + if len(members) == 0 { + return nil + } + if health == nil { + health = NewRegistry(nil, nil) + } + return &Pool{members: members, health: health} +} + +// Members returns the chain in order (read-only). +func (p *Pool) Members() []*Member { return p.members } + +// Head returns the first member of the chain. +func (p *Pool) Head() *Member { return p.members[0] } + +// Stream implements llm.Provider with failover. +// +// The one hard rule: a member may only be abandoned BEFORE it has yielded any +// event. Once text or a tool_use has reached the caller, re-sending the same +// request to another model would duplicate output and corrupt the conversation +// history — so a mid-stream failure is surfaced as-is and left to the agent +// harness's resume logic. Fortunately the failures this exists for (402 no +// credit, 401 bad key, 429, 5xx) all surface during request establishment, +// before the body is read, so they always land in the safe window. +func (p *Pool) Stream(ctx context.Context, req llm.CompletionRequest) iter.Seq2[llm.StreamEvent, error] { + order := p.order(req) + return func(yield func(llm.StreamEvent, error) bool) { + var lastErr error + for i, m := range order { + emitted := false + var failed error + for ev, err := range m.Prov.Stream(ctx, req) { + if err != nil && !emitted && shouldFailover(ctx, err) { + failed = err + break // safe window: nothing reached the caller yet + } + emitted = true + if !yield(ev, err) { + return // caller stopped consuming (cancel / early exit) + } + if err != nil { + return // terminal error already handed to the caller + } + } + if failed == nil { + p.health.Pass(m.ID) // completed (or failed in a non-failover way) + return + } + lastErr = failed + hard := isHardFailure(failed) + if p.health.Trip(m.ID, trimErr(failed), hard) { + log.Printf("[llmpool] 配置 %q(%s) 已熔断:%s", m.Name, m.Model, trimErr(failed)) + } + if i+1 < len(order) { + n := order[i+1] + log.Printf("[llmpool] LLM 故障转移:%q(%s) → %q(%s),原因:%s", + m.Name, m.Model, n.Name, n.Model, trimErr(failed)) + } + } + if lastErr == nil { + lastErr = ErrExhausted + } + log.Printf("[llmpool] 轮询链已耗尽(%d 个配置全部失败),最后错误:%s", len(order), trimErr(lastErr)) + yield(llm.StreamEvent{}, fmt.Errorf("%w:%v", ErrExhausted, lastErr)) + } +} + +// order picks the members to try, in order: skip those in a cooling-off window +// and those whose context window can't hold this request, then rotate within each +// equal-priority group so same-priority profiles share the load. Never returns an +// empty slice — if everything is filtered out, the head of the chain is tried +// anyway, since stalling the engine is worse than one more failed request. +func (p *Pool) order(req llm.CompletionRequest) []*Member { + est := estimateTokens(req) + var open []*Member + for _, m := range p.members { + if p.health.IsOpen(m.ID) { + continue + } + if m.WindowTokens > 0 && est > m.WindowTokens { + continue // would 400 on length — not a useful failover target + } + open = append(open, m) + } + if len(open) == 0 { + return p.members[:1] // last resort: probe the head rather than stall + } + return rotateGroups(open, p.rr.Add(1)-1) +} + +// rotateGroups rotates each run of equal-ranked members by n, so profiles sharing +// a priority take turns going first (free load-spreading across duplicate keys). +// The chain head has its own Rank and always stays at the front. +func rotateGroups(in []*Member, n uint64) []*Member { + out := make([]*Member, 0, len(in)) + for i := 0; i < len(in); { + j := i + 1 + for j < len(in) && in[j].Rank == in[i].Rank { + j++ + } + g := in[i:j] + if len(g) > 1 { + off := int(n % uint64(len(g))) + for k := range g { + out = append(out, g[(k+off)%len(g)]) + } + } else { + out = append(out, g...) + } + i = j + } + return out +} + +// statusRe pulls the HTTP status out of the SDK's error text, which is formatted +// as ": status : " (norma/llm/retry.go). The SDK exposes no +// typed error, so the string is what we have. +var statusRe = regexp.MustCompile(`status (\d{3})`) + +// statusOf returns the HTTP status carried by err, or 0 if it isn't one. +func statusOf(err error) int { + m := statusRe.FindStringSubmatch(err.Error()) + if m == nil { + return 0 + } + code, _ := strconv.Atoi(m[1]) + return code +} + +// shouldFailover reports whether err justifies trying the next profile. +// +// Never fails over on context cancellation — that's the user stopping a task or a +// task-level timeout, and burning a backup key on it would both waste credit and +// pollute the run's termination diagnosis. Never on 400 either: a malformed or +// over-long request fails identically everywhere. +func shouldFailover(ctx context.Context, err error) bool { + if err == nil { + return false + } + if ctx.Err() != nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return false + } + switch code := statusOf(err); { + case code == 0: + return true // no status → transport-level failure (reset / DNS / timeout) + case code == 400: + return false // bad or over-long request: identical everywhere + case code == 401, code == 402, code == 403, code == 404, code == 408, code == 429: + return true + case code >= 500: + return true + } + return false +} + +// isHardFailure reports whether the failure is deterministic (the profile will +// keep failing until a human fixes it) rather than transient. Hard failures open +// the breaker on the first occurrence. +func isHardFailure(err error) bool { + switch statusOf(err) { + case 401, 402, 403, 404: + return true + } + return false +} + +// trimErr shortens an error for logs/UI — provider bodies can be long. +func trimErr(err error) string { + s := strings.TrimSpace(strings.ReplaceAll(err.Error(), "\n", " ")) + if len(s) > 300 { + s = s[:300] + "…" + } + return s +} + +// estimateTokens roughly sizes a request so members whose context window clearly +// can't hold it are skipped. Deliberately crude (~3.5 chars/token) and biased to +// over-estimate slightly; it only needs to tell "fits" from "nowhere near". +func estimateTokens(req llm.CompletionRequest) int { + n := 0 + for _, s := range req.System { + n += len(s) + } + for _, m := range req.Messages { + n += blocksLen(m.Content) + } + for _, t := range req.Tools { + // InputSchema is a decoded map, so its serialized size isn't available + // cheaply — charge a flat ~120 chars per top-level property instead. + n += len(t.Name) + len(t.Description) + len(t.InputSchema)*120 + } + return n * 2 / 7 // ≈ len/3.5 +} + +func blocksLen(bs []llm.ContentBlock) int { + n := 0 + for _, b := range bs { + n += len(b.Text) + len(b.Thinking) + len(b.Input) + len(b.Name) + if len(b.Content) > 0 { + n += blocksLen(b.Content) + } + } + return n +} diff --git a/llmpool/pool_test.go b/llmpool/pool_test.go new file mode 100644 index 0000000..b1e69c8 --- /dev/null +++ b/llmpool/pool_test.go @@ -0,0 +1,359 @@ +package llmpool + +import ( + "context" + "errors" + "fmt" + "iter" + "strings" + "testing" + "time" + + "github.com/Autumn-27/norma/llm" +) + +// fakeProv is a scripted provider: script[i] is what the i-th call yields — +// some events, then optionally an error. +type fakeProv struct { + name string + calls int + events [][]llm.StreamEvent // events emitted before the error, per call + errs []error // error to end each call with (nil = clean finish) +} + +// at returns script entry n, repeating the last one once the script runs out, so +// a provider defined as "always succeeds" / "always 402" keeps behaving that way +// across repeated calls. +func at[T any](s []T, n int) (T, bool) { + var zero T + if len(s) == 0 { + return zero, false + } + if n >= len(s) { + n = len(s) - 1 + } + return s[n], true +} + +func (f *fakeProv) Stream(ctx context.Context, req llm.CompletionRequest) iter.Seq2[llm.StreamEvent, error] { + n := f.calls + f.calls++ + return func(yield func(llm.StreamEvent, error) bool) { + if evs, ok := at(f.events, n); ok { + for _, e := range evs { + if !yield(e, nil) { + return + } + } + } + err, _ := at(f.errs, n) + if err != nil { + yield(llm.StreamEvent{}, err) + return + } + yield(llm.StreamEvent{Type: llm.SEMessageStop}, nil) + } +} + +// ok builds a provider that always succeeds with one text delta. +func okProv(name, text string) *fakeProv { + return &fakeProv{name: name, events: [][]llm.StreamEvent{{{Type: llm.SETextDelta, Text: text}}}} +} + +// failProv builds a provider that fails immediately (before any event) with status. +func failProv(name string, status int) *fakeProv { + return &fakeProv{name: name, errs: []error{fmt.Errorf("anthropic: status %d: nope", status)}} +} + +func member(id int64, name string, rank int, p llm.Provider) *Member { + return &Member{ID: id, Name: name, Model: "m" + name, Rank: rank, Prov: p} +} + +// drain consumes a stream, returning the concatenated text and terminal error. +func drain(seq iter.Seq2[llm.StreamEvent, error]) (string, error) { + var sb strings.Builder + for ev, err := range seq { + if err != nil { + return sb.String(), err + } + if ev.Type == llm.SETextDelta { + sb.WriteString(ev.Text) + } + } + return sb.String(), nil +} + +func TestFailoverOnNoCredit(t *testing.T) { + a, b := failProv("a", 402), okProv("b", "hello") + p := New([]*Member{member(1, "a", 10, a), member(2, "b", 5, b)}, NewRegistry(nil, nil)) + + got, err := drain(p.Stream(context.Background(), llm.CompletionRequest{})) + if err != nil { + t.Fatalf("expected failover to succeed, got %v", err) + } + if got != "hello" { + t.Fatalf("text = %q, want %q", got, "hello") + } + if a.calls != 1 || b.calls != 1 { + t.Fatalf("calls: a=%d b=%d, want 1/1", a.calls, b.calls) + } +} + +// A 402 is deterministic: one failure must open the breaker, so the NEXT request +// skips that profile entirely instead of paying for another round-trip. +func TestHardFailureTripsBreakerImmediately(t *testing.T) { + a, b := failProv("a", 402), okProv("b", "x") + reg := NewRegistry(nil, nil) + p := New([]*Member{member(1, "a", 10, a), member(2, "b", 5, b)}, reg) + + _, _ = drain(p.Stream(context.Background(), llm.CompletionRequest{})) + if !reg.IsOpen(1) { + t.Fatal("402 should have tripped the breaker on the first failure") + } + _, _ = drain(p.Stream(context.Background(), llm.CompletionRequest{})) + if a.calls != 1 { + t.Fatalf("tripped profile was called again: calls=%d, want 1", a.calls) + } + if b.calls != 2 { + t.Fatalf("fallback calls=%d, want 2", b.calls) + } +} + +// 429 is transient: the SDK already retried, but we shouldn't write a profile off +// until it fails repeatedly. +func TestSoftFailureNeedsRepeats(t *testing.T) { + reg := NewRegistry(nil, nil) + for i := 1; i < softTripAfter; i++ { + if reg.Trip(1, "429", false) { + t.Fatalf("tripped after %d transient failures, want %d", i, softTripAfter) + } + } + if !reg.Trip(1, "429", false) { + t.Fatalf("should trip on failure #%d", softTripAfter) + } + if !reg.IsOpen(1) { + t.Fatal("breaker should be open") + } +} + +// A success must fully clear the counters, so an intermittent profile never +// accumulates its way to a trip. +func TestPassResetsCounters(t *testing.T) { + reg := NewRegistry(nil, nil) + reg.Trip(1, "429", false) + reg.Trip(1, "429", false) + reg.Pass(1) + if got := reg.Get(1).Fails; got != 0 { + t.Fatalf("fails=%d after Pass, want 0", got) + } + if reg.Trip(1, "429", false) { + t.Fatal("tripped immediately after a success — counters were not reset") + } +} + +func TestBackoffLadderGrows(t *testing.T) { + reg := NewRegistry(nil, nil) + var prev time.Duration + for i := range 4 { + reg.Reset(1) + st := State{Trips: i} + reg.Restore(1, st) + reg.Trip(1, "402", true) + d := time.Until(reg.Get(1).OpenUntil) + // Non-decreasing, with a second of slack: the ladder plateaus at its last + // rung, and each Trip stamps its own time.Now(). + if i > 0 && d < prev-time.Second { + t.Fatalf("trip #%d cools for %v, shorter than the previous %v", i+1, d, prev) + } + prev = d + } + if prev < 25*time.Minute { + t.Fatalf("ladder tops out at %v, want ~30m", prev) + } +} + +// The safety rule: once output has reached the caller, a mid-stream failure must +// NOT be retried on another model — that would duplicate the assistant turn. +func TestNoFailoverAfterEmit(t *testing.T) { + a := &fakeProv{ + name: "a", + events: [][]llm.StreamEvent{{{Type: llm.SETextDelta, Text: "partial"}}}, + errs: []error{fmt.Errorf("anthropic: status 500: mid-stream drop")}, + } + b := okProv("b", "full") + p := New([]*Member{member(1, "a", 10, a), member(2, "b", 5, b)}, NewRegistry(nil, nil)) + + got, err := drain(p.Stream(context.Background(), llm.CompletionRequest{})) + if err == nil { + t.Fatal("mid-stream error should surface, not be swallowed by a failover") + } + if got != "partial" { + t.Fatalf("text = %q, want the partial output %q", got, "partial") + } + if b.calls != 0 { + t.Fatalf("fell over to the backup after emitting output (calls=%d) — would duplicate the turn", b.calls) + } +} + +// Cancelling a task must not burn a backup key, and must not be diagnosed as an +// LLM fault. +func TestNoFailoverOnCancel(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + a := &fakeProv{name: "a", errs: []error{context.Canceled}} + b := okProv("b", "x") + reg := NewRegistry(nil, nil) + p := New([]*Member{member(1, "a", 10, a), member(2, "b", 5, b)}, reg) + + _, err := drain(p.Stream(ctx, llm.CompletionRequest{})) + if !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v, want context.Canceled", err) + } + if b.calls != 0 { + t.Fatalf("failed over on cancellation (backup calls=%d)", b.calls) + } + if reg.Get(1).Fails != 0 { + t.Fatal("cancellation counted as a profile failure") + } +} + +func TestShouldFailoverByStatus(t *testing.T) { + cases := []struct { + status int + want bool + }{ + {400, false}, // bad/over-long request fails identically everywhere + {401, true}, {402, true}, {403, true}, {404, true}, + {408, true}, {429, true}, {500, true}, {503, true}, + {200, false}, + } + for _, c := range cases { + err := fmt.Errorf("openai: status %d: body", c.status) + if got := shouldFailover(context.Background(), err); got != c.want { + t.Errorf("status %d: shouldFailover=%v, want %v", c.status, got, c.want) + } + } + // A transport error carries no status and must fail over. + if !shouldFailover(context.Background(), errors.New("dial tcp: connection reset by peer")) { + t.Error("network error should fail over") + } +} + +func TestHardVsSoftClassification(t *testing.T) { + for _, s := range []int{401, 402, 403, 404} { + if !isHardFailure(fmt.Errorf("status %d: x", s)) { + t.Errorf("status %d should be a hard failure", s) + } + } + for _, s := range []int{408, 429, 500, 502, 503} { + if isHardFailure(fmt.Errorf("status %d: x", s)) { + t.Errorf("status %d should be transient, not hard", s) + } + } +} + +// A member whose context window can't hold the request is a guaranteed 400 — +// skip it rather than spend a round-trip proving it. +func TestSkipsMembersTooSmallForRequest(t *testing.T) { + small, big := okProv("small", "s"), okProv("big", "b") + ms := member(1, "small", 10, small) + ms.WindowTokens = 1000 + mb := member(2, "big", 5, big) + mb.WindowTokens = 1_000_000 + p := New([]*Member{ms, mb}, NewRegistry(nil, nil)) + + req := llm.CompletionRequest{Messages: []llm.Message{llm.UserText(strings.Repeat("x", 100_000))}} + got, err := drain(p.Stream(context.Background(), req)) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got != "b" { + t.Fatalf("text=%q, want the large-window member to serve it", got) + } + if small.calls != 0 { + t.Fatalf("sent a request to a member that can't hold it (calls=%d)", small.calls) + } +} + +// Everything tripped: probing the head beats stalling the engine outright. +func TestAllTrippedStillProbesHead(t *testing.T) { + a, b := okProv("a", "a"), okProv("b", "b") + reg := NewRegistry(nil, nil) + reg.Trip(1, "402", true) + reg.Trip(2, "402", true) + p := New([]*Member{member(1, "a", 10, a), member(2, "b", 5, b)}, reg) + + got, err := drain(p.Stream(context.Background(), llm.CompletionRequest{})) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got != "a" || a.calls != 1 { + t.Fatalf("text=%q a.calls=%d — want the head probed as a last resort", got, a.calls) + } +} + +func TestExhaustedChainReportsClearly(t *testing.T) { + a, b := failProv("a", 402), failProv("b", 402) + p := New([]*Member{member(1, "a", 10, a), member(2, "b", 5, b)}, NewRegistry(nil, nil)) + + _, err := drain(p.Stream(context.Background(), llm.CompletionRequest{})) + if !errors.Is(err, ErrExhausted) { + t.Fatalf("err = %v, want it to wrap ErrExhausted", err) + } +} + +// Equal-rank members take turns leading, so duplicate keys share the load. +func TestEqualRankRotates(t *testing.T) { + a, b := okProv("a", "a"), okProv("b", "b") + head := member(1, "head", RankActive, failProv("head", 402)) + p := New([]*Member{head, member(2, "a", 5, a), member(3, "b", 5, b)}, NewRegistry(nil, nil)) + + first := map[string]int{} + for range 4 { + txt, _ := drain(p.Stream(context.Background(), llm.CompletionRequest{})) + first[txt]++ + } + if first["a"] == 0 || first["b"] == 0 { + t.Fatalf("same-rank members did not rotate: %v", first) + } +} + +// The head keeps its own rank and never joins a rotation group. +func TestHeadAlwaysFirst(t *testing.T) { + head := okProv("head", "H") + p := New([]*Member{ + member(1, "head", RankActive, head), + member(2, "a", 5, okProv("a", "a")), + member(3, "b", 5, okProv("b", "b")), + }, NewRegistry(nil, nil)) + for range 5 { + if txt, _ := drain(p.Stream(context.Background(), llm.CompletionRequest{})); txt != "H" { + t.Fatalf("head was not tried first, got %q", txt) + } + } +} + +// A single-member chain must behave exactly like the bare provider. +func TestSingleMemberPassthrough(t *testing.T) { + a := okProv("a", "solo") + p := New([]*Member{member(1, "a", RankActive, a)}, NewRegistry(nil, nil)) + got, err := drain(p.Stream(context.Background(), llm.CompletionRequest{})) + if err != nil || got != "solo" { + t.Fatalf("got %q / %v, want solo / nil", got, err) + } +} + +func TestNewEmptyChainIsNil(t *testing.T) { + if New(nil, nil) != nil { + t.Fatal("empty chain should yield a nil pool so callers use the bare provider") + } +} + +// An expired cooling-off window must not keep a profile out of the chain. +func TestExpiredWindowIsClosed(t *testing.T) { + reg := NewRegistry(nil, nil) + reg.Restore(1, State{Trips: 1, OpenUntil: time.Now().Add(-time.Second)}) + if reg.IsOpen(1) { + t.Fatal("expired window should read as closed") + } +} diff --git a/server/dto.go b/server/dto.go index 56b056a..1f7f0cd 100644 --- a/server/dto.go +++ b/server/dto.go @@ -436,6 +436,10 @@ type LLMProfileDTO struct { ContextWindowK int `json:"context_window_k"` ReasoningEffort string `json:"reasoning_effort"` IsDefault bool `json:"is_default"` + // 轮询(故障转移)参数:priority 越大越先被选中(激活配置恒为链首); + // pool_exclude=true 则不作为故障转移目标,但仍可被 agent/任务显式绑定。 + Priority int `json:"priority"` + PoolExclude bool `json:"pool_exclude"` } func llmProfileDTO(p *db.LLMProfile) LLMProfileDTO { @@ -452,6 +456,8 @@ func llmProfileDTO(p *db.LLMProfile) LLMProfileDTO { ContextWindowK: p.ContextWindowK, ReasoningEffort: p.ReasoningEffort, IsDefault: p.IsDefault, + Priority: p.Priority, + PoolExclude: p.PoolExclude, } } diff --git a/server/goals.go b/server/goals.go index 2a72c60..41c4e99 100644 --- a/server/goals.go +++ b/server/goals.go @@ -10,6 +10,7 @@ import ( "github.com/Autumn-27/artex/agent" "github.com/Autumn-27/artex/db" + "github.com/Autumn-27/norma/llm" ) type goalSpec struct { @@ -188,7 +189,7 @@ func (s *Server) createGoals(ctx context.Context, t *Task, emit func(db.Activity // so FromEnv returned empty and every task silently fell back to the crude rule // splitter (which shredded URLs / made meaningless 2-way splits). var specs []goalSpec - if cfg, ok := s.goalsLLMConfig(t); ok { + if prov, ok := s.goalsProvider(t); ok { var as *db.AssetStore if s.m != nil { as = s.m.Assets() @@ -197,7 +198,7 @@ func (s *Server) createGoals(ctx context.Context, t *Task, emit func(db.Activity // DecomposeGoals persists each decomposed goal via the set_goals tool straight // into t.Store; it returns the specs it wrote so we can emit activity + spot the // empty (LLM error / no provider) case below. No write-back needed here anymore. - for _, g := range agent.DecomposeGoals(ctx, cfg, s.m.dir, t.Goal, t.Description, as, t.Store, taskID, emit) { + for _, g := range agent.DecomposeGoals(ctx, prov, s.m.dir, t.Goal, t.Description, as, t.Store, taskID, emit) { if strings.TrimSpace(g.Text) != "" { specs = append(specs, goalSpec{Text: g.Text, VulnClass: g.VulnClass}) } @@ -220,18 +221,22 @@ func (s *Server) createGoals(ctx context.Context, t *Task, emit func(db.Activity return specs } -// goalsLLMConfig resolves the LLM config for goal decomposition by the standard -// precedence: the goals agent's own binding → the task's pinned profile → the global -// active profile (same source the engine runs on). -func (s *Server) goalsLLMConfig(t *Task) (agent.Config, bool) { +// goalsProvider resolves the provider for goal decomposition by the standard +// precedence: the goals agent's own binding → the task's pinned profile → the +// global active provider (the very instance the engine runs on, so decomposition +// shares its rate limiter, gets recorded, and follows LLM failover). +func (s *Server) goalsProvider(t *Task) (llm.Provider, bool) { var pin *int64 if t != nil { pin = t.LLMProfileID } if eff := s.effectiveProfileForAgent("goals", pin); eff != nil { - if cfg, ok := s.loadProfileConfig(*eff); ok { - return cfg, true + if prov, cfg, ok := s.providerForProfile(*eff); ok { + return s.poolForBinding(*eff, prov, cfg), true } } - return s.loadLLMConfig() + s.cfgMu.Lock() + prov, on := s.llmProv, s.llmOn + s.cfgMu.Unlock() + return prov, on && prov != nil } diff --git a/server/llmpool.go b/server/llmpool.go new file mode 100644 index 0000000..527e09c --- /dev/null +++ b/server/llmpool.go @@ -0,0 +1,239 @@ +package server + +import ( + "log" + "time" + + "github.com/Autumn-27/artex/agent" + "github.com/Autumn-27/artex/db" + "github.com/Autumn-27/artex/llmpool" + "github.com/Autumn-27/norma/llm" +) + +// LLM 轮询(故障转移)的服务端接线。设计见 docs/LLM轮询设计.md: +// - 全局激活配置这条路径(agent 未绑定、任务未 pin)才轮询; +// - 绑定/pin 的路径默认独占该配置,失败即失败(可由 llm_pool_bind_fallback 打开兜底); +// - 链序 = 激活配置 → 其余按 priority DESC,排除 pool_exclude 的; +// - 熔断状态进程级共享(s.llmHealth),重建 pool 不清空。 + +// newLLMHealthRegistry builds the process-wide circuit-breaker registry, mirroring +// state into PG so a cooling-off window survives a restart. Writes are async and +// best-effort — the in-memory copy is authoritative. +func newLLMHealthRegistry(pg *db.DB) *llmpool.Registry { + if pg == nil { + return llmpool.NewRegistry(nil, nil) + } + persist := func(id int64, st llmpool.State) { + h := db.LLMHealth{ProfileID: id, Fails: st.Fails, Trips: st.Trips, LastError: st.LastError} + if !st.OpenUntil.IsZero() { + t := st.OpenUntil + h.OpenUntil = &t + } + go func() { + if err := pg.SaveLLMHealth(h); err != nil { + log.Printf("[llmpool] 熔断状态落库失败: %v", err) + } + }() + } + forget := func(id int64) { + go func() { _ = pg.ClearLLMHealth(id) }() + } + reg := llmpool.NewRegistry(persist, forget) + // Restore only windows that haven't expired (LoadLLMHealth filters), so a + // profile that finished cooling while we were down comes back healthy. + if rows, err := pg.LoadLLMHealth(); err == nil { + for _, h := range rows { + st := llmpool.State{Fails: h.Fails, Trips: h.Trips, LastError: h.LastError, LastAt: h.LastAt} + if h.OpenUntil != nil { + st.OpenUntil = *h.OpenUntil + } + reg.Restore(h.ProfileID, st) + log.Printf("[llmpool] 恢复熔断状态: 配置 #%d 冷却至 %s", h.ProfileID, st.OpenUntil.Format(time.RFC3339)) + } + } + return reg +} + +// poolMember builds one chain member from a profile, reusing the per-profile +// provider cache so every agent pointing at the same profile shares one provider +// (hence one rate limiter). Returns nil when the profile can't be built. +func (s *Server) poolMember(p *db.LLMProfile, rank int) *llmpool.Member { + prov, cfg, ok := s.providerForProfile(p.ID) + if !ok { + return nil + } + return &llmpool.Member{ + ID: p.ID, Name: p.Name, Model: p.Model, Format: p.Format, + Priority: p.Priority, Active: p.IsDefault, Rank: rank, + WindowTokens: cfg.CompactionWindow(), Prov: prov, + } +} + +// poolChain reads the failover chain from the DB. headID/headProv/headCfg describe +// the member that must lead the chain — the globally active profile on the normal +// path, or an explicitly bound one when bound-profile fallback is on. The head is +// passed in (rather than resolved through the cache) so applyLLM's already-built +// provider is reused instead of duplicated. +// +// Returns nil when failover is off or the chain has fewer than two usable members +// — callers then use the bare provider, which is byte-for-byte the old behavior. +func (s *Server) poolChain(headID int64, headProv llm.Provider, headCfg agent.Config) *llmpool.Pool { + if s.m == nil || s.m.pg == nil || !s.m.LLMPoolEnabled() { + return nil + } + profs, err := s.m.pg.PoolProfiles() + if err != nil { + log.Printf("[llmpool] 读取轮询链失败: %v", err) + return nil + } + var head *db.LLMProfile + for _, p := range profs { + if p.ID == headID { + head = p + break + } + } + if head == nil { // head excluded from the chain (bound + pool_exclude) — still leads + if p, err := s.m.pg.ProfileByID(headID); err == nil && p != nil { + head = p + } else { + return nil + } + } + members := []*llmpool.Member{{ + ID: head.ID, Name: head.Name, Model: head.Model, Format: head.Format, + Priority: head.Priority, Active: head.IsDefault, Rank: llmpool.RankActive, + WindowTokens: headCfg.CompactionWindow(), Prov: headProv, + }} + for _, p := range profs { + if p.ID == headID { + continue + } + // The globally active profile outranks the others when it isn't the head + // (bound-fallback chains), so it stays the first fallback tried. + rank := p.Priority + if p.IsDefault { + rank = llmpool.RankActive - 1 + } + if m := s.poolMember(p, rank); m != nil { + members = append(members, m) + } + } + if len(members) < 2 { + return nil // nothing to fail over to + } + return llmpool.New(members, s.llmHealth) +} + +// poolForActive wraps the globally-active provider in the failover chain. Returns +// prov unchanged when failover is off or there's nothing to fall back to. +func (s *Server) poolForActive(activeID int64, prov llm.Provider, cfg agent.Config) llm.Provider { + pool := s.poolChain(activeID, prov, cfg) + if pool == nil { + return prov + } + names := make([]string, 0, len(pool.Members())) + for _, m := range pool.Members() { + names = append(names, m.Name+"/"+m.Model) + } + log.Printf("[llmpool] LLM 轮询已启用,链路(%d): %v", len(names), names) + return pool +} + +// poolForBinding wraps a BOUND profile's provider so it falls back to the chain. +// Only active when both the failover master switch and the bound-fallback switch +// are on; otherwise a bound profile stays exclusive (fails hard), which is the +// default and the documented precedence. +func (s *Server) poolForBinding(id int64, prov llm.Provider, cfg agent.Config) llm.Provider { + if s.m == nil || !s.m.LLMPoolEnabled() || !s.m.LLMPoolBindFallback() { + return prov + } + if pool := s.poolChain(id, prov, cfg); pool != nil { + return pool + } + return prov +} + +// LLMPoolMemberStatus is one chain entry as shown in the UI. +type LLMPoolMemberStatus struct { + ProfileID string `json:"profile_id"` + Name string `json:"name"` + Model string `json:"model"` + Format string `json:"format"` + Priority int `json:"priority"` + Active bool `json:"active"` // is the globally-activated profile + Excluded bool `json:"excluded"` // pool_exclude — not a failover target + // Health: state is "ok" | "degraded" (failing but not tripped) | "tripped". + State string `json:"state"` + Fails int `json:"fails"` + Trips int `json:"trips"` + CooldownSecs int `json:"cooldown_secs"` // remaining cooling-off seconds; 0 = none + LastError string `json:"last_error,omitempty"` + LastAt string `json:"last_at,omitempty"` +} + +// llmPoolStatus reports the whole picture for the LLM page: whether failover is +// on, the resolved chain order, and each profile's breaker state. Excluded +// profiles are listed too (flagged), so the user can see why one isn't in line. +func (s *Server) llmPoolStatus() map[string]any { + out := map[string]any{ + "enabled": false, + "bind_fallback": false, + "chain": []LLMPoolMemberStatus{}, + } + if s.m == nil || s.m.pg == nil { + return out + } + out["enabled"] = s.m.LLMPoolEnabled() + out["bind_fallback"] = s.m.LLMPoolBindFallback() + + all, err := s.m.pg.ListProfiles() + if err != nil { + return out + } + health := s.llmHealth.Snapshot() + now := time.Now() + // Chain order: active first, then priority DESC / id ASC — the same ordering + // PoolProfiles applies, recomputed here so excluded ones can be shown in place. + chain := make([]LLMPoolMemberStatus, 0, len(all)) + for _, p := range all { + st := health[p.ID] + m := LLMPoolMemberStatus{ + ProfileID: i64s(p.ID), Name: p.Name, Model: p.Model, Format: p.Format, + Priority: p.Priority, Active: p.IsDefault, Excluded: p.PoolExclude, + State: "ok", Fails: st.Fails, Trips: st.Trips, + LastError: st.LastError, + } + if st.Open() { + m.State = "tripped" + m.CooldownSecs = int(st.OpenUntil.Sub(now).Seconds()) + 1 + } else if st.Fails > 0 { + m.State = "degraded" + } + if !st.LastAt.IsZero() { + m.LastAt = st.LastAt.Format(time.RFC3339) + } + chain = append(chain, m) + } + sortPoolStatus(chain) + out["chain"] = chain + return out +} + +// sortPoolStatus orders the status rows exactly like the live chain: active first, +// then priority DESC, then id ASC (the incoming slice is already id-ordered, so a +// stable insertion sort on the first two keys is enough). +func sortPoolStatus(in []LLMPoolMemberStatus) { + for i := 1; i < len(in); i++ { + for j := i; j > 0 && poolLess(in[j], in[j-1]); j-- { + in[j], in[j-1] = in[j-1], in[j] + } + } +} + +func poolLess(a, b LLMPoolMemberStatus) bool { + if a.Active != b.Active { + return a.Active + } + return a.Priority > b.Priority +} diff --git a/server/llmpool_test.go b/server/llmpool_test.go new file mode 100644 index 0000000..0927b86 --- /dev/null +++ b/server/llmpool_test.go @@ -0,0 +1,41 @@ +package server + +import "testing" + +// The status list drives the "轮询顺序" strip in the UI, so it has to match the +// order PoolProfiles actually runs: active first, then priority DESC, then id ASC +// (input arrives id-ordered, so equal priorities must keep their relative order). +func TestSortPoolStatusMatchesChainOrder(t *testing.T) { + in := []LLMPoolMemberStatus{ + {ProfileID: "1", Name: "low", Priority: 0}, + {ProfileID: "2", Name: "active", Priority: 0, Active: true}, + {ProfileID: "3", Name: "high", Priority: 10}, + {ProfileID: "4", Name: "mid-a", Priority: 5}, + {ProfileID: "5", Name: "mid-b", Priority: 5}, + } + sortPoolStatus(in) + + want := []string{"active", "high", "mid-a", "mid-b", "low"} + for i, w := range want { + if in[i].Name != w { + got := make([]string, len(in)) + for j, m := range in { + got[j] = m.Name + } + t.Fatalf("order = %v, want %v", got, want) + } + } +} + +// The active profile heads the chain no matter how low its own priority is — +// that's the documented precedence, and the UI must not imply otherwise. +func TestActiveProfileHeadsStatusList(t *testing.T) { + in := []LLMPoolMemberStatus{ + {ProfileID: "1", Name: "loud", Priority: 999}, + {ProfileID: "2", Name: "active", Priority: -5, Active: true}, + } + sortPoolStatus(in) + if in[0].Name != "active" { + t.Fatalf("head = %q, want the active profile regardless of priority", in[0].Name) + } +} diff --git a/server/manager.go b/server/manager.go index 454b229..78e403d 100644 --- a/server/manager.go +++ b/server/manager.go @@ -86,6 +86,12 @@ const ( settingWebSearchProxy = "web_search_proxy" settingWorkers = "workers" settingLLMRecord = "llm_record" + // LLM 轮询(故障转移)。默认关闭——开启后走「全局激活配置」的 agent 在当前配置 + // 不可用(余额不足/key 失效/限流/服务异常)时自动切到下一个配置。 + // settingLLMPoolBindFallback 仅在轮询开启时有意义:默认关闭,即 agent/任务显式 + // 绑定了某个配置就只用它、失败即失败;开启后绑定的配置失败也会回落到轮询链。 + settingLLMPoolOn = "llm_pool_enabled" + settingLLMPoolBindFallback = "llm_pool_bind_fallback" // 任务并发上限:开关 + 上限数。默认关闭;开启后默认上限 5(见 defaultConcurrencyLimit)。 settingConcurrencyOn = "task_concurrency_enabled" settingConcurrencyLimit = "task_concurrency_limit" @@ -266,6 +272,36 @@ func (m *Manager) SetLLMRecordEnabled(on bool) error { return nil } +// LLMPoolEnabled reports whether LLM failover ("轮询") is on (默认关; +// settings.llm_pool_enabled). Read when the provider chain is built (applyLLM), +// so a change requires a rebuild — putSettings does that. +func (m *Manager) LLMPoolEnabled() bool { + if m.pg == nil { + return false + } + return m.pg.GetBool(settingLLMPoolOn, false) +} + +// SetLLMPoolEnabled persists the failover toggle. Callers rebuild agents +// (applyLLM) afterwards so it takes effect. +func (m *Manager) SetLLMPoolEnabled(on bool) error { return m.pg.SetBool(settingLLMPoolOn, on) } + +// LLMPoolBindFallback reports whether an agent/task that is BOUND to a specific +// profile still falls back to the chain when that profile fails (默认关:绑定即 +// 独占,失败即失败). Only meaningful while LLMPoolEnabled. +func (m *Manager) LLMPoolBindFallback() bool { + if m.pg == nil { + return false + } + return m.pg.GetBool(settingLLMPoolBindFallback, false) +} + +// SetLLMPoolBindFallback persists the bound-profile fallback toggle. Callers +// rebuild agents (applyLLM) afterwards. +func (m *Manager) SetLLMPoolBindFallback(on bool) error { + return m.pg.SetBool(settingLLMPoolBindFallback, on) +} + // WebSearch returns the current web-search config: whether it is enabled, the // backend ("ddgs" | "brave-free" | "tavily"), the Brave API key, the Tavily API // key (each empty unless set), and the dedicated egress proxy (empty = direct). diff --git a/server/server.go b/server/server.go index 4c68585..2fe0d1d 100644 --- a/server/server.go +++ b/server/server.go @@ -19,6 +19,7 @@ import ( "github.com/Autumn-27/artex/agent" "github.com/Autumn-27/artex/db" + "github.com/Autumn-27/artex/llmpool" "github.com/Autumn-27/artex/llmrec" "github.com/Autumn-27/artex/report" "github.com/Autumn-27/norma/llm" @@ -53,6 +54,11 @@ type Server struct { llmCfg agent.Config // current LLM config (key not exposed) llmOn bool llmProf string // active LLM profile name (for llmrec tagging) + // llmProv is the fully-decorated global provider (recorder + failover chain) + // applyLLM installed. Kept so one-off callers (goal decomposition) reuse the + // same instance instead of building a private one that skips the rate limiter, + // the recorder and failover. + llmProv llm.Provider // chatBusy guards the per-task main-agent run: the chat handler launches the // agent on the server's background ctx (not the request ctx) and returns @@ -97,6 +103,12 @@ type Server struct { // leaving it unset), just not deduped. Cleared alongside profAgents on profile edits. provCacheMu sync.Mutex provByProfile map[int64]*provEntry + + // llmHealth is the process-wide circuit-breaker state for LLM failover (轮询). + // It deliberately lives OUTSIDE the provider caches: rebuilding the chain + // (saving an unrelated profile, flipping a setting) must not erase what we + // learned about which backends are out of credit / rate-limited. + llmHealth *llmpool.Registry } // profBundle is a planner/worker pair built from one LLM profile. @@ -131,7 +143,7 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi chatCancel: map[string]context.CancelCauseFunc{}, triggerQ: map[string][]triggeredRun{}, triggerActive: map[string]int{}, triggerCfg: map[string]triggerBehavior{}, profAgents: map[int64]*profBundle{}, profChatAgents: map[int64]*agent.ChatAgent{}, - provByProfile: map[int64]*provEntry{}} + provByProfile: map[int64]*provEntry{}, llmHealth: newLLMHealthRegistry(m.pg)} // per-task LLM: a task pinned to a specific profile runs on that profile's // dedicated planner/worker; unpinned tasks fall back to the global active pair. s.engine.SetAgentResolver(func(t *Task) (*agent.Planner, *agent.Worker) { @@ -291,10 +303,15 @@ func (s *Server) saveLLMConfig(cfg agent.Config) { format = "openai" } var id int64 + // agent.Config carries no failover fields, so carry the stored ones forward — + // otherwise this legacy endpoint would silently reset the profile's priority + // and pool_exclude to zero values on every save. + var priority int + var poolExclude bool if profs, _ := s.m.pg.ListProfiles(); profs != nil { for _, p := range profs { if p.Name == "default" { - id = p.ID + id, priority, poolExclude = p.ID, p.Priority, p.PoolExclude break } } @@ -303,6 +320,7 @@ func (s *Server) saveLLMConfig(cfg agent.Config) { ID: id, Name: "default", Format: format, Model: cfg.Model, BaseURL: cfg.BaseURL, Proxy: cfg.Proxy, APIKey: cfg.APIKey, RatePerSecond: cfg.RatePerSecond, RatePerMinute: cfg.RatePerMinute, ContextWindowK: cfg.ContextWindowK, ReasoningEffort: cfg.ReasoningEffort, IsDefault: true, + Priority: priority, PoolExclude: poolExclude, }) if err == nil { _ = s.m.pg.SetActiveProfile(newID) @@ -371,6 +389,12 @@ func (s *Server) applyLLM(cfg agent.Config) error { profName := s.llmProf s.cfgMu.Unlock() prov = llmrec.Wrap(prov, s.m.PG(), cfg.Model, profName, s.m.LLMRecordEnabled) + // LLM 轮询(默认关):把激活配置包进故障转移链,当前配置不可用时自动切下一个。 + // 只影响「走全局激活配置」的这条路径——agent 绑定 / 任务 pin 的走 providerForProfile, + // 默认仍然独占该配置(见 poolForBinding)。关闭或无备选时返回原 provider,行为不变。 + if act, err := s.m.pg.ActiveProfile(); err == nil && act != nil { + prov = s.poolForActive(act.ID, prov, cfg) + } pl, wk := s.buildPlannerWorker(nil, prov, cfg) s.engine.UseLLM(pl, wk) @@ -391,6 +415,7 @@ func (s *Server) applyLLM(cfg agent.Config) error { s.chatAgent.SetWebSearch(s.m.WebSearchOpts()) s.chatAgent.SetGuard(s.chatGuard()) s.llmCfg = cfg + s.llmProv = prov s.llmOn = true s.cfgMu.Unlock() @@ -468,10 +493,14 @@ func (s *Server) providerForProfile(id int64) (llm.Provider, agent.Config, bool) // providerForAgent returns the provider+cfg one agent should run on: its bound/pinned // profile if that resolves, else the passed global fallback (gProv/gCfg). +// A bound/pinned profile is EXCLUSIVE by default: it does not fail over, so an +// agent deliberately put on a specific model never silently runs on another one. +// Turning on llm_pool_bind_fallback relaxes that — poolForBinding then appends the +// failover chain behind it. func (s *Server) providerForAgent(agentKey string, pinID *int64, gProv llm.Provider, gCfg agent.Config) (llm.Provider, agent.Config) { if eff := s.effectiveProfileForAgent(agentKey, pinID); eff != nil { if prov, cfg, ok := s.providerForProfile(*eff); ok { - return prov, cfg + return s.poolForBinding(*eff, prov, cfg), cfg } } return gProv, gCfg @@ -492,7 +521,9 @@ func (s *Server) agentsForProfile(id int64) (*agent.Planner, *agent.Worker) { if !ok { return nil, nil } - pl, wk := s.buildPlannerWorker(&id, prov, cfg) + // The pin is the fallback for agents without their own binding — same + // exclusive-by-default rule, so route it through poolForBinding too. + pl, wk := s.buildPlannerWorker(&id, s.poolForBinding(id, prov, cfg), cfg) s.profMu.Lock() if ex := s.profAgents[id]; ex != nil { // lost the race → keep the winner pl, wk = ex.pl, ex.wk @@ -517,7 +548,7 @@ func (s *Server) chatAgentForProfile(id int64) *agent.ChatAgent { return nil } tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts")) - ca := agent.NewChatAgent(prov, cfg.Model, s.m.dir, tx, cfg.CompactionWindow()) + ca := agent.NewChatAgent(s.poolForBinding(id, prov, cfg), cfg.Model, s.m.dir, tx, cfg.CompactionWindow()) ca.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) ca.SetWebSearch(s.m.WebSearchOpts()) ca.SetGuard(s.chatGuard()) @@ -726,6 +757,8 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("POST /api/llm/profiles", s.pgSaveProfile) mux.HandleFunc("DELETE /api/llm/profiles/{id}", s.pgDeleteProfile) mux.HandleFunc("POST /api/llm/profiles/active", s.pgActivateProfile) + mux.HandleFunc("GET /api/llm/pool", s.pgLLMPoolStatus) + mux.HandleFunc("POST /api/llm/pool/reset", s.pgLLMPoolReset) mux.HandleFunc("POST /api/llm/models", s.pgListModels) // 拦截规则管理 @@ -2191,6 +2224,10 @@ func (s *Server) settingsPayload() map[string]any { "workers": s.m.Workers(), // 并发工作 agent 数(默认3);对之后启动的任务生效 "task_concurrency_enabled": concOn, // 任务并发上限开关(默认关) "task_concurrency_limit": concLimit, // 同时运行任务上限(开启后默认5) + // LLM 轮询(故障转移)。默认关;开启后走全局激活配置的 agent 在当前配置不可用时 + // 自动切到下一个配置。bind_fallback 仅在轮询开启时有意义(默认关)。 + "llm_pool_enabled": s.m.LLMPoolEnabled(), + "llm_pool_bind_fallback": s.m.LLMPoolBindFallback(), } } @@ -2227,6 +2264,10 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) { // 任务并发上限:同时「运行中」的任务数上限。关闭=不限;开启后新建任务超限则排队,有空位自动启动。 ConcurrencyEnabled *bool `json:"task_concurrency_enabled"` ConcurrencyLimit *int `json:"task_concurrency_limit"` + // LLM 轮询(故障转移)开关 + 「绑定配置失败也兜底回轮询链」开关。两者都需要 + // 重建 provider 链才生效,走下面的 changed → applyLLM 路径。 + LLMPoolEnabled *bool `json:"llm_pool_enabled"` + LLMPoolBindFallback *bool `json:"llm_pool_bind_fallback"` } if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeErr(w, 400, err.Error()) @@ -2272,6 +2313,25 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) { } } changed := false + if req.LLMPoolEnabled != nil { + if err := s.m.SetLLMPoolEnabled(*req.LLMPoolEnabled); err != nil { + writeErr(w, 500, err.Error()) + return + } + changed = true // the provider chain itself changes shape → rebuild + } + if req.LLMPoolBindFallback != nil { + if err := s.m.SetLLMPoolBindFallback(*req.LLMPoolBindFallback); err != nil { + writeErr(w, 500, err.Error()) + return + } + changed = true + } + if req.LLMPoolEnabled != nil || req.LLMPoolBindFallback != nil { + // Pinned tasks' planner/worker and per-profile chat agents hold providers + // built under the OLD switch state — drop them so they pick up the new one. + s.invalidateProfileAgents() + } if req.TrafficCapture != nil { if err := s.m.SetTrafficEnabled(*req.TrafficCapture); err != nil { writeErr(w, 500, err.Error()) diff --git a/server/server_mgmt.go b/server/server_mgmt.go index 0595787..eef78bb 100644 --- a/server/server_mgmt.go +++ b/server/server_mgmt.go @@ -1609,10 +1609,10 @@ func (s *Server) pgSaveProfile(w http.ResponseWriter, r *http.Request) { } // Editing a profile rebuilds any task pinned to it on its next round. s.invalidateProfileAgents() - // If we just edited the active profile, hot-apply so changes take effect now. - if act, _ := pg.ActiveProfile(); act != nil && act.ID == id { - s.reapplyActiveProfile() - } + // Hot-apply. Not just when the ACTIVE profile changed: with failover on, any + // profile's key/model/priority/pool_exclude edit reshapes the chain, so the + // engine's provider has to be rebuilt either way. + s.reapplyActiveProfile() writeJSON(w, 200, map[string]any{"id": id}) } @@ -1627,6 +1627,8 @@ func (s *Server) pgDeleteProfile(w http.ResponseWriter, r *http.Request) { return } s.invalidateProfileAgents() // drop cached agents for the removed profile + s.llmHealth.Reset(id) // its breaker state is meaningless now (row is FK-cascaded away) + s.reapplyActiveProfile() // the deleted profile may have been in the failover chain writeJSON(w, 200, map[string]any{"deleted": id}) } @@ -1651,6 +1653,40 @@ func (s *Server) pgActivateProfile(w http.ResponseWriter, r *http.Request) { writeJSON(w, 200, map[string]any{"ok": true}) } +// pgLLMPoolStatus reports the failover ("轮询") switches, the resolved chain order +// and every profile's circuit-breaker state — what the LLM page renders as the +// "轮询顺序" strip and the per-card health badges. +func (s *Server) pgLLMPoolStatus(w http.ResponseWriter, r *http.Request) { + if s.pg(w) == nil { + return + } + writeJSON(w, 200, s.llmPoolStatus()) +} + +// pgLLMPoolReset clears a tripped profile's circuit breaker so the next call +// tries it again immediately ("立即恢复"). id=0 clears every profile. +func (s *Server) pgLLMPoolReset(w http.ResponseWriter, r *http.Request) { + pg := s.pg(w) + if pg == nil { + return + } + var body struct { + ID int64 `json:"id"` // 0 / omitted = all + } + if err := decode(r, &body); err != nil { + writeErr(w, 400, err.Error()) + return + } + if body.ID > 0 { + s.llmHealth.Reset(body.ID) + } else { + for id := range s.llmHealth.Snapshot() { + s.llmHealth.Reset(id) + } + } + writeJSON(w, 200, s.llmPoolStatus()) +} + // pgListModels fetches available models from the provider's API endpoint. // Supports OpenAI-format (GET /models) and Anthropic-format (GET /v1/models); for // Anthropic-compatible third parties (e.g. DeepSeek) whose model list lives only on diff --git a/web/src/app/(main)/system/llm/page.tsx b/web/src/app/(main)/system/llm/page.tsx index f01ac53..6de5007 100644 --- a/web/src/app/(main)/system/llm/page.tsx +++ b/web/src/app/(main)/system/llm/page.tsx @@ -2,7 +2,17 @@ import * as React from "react"; -import { Loader2Icon, PlugZapIcon, PlusIcon, RefreshCwIcon, SaveIcon, StarIcon, Trash2Icon } from "lucide-react"; +import { + Loader2Icon, + PlugZapIcon, + PlusIcon, + RefreshCwIcon, + RotateCcwIcon, + SaveIcon, + StarIcon, + Trash2Icon, + ZapIcon, +} from "lucide-react"; import { toast } from "sonner"; import { Badge } from "@/components/ui/badge"; @@ -23,8 +33,9 @@ import { Label } from "@/components/ui/label"; import { Popover, PopoverContent, PopoverTrigger } from "@/components/ui/popover"; import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select"; import { Separator } from "@/components/ui/separator"; +import { Switch } from "@/components/ui/switch"; import { api } from "@/lib/api"; -import type { LLMProfile } from "@/lib/types"; +import type { LLMPoolStatus, LLMProfile } from "@/lib/types"; import { cn } from "@/lib/utils"; // 思考强度档位(仅在开关打开时生效)。 @@ -48,6 +59,203 @@ const toStore = (mode: ThinkMode, effort: string) => (mode === "on" ? effort : m const modeFromStore = (v?: string): ThinkMode => (!v ? "none" : v === "off" ? "off" : "on"); const effortFromStore = (v?: string) => (v && v !== "off" ? v : "high"); +// 熔断状态 → 徽章样式。degraded 是「在失败但还没到熔断阈值」的中间态。 +const STATE_BADGE: Record = { + ok: { label: "正常", cls: "border-emerald-500/50 text-emerald-600 dark:text-emerald-400" }, + degraded: { label: "异常", cls: "border-amber-500/50 text-amber-600 dark:text-amber-400" }, + tripped: { label: "已熔断", cls: "border-destructive/50 text-destructive" }, +}; + +function cooldownText(secs: number) { + if (secs <= 0) return ""; + if (secs < 60) return `${secs}s`; + return `${Math.ceil(secs / 60)}min`; +} + +// PoolCard 是轮询功能的总控:主开关、绑定兜底开关,以及后端算出来的**实际**链路顺序。 +// 顺序预览很关键——「激活 + 优先级 + 不参与轮询」三个因素叠加后,光看单张配置卡片 +// 是猜不出真实顺序的。 +function PoolCard({ refreshKey }: { refreshKey: number }) { + const [pool, setPool] = React.useState(null); + const [busy, setBusy] = React.useState(false); + + const load = React.useCallback(async () => { + try { + setPool(await api.llmPool()); + } catch { + /* ignore */ + } + }, []); + // biome-ignore lint/correctness/useExhaustiveDependencies: refreshKey is a prop — bumping it is exactly how the parent asks for a refetch after a profile edit + React.useEffect(() => { + void load(); + }, [load, refreshKey]); + // 冷却倒计时是后端算的剩余秒数,定时拉一次让它走起来(只在有熔断时才轮询)。 + React.useEffect(() => { + if (!pool?.enabled || !pool.chain.some((m) => m.state !== "ok")) return; + const t = setInterval(() => void load(), 10_000); + return () => clearInterval(t); + }, [pool, load]); + + async function toggle(patch: { llm_pool_enabled?: boolean; llm_pool_bind_fallback?: boolean }) { + if (busy) return; + setBusy(true); + try { + await api.setSettings(patch); + await load(); + if (patch.llm_pool_enabled !== undefined) { + toast.success(patch.llm_pool_enabled ? "已开启 LLM 轮询" : "已关闭 LLM 轮询"); + } else { + toast.success("已更新绑定兜底设置"); + } + } catch (e) { + toast.error(`设置失败:${(e as Error).message}`); + } finally { + setBusy(false); + } + } + + async function recover(id?: string) { + try { + setPool(await api.resetLLMPool(id)); + toast.success(id ? "已恢复该配置" : "已恢复全部配置"); + } catch (e) { + toast.error(`恢复失败:${(e as Error).message}`); + } + } + + const enabled = pool?.enabled ?? false; + // 链路顺序 = 后端返回的顺序;不参与轮询的配置仍会列出(灰显并标注), + // 这样用户能看见「它为什么不在队列里」,而不是凭空消失。 + const chain = pool?.chain ?? []; + const inChain = chain.filter((m) => m.active || !m.excluded); + const tripped = chain.filter((m) => m.state === "tripped"); + + return ( + + + + LLM 轮询 · 故障转移 + + + 开启后,未指定模型的 Agent 在当前配置不可用(余额不足 / Key 失效 / 限流 / + 服务异常)时自动切换到下一个配置。 指定了模型的 Agent 与任务默认不参与轮询。 + + + void toggle({ llm_pool_enabled: v })} + aria-label="LLM 轮询开关" + /> + + + {enabled && ( + +
+
+ +

+ 默认关闭:Agent 或任务指定了某个配置就只用它,失败即失败(不会悄悄换成别的模型)。 + 开启后,指定的配置失败时也会回落到下面的轮询链。 +

+
+ void toggle({ llm_pool_bind_fallback: v })} + aria-label="绑定配置失败兜底开关" + /> +
+ +
+
+ + {tripped.length > 0 && ( + + )} +
+ {inChain.length < 2 && ( +

+ 当前只有 {inChain.length} 个可用配置,轮询不会生效——至少需要 2 个已填 API Key 且参与轮询的配置。 +

+ )} +
+ {chain.map((m) => { + const badge = STATE_BADGE[m.state] ?? STATE_BADGE.ok; + const order = + m.excluded && !m.active ? null : inChain.findIndex((x) => x.profile_id === m.profile_id) + 1; + return ( +
+ + {order ?? "—"} + + {m.name} + {m.model} + {m.active && ( + + 激活 + + )} + {m.excluded && !m.active && 不参与轮询} + {!m.active && 优先级 {m.priority}} +
+ {m.state === "tripped" && m.cooldown_secs > 0 && ( + 冷却 {cooldownText(m.cooldown_secs)} + )} + {m.state === "degraded" && ( + 连续失败 {m.fails} 次 + )} + + {badge.label} + + {m.state !== "ok" && ( + + )} +
+ {m.last_error && ( +

+ {m.last_error} +

+ )} +
+ ); + })} + {chain.length === 0 && ( +
+ 暂无配置 +
+ )} +
+

+ 激活配置恒为第 1 顺位,其余按优先级从高到低。某个配置失败后进入冷却(60s → 5min → 30min), + 冷却期内被跳过;恢复后自动切回。上下文窗口装不下当前请求的配置会被跳过。 +

+
+
+ )} +
+ ); +} + function NewProfileDialog({ onCreated }: { onCreated: (id: string) => void }) { const [open, setOpen] = React.useState(false); const [name, setName] = React.useState(""); @@ -61,6 +269,8 @@ function NewProfileDialog({ onCreated }: { onCreated: (id: string) => void }) { const [cw, setCw] = React.useState("0"); const [thinkMode, setThinkMode] = React.useState("none"); const [effort, setEffort] = React.useState("high"); + const [priority, setPriority] = React.useState("0"); + const [poolExclude, setPoolExclude] = React.useState(false); const [models, setModels] = React.useState([]); const [loadingModels, setLoadingModels] = React.useState(false); const [modelsOpen, setModelsOpen] = React.useState(false); @@ -77,6 +287,8 @@ function NewProfileDialog({ onCreated }: { onCreated: (id: string) => void }) { setCw("0"); setThinkMode("none"); setEffort("high"); + setPriority("0"); + setPoolExclude(false); setModels([]); setModelsOpen(false); } @@ -118,6 +330,8 @@ function NewProfileDialog({ onCreated }: { onCreated: (id: string) => void }) { rate_per_minute: Number(rpm) || 0, context_window_k: Number(cw) || 0, reasoning_effort: toStore(thinkMode, effort), + priority: Number(priority) || 0, + pool_exclude: poolExclude, }); toast.success(`已新建 Profile:${name.trim()}(在列表中「设为激活」以启用)`); reset(); @@ -275,12 +489,39 @@ function NewProfileDialog({ onCreated }: { onCreated: (id: string) => void }) { /> +
+
+
+ +

数字越大越先被选中;激活配置恒为第 1 顺位。

+
+ setPriority(e.target.value)} + /> +
+
+
+ +

+ 开启后不会被当作故障转移目标(仍可被 Agent / 任务显式指定使用)。 +

+
+ +
+

- 不发送=不带思考字段(兼容 MiniMax 等不支持该字段的模型);显式关闭=发 disabled(默认开思考的模型才需要);开启=先推理再作答。 + 不发送=不带思考字段(兼容 MiniMax 等不支持该字段的模型);显式关闭=发 + disabled(默认开思考的模型才需要);开启=先推理再作答。

setPriority(e.target.value)} + /> +
+
+
+ +

+ 开启后不会被当作故障转移目标(仍可被 Agent / 任务显式指定使用)。 适合「只给某个 Agent + 专用、不希望别人失败时烧掉」的昂贵配置。 +

+
+ +
+
+

- 不发送=不带思考字段(兼容 MiniMax 等不支持该字段的模型);显式关闭=发 disabled(默认开思考的模型才需要);开启=先推理再作答。 + 不发送=不带思考字段(兼容 MiniMax 等不支持该字段的模型);显式关闭=发 + disabled(默认开思考的模型才需要);开启=先推理再作答。