End streamed answers that generate below an upstream's minimum speed

crates/tw-api/msg-codes.txt

@@ -90,6 +90,7 @@ config.manual_model_duplicate
 config.manual_model_padded
 config.manual_model_too_long
 config.manual_model_wildcard
+config.min_speed_range
 config.model_spec_blank_model
 config.model_spec_nothing_set
 config.model_spec_wildcard
@@ -458,6 +459,7 @@ gw.upstream.stream_error
 gw.upstream.stream_exception
 gw.upstream.stream_opening_error
 gw.upstream.timeout
+gw.upstream.too_slow
 gw.upstream.unreachable
 gw.upstream.window_full
 gw.wait.mixed
@@ -654,6 +656,7 @@ t.broke test
 t.limited test
 t.plugin_threw test
 t.prefs test
+t.slow test
 t.upstream test
 t.x test
 yaml.anchor_or_alias

crates/tw-api/src/lib.rs

@@ -1131,7 +1131,18 @@ pub const MSG_CODES: &str = include_str!("../msg-codes.txt");
 /// `security.unknown_event`。新端点 `GET /security/counts`（[`SecurityRuleHitsQuery`] →
 /// [`SecurityRuleHits`] 的列表）：一项防护从某个时刻起每条规则命中了几次，各条记录的 `count`
 /// 加起来，规则表的组头用
-pub const CONTROL_API_VERSION: u32 = 57;
+///
+/// **58 起上游可以设最低生成速度**（配置的 `providers[].min_speed`）：[`ProviderView`] 多了
+/// `min_speed`（[`MinSpeedView`]，不管是空），[`ProviderInput`] 多了 `min_speed`
+/// （[`MinSpeedInput`]，不给就是不管）。流式回答从第一个 token 起持续低于它（最近 `window_secs`
+/// 秒的平均速度，中途 5 秒以上的静默不算），回答以一条 503 的错误结束（Anthropic 的
+/// `overloaded_error`、Responses 的 `response.failed` / `server_error`、Gemini 的 `UNAVAILABLE`），
+/// 请求失败，来源 `upstream`，码 `gw.upstream.too_slow`（`upstream`、`rate`、`min`、`secs`），这家
+/// 记一次失败。[`Event::RequestFailed`] 多了 `tokens_per_sec`：只有这种失败带着，是结束那一刻的
+/// 速度，请求记录里照记。新码：`gw.upstream.too_slow`、`config.min_speed_range`（`upstream`、
+/// `field`、`value`、`min`、`max`）；一键修复修得了它：速度越界删掉 `min_speed`，判定时长越界改回
+/// 20。请求库 SCHEMA 不变
+pub const CONTROL_API_VERSION: u32 = 58;
 
 /// `/live` 的速度看最近多久（见 [`LiveView::tokens_per_sec`]），毫秒
 pub const LIVE_RATE_WINDOW_MS: u64 = 3_000;
@@ -1435,6 +1446,11 @@ pub enum Event {
         /// 失败之前上游在回答里写的模型名（见 `RequestFinished::answered_model`）
         #[serde(default, skip_serializing_if = "Option::is_none")]
         answered_model: Option<String>,
+        /// 生成速度，token/秒。**失败的请求一般没有**，只有因为生成太慢被结束的
+        /// （`gw.upstream.too_slow`，见 [`MinSpeedView`]）有：那是结束那一刻判定窗口里的速度，
+        /// 估出来的，和 [`RunningView::tokens_per_sec`] 同一个估法
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        tokens_per_sec: Option<u32>,
     },
     /// 客户端没等到响应结束就走了（Claude Code 里按一下 Esc）。
     ///
