新增: 弱网韧性: send 网络感知重试与基准面探针

- connect 类失败重试: 10 分钟窗口+基准面感知(IP 直连与 DNS 双测判本机断网), 断网进长等待自适应探测(5s 快探到 30s 封顶省电), 恢复即重试; 单点故障指数短退避 1s→8s
- 两协议(anthropic/openai)同构, 隧道级分钟级弱网不再直接吐中断
This commit is contained in:
lxy
2026-10-07 18:33:41 +08:00
parent 76ace953be
commit c2382325cb
4 changed files with 230 additions and 72 deletions
+50 -3
View File
@@ -743,19 +743,64 @@ impl LlmProvider for AnthropicCompatProvider {
// 流式长生成任务会被误砍。改用 tokio::time::timeout // 流式长生成任务会被误砍。改用 tokio::time::timeout
// 包裹 send().await,只管建连+首响应头,不管后续 body 读取(后续由 stream_llm idle timeout 兜底)。 // 包裹 send().await,只管建连+首响应头,不管后续 body 读取(后续由 stream_llm idle timeout 兜底)。
// 60s 选型:正常 send(建连+收 200 headers)<5s,60s 足够宽容。 // 60s 选型:正常 send(建连+收 200 headers)<5s,60s 足够宽容。
// send 阶段弱网退避: connect 类失败(隧道断网/基站切换/DNS 抖动)请求未发出, 重试零副作用。
// 网络感知分速(net_probe 基准面): 本机断网进长等待按自适应间隔探测恢复(隧道级分钟级波动),
// 单点故障指数短退避 1s→8s; 10 分钟窗口耗尽才放弃。timeout/decode/HTTP 状态错不重试。
// 与 openai_compat 同构。
let resp = {
let start = std::time::Instant::now();
let mut offline_logged = false;
let mut offline_polls = 0usize;
let mut attempt = 0usize;
loop {
attempt += 1;
let send_future = self let send_future = self
.auth_headers(self.client.post(self.messages_url())) .auth_headers(self.client.post(self.messages_url()))
.json(&body) .json(&body)
.version(reqwest::Version::HTTP_11) .version(reqwest::Version::HTTP_11)
.send(); .send();
let resp = match tokio::time::timeout(Duration::from_secs(60), send_future).await { let outcome = tokio::time::timeout(Duration::from_secs(60), send_future).await;
Ok(Ok(r)) => r, match outcome {
Ok(Ok(r)) => break r,
Ok(Err(e)) => { Ok(Err(e)) => {
// 谓词结构化分类(对齐 openai 路径,共用 crate::send_error_class)。原 bail 串 // 谓词结构化分类(对齐 openai 路径,共用 crate::send_error_class)。原 bail 串
// [timeout={} connect={}=...] 的字面词让一切错误(含 builder 错)侥幸可重试。 // [timeout={} connect={}=...] 的字面词让一切错误(含 builder 错)侥幸可重试。
// 类词改由谓词决定置串首;谓词转储缩写为 t/c/b/r 防误触分类词表,source 链 // 类词改由谓词决定置串首;谓词转储缩写为 t/c/b/r 防误触分类词表,source 链
// 保留真实传输词(reset/broken pipe 等)作串兜底。 // 保留真实传输词(reset/broken pipe 等)作串兜底。
let class = crate::send_error_class(crate::SendErrPreds::of(&e)); let preds = crate::SendErrPreds::of(&e);
let retryable = preds.is_connect || (preds.is_request && !preds.is_builder);
if retryable && start.elapsed() < crate::net_probe::SEND_RETRY_WINDOW {
// 本机断网: 等基准面恢复(自适应间隔), 不消耗退避档位
if !crate::net_probe::baseline_alive().await {
offline_polls += 1;
let wait = crate::net_probe::offline_probe_wait(offline_polls);
if !offline_logged {
warn!(
url = %self.messages_url(),
window_secs = crate::net_probe::SEND_RETRY_WINDOW.as_secs(),
"本机网络不可达(疑似隧道/电梯弱网), 按自适应间隔探测恢复, 恢复后立即重试"
);
offline_logged = true;
}
tokio::time::sleep(wait).await;
continue;
}
offline_polls = 0;
offline_logged = false;
let delay = Duration::from_millis(
1000u64.saturating_mul(1 << (attempt - 1).min(3)).min(8000),
);
warn!(
url = %self.messages_url(),
attempt,
delay_ms = delay.as_millis() as u64,
error = %e,
"流式 send 连接失败, 退避重试"
);
tokio::time::sleep(delay).await;
continue;
}
let class = crate::send_error_class(preds);
tracing::error!( tracing::error!(
is_timeout = e.is_timeout(), is_timeout = e.is_timeout(),
is_connect = e.is_connect(), is_connect = e.is_connect(),
@@ -785,6 +830,8 @@ impl LlmProvider for AnthropicCompatProvider {
); );
anyhow::bail!("流式请求超时(60秒未收到 HTTP 响应,可能服务不可达或被防火墙拦截)"); anyhow::bail!("流式请求超时(60秒未收到 HTTP 响应,可能服务不可达或被防火墙拦截)");
} }
}
}
}; };
if !resp.status().is_success() { if !resp.status().is_success() {
+1
View File
@@ -11,6 +11,7 @@ pub mod anthropic_helpers;
pub mod context; pub mod context;
pub mod context_helpers; pub mod context_helpers;
pub mod memory_adapter; pub mod memory_adapter;
pub(crate) mod net_probe;
// 多 Agent 协作调度中心(Phase 1 规则驱动, Phase 2 LLM 扩展)。 // 多 Agent 协作调度中心(Phase 1 规则驱动, Phase 2 LLM 扩展)。
pub mod coordinator; pub mod coordinator;
// 会话意图识别层(纯函数,不接入 agentic loop)。依据 docs/02-架构设计/构想审查/ // 会话意图识别层(纯函数,不接入 agentic loop)。依据 docs/02-架构设计/构想审查/
+64
View File
@@ -0,0 +1,64 @@
//! 本机连通性基准面探测(弱网重试的感知依据)。
//!
//! 场景: 隧道/电梯等移动网络波动可达分钟级, 固定次数重试必然失败。send 失败时探
//! 基准面区分「本机断网」与「单点故障」: 前者进长等待(自适应间隔探测, 恢复即重试),
//! 后者短退避即可。纯 IP TCP 与系统 DNS 解析双测, 任一通过即判在线
//! (DNS 单挂时 IP 直连上游仍可能可用, 双测互为盲区互补)。
//!
//! 演进路线(P2, 需求触发再上): 系统网络事件订阅双通道——事件源治接口级变化
//! (WiFi 断开/切蜂窝)零延迟, 本探针治链路级不通(隧道半断/DNS 故障, 系统事件
//! 不触发的盲区)。跨平台无统一事件 crate(Windows=INetworkListManager COM,
//! macOS=NWPathMonitor, Linux=netlink), 届时以 NetEventSource trait 抽象,
//! 按 devflow 实际部署面(Windows 桌面+Linux 云端)逐平台实现; Linux 服务端
//! 网络稳定, 纯探针已足够。
use std::net::ToSocketAddrs;
use std::time::Duration;
/// 探测超时(单次解析或建连)。弱网下慢响应不代表断网, 2s 内完成才计可用。
const PROBE_TIMEOUT: Duration = Duration::from_secs(2);
/// send 弱网重试总窗口(隧道/电梯级分钟级波动覆盖)。
pub(crate) const SEND_RETRY_WINDOW: Duration = Duration::from_secs(600);
/// 脱网确认后的探针间隔表(秒): 首段 5s 快探保即时性(网络抖动常秒级恢复),
/// 确认持续脱网后间隔拉长到 30s 封顶——蜂窝 radio 休眠省电, 分钟级隧道场景无感差异。
pub(crate) const OFFLINE_PROBE_SCHEDULE_SECS: &[u64] = &[5, 5, 5, 5, 10, 15, 30];
/// 按脱网轮次取探针间隔(越界取表尾封顶值)。
pub(crate) fn offline_probe_wait(poll_count: usize) -> Duration {
let idx = poll_count.saturating_sub(1).min(OFFLINE_PROBE_SCHEDULE_SECS.len() - 1);
Duration::from_secs(OFFLINE_PROBE_SCHEDULE_SECS[idx])
}
/// IP 直连探针目标(公共 DNS 服务商的 443, 不依赖 DNS)。
const PROBE_IPS: &[&str] = &["223.5.5.5:443", "119.29.29.29:443"];
/// DNS 面探针目标(公共域, 走系统 resolver)。
const PROBE_DNS_HOST: &str = "www.aliyun.com:443";
async fn tcp_alive(addr: &str) -> bool {
let target = match addr.to_socket_addrs() {
Ok(mut it) => match it.next() {
Some(a) => a,
None => return false,
},
Err(_) => return false,
};
tokio::time::timeout(PROBE_TIMEOUT, tokio::net::TcpStream::connect(target))
.await
.is_ok()
}
/// 本机网络基准面是否可用。任一探针通过即在线。
pub(crate) async fn baseline_alive() -> bool {
for ip in PROBE_IPS {
if tcp_alive(ip).await {
return true;
}
}
matches!(
tokio::time::timeout(PROBE_TIMEOUT, tokio::net::lookup_host(PROBE_DNS_HOST)).await,
Ok(Ok(_))
)
}
+49 -3
View File
@@ -593,6 +593,16 @@ impl LlmProvider for OpenAICompatProvider {
// send 阶段需 timeout 防 hang(同 Anthropic 路径)。 // send 阶段需 timeout 防 hang(同 Anthropic 路径)。
// 不能用 reqwest .timeout()(会砍流式 body),改用 tokio::time::timeout 包裹 send。 // 不能用 reqwest .timeout()(会砍流式 body),改用 tokio::time::timeout 包裹 send。
// send 阶段弱网退避: connect 类失败重试零副作用; 网络感知分速(net_probe 基准面):
// 本机断网长等待按自适应间隔探测恢复, 单点故障指数短退避 1s→8s, 10 分钟窗口耗尽才放弃。
// 与 anthropic_compat 同构。
let resp = {
let start = std::time::Instant::now();
let mut offline_logged = false;
let mut offline_polls = 0usize;
let mut attempt = 0usize;
loop {
attempt += 1;
let send_future = self let send_future = self
.client .client
.post(self.chat_url()) .post(self.chat_url())
@@ -600,13 +610,47 @@ impl LlmProvider for OpenAICompatProvider {
.header("Content-Type", "application/json") .header("Content-Type", "application/json")
.json(&openai_req) .json(&openai_req)
.send(); .send();
let resp = match tokio::time::timeout(Duration::from_secs(60), send_future).await { let outcome = tokio::time::timeout(Duration::from_secs(60), send_future).await;
Ok(Ok(r)) => r, match outcome {
Ok(Ok(r)) => break r,
Ok(Err(e)) => { Ok(Err(e)) => {
// 谓词结构化分类:Display 串不含 source 关键词("error sending request for url"), // 谓词结构化分类:Display 串不含 source 关键词("error sending request for url"),
// 纯串分类必 unknown → Fatal 零重试(陈旧连接首包 RST 就此裸奔)。类词置串首, // 纯串分类必 unknown → Fatal 零重试(陈旧连接首包 RST 就此裸奔)。类词置串首,
// source 链兜底真实传输词(reset/broken pipe 等),下游按类词走可重试。 // source 链兜底真实传输词(reset/broken pipe 等),下游按类词走可重试。
let class = crate::send_error_class(crate::SendErrPreds::of(&e)); let preds = crate::SendErrPreds::of(&e);
let retryable = preds.is_connect || (preds.is_request && !preds.is_builder);
if retryable && start.elapsed() < crate::net_probe::SEND_RETRY_WINDOW {
// 本机断网: 等基准面恢复, 自适应间隔探测(首段 5s 快探, 确认脱网后 30s 封顶省电)
if !crate::net_probe::baseline_alive().await {
offline_polls += 1;
let wait = crate::net_probe::offline_probe_wait(offline_polls);
if !offline_logged {
warn!(
url = %self.chat_url(),
window_secs = crate::net_probe::SEND_RETRY_WINDOW.as_secs(),
"本机网络不可达(疑似隧道/电梯弱网), 按自适应间隔探测恢复, 恢复后立即重试"
);
offline_logged = true;
}
tokio::time::sleep(wait).await;
continue;
}
offline_polls = 0;
offline_logged = false;
let delay = Duration::from_millis(
1000u64.saturating_mul(1 << (attempt - 1).min(3)).min(8000),
);
warn!(
url = %self.chat_url(),
attempt,
delay_ms = delay.as_millis() as u64,
error = %e,
"流式 send 连接失败, 退避重试"
);
tokio::time::sleep(delay).await;
continue;
}
let class = crate::send_error_class(preds);
tracing::error!( tracing::error!(
error = %e, error = %e,
class = class, class = class,
@@ -625,6 +669,8 @@ impl LlmProvider for OpenAICompatProvider {
tracing::error!(url = %self.chat_url(), "OpenAI 流式 send 超时(60s 未返回响应头)"); tracing::error!(url = %self.chat_url(), "OpenAI 流式 send 超时(60s 未返回响应头)");
anyhow::bail!("流式请求超时(60秒未收到 HTTP 响应,可能服务不可达或被防火墙拦截)"); anyhow::bail!("流式请求超时(60秒未收到 HTTP 响应,可能服务不可达或被防火墙拦截)");
} }
}
}
}; };
if !resp.status().is_success() { if !resp.status().is_success() {