From c2382325cbd00b19001648e7d18229362f5213fb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com> Date: Wed, 7 Oct 2026 18:33:41 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E:=20=E5=BC=B1=E7=BD=91?= =?UTF-8?q?=E9=9F=A7=E6=80=A7:=20send=20=E7=BD=91=E7=BB=9C=E6=84=9F?= =?UTF-8?q?=E7=9F=A5=E9=87=8D=E8=AF=95=E4=B8=8E=E5=9F=BA=E5=87=86=E9=9D=A2?= =?UTF-8?q?=E6=8E=A2=E9=92=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - connect 类失败重试: 10 分钟窗口+基准面感知(IP 直连与 DNS 双测判本机断网), 断网进长等待自适应探测(5s 快探到 30s 封顶省电), 恢复即重试; 单点故障指数短退避 1s→8s - 两协议(anthropic/openai)同构, 隧道级分钟级弱网不再直接吐中断 --- crates/df-ai/src/anthropic_compat.rs | 129 ++++++++++++++++++--------- crates/df-ai/src/lib.rs | 1 + crates/df-ai/src/net_probe.rs | 64 +++++++++++++ crates/df-ai/src/openai_compat.rs | 108 +++++++++++++++------- 4 files changed, 230 insertions(+), 72 deletions(-) create mode 100644 crates/df-ai/src/net_probe.rs diff --git a/crates/df-ai/src/anthropic_compat.rs b/crates/df-ai/src/anthropic_compat.rs index 9d23bb2..40b6ca0 100644 --- a/crates/df-ai/src/anthropic_compat.rs +++ b/crates/df-ai/src/anthropic_compat.rs @@ -743,47 +743,94 @@ impl LlmProvider for AnthropicCompatProvider { // 流式长生成任务会被误砍。改用 tokio::time::timeout // 包裹 send().await,只管建连+首响应头,不管后续 body 读取(后续由 stream_llm idle timeout 兜底)。 // 60s 选型:正常 send(建连+收 200 headers)<5s,60s 足够宽容。 - let send_future = self - .auth_headers(self.client.post(self.messages_url())) - .json(&body) - .version(reqwest::Version::HTTP_11) - .send(); - let resp = match tokio::time::timeout(Duration::from_secs(60), send_future).await { - Ok(Ok(r)) => r, - Ok(Err(e)) => { - // 谓词结构化分类(对齐 openai 路径,共用 crate::send_error_class)。原 bail 串 - // [timeout={} connect={}=...] 的字面词让一切错误(含 builder 错)侥幸可重试。 - // 类词改由谓词决定置串首;谓词转储缩写为 t/c/b/r 防误触分类词表,source 链 - // 保留真实传输词(reset/broken pipe 等)作串兜底。 - let class = crate::send_error_class(crate::SendErrPreds::of(&e)); - tracing::error!( - is_timeout = e.is_timeout(), - is_connect = e.is_connect(), - is_body = e.is_body(), - is_request = e.is_request(), - class = class, - url = %self.messages_url(), - source = ?std::error::Error::source(&e), - "Anthropic 流式 send 失败" - ); - anyhow::bail!( - "send 失败({}): {} [t={} c={} b={} r={} src={:?}]", - class, - e, - e.is_timeout(), - e.is_connect(), - e.is_body(), - e.is_request(), - std::error::Error::source(&e) - ); - } - Err(_elapsed) => { - // send 阶段超时(60s 未返回 HTTP 响应头):GLM 端点可能不可达或极慢 - tracing::error!( - url = %self.messages_url(), - "Anthropic 流式 send 超时(60s 未返回响应头)" - ); - anyhow::bail!("流式请求超时(60秒未收到 HTTP 响应，可能服务不可达或被防火墙拦截)"); + // 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 + .auth_headers(self.client.post(self.messages_url())) + .json(&body) + .version(reqwest::Version::HTTP_11) + .send(); + let outcome = tokio::time::timeout(Duration::from_secs(60), send_future).await; + match outcome { + Ok(Ok(r)) => break r, + Ok(Err(e)) => { + // 谓词结构化分类(对齐 openai 路径,共用 crate::send_error_class)。原 bail 串 + // [timeout={} connect={}=...] 的字面词让一切错误(含 builder 错)侥幸可重试。 + // 类词改由谓词决定置串首;谓词转储缩写为 t/c/b/r 防误触分类词表,source 链 + // 保留真实传输词(reset/broken pipe 等)作串兜底。 + 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!( + is_timeout = e.is_timeout(), + is_connect = e.is_connect(), + is_body = e.is_body(), + is_request = e.is_request(), + class = class, + url = %self.messages_url(), + source = ?std::error::Error::source(&e), + "Anthropic 流式 send 失败" + ); + anyhow::bail!( + "send 失败({}): {} [t={} c={} b={} r={} src={:?}]", + class, + e, + e.is_timeout(), + e.is_connect(), + e.is_body(), + e.is_request(), + std::error::Error::source(&e) + ); + } + Err(_elapsed) => { + // send 阶段超时(60s 未返回 HTTP 响应头):GLM 端点可能不可达或极慢 + tracing::error!( + url = %self.messages_url(), + "Anthropic 流式 send 超时(60s 未返回响应头)" + ); + anyhow::bail!("流式请求超时(60秒未收到 HTTP 响应，可能服务不可达或被防火墙拦截)"); + } + } } }; diff --git a/crates/df-ai/src/lib.rs b/crates/df-ai/src/lib.rs index a11ad00..6882b9b 100644 --- a/crates/df-ai/src/lib.rs +++ b/crates/df-ai/src/lib.rs @@ -11,6 +11,7 @@ pub mod anthropic_helpers; pub mod context; pub mod context_helpers; pub mod memory_adapter; +pub(crate) mod net_probe; // 多 Agent 协作调度中心(Phase 1 规则驱动, Phase 2 LLM 扩展)。 pub mod coordinator; // 会话意图识别层(纯函数,不接入 agentic loop)。依据 docs/02-架构设计/构想审查/ diff --git a/crates/df-ai/src/net_probe.rs b/crates/df-ai/src/net_probe.rs new file mode 100644 index 0000000..f219c2b --- /dev/null +++ b/crates/df-ai/src/net_probe.rs @@ -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(_)) + ) +} diff --git a/crates/df-ai/src/openai_compat.rs b/crates/df-ai/src/openai_compat.rs index 93fcdf5..69d9fac 100644 --- a/crates/df-ai/src/openai_compat.rs +++ b/crates/df-ai/src/openai_compat.rs @@ -593,37 +593,83 @@ impl LlmProvider for OpenAICompatProvider { // send 阶段需 timeout 防 hang(同 Anthropic 路径)。 // 不能用 reqwest .timeout()(会砍流式 body),改用 tokio::time::timeout 包裹 send。 - let send_future = self - .client - .post(self.chat_url()) - .header("Authorization", format!("Bearer {}", self.api_key)) - .header("Content-Type", "application/json") - .json(&openai_req) - .send(); - let resp = match tokio::time::timeout(Duration::from_secs(60), send_future).await { - Ok(Ok(r)) => r, - Ok(Err(e)) => { - // 谓词结构化分类:Display 串不含 source 关键词("error sending request for url"), - // 纯串分类必 unknown → Fatal 零重试(陈旧连接首包 RST 就此裸奔)。类词置串首, - // source 链兜底真实传输词(reset/broken pipe 等),下游按类词走可重试。 - let class = crate::send_error_class(crate::SendErrPreds::of(&e)); - tracing::error!( - error = %e, - class = class, - url = %self.chat_url(), - source = ?std::error::Error::source(&e), - "OpenAI 流式 send 失败" - ); - anyhow::bail!( - "流式 send 失败({}): {} [src={:?}]", - class, - e, - std::error::Error::source(&e) - ); - } - Err(_elapsed) => { - tracing::error!(url = %self.chat_url(), "OpenAI 流式 send 超时(60s 未返回响应头)"); - anyhow::bail!("流式请求超时(60秒未收到 HTTP 响应，可能服务不可达或被防火墙拦截)"); + // 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 + .client + .post(self.chat_url()) + .header("Authorization", format!("Bearer {}", self.api_key)) + .header("Content-Type", "application/json") + .json(&openai_req) + .send(); + let outcome = tokio::time::timeout(Duration::from_secs(60), send_future).await; + match outcome { + Ok(Ok(r)) => break r, + Ok(Err(e)) => { + // 谓词结构化分类:Display 串不含 source 关键词("error sending request for url"), + // 纯串分类必 unknown → Fatal 零重试(陈旧连接首包 RST 就此裸奔)。类词置串首, + // source 链兜底真实传输词(reset/broken pipe 等),下游按类词走可重试。 + 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!( + error = %e, + class = class, + url = %self.chat_url(), + source = ?std::error::Error::source(&e), + "OpenAI 流式 send 失败" + ); + anyhow::bail!( + "流式 send 失败({}): {} [src={:?}]", + class, + e, + std::error::Error::source(&e) + ); + } + Err(_elapsed) => { + tracing::error!(url = %self.chat_url(), "OpenAI 流式 send 超时(60s 未返回响应头)"); + anyhow::bail!("流式请求超时(60秒未收到 HTTP 响应，可能服务不可达或被防火墙拦截)"); + } + } } };