@@ -2711,6 +2727,9 @@ pub struct ProviderView {
     /// 同时最多发给这家几个请求。不限是空
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub max_concurrent: Option<u32>,
+    /// 最低生成速度（配置里的 `min_speed`）。不管是空
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub min_speed: Option<MinSpeedView>,
     /// 用量上限，按配置里的顺序，各带此刻用了多少（发出去还没结束的按估算算）。没设的是空的
     pub limits: Vec<KeyLimitView>,
     /// 配置里的 `balance:`，原样：编辑对话框保存时交回去（[`ProviderInput::balance`]）
@@ -3691,6 +3710,9 @@ pub struct ProviderInput {
     /// 同时最多发给这家几个请求，1 到 1000。不给就是不限
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub max_concurrent: Option<u32>,
+    /// 最低生成速度。不给就是不管
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub min_speed: Option<MinSpeedInput>,
     /// 用量上限，整份替换。不给 = 一条都没有
     #[serde(default)]
     pub limits: Vec<KeyLimitInput>,
@@ -3702,6 +3724,31 @@ pub struct ProviderInput {
     pub balance: Option<BalanceSetting>,
 }
 
+/// 一家上游的最低生成速度（配置的 `providers[].min_speed`）。
+///
+/// 流式回答从第一个 token 起，每秒看一次最近 `window_secs` 秒的平均生成速度（和
+/// [`RunningView::tokens_per_sec`] 同一个估法；中途连续 5 秒以上没有内容的那一段不算），低于
+/// `tokens_per_sec` 就结束回答：客户端收到一条 503 的错误（可以重试），请求失败，码是
+/// `gw.upstream.too_slow`，这家记一次失败。非流式的回答、WebSocket 上的不管
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[cfg_attr(feature = "ts", derive(ts_rs::TS))]
+pub struct MinSpeedView {
+    /// 每秒至少多少 token
+    pub tokens_per_sec: u32,
+    /// 按最近多少秒的平均速度判断。配置里没写的是 20
+    pub window_secs: u32,
+}
+
+/// 保存上游时交的最低生成速度（[`ProviderInput::min_speed`]）。`tokens_per_sec` 是 1 到 1000，
+/// `window_secs` 是 5 到 120（界面显式填好默认的 20）；越界的保存被拒，码是
+/// `config.min_speed_range`
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[cfg_attr(feature = "ts", derive(ts_rs::TS))]
+pub struct MinSpeedInput {
+    pub tokens_per_sec: u32,
+    pub window_secs: u32,
+}
+
 /// 按接口地址自动识别的结果。给编辑中、还没保存的上游显示「自动识别」
 /// 会选成什么。
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -7691,6 +7738,7 @@ mod tests {
                 answered_model: None,
             },
             Event::RequestFailed {
+                tokens_per_sec: None,
                 id: 7,
                 model: String::new(),
                 source: FailureSource::Upstream,
@@ -7842,6 +7890,7 @@ mod tests {
         ));
 
         let silent = Event::RequestFailed {
+            tokens_per_sec: None,
             id: 2,
             model: String::new(),
             source: FailureSource::Upstream,

crates/tw-config/src/lib.rs

@@ -200,6 +200,7 @@ impl Default for Provider {
             pricing: None,
             model_specs: std::collections::BTreeMap::new(),
             max_concurrent: None,
+            min_speed: None,
             limits: Vec::new(),
             disabled: false,
             balance: BalanceSetting::Auto,
@@ -850,6 +851,10 @@ pub struct Provider {
     /// 这家的对话等它空出来，别的请求换下一家（见 `tw_gateway::slots`）
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub max_concurrent: Option<u32>,
+    /// 最低生成速度：流式回答在第一个 token 之后持续低于它，就以一条可重试的错误结束回答（见
+    /// `tw_gateway::slow`）。不写就是不管
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub min_speed: Option<MinSpeed>,
     /// 用量上限：每 N 分钟、每 N 小时、每天、每周、每月最多发给这家多少个请求、多少 token、
     /// 花多少钱。**每一条都要过**，不写就不限。写法和密钥的一样（[`limits`]）。
     ///
@@ -958,6 +963,39 @@ impl From<tw_api::BalanceSetting> for BalanceSetting {
 /// 一家上游的并发上限最多写多少
 pub const MAX_PROVIDER_CONCURRENCY: u32 = 1000;
 
+/// 一家上游的最低生成速度（`providers[].min_speed`）。
+///
+/// **只管流式回答的生成阶段**：从第一个 token 起，每秒看一次最近 `window_secs` 秒的平均速度，
+/// 低于 `tokens_per_sec` 就结束这个回答，客户端收到一条可以重试的错误（503）。速度和流量页、
+/// 菜单栏上的实时速度是同一个估法；中途连续几秒没有内容的那一段不算进去，由无响应超时管。
+///
+/// ```yaml
+/// providers:
+///   - name: local-vllm
+///     min_speed: { tokens_per_sec: 15, window_secs: 20 }
+/// ```
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct MinSpeed {
+    /// 每秒至少多少 token，1 到 [`MIN_SPEED_MAX_RATE`]
+    pub tokens_per_sec: u32,
+    /// 按最近多少秒的平均速度判断，[`MIN_SPEED_WINDOW_SECS`] 之内。不写是
+    /// [`DEFAULT_MIN_SPEED_WINDOW_SECS`]；**写回配置时照写**，等于默认值也写，一眼看得出按多久判断
+    #[serde(default = "d_min_speed_window")]
+    pub window_secs: u32,
+}
+
+/// 最低生成速度最多写多少 token/秒
+pub const MIN_SPEED_MAX_RATE: u32 = 1000;
+/// 最低生成速度的判定时长能写的范围，秒
+pub const MIN_SPEED_WINDOW_SECS: (u32, u32) = (5, 120);
+/// 最低生成速度的判定时长不写时是多少秒
+pub const DEFAULT_MIN_SPEED_WINDOW_SECS: u32 = 20;
+
+fn d_min_speed_window() -> u32 {
+    DEFAULT_MIN_SPEED_WINDOW_SECS
+}
+
 /// 手动添加的一个模型 ID 最多多少个字符。最长的真名字是 Bedrock 应用推理配置的 ARN，
 /// 一百来个字符
 pub const MANUAL_MODEL_MAX: usize = 256;

crates/tw-config/src/repair.rs

@@ -5,6 +5,10 @@
 //! —— 而它们的修法不需要猜用户想写什么。语法错（要看标点）、缺字段（没有默认值可回）、语义错
 //! （整份配置的事）都不碰。
 //!
+//! 语义错里只修一种：上游的最低生成速度越界（`config.min_speed_range`）。速度越界删掉整项
+//! （不管），判定时长越界改回默认的 20 秒 —— 改值而不是删键，因为它常写在行内映射里
+//! （`min_speed: { tokens_per_sec: 15, window_secs: 3 }`），行内的键删不了。
+//!
 //! **修完整份配置读得进来才算数**：删掉一处之后冒出别的错（比如那个字段没有默认值，删了就
 //! 成了缺字段），整个修复就不给。半修好的配置还是读不进来，给了只会让人以为好了。
 //!
@@ -61,16 +65,25 @@ pub fn repair(text: &str) -> Option<Repair> {
             }
             Err(r) => r,
         };
-        let (kind, field) = fixable(&r)?;
+        let Action {
+            kind,
+            field,
+            set,
+            shown,
+        } = fixable(&r, &out)?;
         let path = steps(&field)?;
-        // 行号和原值都按**修之前的原文**找：前面删过行，`r.line` 已经对不上了
-        let before = tw_yaml::find(text, &path).ok();
+        // 行号和原值都按**修之前的原文**找：前面删过行，`r.line` 已经对不上了。删掉整项的，
+        // 原值给人看的是越界的那一个（`shown`）
+        let before = tw_yaml::find(text, &steps(shown.as_deref().unwrap_or(&field))?).ok();
         let line = before
             .as_ref()
             .map(|f| text[..f.bytes.start].matches('\n').count() + 1)
             .or(r.line);
         let value = before.map(|f| masked(&path, &f.value));
-        out = tw_yaml::remove_key(&out, &path).ok()?;
+        out = match set {
+            Some(v) => tw_yaml::set(&out, &path, &v).ok()?,
+            None => tw_yaml::remove_key(&out, &path).ok()?,
+        };
         found.push(Fix {
             kind,
             field,
@@ -97,14 +110,57 @@ fn masked(path: &[Step], v: &str) -> String {
     }
 }
 
-/// 这一处能不能修，能修的话是哪一种、哪个字段
-fn fixable(r: &Rejected) -> Option<(FixKind, String)> {
+/// 一处怎么修
+struct Action {
+    kind: FixKind,
+    /// 改的是哪个字段
+    field: String,
+    /// 改成什么。`None` 是删掉
+    set: Option<tw_yaml::Scalar>,
+    /// 原值和行号从哪个字段读。`None` 就是 `field`
+    shown: Option<String>,
+}
+
+/// 这一处能不能修，能修的话怎么修。`text` 是此刻的原文
+fn fixable(r: &Rejected, text: &str) -> Option<Action> {
     let kind = match r.message.code.as_str() {
         "config.unknown_variant" => FixKind::UnknownValue,
         "config.unknown_field" => FixKind::UnknownField,
+        "config.min_speed_range" => return min_speed(r, text),
         _ => return None,
     };
-    Some((kind, r.message.args.get("field")?.clone()))
+    Some(Action {
+        kind,
+        field: r.message.args.get("field")?.clone(),
+        set: None,
+        shown: None,
+    })
+}
+
+/// 最低生成速度越界：速度不对删掉整项，判定时长不对改回默认值。上游按名字找它在第几个
+/// （到了校验这一步，原文是读得进来的）
+fn min_speed(r: &Rejected, text: &str) -> Option<Action> {
+    let cfg: Config = serde_yaml_ng::from_str(text).ok()?;
+    let upstream = r.message.args.get("upstream")?;
+    let i = cfg.providers.iter().position(|p| &p.name == upstream)?;
+    let at = format!("providers[{i}].min_speed");
+    Some(match r.message.args.get("field")?.as_str() {
+        "tokens_per_sec" => Action {
+            kind: FixKind::UnknownValue,
+            shown: Some(format!("{at}.tokens_per_sec")),
+            field: at,
+            set: None,
+        },
+        "window_secs" => Action {
+            kind: FixKind::UnknownValue,
+            field: format!("{at}.window_secs"),
+            set: Some(tw_yaml::Scalar::Int(i64::from(
+                crate::DEFAULT_MIN_SPEED_WINDOW_SECS,
+            ))),
+            shown: None,
+        },
+        _ => return None,
+    })
 }
 
 /// 取值改回默认值的那几处，查出默认值是什么，给人看
@@ -355,6 +411,59 @@ mod tests {
         assert!(!shown.contains("abcdefghijklmnopqrstuvwxyz"), "{shown}");
     }
 
+    /// 最低生成速度越界：速度不对删掉整项，判定时长不对改回 20（行内映射里也改得了）
+    #[test]
+    fn a_minimum_speed_out_of_range_is_dropped_or_its_window_reset() {
+        let ups = |min_speed: &str| {
+            format!(
+                "{GOOD}providers:\n  - name: a\n    base_url: https://x\n  - name: b\n    base_url: https://y\n{min_speed}"
+            )
+        };
+        let bad = ups("    min_speed: { tokens_per_sec: 0, window_secs: 20 }\n");
+        let r = repair(&bad).expect("该修得了");
+        assert_eq!(
+            r.fixes,
+            vec![Fix {
+                kind: FixKind::UnknownValue,
+                field: "providers[1].min_speed".into(),
+                line: Some(13),
+                value: Some("0".into()),
+                now: None,
+            }]
+        );
+        assert_eq!(r.text, ups(""));
+        assert!(try_parse(&r.text).unwrap().providers[1].min_speed.is_none());
+
+        let bad = ups("    min_speed: { tokens_per_sec: 15, window_secs: 3 }\n");
+        let r = repair(&bad).expect("该修得了");
+        assert_eq!(
+            r.fixes,
+            vec![Fix {
+                kind: FixKind::UnknownValue,
+                field: "providers[1].min_speed.window_secs".into(),
+                line: Some(13),
+                value: Some("3".into()),
+                now: Some("20".into()),
+            }]
+        );
+        assert_eq!(
+            r.text,
+            ups("    min_speed: { tokens_per_sec: 15, window_secs: 20 }\n")
+        );
+
+        // 块式的照样改
+        let bad = ups("    min_speed:\n      tokens_per_sec: 15\n      window_secs: 999\n");
+        let r = repair(&bad).expect("该修得了");
+        assert_eq!(r.fixes[0].line, Some(15));
+        assert_eq!(
+            try_parse(&r.text).unwrap().providers[1].min_speed,
+            Some(crate::MinSpeed {
+                tokens_per_sec: 15,
+                window_secs: 20
+            })
+        );
+    }
+
     #[test]
     fn paths_with_indices_are_understood_and_odd_ones_are_left_alone() {
         assert_eq!(