Files
ARTEX/llmpool/health.go
T
Autumn-27 0c653a2135 feat(llm): LLM 配置轮询(故障转移),某个配置没额度/失效时自动切下一个
默认关闭。开启后「未指定模型」的 agent 在当前配置不可用(402 余额不足 /
401 key 失效 / 429 限流 / 5xx)时自动切到下一个配置;agent 绑定或任务 pin
了配置的默认不轮询(绑定即独占,可由 llm_pool_bind_fallback 打开兜底)。

链序 = 激活配置 + 其余按 priority DESC,排除 pool_exclude 的。只给
llm_profiles 加两列,不引入关联表,所以 agents/tasks/conversations 对
llm_profile_id 的四处引用一行未改。

核心约束:只在本次调用尚未 yield 出任何事件时切换成员。已经吐出
text_delta / tool_use_start 后中途断流一律原样上抛(交给 agentcore 的
resume),否则重发同一 request 会重复输出、污染会话历史。SDK 的 doStream
保证 4xx/5xx 在读 body 前返回,所以目标场景都落在这个安全窗口内。

context.Canceled / DeadlineExceeded 绝不触发切换——那是用户取消或任务
超时,误判会白烧备用 key 并污染 a6bd180 的终止原因诊断。400 也不切
(请求本身有问题,换模型结果一样)。

熔断:402/401/403/404 一次即熔断,429/5xx/网络错误连续 3 次;冷却
60s→5min→30min 退避,到期半开试探,成功清零。Registry 是进程级的,独立
于 provider 缓存,重建链路不丢状态,且全进程共享(A 任务撞见的 402,B 任务
立刻知道)。状态旁路落库,重启只恢复尚未到期的冷却窗口。

上下文窗口装不下当前请求的成员直接跳过(必然 400),不把主力模型的窗口
压到全链最小值。全员熔断时仍探链首,不让引擎停摆。

Pool 包在 llmrec 外面、每个成员自带 recorder,失败调用会以 status=error
+ 正确的 profile 名进 llm_records;切换/熔断走 log.Printf,自动出现在
系统日志页。

顺带修两个既有问题:
- DecomposeGoals 原本自己 NewProvider(),绕过限速器/录制/轮询,且每个
  任务新建一个实例——改为由 server 传入共享 provider
- saveLLMConfig 会把 priority/pool_exclude 重置为零值

前端:LLM 页新增轮询卡片(总开关/兜底开关/实际链路顺序/健康徽章/冷却
倒计时/立即恢复),配置编辑区加优先级与「不参与轮询」。

只有一个配置时链长度为 1,poolChain 返回 nil 走裸 provider,行为与开关
打开前完全一致。设计见 docs/LLM轮询设计.md。
2026-08-19 11:07:53 -04:00

150 lines
4.6 KiB
Go

// 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
}