From a50bf02428b3f8c7ffa9bf99af7316ca743ea20a Mon Sep 17 00:00:00 2001 From: Autumn-27 <53844290+Autumn-27@users.noreply.github.com> Date: Wed, 9 Sep 2026 13:35:04 -0400 Subject: [PATCH] =?UTF-8?q?feat(llm):=20=E4=BA=94=E5=B1=82=E9=87=8D?= =?UTF-8?q?=E8=AF=95=E7=9A=84=E6=AC=A1=E6=95=B0=E4=B8=8E=E9=97=B4=E9=9A=94?= =?UTF-8?q?=E5=8F=AF=E9=85=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 一次模型调用的失败由内到外经过五层重试:建连(SDK)、空响应(SDK)、同 provider 安全窗口、轮询熔断、意图重跑。此前五层的次数与间隔全是写死的常量,遇到脾气各异 的中转站只能改代码。现在都能在 LLM 页配置,不配就是原来的默认值。 配置语义五层统一(db.RetryRule):次数 0=内置默认 / -1=关掉这层 / >0=该值; 间隔 0=原本的指数退避 / >0=改用固定间隔。0 被"未配置"占用,故关闭用 -1。 写库前夹紧到 次数 [-1,20]、间隔 [0,1h],DB 另有 CHECK 兜底。 粒度按层的性质分:前三层跟着端点走,可在每个模型配置里逐字段覆盖全局默认(只钉 间隔的仍继承全局次数);熔断与意图重跑是进程级语义,只有全局一份。 生效路径:①②是 provider 构建参数,保存后重建缓存;③随每次 hooks.current() 解析,故障转移换配置时重试节奏跟着换;④在失败路径上每次都要读,由 applyRetryPolicy 推给 llmpool.Registry;⑤每条意图收尾读一次。全部热生效,无需重启。 连接测试刻意不带这些参数——它有 30s 硬超时,叠上用户配的重试只会把本来能用的 端点测成超时失败。 norma 升 v0.3.5(llm.Config 新增 RetryInterval / EmptyResponseRetries / EmptyResponseInterval)。旧库升级后行为逐字节不变:新列默认 0,settings 键不 存在即全默认。设计见 docs/LLM重试设计.md。 --- CHANGELOG.md | 6 + agent/provider.go | 28 ++ db/config.go | 42 ++- db/llmretry.go | 127 +++++++ db/llmretry_test.go | 145 ++++++++ db/schema.sql | 26 ++ go.mod | 2 +- go.sum | 4 +- llmpool/health.go | 51 ++- llmpool/health_policy_test.go | 67 ++++ server/dto.go | 5 + server/engine.go | 7 +- server/llmretry.go | 80 +++++ server/llmretry_test.go | 97 ++++++ server/server.go | 13 + server/server_mgmt.go | 33 ++ server/task_llm.go | 78 +++-- .../(main)/system/llm/_components/retry.tsx | 310 ++++++++++++++++++ web/src/app/(main)/system/llm/page.tsx | 209 ++++++------ web/src/lib/api.ts | 6 + web/src/lib/types.ts | 26 ++ 21 files changed, 1216 insertions(+), 146 deletions(-) create mode 100644 db/llmretry.go create mode 100644 db/llmretry_test.go create mode 100644 llmpool/health_policy_test.go create mode 100644 server/llmretry.go create mode 100644 server/llmretry_test.go create mode 100644 web/src/app/(main)/system/llm/_components/retry.tsx diff --git a/CHANGELOG.md b/CHANGELOG.md index faf9c50..21dbe78 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ ## [Unreleased] +### LLM + +#### 新增的功能 + +- LLM 页新增「重试与退避」标签页：五层重试的**次数**与**间隔**都可以配。一次模型调用的失败由内到外经过建连重试（SDK，流开始前的连接重置 / 超时 / 429 / 5xx）、空响应重试（SDK，正常结束却没有任何内容，仅 openai 格式）、同 provider 安全窗口重试（未向调用方交付任何输出前的断流重放）、轮询熔断（连续失败到阈值就冷却跳过）、意图重跑（worker 以 model_error 收场后整条意图重跑）五层，内层用尽才轮到外层。每层两个旋钮，语义统一：留空 = 用原本的默认次数与指数退避；填次数就用该次数；填间隔就把指数退避换成固定间隔；填 -1 = 关掉这层重试。前三层跟着端点走，可在每个模型配置里逐字段覆盖全局默认（只钉间隔的仍继承全局次数）；熔断与意图重跑是进程级语义，只有全局一份。保存后热生效，不需要重启。全部留空即当前行为，旧库升级后逐字节不变（新列默认 0、settings 键不存在即全默认）。连接测试刻意不带这些参数——它有 30s 硬超时，叠上用户配的重试只会把能用的端点测成超时失败。设计见 `docs/LLM重试设计.md`。 + ## [0.3.7] - 2026-08-31 ### LLM diff --git a/agent/provider.go b/agent/provider.go index f0447b7..14b0ddd 100644 --- a/agent/provider.go +++ b/agent/provider.go @@ -74,6 +74,29 @@ type Config struct { // 空 = 不发送。值由 transcript.WithSessionID 挂在请求 context 上,由 RoundTripper // 读取填入,因此同一共享 provider 也能按会话发出不同的头值。 SessionHeaderKey string + // Retry 是该配置解析后的重试参数(profile 覆盖 → 全局策略 → 内置默认,由 + // server 侧解析)。三层的含义见 RetryConfig;零值 = 完全沿用内置默认。 + Retry RetryConfig +} + +// RetryConfig 是随一个 LLM 配置走的重试参数。每层的「次数」统一语义: +// 0 = 用内置默认次数;负数 = 关闭该层重试;>0 = 用该值。每层的「间隔」: +// 0 = 用该层原本的指数退避;>0 = 改用这个固定间隔。 +type RetryConfig struct { + // ConnectAttempts/ConnectInterval:SDK 建连重试(连接重置/超时/429/5xx,流开始前), + // 直接映射为 llm.Config.MaxRetries / RetryInterval。默认 3 次、0.5s 起指数(封顶 8s)。 + ConnectAttempts int + ConnectInterval time.Duration + // EmptyAttempts/EmptyInterval:SDK 空响应重试(完成但无 content block,仅 openai + // 格式),映射为 llm.Config.EmptyResponseRetries / EmptyResponseInterval。 + // 默认 2 次、同一条指数梯度。 + EmptyAttempts int + EmptyInterval time.Duration + // StreamAttempts/StreamInterval:同 provider 安全窗口重试——本项目在 SDK 之上补的 + // 一层,只在「还没向调用方交付任何输出」时重放断流/过载/流内 429。SDK 看不到它, + // 由 server/task_llm.go 消费。默认 2 次、0.5s 起指数(封顶 4s)。 + StreamAttempts int + StreamInterval time.Duration } // compaction window resolution bounds (in K tokens). Below the floor the @@ -249,6 +272,11 @@ func (c Config) NewProvider() (llm.Provider, error) { // 输出上限的字段名选择(空 = 用 max_tokens)。上限的「值」不在这里:它每轮随 // agentcore.Options.MaxTokens 走,provider 只决定把它塞进哪个键。 lc.MaxTokensField = c.MaxTokensField + // 重试参数与 SDK 同语义(次数 0=默认/负=关闭,间隔 0=指数退避/>0=固定),原样透传。 + lc.MaxRetries = c.Retry.ConnectAttempts + lc.RetryInterval = c.Retry.ConnectInterval + lc.EmptyResponseRetries = c.Retry.EmptyAttempts + lc.EmptyResponseInterval = c.Retry.EmptyInterval if c.RatePerSecond > 0 || c.RatePerMinute > 0 { lc.RateLimit = &llm.RateLimit{PerSecond: c.RatePerSecond, PerMinute: c.RatePerMinute} } diff --git a/db/config.go b/db/config.go index e1d2024..32c605b 100644 --- a/db/config.go +++ b/db/config.go @@ -61,19 +61,37 @@ type LLMProfile struct { // or sticky routing off a session-id header. "" = not sent. Maps to // agent.Config.SessionHeaderKey. SessionHeaderKey string `json:"session_header_key"` + // Retry overrides this profile's share of the retry ladder. Zero value = + // inherit the global policy (LLMRetryPolicy), so an untouched profile behaves + // exactly as before. See RetryOverride. + Retry RetryOverride `json:"retry"` +} + +// RetryOverride is one profile's optional override of the three retry layers +// that are per-endpoint: 建连(connect) / 空响应(empty) / 同 provider 安全窗口 +// (stream). Each rule's zero value means "inherit the global policy"; see +// RetryRule for the -1 / 0 / >0 semantics. +type RetryOverride struct { + Connect RetryRule `json:"connect"` + Empty RetryRule `json:"empty"` + Stream RetryRule `json:"stream"` } // 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,COALESCE(thinking_type,''),COALESCE(streaming,true),COALESCE(max_tokens,0),COALESCE(max_tokens_field,''),COALESCE(session_header_key,'')` -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,COALESCE(thinking_type,''),COALESCE(streaming,true),COALESCE(max_tokens,0),COALESCE(max_tokens_field,''),COALESCE(session_header_key,'')` +const profileRetryCols = `COALESCE(retry_connect_attempts,0),COALESCE(retry_connect_interval_ms,0),COALESCE(retry_empty_attempts,0),COALESCE(retry_empty_interval_ms,0),COALESCE(retry_stream_attempts,0),COALESCE(retry_stream_interval_ms,0)` +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,COALESCE(thinking_type,''),COALESCE(streaming,true),COALESCE(max_tokens,0),COALESCE(max_tokens_field,''),COALESCE(session_header_key,''),` + profileRetryCols +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,COALESCE(thinking_type,''),COALESCE(streaming,true),COALESCE(max_tokens,0),COALESCE(max_tokens_field,''),COALESCE(session_header_key,''),` + profileRetryCols // 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, &p.ThinkingType, &p.Streaming, - &p.MaxTokens, &p.MaxTokensField, &p.SessionHeaderKey) + &p.MaxTokens, &p.MaxTokensField, &p.SessionHeaderKey, + &p.Retry.Connect.Attempts, &p.Retry.Connect.IntervalMS, + &p.Retry.Empty.Attempts, &p.Retry.Empty.IntervalMS, + &p.Retry.Stream.Attempts, &p.Retry.Stream.IntervalMS) } func (d *DB) ListProfiles() ([]*LLMProfile, error) { @@ -144,20 +162,24 @@ func (d *DB) SaveProfile(p *LLMProfile) (int64, error) { if len(p.APIKey) >= 4 { hint = "…" + p.APIKey[len(p.APIKey)-4:] } + r := p.Retry.Clamped() 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,priority,pool_exclude,thinking_type,streaming,max_tokens,max_tokens_field,session_header_key) -VALUES ($1,$2,NULLIF($3,''),NULLIF($4,''),$5,NULLIF($6,''),NULLIF($7,''),$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18) 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, p.ThinkingType, p.Streaming, p.MaxTokens, p.MaxTokensField, p.SessionHeaderKey).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,thinking_type,streaming,max_tokens,max_tokens_field,session_header_key,retry_connect_attempts,retry_connect_interval_ms,retry_empty_attempts,retry_empty_interval_ms,retry_stream_attempts,retry_stream_interval_ms) +VALUES ($1,$2,NULLIF($3,''),NULLIF($4,''),$5,NULLIF($6,''),NULLIF($7,''),$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24) 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, p.ThinkingType, p.Streaming, p.MaxTokens, p.MaxTokensField, p.SessionHeaderKey, + r.Connect.Attempts, r.Connect.IntervalMS, r.Empty.Attempts, r.Empty.IntervalMS, r.Stream.Attempts, r.Stream.IntervalMS).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,priority=$10,pool_exclude=$11,thinking_type=$12,streaming=$13,max_tokens=$14,max_tokens_field=$15,session_header_key=$16 WHERE id=$17`, - p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType, p.Streaming, p.MaxTokens, p.MaxTokensField, p.SessionHeaderKey, 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,thinking_type=$12,streaming=$13,max_tokens=$14,max_tokens_field=$15,session_header_key=$16,retry_connect_attempts=$17,retry_connect_interval_ms=$18,retry_empty_attempts=$19,retry_empty_interval_ms=$20,retry_stream_attempts=$21,retry_stream_interval_ms=$22 WHERE id=$23`, + p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType, p.Streaming, p.MaxTokens, p.MaxTokensField, p.SessionHeaderKey, + r.Connect.Attempts, r.Connect.IntervalMS, r.Empty.Attempts, r.Empty.IntervalMS, r.Stream.Attempts, r.Stream.IntervalMS, 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,priority=$12,pool_exclude=$13,thinking_type=$14,streaming=$15,max_tokens=$16,max_tokens_field=$17,session_header_key=$18 WHERE id=$19`, - 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.ThinkingType, p.Streaming, p.MaxTokens, p.MaxTokensField, p.SessionHeaderKey, 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,thinking_type=$14,streaming=$15,max_tokens=$16,max_tokens_field=$17,session_header_key=$18,retry_connect_attempts=$19,retry_connect_interval_ms=$20,retry_empty_attempts=$21,retry_empty_interval_ms=$22,retry_stream_attempts=$23,retry_stream_interval_ms=$24 WHERE id=$25`, + 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.ThinkingType, p.Streaming, p.MaxTokens, p.MaxTokensField, p.SessionHeaderKey, + r.Connect.Attempts, r.Connect.IntervalMS, r.Empty.Attempts, r.Empty.IntervalMS, r.Stream.Attempts, r.Stream.IntervalMS, p.ID) return p.ID, err } diff --git a/db/llmretry.go b/db/llmretry.go new file mode 100644 index 0000000..49354a9 --- /dev/null +++ b/db/llmretry.go @@ -0,0 +1,127 @@ +package db + +import ( + "encoding/json" + "time" +) + +// LLM 重试策略：五层重试的「次数 + 间隔」全局配置，见 docs/LLM重试设计.md。 +// 存在 settings 表的一个 JSON 值里 —— 它是整机一份的运行参数，不值得为它开一张表； +// 读取走内置默认兜底，所以键不存在(全新库/从未配置过)时行为与写死常量时代完全一致。 + +const settingLLMRetryPolicy = "llm_retry_policy" + +// RetryRule is one layer's knob pair. The zero value means "unset": +// +// Attempts 0 = 用内置默认次数; -1 = 关闭该层重试; >0 = 用该值 +// IntervalMS 0 = 用该层原本的间隔策略(通常是指数退避); >0 = 改用固定毫秒间隔 +// +// -1 是「显式关掉」而不是「0 次」，因为 0 已经被「未配置」占用了。 +type RetryRule struct { + Attempts int `json:"attempts"` + IntervalMS int `json:"interval_ms"` +} + +// Interval returns the configured fixed interval, or 0 when unset (caller keeps +// its own default ladder). +func (r RetryRule) Interval() time.Duration { + if r.IntervalMS <= 0 { + return 0 + } + return time.Duration(r.IntervalMS) * time.Millisecond +} + +// Or returns the rule with each unset field filled in from fallback. Used to +// layer a profile override on top of the global policy field by field, so a +// profile that only pins the interval still inherits the global count. +func (r RetryRule) Or(fallback RetryRule) RetryRule { + if r.Attempts == 0 { + r.Attempts = fallback.Attempts + } + if r.IntervalMS == 0 { + r.IntervalMS = fallback.IntervalMS + } + return r +} + +// retry knob bounds. A count above the cap turns a blip into a token bonfire; +// an interval above an hour outlives any transient failure worth waiting out. +const ( + maxRetryAttempts = 20 + maxRetryIntervalMS = 3600_000 // 1h +) + +// Clamped returns the rule with out-of-range values pulled back into the sane +// band (attempts within [-1, 20], interval within [0, 1h]). +func (r RetryRule) Clamped() RetryRule { + if r.Attempts < -1 { + r.Attempts = -1 + } + if r.Attempts > maxRetryAttempts { + r.Attempts = maxRetryAttempts + } + if r.IntervalMS < 0 { + r.IntervalMS = 0 + } + if r.IntervalMS > maxRetryIntervalMS { + r.IntervalMS = maxRetryIntervalMS + } + return r +} + +// Clamped bounds a profile's override the same way the global policy is bounded, +// so a hand-crafted API payload can't land a value the CHECK constraint rejects. +func (o RetryOverride) Clamped() RetryOverride { + o.Connect, o.Empty, o.Stream = o.Connect.Clamped(), o.Empty.Clamped(), o.Stream.Clamped() + return o +} + +// LLMRetryPolicy holds the五层 retry configuration. Connect/Empty/Stream are the +// per-request layers (a profile may override them, see LLMProfile.Retry); +// Breaker and Intent are process-wide by nature and live only here. +type LLMRetryPolicy struct { + // Connect：SDK 建连重试(连接重置/超时/429/5xx，流开始前)。默认 3 次、指数退避。 + Connect RetryRule `json:"connect"` + // Empty：SDK 空响应重试(完成但无 content block，仅 openai 格式)。默认 2 次、指数退避。 + Empty RetryRule `json:"empty"` + // Stream：同 provider 安全窗口重试(未交付输出前的断流重放)。默认 2 次、0.5s 起指数(封顶 4s)。 + Stream RetryRule `json:"stream"` + // Breaker：轮询熔断。Attempts=连续几次瞬时失败触发熔断(默认 3，-1=瞬时失败不熔断， + // 硬失败如余额不足/密钥失效仍立即熔断)；IntervalMS=固定冷却时长(0=默认 1/5/30min 梯度)。 + Breaker RetryRule `json:"breaker"` + // Intent：worker 以 model_error 收场后的整条意图重跑。默认 2 次、固定 3s。 + Intent RetryRule `json:"intent"` +} + +// Clamped returns the policy with every rule clamped. +func (p LLMRetryPolicy) Clamped() LLMRetryPolicy { + p.Connect, p.Empty, p.Stream = p.Connect.Clamped(), p.Empty.Clamped(), p.Stream.Clamped() + p.Breaker, p.Intent = p.Breaker.Clamped(), p.Intent.Clamped() + return p +} + +// LLMRetryPolicy reads the global retry policy. A missing or unparseable value +// yields the zero policy — i.e. every layer on its built-in default. +func (d *DB) LLMRetryPolicy() LLMRetryPolicy { + var p LLMRetryPolicy + if d == nil { + return p + } + raw, ok, err := d.GetSetting(settingLLMRetryPolicy) + if err != nil || !ok || raw == "" { + return p + } + if err := json.Unmarshal([]byte(raw), &p); err != nil { + return LLMRetryPolicy{} + } + return p.Clamped() +} + +// SetLLMRetryPolicy persists the global retry policy (values are clamped first). +func (d *DB) SetLLMRetryPolicy(p LLMRetryPolicy) error { + raw, err := json.Marshal(p.Clamped()) + if err != nil { + return err + } + return d.SetSetting(settingLLMRetryPolicy, string(raw)) +} diff --git a/db/llmretry_test.go b/db/llmretry_test.go new file mode 100644 index 0000000..cf83703 --- /dev/null +++ b/db/llmretry_test.go @@ -0,0 +1,145 @@ +package db + +import ( + "testing" + "time" +) + +// A profile's retry override must survive a full round trip through the real +// column list — this is what catches a mis-ordered scan/insert after adding six +// columns at once. Also pins the "empty api key on update keeps the row usable" +// path, since that UPDATE has its own parameter numbering. +func TestProfileRetryRoundTrip(t *testing.T) { + d, err := Open(testDSN(t)) + if err != nil { + t.Skipf("postgres unavailable (%v) — skipping", err) + } + defer d.Close() + + want := RetryOverride{ + Connect: RetryRule{Attempts: 5, IntervalMS: 2000}, + Empty: RetryRule{Attempts: -1}, + Stream: RetryRule{IntervalMS: 250}, + } + id, err := d.SaveProfile(&LLMProfile{ + Name: "t-retry-roundtrip", Format: "openai", Model: "m", APIKey: "k", Retry: want, + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { d.Exec(`DELETE FROM llm_profiles WHERE id=$1`, id) }) + + got, err := d.ProfileByID(id) + if err != nil || got == nil { + t.Fatalf("ProfileByID: %v", err) + } + if got.Retry != want { + t.Fatalf("retry=%+v, want %+v", got.Retry, want) + } + // Keyless update path (the UI sends no key when the user didn't retype it). + got.APIKey = "" + got.Retry.Stream = RetryRule{Attempts: 3, IntervalMS: 700} + if _, err := d.SaveProfile(got); err != nil { + t.Fatal(err) + } + after, err := d.ProfileByID(id) + if err != nil || after == nil { + t.Fatalf("ProfileByID after update: %v", err) + } + if after.Retry.Stream != (RetryRule{Attempts: 3, IntervalMS: 700}) { + t.Fatalf("stream=%+v after update", after.Retry.Stream) + } + if after.Retry.Connect != want.Connect || after.Retry.Empty != want.Empty { + t.Fatalf("untouched rules changed: %+v", after.Retry) + } + // The listing query reads a different column list — it must agree. + profiles, err := d.ListProfiles() + if err != nil { + t.Fatal(err) + } + for _, p := range profiles { + if p.ID == id && p.Retry.Connect != want.Connect { + t.Fatalf("ListProfiles retry=%+v, want %+v", p.Retry, want) + } + } +} + +// Out-of-range values are clamped on the way in, so the DB CHECK constraint is +// never what the user hears about. +func TestProfileRetryClampedOnSave(t *testing.T) { + d, err := Open(testDSN(t)) + if err != nil { + t.Skipf("postgres unavailable (%v) — skipping", err) + } + defer d.Close() + + id, err := d.SaveProfile(&LLMProfile{ + Name: "t-retry-clamp", Format: "openai", Model: "m", APIKey: "k", + Retry: RetryOverride{Connect: RetryRule{Attempts: -99, IntervalMS: -5}}, + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { d.Exec(`DELETE FROM llm_profiles WHERE id=$1`, id) }) + got, err := d.ProfileByID(id) + if err != nil || got == nil { + t.Fatalf("ProfileByID: %v", err) + } + if got.Retry.Connect != (RetryRule{Attempts: -1}) { + t.Fatalf("connect=%+v, want attempts -1 and no interval", got.Retry.Connect) + } +} + +// The global policy round-trips through settings, and an unset key reads back as +// "everything on its built-in default". +func TestLLMRetryPolicyRoundTrip(t *testing.T) { + d, err := Open(testDSN(t)) + if err != nil { + t.Skipf("postgres unavailable (%v) — skipping", err) + } + defer d.Close() + + before, hadBefore, err := d.GetSetting(settingLLMRetryPolicy) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if hadBefore { + d.SetSetting(settingLLMRetryPolicy, before) + } else { + d.Exec(`DELETE FROM settings WHERE key=$1`, settingLLMRetryPolicy) + } + }) + + want := LLMRetryPolicy{ + Connect: RetryRule{Attempts: 4, IntervalMS: 1500}, + Breaker: RetryRule{Attempts: 2, IntervalMS: 90_000}, + Intent: RetryRule{Attempts: -1}, + } + if err := d.SetLLMRetryPolicy(want); err != nil { + t.Fatal(err) + } + got := d.LLMRetryPolicy() + if got != want { + t.Fatalf("policy=%+v, want %+v", got, want) + } + if d := got.Breaker.Interval(); d != 90*time.Second { + t.Fatalf("breaker interval=%v, want 90s", d) + } + + // 越界值写进去也会被夹回区间，读出来是夹紧后的值。 + if err := d.SetLLMRetryPolicy(LLMRetryPolicy{Stream: RetryRule{Attempts: 999, IntervalMS: 99_999_999}}); err != nil { + t.Fatal(err) + } + if got := d.LLMRetryPolicy().Stream; got.Attempts != 20 || got.IntervalMS != 3600_000 { + t.Fatalf("stream=%+v, want the 20 / 1h caps", got) + } + + // 键不存在 = 全默认。 + if _, err := d.Exec(`DELETE FROM settings WHERE key=$1`, settingLLMRetryPolicy); err != nil { + t.Fatal(err) + } + if got := d.LLMRetryPolicy(); got != (LLMRetryPolicy{}) { + t.Fatalf("unset policy=%+v, want zero", got) + } +} diff --git a/db/schema.sql b/db/schema.sql index 4a25595..4248569 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -367,6 +367,14 @@ CREATE TABLE IF NOT EXISTS llm_profiles ( -- 自定义会话头：非空时每次请求带一个该名字的 HTTP 头，头值=当前运行的 session id -- (chat 会话/worker 意图)。用于某些按 session-id 头做提示缓存/粘性路由的网关。''=不发送。 session_header_key TEXT NOT NULL DEFAULT '', + -- 重试覆盖：次数 0=用全局默认/-1=关闭/>0=该值；间隔 0=用默认指数退避/>0=固定毫秒。 + -- 三组分别对应建连重试、空响应重试、同 provider 安全窗口重试，详见下方 ALTER 处注释。 + retry_connect_attempts INTEGER NOT NULL DEFAULT 0, + retry_connect_interval_ms INTEGER NOT NULL DEFAULT 0, + retry_empty_attempts INTEGER NOT NULL DEFAULT 0, + retry_empty_interval_ms INTEGER NOT NULL DEFAULT 0, + retry_stream_attempts INTEGER NOT NULL DEFAULT 0, + retry_stream_interval_ms INTEGER NOT NULL DEFAULT 0, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); @@ -399,6 +407,24 @@ ALTER TABLE llm_profiles ADD CONSTRAINT llm_profiles_max_tokens_check -- 自定义会话头名；补旧库。默认 '' = 不发送，旧配置行为不变。 ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS session_header_key TEXT NOT NULL DEFAULT ''; +-- 单配置的重试覆盖（见 docs/LLM重试设计.md）。三组各自一对「次数 + 固定间隔」， +-- 语义统一：次数 0=沿用全局默认、-1=关闭该层重试、>0=用该值；间隔 0=沿用该层的 +-- 默认指数退避、>0=改用这个固定毫秒数。全部默认 0，所以旧库/旧配置行为不变。 +-- connect = 建连重试（SDK doStream：连接重置/超时/429/5xx，流开始前） +-- empty = 空响应重试（SDK：完成但没有任何 content block，仅 openai 格式） +-- stream = 同 provider 安全窗口重试（本项目 task_llm：未交付输出前的断流重放） +ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS retry_connect_attempts INTEGER NOT NULL DEFAULT 0; +ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS retry_connect_interval_ms INTEGER NOT NULL DEFAULT 0; +ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS retry_empty_attempts INTEGER NOT NULL DEFAULT 0; +ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS retry_empty_interval_ms INTEGER NOT NULL DEFAULT 0; +ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS retry_stream_attempts INTEGER NOT NULL DEFAULT 0; +ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS retry_stream_interval_ms INTEGER NOT NULL DEFAULT 0; +-- 同 format：先删再建，保证每次启动幂等。次数下限 -1(关闭)，间隔不能为负。 +ALTER TABLE llm_profiles DROP CONSTRAINT IF EXISTS llm_profiles_retry_check; +ALTER TABLE llm_profiles ADD CONSTRAINT llm_profiles_retry_check CHECK ( + retry_connect_attempts >= -1 AND retry_empty_attempts >= -1 AND retry_stream_attempts >= -1 + AND retry_connect_interval_ms >= 0 AND retry_empty_interval_ms >= 0 AND retry_stream_interval_ms >= 0); + -- 思考开关字段 thinking_type，从旧的单一 reasoning_effort 语义一次性拆分而来。 -- schema.sql 每次启动都执行，故迁移必须只跑一次：仅当该列尚不存在时才回填， -- 否则每次启动都会把用户后来手动设的组合覆盖回去。旧 reasoning_effort 语义： diff --git a/go.mod b/go.mod index 09dfd8f..b13f25d 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/Autumn-27/artex go 1.26.3 require ( - github.com/Autumn-27/norma v0.3.4 + github.com/Autumn-27/norma v0.3.5 github.com/golang-jwt/jwt/v5 v5.3.1 github.com/jackc/pgx/v5 v5.10.0 github.com/klauspost/compress v1.18.6 diff --git a/go.sum b/go.sum index 5c3889a..3029a38 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,5 @@ -github.com/Autumn-27/norma v0.3.4 h1:695ceEU5dKaf43xWlcqdYKdmGG2pgqhaKmTQy8rTD9A= -github.com/Autumn-27/norma v0.3.4/go.mod h1:BudWrf7iHl96FgHqFr/HwN3mKYmGjJDt+xL2ZHXfFZw= +github.com/Autumn-27/norma v0.3.5 h1:zYJKoN+SNi3iscednrw21xsW48jUD55CT6RdySP15MA= +github.com/Autumn-27/norma v0.3.5/go.mod h1:BudWrf7iHl96FgHqFr/HwN3mKYmGjJDt+xL2ZHXfFZw= github.com/andybalholm/brotli v1.2.1 h1:R+f5xP285VArJDRgowrfb9DqL18yVK0gKAW/F+eTWro= github.com/andybalholm/brotli v1.2.1/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY= github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 h1:DklsrG3dyBCFEj5IhUbnKptjxatkF07cF2ak3yi77so= diff --git a/llmpool/health.go b/llmpool/health.go index 80ce1f4..f95fc2d 100644 --- a/llmpool/health.go +++ b/llmpool/health.go @@ -32,8 +32,9 @@ type State struct { // 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. +// softTripAfter is the DEFAULT number of consecutive TRANSIENT failures (429 / +// 5xx / network) that trip the breaker. Deterministic failures (no credit, bad +// key) trip on the first regardless. Overridable via Registry.SetPolicy. const softTripAfter = 3 // Registry holds the circuit-breaker state of every profile, keyed by profile id. @@ -48,6 +49,43 @@ type Registry struct { // restart. Both may be nil (no DB); both are called off the hot path. persist func(id int64, st State) forget func(id int64) + + // softTrip / cooldown are the operator-set overrides (0 = use the built-in + // default / ladder). They live here rather than being read per failure because + // Trip runs on the failure path of every request. + softTrip int + cooldown time.Duration +} + +// SetPolicy overrides the breaker's two knobs. softTrip: how many consecutive +// transient failures trip it (0 = default softTripAfter; negative = transient +// failures never trip it, leaving only the deterministic ones). cooldown: a fixed +// cooling-off window (0 = the 1min/5min/30min ladder). Safe to call at any time. +func (r *Registry) SetPolicy(softTrip int, cooldown time.Duration) { + r.mu.Lock() + defer r.mu.Unlock() + r.softTrip, r.cooldown = softTrip, cooldown +} + +// tripAfter is the effective consecutive-transient-failure threshold. Callers +// hold r.mu. A negative override yields 0, which Trip reads as "never soft-trip". +func (r *Registry) tripAfter() int { + if r.softTrip == 0 { + return softTripAfter + } + return max(r.softTrip, 0) +} + +// coolFor is the cooling-off window for the trips-th trip (1-based): the fixed +// override when set, else the ladder (last rung repeats). Callers hold r.mu. +func (r *Registry) coolFor(trips int) time.Duration { + if r.cooldown > 0 { + return r.cooldown + } + if trips-1 < len(backoff) { + return backoff[trips-1] + } + return backoff[len(backoff)-1] } // NewRegistry builds an empty registry. persist/forget may be nil. @@ -107,13 +145,10 @@ func (r *Registry) Trip(id int64, errMsg string, hard bool) (tripped bool) { st.Fails++ st.LastError = errMsg st.LastAt = time.Now() - if hard || st.Fails >= softTripAfter { + softTrip := r.tripAfter() + if hard || (softTrip > 0 && st.Fails >= softTrip) { 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.OpenUntil = time.Now().Add(r.coolFor(st.Trips)) st.Fails = 0 // counted into this trip; start fresh for the half-open probe tripped = true } diff --git a/llmpool/health_policy_test.go b/llmpool/health_policy_test.go new file mode 100644 index 0000000..09783c0 --- /dev/null +++ b/llmpool/health_policy_test.go @@ -0,0 +1,67 @@ +package llmpool + +import ( + "testing" + "time" +) + +// A configured soft-trip threshold replaces the default: two transient failures +// are enough, and the fixed cooldown replaces the 1/5/30min ladder. +func TestSetPolicyOverridesThresholdAndCooldown(t *testing.T) { + reg := NewRegistry(nil, nil) + reg.SetPolicy(2, 90*time.Second) + if reg.Trip(1, "429", false) { + t.Fatal("tripped on the first transient failure, want the second") + } + if !reg.Trip(1, "429", false) { + t.Fatal("should trip on transient failure #2") + } + st := reg.Get(1) + if d := time.Until(st.OpenUntil); d < 80*time.Second || d > 90*time.Second { + t.Fatalf("cooldown=%v, want ~90s", d) + } + // Every later trip keeps the same fixed window instead of climbing the ladder. + reg.Trip(1, "429", true) + st = reg.Get(1) + if d := time.Until(st.OpenUntil); d < 80*time.Second || d > 90*time.Second { + t.Fatalf("second cooldown=%v, want ~90s (fixed)", d) + } +} + +// A negative threshold turns off soft tripping entirely: transient failures never +// open the breaker, deterministic ones still do immediately. +func TestSetPolicyDisablesSoftTrip(t *testing.T) { + reg := NewRegistry(nil, nil) + reg.SetPolicy(-1, 0) + for i := range 10 { + if reg.Trip(1, "429", false) { + t.Fatalf("transient failure #%d tripped the breaker, want never", i+1) + } + } + if reg.IsOpen(1) { + t.Fatal("breaker should stay closed for transient failures") + } + if !reg.Trip(1, "no credit", true) { + t.Fatal("a hard failure must still trip immediately") + } + if d := time.Until(reg.Get(1).OpenUntil); d < 50*time.Second || d > 60*time.Second { + t.Fatalf("cooldown=%v, want the default first rung (~1min)", d) + } +} + +// The zero policy is the historical behaviour: 3 transient failures, ladder cooldown. +func TestZeroPolicyKeepsDefaults(t *testing.T) { + reg := NewRegistry(nil, nil) + reg.SetPolicy(0, 0) + 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 d := time.Until(reg.Get(1).OpenUntil); d < 50*time.Second || d > 60*time.Second { + t.Fatalf("cooldown=%v, want the first ladder rung (~1min)", d) + } +} diff --git a/server/dto.go b/server/dto.go index 2c77971..368c9c0 100644 --- a/server/dto.go +++ b/server/dto.go @@ -591,6 +591,10 @@ type LLMProfileDTO struct { // 自定义会话头名：非空时每次请求带该 HTTP 头，头值=当前会话/意图的 session id。 // ''=不发送。用于按 session-id 头做提示缓存/粘性路由的网关。 SessionHeaderKey string `json:"session_header_key"` + // 本配置对重试的覆盖(建连/空响应/同 provider 安全窗口)。每项 attempts: + // 0=继承全局策略 | -1=关闭该层重试 | >0=次数；interval_ms: 0=用默认指数退避 | + // >0=改用该固定毫秒间隔。全 0 = 完全跟随全局，即历史行为。 + Retry db.RetryOverride `json:"retry"` } func llmProfileDTO(p *db.LLMProfile) LLMProfileDTO { @@ -614,6 +618,7 @@ func llmProfileDTO(p *db.LLMProfile) LLMProfileDTO { MaxTokens: p.MaxTokens, MaxTokensField: p.MaxTokensField, SessionHeaderKey: p.SessionHeaderKey, + Retry: p.Retry, } } diff --git a/server/engine.go b/server/engine.go index b0b0a56..1b4ba10 100644 --- a/server/engine.go +++ b/server/engine.go @@ -972,12 +972,13 @@ func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *ag // model_error 收场 → 额外重跑几次（退避后再试）。仅在意图仍属本 work、任务 // 未暂停/未终止/未取消【且未进入收尾】时重试；否则让位给对应分支处理(收尾期不 // 再重试,避免退避挤占其他 worker 的优雅收尾窗口)。 - for attempt := 1; attempt <= modelErrorRetries && + maxRetries, retryBackoff := e.modelErrorRetryPolicy() + for attempt := 1; attempt <= maxRetries && retryableWorkerModelError(reason, err) && workCtx.Err() == nil && ectx.Err() == nil && !e.IsPaused(t.ID) && !e.isSettling(t.ID); attempt++ { log.Printf("[worker %s] task %s 意图 #%d model_error 收场，%v 后重试 (%d/%d)", - name, t.ID, intent.ID, modelErrorRetryBackoff, attempt, modelErrorRetries) - if sleepCtx(workCtx, modelErrorRetryBackoff) { + name, t.ID, intent.ID, retryBackoff, attempt, maxRetries) + if sleepCtx(workCtx, retryBackoff) { break // 退避期间被取消（终止/暂停）→ 交给下方分支处理 } e.BeginLLMCall(t.ID) diff --git a/server/llmretry.go b/server/llmretry.go new file mode 100644 index 0000000..53f74eb --- /dev/null +++ b/server/llmretry.go @@ -0,0 +1,80 @@ +package server + +import ( + "time" + + "github.com/Autumn-27/artex/agent" + "github.com/Autumn-27/artex/db" +) + +// 重试策略的服务端解析，见 docs/LLM重试设计.md。五层里： +// - 建连 / 空响应 / 同 provider 安全窗口 是「跟着端点走」的，每个 LLM 配置可以覆盖 +// 全局默认（profile 的某项留空就继承全局，全局也没配就用内置默认）； +// - 熔断 / 意图重跑 是进程级的，只有全局一份。 +// +// 全局策略读一次 DB 一行 settings，调用点都在低频路径（构建 provider、work 收尾、 +// 保存配置），不值得再加一层缓存；熔断参数是例外——它在失败路径上每次都要读，所以 +// 由 applyRetryPolicy 推给 Registry 保存。 + +// retryPolicy reads the global policy; a nil DB yields the zero policy (all +// layers on their built-in defaults). +func (s *Server) retryPolicy() db.LLMRetryPolicy { + if s.m == nil || s.m.pg == nil { + return db.LLMRetryPolicy{} + } + return s.m.pg.LLMRetryPolicy() +} + +// resolveRetry layers one profile's override on top of the global policy and +// converts the result into the form agent.Config carries. Rules combine field by +// field, so a profile that only pins an interval still inherits the global count. +func resolveRetry(o db.RetryOverride, pol db.LLMRetryPolicy) agent.RetryConfig { + connect := o.Connect.Or(pol.Connect) + empty := o.Empty.Or(pol.Empty) + stream := o.Stream.Or(pol.Stream) + return agent.RetryConfig{ + // 次数在这里保持「0=默认 / 负=关闭」的原始语义:SDK 的 MaxRetries / + // EmptyResponseRetries 与之完全同构,交给它自己解析即可。 + ConnectAttempts: connect.Attempts, ConnectInterval: connect.Interval(), + EmptyAttempts: empty.Attempts, EmptyInterval: empty.Interval(), + StreamAttempts: stream.Attempts, StreamInterval: stream.Interval(), + } +} + +// applyProfileRetry fills cfg.Retry for a profile read from the DB. +func (s *Server) applyProfileRetry(cfg *agent.Config, p *db.LLMProfile) { + if p == nil { + return + } + cfg.Retry = resolveRetry(p.Retry, s.retryPolicy()) +} + +// 熔断(轮询冷却)的默认值,与 llmpool 内置的一致 —— 这里只在「用户配了值」时才覆盖。 +// 意图重跑的默认值见 engine.go 的 modelErrorRetries / modelErrorRetryBackoff。 + +// applyRetryPolicy pushes the process-wide layers of the policy into the objects +// that consume them on a hot path: the circuit-breaker registry. Called at +// startup and whenever the policy is saved. +func (s *Server) applyRetryPolicy() { + pol := s.retryPolicy() + if s.llmHealth != nil { + s.llmHealth.SetPolicy(pol.Breaker.Attempts, pol.Breaker.Interval()) + } +} + +// modelErrorRetryPolicy resolves the intent-level replay knobs (layer ⑤): how +// many times a model_error work is re-run and how long to back off between runs. +func (e *Engine) modelErrorRetryPolicy() (retries int, backoff time.Duration) { + retries, backoff = modelErrorRetries, modelErrorRetryBackoff + if e == nil || e.m == nil || e.m.pg == nil { + return retries, backoff + } + rule := e.m.pg.LLMRetryPolicy().Intent + if rule.Attempts != 0 { + retries = max(rule.Attempts, 0) + } + if d := rule.Interval(); d > 0 { + backoff = d + } + return retries, backoff +} diff --git a/server/llmretry_test.go b/server/llmretry_test.go new file mode 100644 index 0000000..582e116 --- /dev/null +++ b/server/llmretry_test.go @@ -0,0 +1,97 @@ +package server + +import ( + "testing" + "time" + + "github.com/Autumn-27/artex/agent" + "github.com/Autumn-27/artex/db" +) + +// 没有任何配置时，解析结果必须是全零 —— 也就是 SDK 与 task_llm 各自的内置默认， +// 与「重试可配」这个特性上线之前逐字节一致。 +func TestResolveRetryUnconfigured(t *testing.T) { + got := resolveRetry(db.RetryOverride{}, db.LLMRetryPolicy{}) + if got != (agent.RetryConfig{}) { + t.Fatalf("resolveRetry=%+v, want zero", got) + } + retries, backoff := sameProviderRetryPolicy(got) + if retries != sameProviderStreamRetries { + t.Fatalf("retries=%d, want %d", retries, sameProviderStreamRetries) + } + if d := backoff(1); d != time.Second { + t.Fatalf("backoff(1)=%v, want 1s (the default ladder)", d) + } +} + +// The global policy applies to a profile that overrides nothing. +func TestResolveRetryFromGlobal(t *testing.T) { + pol := db.LLMRetryPolicy{ + Connect: db.RetryRule{Attempts: 5, IntervalMS: 2000}, + Empty: db.RetryRule{Attempts: -1}, + Stream: db.RetryRule{Attempts: 4, IntervalMS: 1500}, + } + got := resolveRetry(db.RetryOverride{}, pol) + want := agent.RetryConfig{ + ConnectAttempts: 5, ConnectInterval: 2 * time.Second, + EmptyAttempts: -1, + StreamAttempts: 4, StreamInterval: 1500 * time.Millisecond, + } + if got != want { + t.Fatalf("resolveRetry=%+v, want %+v", got, want) + } + retries, backoff := sameProviderRetryPolicy(got) + if retries != 4 { + t.Fatalf("retries=%d, want 4", retries) + } + for _, attempt := range []int{0, 3} { + if d := backoff(attempt); d != 1500*time.Millisecond { + t.Fatalf("backoff(%d)=%v, want a fixed 1.5s", attempt, d) + } + } +} + +// A profile override wins field by field: the pinned count replaces the global +// one, while the interval it left alone still comes from the global policy. +func TestResolveRetryProfileOverridesPerField(t *testing.T) { + pol := db.LLMRetryPolicy{ + Connect: db.RetryRule{Attempts: 5, IntervalMS: 2000}, + Stream: db.RetryRule{Attempts: 4, IntervalMS: 1500}, + } + over := db.RetryOverride{ + Connect: db.RetryRule{Attempts: 1}, + Stream: db.RetryRule{IntervalMS: 250}, + } + got := resolveRetry(over, pol) + if got.ConnectAttempts != 1 || got.ConnectInterval != 2*time.Second { + t.Fatalf("connect=%d/%v, want 1 attempt inheriting the 2s interval", got.ConnectAttempts, got.ConnectInterval) + } + if got.StreamAttempts != 4 || got.StreamInterval != 250*time.Millisecond { + t.Fatalf("stream=%d/%v, want the global 4 attempts with a pinned 250ms", got.StreamAttempts, got.StreamInterval) + } +} + +// -1 turns the same-provider layer off entirely (0 retries), rather than being +// mistaken for "unset". +func TestSameProviderRetryDisabled(t *testing.T) { + retries, _ := sameProviderRetryPolicy(agent.RetryConfig{StreamAttempts: -1}) + if retries != 0 { + t.Fatalf("retries=%d, want 0 (disabled)", retries) + } +} + +// Values beyond the sane range are clamped on the way into the DB, so a hostile +// or fat-fingered payload can't park a worker for a day. +func TestRetryRuleClamp(t *testing.T) { + pol := db.LLMRetryPolicy{ + Connect: db.RetryRule{Attempts: -50, IntervalMS: -1}, + Stream: db.RetryRule{Attempts: 9999, IntervalMS: 99_999_999}, + } + got := resolveRetry(db.RetryOverride{}, pol.Clamped()) + if got.ConnectAttempts != -1 || got.ConnectInterval != 0 { + t.Fatalf("connect=%d/%v, want -1 attempts and no interval", got.ConnectAttempts, got.ConnectInterval) + } + if got.StreamAttempts != 20 || got.StreamInterval != time.Hour { + t.Fatalf("stream=%d/%v, want the 20-attempt / 1h caps", got.StreamAttempts, got.StreamInterval) + } +} diff --git a/server/server.go b/server/server.go index 2d103ad..3f31803 100644 --- a/server/server.go +++ b/server/server.go @@ -157,6 +157,9 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi profAgents: map[int64]*profBundle{}, profChatAgents: map[int64]*agent.ChatAgent{}, provByProfile: map[int64]*provEntry{}, llmHealth: newLLMHealthRegistry(m.pg), taskAgents: map[string]*taskAgentBundle{}, archiveWake: make(chan struct{}, 1)} + // 熔断阈值/冷却是失败路径上的热参数，启动时把全局重试策略推给 Registry 一次； + // 之后每次保存策略再推一次（saveLLMRetryPolicy）。 + s.applyRetryPolicy() // Every task uses a stable task router. An empty explicit chain is resolved by // that router through Agent bindings and then the global provider, so adding a // first chain to a running task takes effect on its very next LLM call. @@ -314,6 +317,7 @@ func (s *Server) loadLLMConfig() (agent.Config, bool) { cfg.Stream = p.Streaming cfg.MaxTokens, cfg.MaxTokensField = p.MaxTokens, p.MaxTokensField cfg.SessionHeaderKey = p.SessionHeaderKey + s.applyProfileRetry(&cfg, p) if cfg.APIKey == "" { return cfg, false } @@ -338,12 +342,14 @@ func (s *Server) saveLLMConfig(cfg agent.Config) error { var maxTokens int var maxTokensField string var sessionHeaderKey string + var retry db.RetryOverride if profs, _ := s.m.pg.ListProfiles(); profs != nil { for _, p := range profs { if p.Name == "default" { id, priority, poolExclude, streaming = p.ID, p.Priority, p.PoolExclude, p.Streaming maxTokens, maxTokensField = p.MaxTokens, p.MaxTokensField sessionHeaderKey = p.SessionHeaderKey + retry = p.Retry break } } @@ -354,6 +360,7 @@ func (s *Server) saveLLMConfig(cfg agent.Config) error { ContextWindowK: cfg.ContextWindowK, ThinkingType: cfg.ThinkingType, ReasoningEffort: cfg.ReasoningEffort, IsDefault: true, Priority: priority, PoolExclude: poolExclude, Streaming: streaming, MaxTokens: maxTokens, MaxTokensField: maxTokensField, SessionHeaderKey: sessionHeaderKey, + Retry: retry, }) if err != nil { return err @@ -516,6 +523,7 @@ func (s *Server) loadProfileConfig(id int64) (agent.Config, bool) { cfg.Stream = p.Streaming cfg.MaxTokens, cfg.MaxTokensField = p.MaxTokens, p.MaxTokensField cfg.SessionHeaderKey = p.SessionHeaderKey + s.applyProfileRetry(&cfg, p) if cfg.APIKey == "" { return cfg, false } @@ -925,6 +933,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/retry-policy", s.pgGetLLMRetryPolicy) + mux.HandleFunc("POST /api/llm/retry-policy", s.pgSaveLLMRetryPolicy) 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) @@ -1411,6 +1421,9 @@ func (s *Server) testLLM(w http.ResponseWriter, r *http.Request) { writeJSON(w, 200, map[string]any{"ok": false, "error": "未提供 API Key"}) return } + // 重试参数【不】带进连接测试:测试有 30s 硬超时,把配置的重试次数/长间隔叠上去 + // 只会让一个本来能用的端点测成"超时失败"。测试看的是"这个端点通不通",重试节奏 + // 是跑起来之后的事。 lat, reply, err := agent.TestConnection(r.Context(), cfg) if err != nil { writeJSON(w, 200, map[string]any{"ok": false, "error": err.Error()}) diff --git a/server/server_mgmt.go b/server/server_mgmt.go index 036e409..6b02b64 100644 --- a/server/server_mgmt.go +++ b/server/server_mgmt.go @@ -1830,6 +1830,39 @@ func (s *Server) pgSaveProfile(w http.ResponseWriter, r *http.Request) { writeJSON(w, 200, map[string]any{"id": id}) } +// pgGetLLMRetryPolicy 返回全局重试策略(五层各自的次数+间隔)。未配置过 → 全零， +// 前端把零显示成「默认」。 +func (s *Server) pgGetLLMRetryPolicy(w http.ResponseWriter, r *http.Request) { + pg := s.pg(w) + if pg == nil { + return + } + writeJSON(w, 200, pg.LLMRetryPolicy()) +} + +// pgSaveLLMRetryPolicy 保存全局重试策略。三个「跟着端点走」的层(建连/空响应/同 +// provider 安全窗口)是 provider 的构建参数或调用参数，改完必须让缓存里的 provider +// 重建；熔断参数则直接推给进程级 Registry。 +func (s *Server) pgSaveLLMRetryPolicy(w http.ResponseWriter, r *http.Request) { + pg := s.pg(w) + if pg == nil { + return + } + var pol db.LLMRetryPolicy + if err := decode(r, &pol); err != nil { + writeErr(w, 400, err.Error()) + return + } + if err := pg.SetLLMRetryPolicy(pol); err != nil { + writeErr(w, 500, err.Error()) + return + } + s.applyRetryPolicy() + s.invalidateProfileAgents() + s.reapplyActiveProfile() + writeJSON(w, 200, pg.LLMRetryPolicy()) +} + func (s *Server) pgDeleteProfile(w http.ResponseWriter, r *http.Request) { pg := s.pg(w) if pg == nil { diff --git a/server/task_llm.go b/server/task_llm.go index c0d4a56..05893fb 100644 --- a/server/task_llm.go +++ b/server/task_llm.go @@ -91,6 +91,9 @@ type taskLLMSelection struct { profileID int64 revision int64 provider llm.Provider + // retry 是这次选中的配置解析出来的重试参数(profile 覆盖 → 全局策略 → 内置默认)。 + // 同 provider 安全窗口重试按它走,所以换了 profile 就换一套重试节奏。 + retry agent.RetryConfig } type taskLLMStreamHooks struct { @@ -105,38 +108,43 @@ type taskLLMStreamHooks struct { // 构建失败时才降级到任务链，任务链为空时再降级到全局配置。 // 返回的 profile id 只有走任务链时才非零 —— streamTaskLLM 以此判断额度错误是否 // 应该推进任务的故障转移状态（绑定/全局路径不改任务链状态，沿用既有语义）。 -func (r *taskLLMRuntime) current() (*Task, int64, int64, llm.Provider, error) { +func (r *taskLLMRuntime) current() (taskLLMSelection, error) { taskNum, err := parseTaskID(r.taskID) if err != nil { - return nil, 0, 0, nil, err + return taskLLMSelection{}, err } pt, err := r.s.m.pg.GetTask(taskNum) if err != nil { - return nil, 0, 0, nil, err + return taskLLMSelection{}, err } if pt == nil { - return nil, 0, 0, nil, fmt.Errorf("task %s not found", r.taskID) + return taskLLMSelection{}, fmt.Errorf("task %s not found", r.taskID) } r.s.syncTaskLLMState(pt) t, _ := r.s.m.Task(r.taskID) - if prov, _, ok := r.s.agentBindingProvider(r.agentKey); ok { - return t, 0, pt.LLMChainRevision, prov, nil + sel := taskLLMSelection{task: t, revision: pt.LLMChainRevision} + if prov, cfg, ok := r.s.agentBindingProvider(r.agentKey); ok { + sel.provider, sel.retry = prov, cfg.Retry + return sel, nil } if len(pt.LLMProfileIDs) > 0 { if pt.ActiveLLMProfileID == nil { - return t, 0, pt.LLMChainRevision, nil, &taskLLMError{taskID: r.taskID, chainExhausted: true, cause: errors.New("all selected profiles are quota exhausted")} + return sel, &taskLLMError{taskID: r.taskID, chainExhausted: true, cause: errors.New("all selected profiles are quota exhausted")} } - prov, _, ok := r.s.providerForProfile(*pt.ActiveLLMProfileID) + sel.profileID = *pt.ActiveLLMProfileID + prov, cfg, ok := r.s.providerForProfile(sel.profileID) if !ok { - return t, *pt.ActiveLLMProfileID, pt.LLMChainRevision, nil, fmt.Errorf("LLM profile #%d is missing or invalid", *pt.ActiveLLMProfileID) + return sel, fmt.Errorf("LLM profile #%d is missing or invalid", sel.profileID) } - return t, *pt.ActiveLLMProfileID, pt.LLMChainRevision, prov, nil + sel.provider, sel.retry = prov, cfg.Retry + return sel, nil } - prov, _, ok := r.s.globalProvider() + prov, cfg, ok := r.s.globalProvider() if !ok { - return t, 0, pt.LLMChainRevision, nil, fmt.Errorf("task %s has no available fallback LLM provider", r.taskID) + return sel, fmt.Errorf("task %s has no available fallback LLM provider", r.taskID) } - return t, 0, pt.LLMChainRevision, prov, nil + sel.provider, sel.retry = prov, cfg.Retry + return sel, nil } // activeCfg resolves the task's currently-active LLM config, mirroring current()'s @@ -194,10 +202,7 @@ func parseTaskID(id string) (int64, error) { // exhausted, and how to emit a failover transition. func (r *taskLLMRuntime) streamHooks() taskLLMStreamHooks { return taskLLMStreamHooks{ - current: func() (taskLLMSelection, error) { - task, profileID, revision, provider, err := r.current() - return taskLLMSelection{task: task, profileID: profileID, revision: revision, provider: provider}, err - }, + current: r.current, exhaust: func(selection taskLLMSelection, cause error) (db.TaskLLMTransition, error) { taskNum, _ := parseTaskID(r.taskID) transition, err := r.s.m.pg.MarkTaskLLMProfileQuotaExhaustedAtRevision(taskNum, selection.profileID, selection.revision, cause.Error()) @@ -244,13 +249,14 @@ func completeTaskLLM(ctx context.Context, taskID string, req llm.CompletionReque ) // 同 provider 安全窗口重试:非流式调用要么整体成功、要么整体失败,没有 // 中途已交付输出的问题,所以任何瞬时失败都可原样重试。 + retries, backoffOf := sameProviderRetryPolicy(selection.retry) for attempt := 0; ; attempt++ { msg, sr, usage, callErr = selection.provider.Complete(ctx, req) if callErr != nil && ctx.Err() == nil && - attempt < sameProviderStreamRetries && isRetryableStreamError(callErr) { - backoff := sameProviderRetryBackoff(attempt) + attempt < retries && isRetryableStreamError(callErr) { + backoff := backoffOf(attempt) log.Printf("[task-llm] task %s 非流式调用失败,%v 后同 provider 重试 (%d/%d): %v", - taskID, backoff, attempt+1, sameProviderStreamRetries, callErr) + taskID, backoff, attempt+1, retries, callErr) if sleepCtx(ctx, backoff) { break // 退避期间 ctx 取消 → 停止重试 } @@ -290,6 +296,7 @@ func streamTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest committed := false var pending []llm.StreamEvent var streamErr error + retries, backoffOf := sameProviderRetryPolicy(selection.retry) // 同 provider 安全窗口重试:committed 之前(还没向调用方交付任何输出) // 的瞬时失败可以原样重放,不会重复模型输出或工具执行。committed 之后、 // ctx 取消、或确定性/额度错误则跳出,交给下方原有的透传/故障转移逻辑。 @@ -320,10 +327,10 @@ func streamTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest } } if streamErr != nil && !committed && ctx.Err() == nil && - attempt < sameProviderStreamRetries && isRetryableStreamError(streamErr) { - backoff := sameProviderRetryBackoff(attempt) + attempt < retries && isRetryableStreamError(streamErr) { + backoff := backoffOf(attempt) log.Printf("[task-llm] task %s 提交前流失败,%v 后同 provider 重试 (%d/%d): %v", - taskID, backoff, attempt+1, sameProviderStreamRetries, streamErr) + taskID, backoff, attempt+1, retries, streamErr) if sleepCtx(ctx, backoff) { break // 退避期间 ctx 取消 → 停止重试 } @@ -388,20 +395,35 @@ func streamEventCommitsOutput(event llm.StreamEvent) bool { } } -// 提交前安全窗口内、对同一 provider 的重试次数。SDK 的 doStream 只重试建连阶段 -// (拿到 200 之前);流一旦开始,中途断流 / overloaded / 流内 429 等瞬时故障会直接 +// 提交前安全窗口内、对同一 provider 的【默认】重试次数。SDK 的 doStream 只重试建连 +// 阶段(拿到 200 之前);流一旦开始,中途断流 / overloaded / 流内 429 等瞬时故障会直接 // 冒泡成 model_error,零重试。只要一个 token 都还没交给调用方(!committed),重放 // 完全相同的请求就不会重复模型输出或工具副作用,因此这里补一层同 provider 退避重试, -// 把这类抖动挡在意图整体重跑之前。 +// 把这类抖动挡在意图整体重跑之前。可被 LLM 配置的重试覆盖/全局重试策略改写。 const sameProviderStreamRetries = 2 -// sameProviderRetryBackoff 是第 attempt 次重试前的退避(0.5s、1s…,上限 4s), -// 与 SDK backoffSleep 同风格但封顶更小,避免拖住 worker 的收尾/取消响应。 +// sameProviderRetryBackoff 是第 attempt 次重试前的【默认】退避(0.5s、1s…,上限 4s), +// 与 SDK 的指数梯度同风格但封顶更小,避免拖住 worker 的收尾/取消响应。 // 以变量形式暴露,便于测试将退避置零。 var sameProviderRetryBackoff = func(attempt int) time.Duration { return min(500*time.Millisecond*(1< 0 { + d := r.StreamInterval + backoff = func(int) time.Duration { return d } + } + return retries, backoff +} + // isRetryableStreamError 判断「提交前的流失败」是否值得在同一 provider 上重放。 // 瞬时的传输中断 / 供应商过载 / 限流会自行恢复,可安全重试;而以下三类不重试: // - 额度耗尽:交给 profile 故障转移处理,别在这里白烧重试 diff --git a/web/src/app/(main)/system/llm/_components/retry.tsx b/web/src/app/(main)/system/llm/_components/retry.tsx new file mode 100644 index 0000000..28f4048 --- /dev/null +++ b/web/src/app/(main)/system/llm/_components/retry.tsx @@ -0,0 +1,310 @@ +"use client"; + +// LLM 重试配置的共用件：五层重试各自的「次数 + 间隔」。 +// +// 五层从内到外：建连(SDK) → 空响应(SDK) → 同 provider 安全窗口 → 轮询熔断 → 意图重跑。 +// 前三层跟着端点走，所以每个模型配置都能覆盖全局默认；后两层是进程级的，只有全局一份。 +// +// 所有输入都遵循同一套「留空 = 不配置」语义，与后端 db.RetryRule 一致： +// 次数 空/0 = 用内置默认 | -1 = 关闭这层重试 | >0 = 用这个次数 +// 间隔 空/0 = 用这层原本的指数退避 | >0 = 改用这个固定毫秒间隔 + +import * as React from "react"; + +import { Loader2Icon, SaveIcon } from "lucide-react"; +import { toast } from "sonner"; + +import { Button } from "@/components/ui/button"; +import { Input } from "@/components/ui/input"; +import { Label } from "@/components/ui/label"; +import { api } from "@/lib/api"; +import type { LLMRetryOverride, LLMRetryPolicy, LLMRetryRule } from "@/lib/types"; + +export const ZERO_RULE: LLMRetryRule = { attempts: 0, interval_ms: 0 }; +export const ZERO_OVERRIDE: LLMRetryOverride = { + connect: ZERO_RULE, + empty: ZERO_RULE, + stream: ZERO_RULE, +}; +const ZERO_POLICY: LLMRetryPolicy = { + ...ZERO_OVERRIDE, + breaker: ZERO_RULE, + intent: ZERO_RULE, +}; + +type LayerMeta = { + title: string; + /** 这层重试发生在哪、由谁执行 */ + where: string; + desc: string; + attemptsLabel: string; + /** 次数留空时的默认值，用于占位符 */ + defAttempts: number; + /** 间隔留空时的默认策略，用于占位符 */ + defInterval: string; + /** 次数填 -1 的含义 */ + offHint: string; +}; + +export const RETRY_LAYERS = { + connect: { + title: "建连重试", + where: "SDK · 拿到 200 之前", + desc: "连接重置 / 超时 / 429 / 5xx 等在流开始前的失败，原样重发请求。流一旦开始，中途断流不归这层管。", + attemptsLabel: "重试次数", + defAttempts: 3, + defInterval: "0.5s→1s→2s 指数（封顶 8s）", + offHint: "-1 = 一次都不重试，失败立刻上抛", + }, + empty: { + title: "空响应重试", + where: "SDK · 仅 openai 格式", + desc: "请求正常完成、但一个内容块都没有（正常 stop 且无输出）时重发整个 prompt。max_tokens 截断不算，不会重试。", + attemptsLabel: "重试次数", + defAttempts: 2, + defInterval: "0.5s→1s→2s 指数（封顶 8s）", + offHint: "-1 = 空响应直接原样交出", + }, + stream: { + title: "同 provider 安全窗口重试", + where: "本项目 · 未交付输出前", + desc: "流中途断开、供应商过载、流内 429——只要一个 token 都还没交给调用方，就在同一个配置上重放，不会重复模型输出或工具执行。额度耗尽、上下文过长、4xx 确定性拒绝不重试。", + attemptsLabel: "重试次数", + defAttempts: 2, + defInterval: "0.5s→1s 指数（封顶 4s）", + offHint: "-1 = 断流直接交给外层的意图重跑", + }, + breaker: { + title: "轮询熔断", + where: "本项目 · 进程级，全局一份", + desc: "某个配置连续失败到阈值就进入冷却，冷却期内轮询直接跳过它。余额不足 / 密钥失效这类确定性失败不看阈值，第一次就熔断。", + attemptsLabel: "连续失败几次熔断", + defAttempts: 3, + defInterval: "1min→5min→30min 梯度", + offHint: "-1 = 瞬时失败永不熔断（确定性失败仍然熔断）", + }, + intent: { + title: "意图重跑", + where: "本项目 · 进程级，全局一份", + desc: "worker 以 model_error 收场（provider 故障，内层重试全都用尽）时，整条意图重跑。重跑期间任务被暂停 / 终止 / 进入收尾则立即让位。", + attemptsLabel: "重跑次数", + defAttempts: 2, + defInterval: "固定 3s", + offHint: "-1 = 不重跑，该意图直接判为 blocked", + }, +} satisfies Record; + +type LayerKey = keyof typeof RETRY_LAYERS; + +/** 毫秒的人话，只用于在输入框旁边回显，免得数零。 */ +function humanMs(ms: number) { + if (!Number.isFinite(ms) || ms <= 0) return ""; + if (ms < 1000) return `${ms}ms`; + if (ms < 60_000) return `${Number((ms / 1000).toFixed(2))}s`; + return `${Number((ms / 60_000).toFixed(2))}min`; +} + +/** 受控数字输入：空串 ↔ 0，中间态（"-"、"1e"）原样留在本地，不打扰父级。 */ +function NumField({ + id, + value, + onChange, + placeholder, + min, +}: { + id: string; + value: number; + onChange: (n: number) => void; + placeholder: string; + min: number; +}) { + const [text, setText] = React.useState(value === 0 ? "" : String(value)); + // 父级换了一整套值（读取到策略、切换配置）时跟上；自己敲字时不会走到这里， + // 因为那时 value 已经等于本地文本 parse 后的结果。 + React.useEffect(() => { + const incoming = value === 0 ? "" : String(value); + setText((cur) => (Number(cur || 0) === value ? cur : incoming)); + }, [value]); + return ( + { + setText(e.target.value); + const n = Number(e.target.value); + onChange(e.target.value.trim() === "" || !Number.isFinite(n) ? 0 : Math.trunc(n)); + }} + /> + ); +} + +/** 一层重试的两个旋钮。idPrefix 用来在同一页出现多次时保住 label 的 htmlFor。 */ +export function RetryRuleFields({ + layer, + idPrefix, + value, + onChange, + compact, +}: { + layer: LayerKey; + idPrefix: string; + value: LLMRetryRule; + onChange: (r: LLMRetryRule) => void; + /** true = 配置抽屉里的紧凑版：只留标题与两个输入 */ + compact?: boolean; +}) { + const meta = RETRY_LAYERS[layer]; + const human = humanMs(value.interval_ms); + return ( +
+
+
+ + {meta.where} +
+ {!compact &&

{meta.desc}

} +
+
+
+ + onChange({ ...value, attempts: n })} + /> +
+
+ + onChange({ ...value, interval_ms: n })} + /> + {human ? `固定 ${human}` : meta.defInterval} +
+
+ {!compact &&

留空 = 用默认；{meta.offHint}。

} +
+ ); +} + +/** 模型配置抽屉里的三层覆盖（跟着端点走的那三层）。 */ +export function ProfileRetryFields({ + value, + onChange, +}: { + value: LLMRetryOverride; + onChange: (o: LLMRetryOverride) => void; +}) { + return ( +
+
+ +

+ 只对这个配置生效，覆盖「重试与退避」里的全局默认。每格留空 = 跟随全局；次数填 -1 = 关掉这层重试； + 间隔填了就用固定间隔取代指数退避。熔断与意图重跑是进程级的，只能在全局那页调。 +

+
+ {(["connect", "empty", "stream"] as const).map((k) => ( +
+ onChange({ ...value, [k]: r })} + /> +
+ ))} +
+ ); +} + +/** 「重试与退避」tab：五层的全局默认值。 */ +export function RetryPolicyPanel() { + const [policy, setPolicy] = React.useState(ZERO_POLICY); + const [loading, setLoading] = React.useState(true); + const [saving, setSaving] = React.useState(false); + + const load = React.useCallback(async () => { + setLoading(true); + try { + const p = await api.llmRetryPolicy(); + setPolicy({ ...ZERO_POLICY, ...p }); + } catch (e) { + toast.error(`读取重试策略失败：${(e as Error).message}`); + } finally { + setLoading(false); + } + }, []); + + React.useEffect(() => { + void load(); + }, [load]); + + async function save() { + if (saving) return; + setSaving(true); + try { + // 后端会把越界值夹回区间并回传，直接用回传值刷新，所见即所存。 + const saved = await api.saveLLMRetryPolicy(policy); + setPolicy({ ...ZERO_POLICY, ...saved }); + toast.success("已保存，即时生效（正在跑的这一轮调用仍用旧参数）"); + } catch (e) { + toast.error(`保存失败：${(e as Error).message}`); + } finally { + setSaving(false); + } + } + + const set = (k: LayerKey) => (r: LLMRetryRule) => setPolicy((p) => ({ ...p, [k]: r })); + + if (loading) { + return ( +
+ 读取重试策略… +
+ ); + } + + return ( +
+
+ 一次模型调用的失败会依次经过五层重试，由内到外： + 建连 → 空响应 → 同 provider 安全窗口 → 轮询熔断 → 意图重跑 + 。内层用尽才轮到外层，所以次数是 + 相乘 + 的——把每层都拉满，一次抖动能烧掉几十次请求。 + 全部留空即当前默认值，与没有这页时的行为完全一致。前三层可以在每个模型配置里单独覆盖。 +
+ +
+ {(Object.keys(RETRY_LAYERS) as LayerKey[]).map((k) => ( + + ))} +
+ +
+ + +
+
+ ); +} diff --git a/web/src/app/(main)/system/llm/page.tsx b/web/src/app/(main)/system/llm/page.tsx index 099bcb0..7dc4052 100644 --- a/web/src/app/(main)/system/llm/page.tsx +++ b/web/src/app/(main)/system/llm/page.tsx @@ -25,10 +25,13 @@ import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@ import { Separator } from "@/components/ui/separator"; import { Sheet, SheetContent, SheetDescription, SheetHeader, SheetTitle } from "@/components/ui/sheet"; import { Switch } from "@/components/ui/switch"; +import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs"; import { api } from "@/lib/api"; -import type { LLMPoolMember, LLMPoolStatus, LLMProfile } from "@/lib/types"; +import type { LLMPoolMember, LLMPoolStatus, LLMProfile, LLMRetryOverride } from "@/lib/types"; import { cn } from "@/lib/utils"; +import { ProfileRetryFields, RetryPolicyPanel, ZERO_OVERRIDE } from "./_components/retry"; + // 思考开关(thinking.type)与思考强度(reasoning_effort)是两个【互相独立】的字段， // 各自单独设置——有些接口没有 thinking 字段、只靠强度参数就能激活思考，故需解耦。 // 存库空字符串 = 该字段【不发送】；Radix Select 不接受空 value，故 UI 用 "none" @@ -325,6 +328,7 @@ function ProfileSheet({ const [maxTokens, setMaxTokens] = React.useState("0"); // 单次回复输出上限;0=不发送 const [maxTokensField, setMaxTokensField] = React.useState(NONE); // 上限用哪个字段名;NONE=max_tokens const [sessionHeaderKey, setSessionHeaderKey] = React.useState(""); // 自定义会话头名;空=不发送 + const [retry, setRetry] = React.useState(ZERO_OVERRIDE); // 本配置的重试覆盖;全 0=跟随全局 const [testing, setTesting] = React.useState(false); const [saving, setSaving] = React.useState(false); const [models, setModels] = React.useState([]); @@ -336,9 +340,7 @@ function ProfileSheet({ React.useEffect(() => { if (!open) return; setName(profile?.name ?? ""); - setFormat( - profile?.format === "openai" || profile?.format === "openai-responses" ? profile.format : "anthropic", - ); + setFormat(profile?.format === "openai" || profile?.format === "openai-responses" ? profile.format : "anthropic"); setModel(profile?.model ?? ""); setBaseUrl(profile?.base_url ?? ""); setProxy(profile?.proxy ?? ""); @@ -353,6 +355,7 @@ function ProfileSheet({ setMaxTokens(String(profile?.max_tokens ?? 0)); setMaxTokensField(fromStore(profile?.max_tokens_field)); setSessionHeaderKey(profile?.session_header_key ?? ""); + setRetry(profile?.retry ?? ZERO_OVERRIDE); setApiKey(""); setKeyHint(profile?.api_key_hint ?? ""); setModels([]); @@ -440,6 +443,7 @@ function ProfileSheet({ // 后端也会再做一次同样的归一化，这里只是别让 UI 送出自相矛盾的值。 max_tokens_field: format === "openai" ? toStore(maxTokensField) : "", session_header_key: sessionHeaderKey.trim(), + retry, }); if (isNew) toast.success(`已新建：${name.trim()}（在卡片上「设为激活」以启用）`); else toast.success(profile?.is_default ? "已保存，激活配置即时生效，无需重启" : "已保存"); @@ -583,9 +587,9 @@ function ProfileSheet({ onChange={(e) => setSessionHeaderKey(e.target.value)} />

