lxy
d406a31722
- 循环主体 catch_unwind 防护,panic 走正典终态不再卡死 - 会话删除停 in-flight 循环+删除侧单点补发终态 - 审批超时到点主动收敛,执行等待循环加删除判死 - 子代理出口整轨迹落库,拒绝工具结构化+报告协议 - 压缩会话 id 修正,begin_turn 喂用户原文 - 远程桥会话列表按 origin 过滤子会话,新增中断续跑命令
668 lines
36 KiB
Rust
668 lines
36 KiB
Rust
//! 单轮流式中继 — candidate 链 fallback + 流式调用 + 流前重试 + MidStream 保文。
|
|
//!
|
|
//! 自 agentic/mod.rs 巨石拆解外移(2026-10-01,纯搬运语义等价,职责分组):
|
|
//! - `StreamOutcome` / `stream_one_provider` / `stream_with_retry`:单 candidate 流式调用
|
|
//! + 流前 InitFailed retryable 重试循环;
|
|
//! - `RoundOutcome` / `stream_round`:单轮流式自包含块(retry_deadline → candidate 链
|
|
//! 构建/降级 → candidate 循环 → epoch 后校验 → MidStream 保文);
|
|
//! - `emit_retry_attempt`:退避日志 + AiStreamRetry emit + sleep;
|
|
//! - `merge_volatile_tail`:Anthropic prompt caching 易变尾段追加到请求消息流末尾。
|
|
//!
|
|
//! 行为红线:session_arc 锁的获取/释放位置、guard.reset/disarm 时序、emit+publish 双写
|
|
//! 顺序、per-provider permit 取放范围均与拆解前逐字一致,本模块不改任何行为。
|
|
|
|
use std::collections::HashMap;
|
|
use std::sync::Arc;
|
|
use std::sync::atomic::Ordering;
|
|
|
|
use tauri::AppHandle;
|
|
use tokio::sync::Mutex;
|
|
|
|
// 复用 retry::backoff_delay(jitter 1s→2s→4s) + is_status_retryable(Fatal 分类)
|
|
// 实现流前失败重试退避对齐,避免重写退避逻辑。
|
|
use df_ai::retry;
|
|
use df_ai::provider::{ChatMessage, CompletionRequest, LlmProvider};
|
|
// 智能路由 helper + TaskRequirements(主对话:needs_tool_use=true,当前仅 Text 模态)。
|
|
use df_ai::router::TaskRequirements;
|
|
|
|
use df_storage::db::Database;
|
|
use df_storage::models::AiProviderRecord;
|
|
|
|
use crate::state::{LlmConcurrency, ProviderAcquire};
|
|
|
|
use crate::commands::ai::conversation::{save_conversation, TokenAccumulator};
|
|
use crate::commands::ai::stream_recv::{stream_llm, StreamResult};
|
|
use crate::commands::ai::{AiChatEvent, AiSession, GoalEntry, ToolCallDraft};
|
|
|
|
use super::guard::GeneratingGuard;
|
|
use super::PROVIDER_SATURATION_DEMOTE_THRESHOLD;
|
|
use super::{check_retry_exhausted, resolve_model_with_override};
|
|
use super::{emit_fatal_error, teardown_round, tokens_snapshot, usage_delta_since};
|
|
use super::super::event_bus::emit_ai_chat_event;
|
|
|
|
// ============================================================
|
|
// 单 Provider 流式结果 + fallback 辅助
|
|
// ============================================================
|
|
|
|
/// 单 candidate provider 一次迭代的流式调用结果。
|
|
///
|
|
/// 显式区分三态,驱动外层 `for candidate` fallback 循环:
|
|
/// - `Success`:`Complete` 或 `Partial`(MidStream 保文)。携带该 provider 路由出的
|
|
/// `resolved_model`(调用方据此更新迭代级 resolved_model,供 push/save)。
|
|
/// - `InitFailedExhausted`:流前 `InitFailed{retryable=true}` 在本 provider 上重试
|
|
/// 耗尽(或预算 30s 耗尽)。retryable=true 即「瞬态错误,换 provider 可能成功」→
|
|
/// 外层 `continue` 切下一 candidate(重建 build_provider + resolved_model 重算)。
|
|
/// 携带 `error`(最后一条诊断文本),供外层「全 candidate 耗尽」时 emit 最终 AiError。
|
|
/// - `Fatal`:流前 `InitFailed{retryable=false}`(4xx 非429/鉴权/参数错)。立即放弃
|
|
/// 整个 fallback(对齐 retry.rs Fatal 分类 + provider_pool 文档「不可重试错误
|
|
/// 立即放弃不浪费备用 provider」)。携带 `error`,stream_one_provider 内 Fatal 分支
|
|
/// emit AiError(终态不切候选,无残留气泡风险),调用方仅做 guard.reset + return。
|
|
/// 空 key(build_provider_for Err)同样归 Fatal 语义——key 缺失/钥匙串损坏非瞬态,
|
|
/// 换 provider 无意义(但实际场景:外层已对 primary 做了启动 Auth 早失败,故此分支
|
|
/// 多见于 secondary 配置不一致,保守 Fatal 收敛)。
|
|
enum StreamOutcome {
|
|
Success {
|
|
text: String,
|
|
tool_calls: std::collections::HashMap<u32, ToolCallDraft>,
|
|
usage: df_ai::provider::TokenUsage,
|
|
incomplete: bool,
|
|
resolved_model: String,
|
|
/// DeepSeek thinking 模式推理内容（需回传到下一轮请求）
|
|
reasoning_content: Option<String>,
|
|
},
|
|
/// error 字段供外层全 candidate 耗尽时 emit 最终 AiError(单气泡聚合)。
|
|
InitFailedExhausted { error: String },
|
|
/// error 字段供 stream_one_provider 内 Fatal 分支 emit AiError。
|
|
Fatal { error: String },
|
|
}
|
|
|
|
/// 单 provider 流式调用 + 重试。
|
|
///
|
|
/// 封装一个 candidate 的:resolve_secret → ensure_resolved_key → build_provider
|
|
/// → select_model_id(resolved_model 在该 candidate.model_configs 上重算)
|
|
/// → 流式 stream_recv + 流前 InitFailed retryable 重试循环。
|
|
///
|
|
/// 与主链共享的退避/分类:`retry::backoff_delay`(1s→2s→4s+jitter) +
|
|
/// `StreamResult::InitFailed{retryable}`(镜像 retry::is_status_retryable) +
|
|
/// 30s 总挂钟预算(`retry_deadline` 由调用方传入,本轮 fallback 各 candidate 共享一个预算)。
|
|
///
|
|
/// **permit 责任**:本函数不取并发 permit——permit 持有期间多 provider 限流不重叠,
|
|
/// 由调用方在进 candidate 前取 global/per_conv(整轮迭代共享) + 本 candidate 的
|
|
/// per-provider permit(切换 candidate 时释放旧取新)。
|
|
///
|
|
/// 返回 [`StreamOutcome`]:Fatal 含已 emit AiError;Success 含保文或正常完成;
|
|
/// InitFailedExhausted 让调用方切下一 candidate。
|
|
async fn stream_one_provider(
|
|
candidate: &AiProviderRecord,
|
|
messages: &[ChatMessage],
|
|
tool_defs: &[df_ai::provider::ToolDefinition],
|
|
app_handle: &AppHandle,
|
|
stop_flag: &std::sync::atomic::AtomicBool,
|
|
notify: &tokio::sync::Notify,
|
|
conv_id: &str,
|
|
max_retries: usize,
|
|
retry_deadline: tokio::time::Instant,
|
|
model_override: &Option<String>,
|
|
agentic_req: &TaskRequirements,
|
|
last_reasoning_content: &Option<String>,
|
|
) -> StreamOutcome {
|
|
// resolve→ensure→build 三步(复用 secret::build_provider_for,DRY)。
|
|
// key 缺失/损坏 → Err → 归 Fatal(stream_llm Fatal 也是 key 类错,语义一致)。
|
|
// 注:primary 候选的启动 Auth 早失败已在外层 run_agentic_loop 顶部处理(emit + return),
|
|
// 本函数到达时 primary 的 key 已验证过;此处 Err 多见于 secondary 配置不一致,
|
|
// 保守归 Fatal 立即放弃(不浪费预算试下一 provider,因 key 错非瞬态)。
|
|
let provider: Arc<dyn LlmProvider> = match crate::commands::ai::secret::build_provider_for(candidate) {
|
|
Ok(p) => Arc::from(p),
|
|
Err(msg) => return StreamOutcome::Fatal { error: msg },
|
|
};
|
|
|
|
// resolved_model 在本 candidate.model_configs 上重算(provider 切换后
|
|
// 模型池不同,必须重选;否则拿主 provider 的 model_id 去打次 provider 会吃 400/404)。
|
|
// model_override 穿透:override 非空且在本 candidate 池中 → 用 override;
|
|
// 否则用 resolved_model(绝不因 override 致无模型)。
|
|
let resolved_model = resolve_model_with_override(agentic_req, candidate, model_override);
|
|
|
|
// 流前 InitFailed 重试循环。
|
|
// 重试逻辑抽至 stream_with_retry helper(扁平重构),本函数仅负责 provider 准备 + model 决议 + 委派。
|
|
stream_with_retry(
|
|
&provider, candidate, messages, tool_defs, app_handle, stop_flag, notify, conv_id,
|
|
max_retries, retry_deadline, &resolved_model, last_reasoning_content,
|
|
).await
|
|
}
|
|
|
|
// ── stream_with_retry: 流式 + 流前 InitFailed retryable 重试循环(扁平抽自原嵌套 5 层 match) ──
|
|
// 返回 StreamOutcome。内部含 retry_attempt 循环 + match StreamResult 三分支:
|
|
// Complete/Partial → Success;InitFailed Fatal → Fatal;InitFailed Retryable → 重试/耗尽。
|
|
// 嵌套 ≤ 3 层:for 内 match 内单层 if(早返回避免嵌套加深)。
|
|
async fn stream_with_retry(
|
|
provider: &Arc<dyn LlmProvider>,
|
|
candidate: &AiProviderRecord,
|
|
messages: &[ChatMessage],
|
|
tool_defs: &[df_ai::provider::ToolDefinition],
|
|
app_handle: &AppHandle,
|
|
stop_flag: &std::sync::atomic::AtomicBool,
|
|
notify: &tokio::sync::Notify,
|
|
conv_id: &str,
|
|
max_retries: usize,
|
|
retry_deadline: tokio::time::Instant,
|
|
resolved_model: &str,
|
|
last_reasoning_content: &Option<String>,
|
|
) -> StreamOutcome {
|
|
// 供应商级采样温度覆盖:provider config JSON 带 "temperature" 字段则用之,
|
|
// 治受限模型(如 Kimi 系经 u-switch 透传)仅接受 temperature=1、其它值 400 的问题;
|
|
// 未配置/非法 JSON 走默认 0.7。candidate 在重试循环内不变,解析提到循环外只做一次。
|
|
let temperature = candidate
|
|
.config
|
|
.as_deref()
|
|
.and_then(|cfg| serde_json::from_str::<serde_json::Value>(cfg).ok())
|
|
.and_then(|v| v.get("temperature").and_then(|t| t.as_f64()))
|
|
.map(|t| t as f32)
|
|
.or(Some(0.7));
|
|
for retry_attempt in 0..=max_retries {
|
|
let retry_request = CompletionRequest {
|
|
model: resolved_model.to_string(),
|
|
messages: messages.to_vec(),
|
|
temperature,
|
|
max_tokens: Some(8192),
|
|
stream: true,
|
|
tools: if tool_defs.is_empty() { None } else { Some(tool_defs.to_vec()) },
|
|
tool_choice: None,
|
|
reasoning_content: last_reasoning_content.clone(),
|
|
};
|
|
|
|
let result = stream_llm(
|
|
Arc::clone(provider), retry_request, app_handle, stop_flag, notify, conv_id,
|
|
).await;
|
|
|
|
match result {
|
|
StreamResult::Complete { text, tool_calls, usage, reasoning_content } => {
|
|
return StreamOutcome::Success {
|
|
text, tool_calls, usage, incomplete: false,
|
|
resolved_model: resolved_model.to_string(), reasoning_content,
|
|
};
|
|
}
|
|
StreamResult::Partial { text, tool_calls, usage, reasoning_content } => {
|
|
// MidStream 保文不重试。携带 incomplete=true 交调用方走保文路径。
|
|
tracing::warn!(
|
|
conv_id = %conv_id,
|
|
text_len = text.len(),
|
|
"[ai] 流中途失败(候选 {}),保文不重试(incomplete=true)",
|
|
candidate.name,
|
|
);
|
|
return StreamOutcome::Success {
|
|
text, tool_calls, usage, incomplete: true,
|
|
resolved_model: resolved_model.to_string(), reasoning_content,
|
|
};
|
|
}
|
|
StreamResult::InitFailed { retryable, error } => {
|
|
// Fatal(4xx 非429/鉴权/参数错)立即放弃整个 fallback。
|
|
if !retryable {
|
|
tracing::warn!(
|
|
conv_id = %conv_id,
|
|
provider = %candidate.name,
|
|
attempt = retry_attempt + 1,
|
|
"[ai] 候选 {} 流前失败 Fatal(4xx/鉴权),立即放弃 fallback",
|
|
candidate.name,
|
|
);
|
|
return StreamOutcome::Fatal { error };
|
|
}
|
|
// Retryable:判断是否重试预算耗尽
|
|
if let Some(exhausted_error) = check_retry_exhausted(
|
|
conv_id, candidate, retry_attempt, max_retries, retry_deadline, &error,
|
|
) {
|
|
return StreamOutcome::InitFailedExhausted { error: exhausted_error };
|
|
}
|
|
// 未耗尽:退避 + emit AiStreamRetry + sleep + continue
|
|
emit_retry_attempt(
|
|
app_handle, conv_id, retry_attempt, max_retries, retry_deadline,
|
|
).await;
|
|
}
|
|
}
|
|
}
|
|
|
|
// 仅 max_retries=0(无重试配置)且首试 InitFailed retryable=true 时到达此处:
|
|
// 0..=0 循环首趟即 is_last → 已在循环内 return InitFailedExhausted。此为防御兜底。
|
|
StreamOutcome::InitFailedExhausted {
|
|
error: "AI 调用失败：重试预算耗尽（防御兜底）".to_string(),
|
|
}
|
|
}
|
|
|
|
// ============================================================
|
|
// stream_round: 单轮流式调用(candidate 链 + stream_one_provider + MidStream 保文)
|
|
// ============================================================
|
|
|
|
/// `run_agentic_loop` for 循环体「单轮流式调用」自包含块的返回结果。
|
|
///
|
|
/// 抽自 L1505-L1756(retry_deadline → candidate_chain 构建/降级 → candidate 循环
|
|
/// (stream_one_provider + match 三态) → epoch 后校验 → MidStream 保文)。
|
|
/// 表达本轮的 4 种结束路径,供 loop 主框架据以 `return` 或继续后续处理。
|
|
///
|
|
/// **不在本函数范围**(loop 内其它段):
|
|
/// - 停止/收敛/审批等待/熔断这些在 process_tool_calls 之后,
|
|
/// 属「工具执行后路径」,与流式调用本身解耦,留 loop 主体。
|
|
/// - 正常完成(loop 外 L1957+):loop 退出后由 run_agentic_loop 直接收尾。
|
|
pub(crate) enum RoundOutcome {
|
|
/// 本轮 stream 成功(Complete,无 incomplete)。携带本轮输出供 loop 后续
|
|
/// push_assistant_message / save / 工具执行处理。
|
|
///
|
|
/// 注:reasoning_content 不在此 variant 携带——stream_round 内已通过
|
|
/// `*last_reasoning_content = rc` 写回 caller 的 loop 级 mut 变量,loop 后续
|
|
/// push_assistant_message 读 last_reasoning_content 即可(零行为变更)。
|
|
Continue {
|
|
full_text: String,
|
|
tool_calls: std::collections::HashMap<u32, ToolCallDraft>,
|
|
/// 本轮补全后 usage(prompt 估算兜底,round_usage_effective 单点派生),
|
|
/// loop 主体直接 add_usage/push,不再字面量重组。
|
|
round_usage: df_ai::provider::TokenUsage,
|
|
/// 本轮 prompt 是否估算兜底(= 进 stream_round 时的 round_usage.prompt_tokens == 0),
|
|
/// 供 loop 主体 push 消息级 is_estimated,与 round_usage 同点派生防漂移。
|
|
round_is_estimated: bool,
|
|
},
|
|
/// 流返回后 stale(epoch 已变,被新 loop 接管):guard 已 disarm,
|
|
/// loop 应直接 return(不再 push/save/执行工具)。
|
|
StaleAfterStream,
|
|
/// MidStream 保文路径已完整收尾(tokens.add + push partial + 系统提示 +
|
|
/// teardown_round save/reset/emit)→ loop 直接 return。
|
|
MidStreamSaved,
|
|
/// Fatal(候选 Fatal 或全候选耗尽):guard.reset + epoch 检查 + emit AiError +
|
|
/// save 已完成 → loop 直接 return。
|
|
Fatal,
|
|
}
|
|
|
|
/// 单轮流式调用:抽自 `run_agentic_loop` for 循环体的「单轮流式」自包含块。
|
|
///
|
|
/// 范围 = retry_deadline 计算 → candidate_chain 构建(G4.1 防抖降级)
|
|
/// → candidate 循环(try_acquire_for_provider + stream_one_provider + match 三态)
|
|
/// → epoch 后 stale 检查 → MidStream 保文路径。返回 [`RoundOutcome`] 表达本轮结果。
|
|
///
|
|
/// **逐字节等价**:原 L1505-L1756 内联块的 save/emit/guard.reset/permit 顺序、
|
|
/// token 累加、push_assistant_message 调用、teardown_round 退出序列全部保留。
|
|
/// 抽取仅改控制流(原各路径 `return` → `return RoundOutcome::Xxx`,loop 据返回值决定 return)。
|
|
///
|
|
/// **借用处理**:
|
|
/// - `tokens: &mut TokenAccumulator` / `saved_token_snapshot: &mut TokenUsage` /
|
|
/// `last_round_estimated: &mut bool`:loop 级累计状态,&mut 借用,函数内修改 caller 可见。
|
|
/// - `resolved_model: &mut String` / `provider_config: &mut AiProviderRecord` /
|
|
/// `last_reasoning_content: &mut Option<String>`:loop 级 mut 变量(被 candidate
|
|
/// 切换覆盖),&mut 借用,函数内赋新值后 loop 后续 push/save 用最新。
|
|
/// - `guard: &mut GeneratingGuard`:RAII 复位,&mut 借用,函数内 guard.reset()/disarm()
|
|
/// 直接作用于 caller 的 guard。
|
|
/// - 其它(session_arc/db/app_handle/conv_id/...):& 引用。
|
|
///
|
|
/// **permit**:per-provider permit 仍在 candidate 循环内取/释放(随作用域 Drop),
|
|
/// 与原实现一致;per_conv permit 由 loop 入口整 loop 持有,本函数不涉及。
|
|
#[allow(clippy::too_many_arguments)]
|
|
pub(crate) async fn stream_round(
|
|
session_arc: &Arc<Mutex<AiSession>>,
|
|
db: &Arc<Database>,
|
|
conv_id: &str,
|
|
messages: &[ChatMessage],
|
|
tool_defs: &[df_ai::provider::ToolDefinition],
|
|
app_handle: &AppHandle,
|
|
stop_flag: &std::sync::atomic::AtomicBool,
|
|
notify: &tokio::sync::Notify,
|
|
max_retries: usize,
|
|
model_override: &Option<String>,
|
|
agentic_req: &TaskRequirements,
|
|
estimated_prompt: u32,
|
|
// &mut loop 级状态(函数内修改,caller 可见)
|
|
last_round_estimated: &mut bool,
|
|
last_reasoning_content: &mut Option<String>,
|
|
resolved_model: &mut String,
|
|
provider_config: &mut AiProviderRecord,
|
|
provider_saturation: &mut HashMap<String, usize>,
|
|
tokens: &mut TokenAccumulator,
|
|
saved_token_snapshot: &mut df_ai::provider::TokenUsage,
|
|
guard: &mut GeneratingGuard,
|
|
// & 引用
|
|
candidates: &[AiProviderRecord],
|
|
llm_concurrency: &LlmConcurrency,
|
|
pinned_goals_snapshot: &[GoalEntry],
|
|
loop_epoch_arc: &std::sync::Arc<std::sync::atomic::AtomicU64>,
|
|
my_epoch: u64,
|
|
) -> RoundOutcome {
|
|
// 重试总预算(挂钟,含 sleep + 各次请求耗时),对齐 retry::MAX_TOTAL_BUDGET 30s。
|
|
// 本轮各 candidate 共享一个 30s 预算(切换 provider 不重置预算,
|
|
// 防多 provider 串行重试累加超过单轮总预算)。超预算直接放弃重试交最终错误/保文路径。
|
|
let retry_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(30);
|
|
|
|
// 外层 for candidate 包裹 stream_one_provider。
|
|
// 顺序:[primary, ...candidates](空池兜底时仅 primary)。
|
|
// - Success → 用结果覆盖迭代级 resolved_model/provider_config(供 push/save/标题);
|
|
// 成功即 break。
|
|
// - InitFailedExhausted(retryable 耗尽)→ continue 切下一 candidate。
|
|
// - Fatal(4xx 非429/鉴权)→ stream_llm 已 emit AiError,guard.reset + return
|
|
// 放弃整轮(对齐 retry.rs Fatal + provider_pool 文档)。
|
|
// 全 candidate 耗尽 → 沿用单 provider 失败语义(guard.reset + return,AiError 已在
|
|
// 各 candidate 的 stream_llm 内 emit 最后一条)。
|
|
// 候选链(拥有 Vec):[primary, ...candidates]。预先 clone primary 入链,
|
|
// 避免借用 provider_config(切换成功后需写回 provider_config = candidate.clone())。
|
|
// candidates 空(单 provider)→ candidate_chain 仅 primary,循环跑一次,耗尽即 return。
|
|
let mut candidate_chain: Vec<AiProviderRecord> = Vec::with_capacity(1 + candidates.len());
|
|
candidate_chain.push(provider_config.clone());
|
|
candidate_chain.extend(candidates.iter().cloned());
|
|
|
|
// 防瞬时抖动(G4.1):连续 PROVIDER_SATURATION_DEMOTE_THRESHOLD 次饱和才把主 provider
|
|
// 降级到链尾。计数器在候选循环内维护(Exhausted +1 / Acquired 清零),跨迭代累计。
|
|
// 链首(当前主 provider)连续 N 次 Exhausted → 移到链尾让 fallback 先试;
|
|
// 恢复后 Acquired 清零 → 下轮回到链首。单次/瞬时饱和(N 次内)不降级,防跳过主 provider。
|
|
// first():链构建总是 push 至少 1 个(provider_config),但防御性取首防空链 panic。
|
|
if let Some(streak) = candidate_chain
|
|
.first()
|
|
.and_then(|c| provider_saturation.get(&c.id))
|
|
.copied()
|
|
.filter(|s| *s >= PROVIDER_SATURATION_DEMOTE_THRESHOLD)
|
|
{
|
|
let saturated = candidate_chain.remove(0);
|
|
candidate_chain.push(saturated.clone());
|
|
tracing::info!(
|
|
conv_id = %conv_id,
|
|
provider = %saturated.name,
|
|
streak = streak,
|
|
threshold = PROVIDER_SATURATION_DEMOTE_THRESHOLD,
|
|
"[ai] 主 provider 连续 {} 次 per-provider 饱和,降级到候选链尾(防抖动跳过)",
|
|
streak,
|
|
);
|
|
}
|
|
|
|
let (full_text, tool_calls_acc, round_usage, incomplete, round_reasoning_content) = {
|
|
// outcome 累积成功结果(Complete/Partial),InitFailed 不写入。
|
|
let mut outcome: Option<(String, std::collections::HashMap<u32, ToolCallDraft>, df_ai::provider::TokenUsage, bool, Option<String>)> = None;
|
|
// 追踪最后一个 Exhausted candidate 的诊断文本,供「全 candidate 耗尽」时 emit 最终 AiError。
|
|
// Fatal 分支即时 emit(终态不切候选),Exhausted 分支仅记录 error 不 emit(可能切下一 candidate 成功,
|
|
// emit 会留残留气泡)。
|
|
let mut last_exhausted_error: Option<String> = None;
|
|
|
|
'candidate: for candidate in &candidate_chain {
|
|
// per-provider permit(非阻塞三态,G4.1):
|
|
// NotConfigured(未配置,无限流)→ 无 permit 直接 proceed(单 provider 零变化);
|
|
// Acquired(permit)→ 持 permit stream,切换 candidate 时随作用域 Drop 释放;
|
|
// Exhausted(占满)→ 连续饱和计数 +1,跳下一 candidate(降级,不阻塞)。
|
|
// 旧 acquire_for_provider(阻塞 acquire_owned)在主 provider 信号量占满时整条
|
|
// fallback 链卡死(降级失效),已由本三态非阻塞版本取代。
|
|
let _provider_permit = match llm_concurrency.try_acquire_for_provider(&candidate.id).await {
|
|
ProviderAcquire::NotConfigured => None,
|
|
ProviderAcquire::Acquired(permit) => {
|
|
// 成功拿到 permit:该 provider 已恢复,连续饱和计数清零(跨迭代防抖)。
|
|
provider_saturation.insert(candidate.id.clone(), 0);
|
|
Some(permit)
|
|
}
|
|
ProviderAcquire::Exhausted => {
|
|
// 占满:连续饱和计数 +1;本次跳下一 candidate(非阻塞,不等待)。
|
|
// 跨迭代防抖:连续 N 次饱和后链首降级(见候选链构建处),防瞬时抖动跳过主 provider。
|
|
*provider_saturation.entry(candidate.id.clone()).or_insert(0) += 1;
|
|
tracing::debug!(
|
|
conv_id = %conv_id,
|
|
provider = %candidate.name,
|
|
streak = provider_saturation.get(&candidate.id).copied().unwrap_or(0),
|
|
"[ai] 候选 {} per-provider 信号量占满,跳过(非阻塞,防单 provider 429)",
|
|
candidate.name,
|
|
);
|
|
continue 'candidate;
|
|
}
|
|
};
|
|
|
|
let stream_outcome = stream_one_provider(
|
|
candidate,
|
|
messages,
|
|
tool_defs,
|
|
app_handle,
|
|
stop_flag,
|
|
notify,
|
|
conv_id,
|
|
max_retries,
|
|
retry_deadline,
|
|
model_override,
|
|
agentic_req,
|
|
last_reasoning_content,
|
|
).await;
|
|
|
|
match stream_outcome {
|
|
StreamOutcome::Success { text, tool_calls, usage, incomplete, resolved_model: m, reasoning_content: rc } => {
|
|
// 成功:更新迭代级 resolved_model + provider_config(供后续 push/save/标题)。
|
|
// mut 解构(已在顶部声明 mut):本轮后续 save/push 用「实际成功所用 provider」
|
|
// 而非主 candidate。compress 已在本轮 stream 之前用过 primary,不受影响。
|
|
*resolved_model = m;
|
|
*provider_config = candidate.clone();
|
|
*last_reasoning_content = rc.clone();
|
|
outcome = Some((text, tool_calls, usage, incomplete, last_reasoning_content.clone()));
|
|
break 'candidate;
|
|
}
|
|
StreamOutcome::InitFailedExhausted { error } => {
|
|
// 本 candidate 重试耗尽(retryable)。切下一 candidate 继续尝试。
|
|
// candidates 空(单 provider)时此即「耗尽 return」语义——
|
|
// 跳出循环后 outcome 仍 None,落入下方「全耗尽 return」兜底。
|
|
// 不在此 emit AiError(可能切下一 candidate 成功留残留气泡),
|
|
// 仅记录 error 供全耗尽兜底 emit 最后一条。
|
|
tracing::warn!(
|
|
conv_id = %conv_id,
|
|
provider = %candidate.name,
|
|
"[ai] 候选 {} 流前失败重试耗尽,切换下一 provider",
|
|
candidate.name,
|
|
);
|
|
last_exhausted_error = Some(error);
|
|
continue 'candidate;
|
|
}
|
|
StreamOutcome::Fatal { error } => {
|
|
// Fatal(4xx 非429/鉴权/参数错):立即放弃整轮 fallback。
|
|
// 统一 emit 最终错误气泡(单气泡聚合)。
|
|
guard.reset().await;
|
|
// F1:旧 loop(被新 loop 接管)不 emit 错误气泡(owner loop 负责呈现)。
|
|
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
|
|
return RoundOutcome::Fatal;
|
|
}
|
|
emit_fatal_error(app_handle, conv_id, &error).await;
|
|
// G1.1(2026-08-05):Fatal 退出也落库 user 消息(镜像下方 Exhausted 分支),
|
|
// 否则重启/切会话从 DB 恢复时末条 user 丢失(与 Exhausted 同根因不对称)。
|
|
// epoch 校验通过后、return 前补 save(owner loop 才写,防 owner 已转移仍写)。
|
|
save_conversation(session_arc, db, conv_id, None, Some(&*resolved_model), true).await;
|
|
return RoundOutcome::Fatal;
|
|
}
|
|
}
|
|
}
|
|
|
|
match outcome {
|
|
Some(r) => r,
|
|
None => {
|
|
// 全 candidate 耗尽(含单 provider 场景):沿用原单 provider 失败语义。
|
|
tracing::warn!(
|
|
conv_id = %conv_id,
|
|
candidates_tried = candidate_chain.len(),
|
|
"[ai] 全 provider 候选流前失败重试耗尽,放弃本轮",
|
|
);
|
|
guard.reset().await;
|
|
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
|
|
return RoundOutcome::Fatal;
|
|
}
|
|
let err_msg = last_exhausted_error
|
|
.unwrap_or_else(|| "AI 调用失败：所有候选 provider 重试耗尽".to_string());
|
|
emit_fatal_error(app_handle, conv_id, &err_msg).await;
|
|
// 全 provider 失败也需落库 user 消息(已 push 内存),
|
|
// 否则切换/重载从 DB 恢复时末条 user 丢失(实测压缩触发失败场景:DB 末条 tool 无后续 user)。
|
|
save_conversation(session_arc, db, conv_id, None, Some(&*resolved_model), true).await;
|
|
return RoundOutcome::Fatal;
|
|
}
|
|
}
|
|
};
|
|
|
|
// G4.3:本轮 token 用量是否估算值(provider 未报 prompt_tokens → estimated_prompt 兜底),
|
|
// 供 loop 内/loop 后各退出路径透传 AiCompleted(is_estimated)仅作展示标注。
|
|
// 语义修正:整 loop 只要任一轮估算即标估算(累计总量含估算成分),非仅末轮。
|
|
// 三真相源收口:本轮 usage 补全 + estimated 判定在此单点派生(round_usage_effective),
|
|
// MidStream 保文分支与 Continue 路径(loop 主体)均消费本结果,不再各自字面量重组。
|
|
let (round_usage_full, round_is_estimated) = round_usage_effective(&round_usage, estimated_prompt);
|
|
*last_round_estimated |= round_is_estimated;
|
|
|
|
// F1 并发 epoch:流返回后若已被新 loop 接管(force_send 等),旧 loop 不再 push 消息 /
|
|
// 保文 / 执行工具,立即退出(guard disarm 跳过复位,防 clobber 新 loop 的 Generating)。
|
|
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
|
|
guard.disarm();
|
|
return RoundOutcome::StaleAfterStream;
|
|
}
|
|
|
|
// CR-30-2 / UX-2025-04 / 决策 a1: MidStream 保文路径——partial_text 已接收,
|
|
// 入库为正常 assistant 消息 + emit AiCompleted(incomplete=true) + 追加系统提示消息。
|
|
// 不走 AiError(非异常中断,已有可用文本),不重试(决策 a1)。
|
|
// 并发时用户停优先:网络断同时用户点停 → stop_flag true 时不走保文路径,
|
|
// 落入下方 ~1508 stop_flag 检查走停止路径(用户意图优先)。
|
|
// partial 文本仍由下方 push_assistant_message(~1485)保文不丢。
|
|
if incomplete && !stop_flag.load(Ordering::SeqCst) {
|
|
// 本轮 usage 补全(prompt 估算兜底,含 cache/reasoning 分项)已由 round_usage_effective 单点派生
|
|
let usage = round_usage_full;
|
|
// 累加本轮全量 usage(含 cache/reasoning 分项)到 tokens 累加器
|
|
tokens.add_usage(&usage);
|
|
|
|
// 追加 partial assistant 消息(若无 tool_calls 且有文本)
|
|
// 退出校验改 conv 存在性 + push 改 per_conv.messages。
|
|
// conv_id 来源:run_agentic_loop 入参。
|
|
{
|
|
let mut session = session_arc.lock().await;
|
|
if !session.per_conv.contains_key(conv_id) {
|
|
tracing::warn!(
|
|
stale_conv = %conv_id,
|
|
"[ai] MidStream 保文后 conv 已删除,丢弃本轮 push"
|
|
);
|
|
return RoundOutcome::MidStreamSaved;
|
|
}
|
|
// 与 push_assistant_message 一致:trim 归一化,纯空白保文(网络中断只有空白)不落库
|
|
let full_text = full_text.trim();
|
|
if !full_text.is_empty() {
|
|
let conv = session.conv(conv_id);
|
|
let mut msg = ChatMessage::assistant(full_text);
|
|
msg.model = Some(resolved_model.clone());
|
|
// MidStream 保文也回填 reasoning_content
|
|
msg.reasoning_content = round_reasoning_content.clone();
|
|
// 消息级 token(对齐 push_assistant_message 双轨持久化):本轮 partial usage
|
|
// (prompt=round 或 estimated 兜底,completion=round)。系统提示消息无 token,不设。
|
|
// 分项 token(2026-08-02):cache/reasoning 透传自 round_usage。
|
|
msg.prompt_tokens = Some(usage.prompt_tokens);
|
|
msg.completion_tokens = Some(usage.completion_tokens);
|
|
msg.prompt_cache_hit_tokens = Some(usage.prompt_cache_hit_tokens);
|
|
msg.prompt_cache_miss_tokens = Some(usage.prompt_cache_miss_tokens);
|
|
msg.reasoning_tokens = Some(usage.reasoning_tokens);
|
|
// 消息级估算标记:本轮 prompt 是否 estimated 兜底,reload 逐条回显对齐 live 态
|
|
msg.is_estimated = Some(round_is_estimated);
|
|
conv.messages.push(msg);
|
|
// 追加系统提示消息:响应因网络中断不完整(对齐决策 a1 系统提示机制)
|
|
let mut notice = ChatMessage::system("⚠ 响应因网络中断不完整,以上为已接收的部分内容。可重新发送以获取完整回复。");
|
|
notice.model = Some(resolved_model.clone());
|
|
conv.messages.push(notice);
|
|
}
|
|
}
|
|
|
|
// 统一走 teardown_round 收尾(save + spawn_title + reset + emit)。
|
|
// 注意:partial 文本+系统提示已先 push(上方 block),此 save 落库含本轮 partial,幂等覆盖。
|
|
// save_usage 用增量(tokens 已累加本轮,减上次快照);emit_usage 用 tokens 快照累计。
|
|
// MidStream 分叉:emit_incomplete=Some(true)(前端标不完整),publish_incomplete=None(总线消费方),
|
|
// do_publish=true(publish 走总线)。spawn_title=true(后台标题,失败 extract 兜底)。
|
|
let save_usage = usage_delta_since(tokens, saved_token_snapshot);
|
|
let emit_usage = tokens_snapshot(tokens);
|
|
teardown_round(
|
|
session_arc, db, conv_id,
|
|
Some(&save_usage), Some(&*resolved_model),
|
|
true,
|
|
provider_config, llm_concurrency,
|
|
guard,
|
|
&emit_usage,
|
|
*last_round_estimated,
|
|
Some(true), None, Some("network"), true,
|
|
pinned_goals_snapshot,
|
|
app_handle,
|
|
loop_epoch_arc, my_epoch,
|
|
true,
|
|
).await;
|
|
return RoundOutcome::MidStreamSaved;
|
|
}
|
|
|
|
// 正常完成本轮 stream(无 incomplete,或 incomplete 但用户停优先走 stop 路径)。
|
|
// 携带本轮输出交 loop 主体处理后续 push/save/工具执行。
|
|
// round_reasoning_content 已通过上方 `*last_reasoning_content = rc` 写回 caller,
|
|
// loop 后续 push_assistant_message 读 last_reasoning_content 即可。
|
|
let _ = round_reasoning_content; // 防未用 warning(MidStream 分支已消费)
|
|
RoundOutcome::Continue {
|
|
full_text,
|
|
tool_calls: tool_calls_acc,
|
|
round_usage: round_usage_full,
|
|
round_is_estimated,
|
|
}
|
|
}
|
|
|
|
// ── round_usage_effective: 本轮 usage 补全 + estimated 判定的单一派生点(三真相源收口) ──
|
|
//
|
|
// 收口前同一事实(「provider 未报 prompt_tokens → estimated_prompt 兜底」)在三处独立派生,
|
|
// 各持一份可漂移的重复逻辑:
|
|
// 1) MidStream 保文分支:TokenUsage 字面量重组(prompt/total 双条件分支)+ msg.is_estimated;
|
|
// 2) loop 主体 Continue 后:round_prompt + round_usage_full 同构字面量重组 + push 实参 estimated;
|
|
// 3) G4.3 累计标注:*last_round_estimated |= round_usage.prompt_tokens == 0。
|
|
// 收口后:此处一次求值返回 (补全 usage, is_estimated),三处消费方全部改为引用/派生,
|
|
// 逐字段等价(prompt=estimated 兜底 / total=estimated+completion / 分项透传),行为零变更。
|
|
fn round_usage_effective(
|
|
round_usage: &df_ai::provider::TokenUsage,
|
|
estimated_prompt: u32,
|
|
) -> (df_ai::provider::TokenUsage, bool) {
|
|
let is_estimated = round_usage.prompt_tokens == 0;
|
|
let usage = df_ai::provider::TokenUsage {
|
|
prompt_tokens: if is_estimated { estimated_prompt } else { round_usage.prompt_tokens },
|
|
completion_tokens: round_usage.completion_tokens,
|
|
total_tokens: if is_estimated {
|
|
estimated_prompt.saturating_add(round_usage.completion_tokens)
|
|
} else {
|
|
round_usage.total_tokens
|
|
},
|
|
// 分项 token(2026-08-02):cache/reasoning 透传自 round_usage,落库 + 累加器都需
|
|
prompt_cache_hit_tokens: round_usage.prompt_cache_hit_tokens,
|
|
prompt_cache_miss_tokens: round_usage.prompt_cache_miss_tokens,
|
|
reasoning_tokens: round_usage.reasoning_tokens,
|
|
};
|
|
(usage, is_estimated)
|
|
}
|
|
|
|
// ── emit_retry_attempt: 退避日志 + emit AiStreamRetry + sleep ──
|
|
// 前置:未耗尽重试预算(check_retry_exhausted 已返 None)。本函数完成退避后返回。
|
|
async fn emit_retry_attempt(
|
|
app_handle: &AppHandle,
|
|
conv_id: &str,
|
|
retry_attempt: usize,
|
|
max_retries: usize,
|
|
retry_deadline: tokio::time::Instant,
|
|
) {
|
|
let now = tokio::time::Instant::now();
|
|
let delay = retry::backoff_delay((retry_attempt + 1) as u32)
|
|
.min(retry_deadline.saturating_duration_since(now));
|
|
let total_retry = retry_attempt + 1;
|
|
tracing::warn!(
|
|
conv_id = %conv_id,
|
|
attempt = total_retry,
|
|
max_attempts = max_retries + 1,
|
|
delay_ms = delay.as_millis() as u64,
|
|
"[ai] 流前失败(Retryable),{}ms 后重试 ({}/{})",
|
|
delay.as_millis(), total_retry, max_retries + 1,
|
|
);
|
|
let ev = AiChatEvent::AiStreamRetry {
|
|
attempt: total_retry as u32,
|
|
max_attempts: (max_retries + 1) as u32,
|
|
conversation_id: Some(conv_id.to_string()),
|
|
};
|
|
emit_ai_chat_event(app_handle, ev);
|
|
tokio::time::sleep(delay).await;
|
|
}
|
|
|
|
/// 把 Anthropic prompt caching 易变尾段追加到请求消息流**末尾**(一条 user 消息)。
|
|
///
|
|
/// 多场景命中率设计要点:
|
|
/// - **放消息流末尾而非 system**:任何位于 system 之后的消息级缓存断点(含未来「多轮对话缓存」
|
|
/// 在末条 user 上打断点)都在易变段之前——易变内容(日期/清单/augmentation/知识)每日/每请求
|
|
/// 变化时,只影响其自身及之后(无),不失效已缓存的 system/历史前缀。若放首条 user,跨天日期
|
|
/// 变化会失效其后全部历史缓存。
|
|
/// - **用 user 角色而非 system**:anthropic_compat::convert_request 会把 System-role 消息抽到
|
|
/// 顶层 system(受 cache_control 缓存),易变段若为 system 会随请求变化失效整段 system 缓存。
|
|
/// 必须 user 承载,convert_request 对连续 user 做 merge_consecutive_users,追加安全。
|
|
/// - **不持久化**:仅注入本轮请求的 messages clone,per_conv.messages 不含易变段 → 每轮重建
|
|
/// 恰好一条,无重复。
|
|
pub(crate) fn merge_volatile_tail(messages: &mut Vec<ChatMessage>, tail: &str) {
|
|
if tail.is_empty() {
|
|
return;
|
|
}
|
|
messages.push(ChatMessage::user(tail));
|
|
}
|