lxy/DevFlow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重构: AI流式断线保文StreamResult三分支+重试对齐决策a1,推进链落df-nodes

Browse Source
This commit is contained in:
绝尘
2026-06-16 19:11:19 +08:00
parent 7d5cd4c89a
commit ba7f35552b
30 changed files with 1325 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files

277
src-tauri/src/commands/ai/stream_recv.rs
View File

@@ -92,12 +92,42 @@ pub(crate) fn fmt_diag(provider_name: &str, kind: DiagKind, status_or_class: &st
format!("[{}] {}({}): {}", provider_name, ctx, status_or_class, raw)
}
/// 流式接收 LLM 响应，返回 (完整文本, 工具调用草稿)
/// 流式接收结果。显式区分三类出口，避免调用方对 None 的歧义推断
/// (UX-2025-04 / CR-30-1 / CR-30-2 / 决策 F-260616-07 a1)。
pub(crate) enum StreamResult {
/// 正常完成（流尽 + finished 信号，或用户主动停止）。incomplete=false。
/// 调用方按正常流程入库 + emit AiCompleted。
Complete {
text: String,
tool_calls: HashMap<u32, ToolCallDraft>,
usage: df_ai::provider::TokenUsage,
},
/// 流中途断（MidStream chunk Err / idle timeout / provider stream-error / 有 partial_text
/// 但流尽未收到 finished）。已 emit 过 AiTextDelta（前端 currentText 已累积），
/// **不 emit AiError**——调用方保文入库 + emit AiCompleted(incomplete=true) + 系统提示。
/// 调用方**不重试**（决策 a1）。
Partial {
text: String,
tool_calls: HashMap<u32, ToolCallDraft>,
usage: df_ai::provider::TokenUsage,
},
/// Init 失败（provider.stream() Err：建连/鉴权/HTTP non-2xx）。已 emit AiError。
/// `retryable`: 据状态码分类(retry::is_status_retryable 镜像)：true=5xx/429/timeout/connect
/// 可重试；false=4xx(非429)/鉴权/参数错 Fatal 立即放弃。调用方据此决定是否重试。
InitFailed { retryable: bool },
}
/// 流式接收 LLM 响应。
///
/// 三类异常处理：
/// - idle timeout（120s 无 chunk）：判定连接静默断，emit AiError 返回 None
/// - 流尽但从未收到 finished 信号：判定异常中断，emit AiError 返回 None（丢弃残缺，不当完整入库）
/// - 用户停止（stop_flag）：break 返回 Some(已收文本)，由调用方入库展示后退出
/// 三类异常处理（返回 StreamResult 显式区分出口）：
/// - Init 失败（provider.stream() Err）：emit AiError + 返回 InitFailed{retryable}。
/// retryable 据状态码分类(retry::is_status_retryable 镜像)：5xx/429/timeout/connect=true，
/// 4xx(非429)/鉴权/参数错=false Fatal。调用方走流前重试(仅 retryable=true,≤max_retries 次,
/// 复用 retry::backoff_delay 退避 + 30s 总预算)。
/// - MidStream 失败（流已建立后断/timeout/err 事件/有 partial_text 但流尽未 finished）：
/// **不 emit AiError**，返回 Partial{...} 保文。调用方不重试，入库 + AiCompleted(incomplete)
/// + 系统提示网络中断。空文本仍 emit AiError + InitFailed{retryable=true}(无文可保,交重试)。
/// - 用户停止（stop_flag）：返回 Complete（incomplete 语义非异常中断）。
///
/// 心跳与停止响应（B-260615-02 / B-260615-04）：单 `tokio::select!` 三分支
/// - `stream.next()`：正常 chunk 处理
@@ -111,7 +141,7 @@ pub(crate) async fn stream_llm(
stop_flag: &AtomicBool,
notify: &tokio::sync::Notify,
conv_id: &str,
) -> Option<(String, HashMap<u32, ToolCallDraft>, df_ai::provider::TokenUsage)> {
) -> StreamResult {
/// 流式读取空闲超时：超过此时长无任何 chunk 即判定连接已断
const STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(120);
/// 心跳间隔：静默期向前端报「LLM 仍在跑」，reset watchdog
@@ -157,13 +187,28 @@ pub(crate) async fn stream_llm(
chunk_result = tokio::time::timeout(STREAM_IDLE_TIMEOUT, stream.next()) => {
match chunk_result {
Err(_elapsed) => {
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
error: "流式响应超时（120 秒无数据，连接可能已断开）".to_string(),
// 120s 无 chunk = idle timeout,归 Timeout
error_type: Some(ErrorType::Timeout),
conversation_id: Some(conv_id.to_string()),
});
return None;
// UX-2025-04 / CR-30-2 / 决策 F-260616-07 a1:
// idle timeout 属 MidStream 类失败——已有文本则保文(Partial),不重试。
// 空文本无文可保,emit AiError + InitFailed{retryable=true}(timeout 瞬态可重试)。
if full_text.is_empty() && tool_calls_acc.is_empty() {
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
error: "流式响应超时（120 秒无数据，连接可能已断开）".to_string(),
error_type: Some(ErrorType::Timeout),
conversation_id: Some(conv_id.to_string()),
});
return StreamResult::InitFailed { retryable: true };
}
warn!(
provider = %provider.name(),
conv_id = %conv_id,
text_len = full_text.len(),
"[ai] 流中途 idle timeout,保文不重试(incomplete)",
);
return StreamResult::Partial {
text: full_text,
tool_calls: tool_calls_acc,
usage: final_usage.unwrap_or_default(),
};
}
Ok(None) => break, // 流正常结束
Ok(Some(chunk_result)) => match chunk_result {
@@ -187,27 +232,35 @@ pub(crate) async fn stream_llm(
final_usage = Some(u.clone());
}
// provider 流式错误事件（Anthropic SSE `type=="error"` 等）：
// 不走 finished 完成路径，发 AiError + 丢弃残缺，与 Err 分支对齐（OpenAI 路径一致性）。
// UX-2025-04 / CR-30-2 / 决策 a1: MidStream 类失败——已有文本则保文(Partial),
// 不重试。空文本无文可保,emit AiError + InitFailed(保守 retryable=true,
// err_msg 不可靠解析状态码,默认按可重试交调用方决定)。
if let Some(err_msg) = &chunk.error {
warn!(
provider = %provider.name(),
conv_id = %conv_id,
error = %err_msg,
text_len = full_text.len(),
"[ai] provider 流式错误事件",
);
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
error: fmt_diag(
provider.name(),
DiagKind::MidStream,
"stream-error",
err_msg,
),
// provider SSE error 事件体可为 401/overloaded 等多种,
// 从文本分类不可靠,归 Unknown(前端可据 error 文本二次判断)
error_type: Some(ErrorType::Unknown),
conversation_id: Some(conv_id.to_string()),
});
return None;
if full_text.is_empty() && tool_calls_acc.is_empty() {
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
error: fmt_diag(
provider.name(),
DiagKind::MidStream,
"stream-error",
err_msg,
),
error_type: Some(ErrorType::Unknown),
conversation_id: Some(conv_id.to_string()),
});
return StreamResult::InitFailed { retryable: true };
}
return StreamResult::Partial {
text: full_text,
tool_calls: tool_calls_acc,
usage: final_usage.unwrap_or_default(),
};
}
if chunk.finished {
finished_received = true;
@@ -222,22 +275,33 @@ pub(crate) async fn stream_llm(
status = %status_or_class,
conv_id = %conv_id,
error = %raw,
text_len = full_text.len(),
"[ai] 流式接收中途错误",
);
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
error: fmt_diag(
provider.name(),
DiagKind::MidStream,
&status_or_class,
&raw,
),
// 流已建立后 next() 返 Err = SSE 传输断/解析错,归 Network
// (extract_error_diag 已抠 status_or_class,但混合源难统一归 auth/timeout,
// 主流为传输断,前端可据 error 文本二次判断 HTTP 4xx 等)
error_type: Some(ErrorType::Network),
conversation_id: Some(conv_id.to_string()),
});
return None;
// UX-2025-04 / CR-30-2 / 决策 F-260616-07 a1: MidStream chunk Err——
// 已有文本则保文(Partial),不重试。空文本无文可保,emit AiError +
// InitFailed{retryable=classify_status_or_class(status_or_class)}
// (4xx Fatal 立即放弃,5xx/429/timeout/connect 可重试)。
if full_text.is_empty() && tool_calls_acc.is_empty() {
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
error: fmt_diag(
provider.name(),
DiagKind::MidStream,
&status_or_class,
&raw,
),
error_type: Some(ErrorType::Network),
conversation_id: Some(conv_id.to_string()),
});
return StreamResult::InitFailed {
retryable: classify_status_or_class(&status_or_class),
};
}
return StreamResult::Partial {
text: full_text,
tool_calls: tool_calls_acc,
usage: final_usage.unwrap_or_default(),
};
}
}
}
@@ -262,26 +326,47 @@ pub(crate) async fn stream_llm(
}
}
// 用户停止：已生成文本（可能残缺）交调用方入库展示
// 用户停止：已生成文本（可能残缺）交调用方入库展示。Complete(incomplete=false)——
// 用户主动停止语义非异常中断,正常入库 + AiCompleted(决策 a1: 不重试路径)。
if stopped {
return Some((full_text, tool_calls_acc, final_usage.unwrap_or_default()));
return StreamResult::Complete {
text: full_text,
tool_calls: tool_calls_acc,
usage: final_usage.unwrap_or_default(),
};
}
// 断连检测：流尽但从未收到 finished 信号 = 异常中断，丢弃残缺不当完整入库。
// B-260615-05:一律判异常(不区分内容空否)——空内容无 finished 同属异常:
// 上游 agentic.rs !has_tool_calls 早 break 会按正常路径 emit AiCompleted,
// 用户看空气泡无错误提示,转 P0 卡死入口。此处统一 emit AiError 拦截。
// 断连检测：流尽但从未收到 finished 信号 = 异常中断。
// B-260615-05:空内容无 finished emit AiError + InitFailed{retryable=true}(无文可保,交重试)。
// UX-2025-04:有 partial_text 则保文(Partial),不 emit AiError,不重试。
if !finished_received {
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
error: "流式响应意外中断（未收到完成信号，已丢弃残缺响应）".to_string(),
// 流尽但未收到 finished = 连接异常中断,归 Network
error_type: Some(ErrorType::Network),
conversation_id: Some(conv_id.to_string()),
});
return None;
if full_text.is_empty() && tool_calls_acc.is_empty() {
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
error: "流式响应意外中断（未收到完成信号，已丢弃残缺响应）".to_string(),
error_type: Some(ErrorType::Network),
conversation_id: Some(conv_id.to_string()),
});
return StreamResult::InitFailed { retryable: true };
}
warn!(
provider = %provider.name(),
conv_id = %conv_id,
text_len = full_text.len(),
"[ai] 流尽未收到 finished 但有 partial_text,保文不重试(incomplete)",
);
return StreamResult::Partial {
text: full_text,
tool_calls: tool_calls_acc,
usage: final_usage.unwrap_or_default(),
};
}
Some((full_text, tool_calls_acc, final_usage.unwrap_or_default()))
// 正常完成：流尽 + finished 信号到位,Complete(incomplete=false)
StreamResult::Complete {
text: full_text,
tool_calls: tool_calls_acc,
usage: final_usage.unwrap_or_default(),
}
}
Err(e) => {
// 诊断：连接/鉴权/HTTP 错误，补 provider 标识 + HTTP 状态码/分类 + 原始文本，
@@ -301,18 +386,53 @@ pub(crate) async fn stream_llm(
&status_or_class,
&raw,
),
// 建连失败混合源:401(auth)/404(provider_config)/connect(network)/timeout/unknown,
// extract_error_diag 已抠 status_or_class 供前端 error 文本展示,但运行时文本分类
// 归一 error_type 不可靠(同 raw 跨多类型),按任务约定难精确分类的旧点填 None,
// 前端可据 error 文本中的 "HTTP 401" 等自行二次判断。
error_type: None,
conversation_id: Some(conv_id.to_string()),
});
None
// CR-30-1: Init 失败分类——retryable 据 status_or_class 镜像 retry::is_status_retryable:
// 5xx/429/timeout/connect=true 可重试;4xx(非429)/鉴权/参数错=false Fatal 立即放弃。
StreamResult::InitFailed {
retryable: classify_status_or_class(&status_or_class),
}
}
}
}
/// 据 extract_error_diag 产出的 status_or_class 字符串镜像 retry::is_status_retryable 分类。
///
/// status_or_class 取值: "HTTP 5xx"/"HTTP 4xx"(白名单码) / "timeout" / "connect" / "unknown"。
/// - HTTP 429 / 5xx → retryable=true(复用 retry::is_status_retryable 逻辑)
/// - HTTP 4xx(非429) → retryable=false Fatal(401/403/404/422 等)
/// - timeout / connect → retryable=true(瞬态,可重试)
/// - unknown / 其他 → retryable=true(保守可重试,避免误判 Fatal 错杀)
///
/// CR-30-1: 抽公共分类函数(决策 F-260616-07 a1 "复用 retry.rs 错误分类逻辑或镜像"),
/// 避免 stream_recv 与 agentic 各自重写状态码解析。
fn classify_status_or_class(status_or_class: &str) -> bool {
use df_ai::retry::is_status_retryable;
// 抠 status_or_class 内三位数字状态码("HTTP 401" → 401)
let chars: Vec<char> = status_or_class.chars().collect();
let mut i = 0;
let n = chars.len();
while i + 3 <= n {
if chars[i].is_ascii_digit()
&& chars[i + 1].is_ascii_digit()
&& chars[i + 2].is_ascii_digit()
{
let code: String = chars[i..i + 3].iter().collect();
if let Ok(status) = code.parse::<u16>() {
return is_status_retryable(status);
}
i += 3;
continue;
}
i += 1;
}
// 无三位数字码:按文本分类(timeout/connect 可重试,unknown/其他保守可重试)
let lower = status_or_class.to_lowercase();
lower.contains("timeout") || lower.contains("connect") || true // unknown 保守可重试
}
// ============================================================
// 单测：诊断提取/格式化（纯函数，不发 HTTP、不依赖 app_handle）
// ============================================================
@@ -320,6 +440,43 @@ pub(crate) async fn stream_llm(
mod tests {
use super::*;
// ---- classify_status_or_class：CR-30-1 retryable 分类(镜像 retry::is_status_retryable) ----
/// 5xx → retryable=true
#[test]
fn classify_5xx_retryable() {
assert!(classify_status_or_class("HTTP 500"));
assert!(classify_status_or_class("HTTP 503"));
}
/// 429 → retryable=true(限流可重试)
#[test]
fn classify_429_retryable() {
assert!(classify_status_or_class("HTTP 429"));
}
/// 4xx(非429) → retryable=false Fatal(鉴权/参数错立即放弃)
#[test]
fn classify_4xx_fatal() {
assert!(!classify_status_or_class("HTTP 401"));
assert!(!classify_status_or_class("HTTP 403"));
assert!(!classify_status_or_class("HTTP 404"));
assert!(!classify_status_or_class("HTTP 422"));
}
/// timeout / connect → retryable=true(瞬态)
#[test]
fn classify_transient_retryable() {
assert!(classify_status_or_class("timeout"));
assert!(classify_status_or_class("connect"));
}
/// unknown → 保守 retryable=true(不误判 Fatal 错杀)
#[test]
fn classify_unknown_retryable() {
assert!(classify_status_or_class("unknown"));
}
// ---- extract_error_diag：业务层 bail（provider 串已含状态码）----
/// provider 在 non-2xx bail 的典型串：抠出 401（鉴权失败/Key 错）