- 填写头名后，每次请求都会带上这个 HTTP 头，头值自动填为{" "} - 当前会话的 session id（chat 会话如 conv-12、worker 如 exp3-worker-i87）。用于按 session-id - 头做提示缓存 / 粘性路由的网关；同一会话多轮稳定、不同会话互不相同。留空则不发送。 + 填写头名后，每次请求都会带上这个 HTTP 头，头值自动填为 当前会话的 session id（chat 会话如 + conv-12、worker 如 exp3-worker-i87）。用于按 session-id 头做提示缓存 / + 粘性路由的网关；同一会话多轮稳定、不同会话互不相同。留空则不发送。

@@ -758,6 +762,8 @@ function ProfileSheet({ + +
@@ -869,97 +875,112 @@ export default function LLMPage() {
-
- {profiles.map((p) => { - const h = healthOf(p, health.get(p.id)); - return ( - // biome-ignore lint/a11y/useSemanticElements: 卡片内含自己的操作按钮，用原生
- -
- {p.api_key_hint && {p.api_key_hint}} - - {p.rate_per_second}/s · {p.rate_per_minute}/min - - {p.proxy && 代理 {p.proxy}} - {p.reasoning_effort && 思考 {p.reasoning_effort === "off" ? "关" : p.reasoning_effort}} - {/* 轮询相关的两个字段只在轮询开着时才有意义，关着时不占版面 */} - {poolOn && - !p.is_default && - (p.pool_exclude ? 不参与轮询 : 优先级 {p.priority ?? 0})} -
+
+ {p.api_key_hint && {p.api_key_hint}} + + {p.rate_per_second}/s · {p.rate_per_minute}/min + + {p.proxy && 代理 {p.proxy}} + {p.reasoning_effort && ( + 思考 {p.reasoning_effort === "off" ? "关" : p.reasoning_effort} + )} + {/* 轮询相关的两个字段只在轮询开着时才有意义，关着时不占版面 */} + {poolOn && + !p.is_default && + (p.pool_exclude ? 不参与轮询 : 优先级 {p.priority ?? 0})} +
-
- - -
- - - ); - })} - {profiles.length === 0 && ( -
- 还没有模型配置，点击右上角「新建」创建第一个。 +
+ + +
+ + + ); + })} + {profiles.length === 0 && ( +
+ 还没有模型配置，点击右上角「新建」创建第一个。 +
+ )}
- )} - + + + + + + void load()} /> diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index 1ded282..87eac85 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -45,6 +45,8 @@ import type { LLMProfile, LLMRecordDetail, LLMRecordItem, + LLMRetryOverride, + LLMRetryPolicy, LLMTask, MCPServer, MCPTool, @@ -724,6 +726,7 @@ export const api = { max_tokens?: number; // 单次回复输出上限；0=不发送，由服务端默认值决定 max_tokens_field?: string; // ""=max_tokens(默认) | "max_completion_tokens"（仅 openai 格式） session_header_key?: string; // 非空=每次请求带该 HTTP 头，头值=当前会话 session id；""=不发送 + retry?: LLMRetryOverride; // 本配置的重试覆盖；各项留 0 = 跟随全局重试策略 }) => post<{ id: number }>("/llm/profiles", p), deleteLLMProfile: (id: string) => del<{ deleted: number }>(`/llm/profiles/${id}`), activateLLMProfile: (id: string) => post<{ ok: boolean }>("/llm/profiles/active", { id: Number(id) }), @@ -731,6 +734,9 @@ export const api = { llmPool: () => get("/llm/pool"), // 清除熔断，让下一次调用立刻重试该配置；不传 id = 全部清除。 resetLLMPool: (id?: string) => post("/llm/pool/reset", { id: id ? Number(id) : 0 }), + // 全局重试策略（五层各自的次数+间隔）。全 0 = 全部走内置默认。 + llmRetryPolicy: () => get("/llm/retry-policy"), + saveLLMRetryPolicy: (p: LLMRetryPolicy) => post("/llm/retry-policy", p), fetchLLMModels: (provider: string, base_url: string, api_key: string, proxy = "", profile_id?: number) => post<{ ok: boolean; error?: string; models?: string[] }>("/llm/models", { provider, diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index 7dc44cc..fdfffd5 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -916,6 +916,32 @@ export interface LLMProfile { // 自定义会话头名：非空时每次请求带该 HTTP 头，头值=当前会话/意图的 session id。 // ""=不发送。用于按 session-id 头做提示缓存/粘性路由的网关。 session_header_key?: string; + // 本配置对重试的覆盖（建连/空响应/同 provider 安全窗口）。留空/全 0 = 跟随全局策略。 + retry?: LLMRetryOverride; +} + +// ---- LLM 重试策略 ---- +// 一层重试的两个旋钮。两者都是「0 = 未配置」： +// attempts 0=用默认次数 | -1=关闭该层重试 | >0=重试次数 +// interval_ms 0=用默认的指数退避 | >0=改用这个固定毫秒间隔 +export interface LLMRetryRule { + attempts: number; + interval_ms: number; +} + +// 单个 LLM 配置能覆盖的三层（都是「跟着端点走」的重试）。 +export interface LLMRetryOverride { + connect: LLMRetryRule; // 建连重试：连接重置/超时/429/5xx，流开始前 + empty: LLMRetryRule; // 空响应重试：完成但没有任何内容（仅 openai 格式） + stream: LLMRetryRule; // 同 provider 安全窗口重试：未交付输出前的断流重放 +} + +// 全局策略 = 上面三层的默认值 + 两层只有全局的： +// breaker 轮询熔断（attempts=连续几次瞬时失败熔断，interval_ms=固定冷却时长） +// intent 意图重跑（worker 以 model_error 收场后整条意图重跑） +export interface LLMRetryPolicy extends LLMRetryOverride { + breaker: LLMRetryRule; + intent: LLMRetryRule; } // ---- LLM 轮询（故障转移）----