重构: 失败采样与停滞判定纯函数提取独立模块

循环健康判定函数群外移,行为零变化
This commit is contained in:
lxy
2026-10-01 02:57:08 +08:00
parent 378047f1ff
commit 47a75a4d49
2 changed files with 295 additions and 278 deletions
+6 -278
View File
@@ -14,10 +14,7 @@ use df_ai::context::ContextManager;
// 改进3 B: 压缩失败兜底关键词摘要(纯函数 extract_keyword_summary)。
// 改进4: 工具结果 view-only 摘要(should_summarize_tool_result / extract_key_info)。
// extract_keyword_summary 随压缩逻辑迁至 context_lifecycle.rs(改进3 B 关键词兑底)。
use df_ai::context_helpers::{
extract_key_info, should_summarize_tool_result,
PLACEHOLDER_INTEGRITY_ENABLED,
};
use df_ai::context_helpers::PLACEHOLDER_INTEGRITY_ENABLED;
// 改进2 B:意图收敛工具(LLM 可见 tool_defs 按 intent 过滤,执行路径仍走完整 registry)
// B 路线 Phase 1:plan_hint 接入主 loop——filter_tool_defs_planned 在 filter_tool_defs
// 收敛的扁平子集之上叠加 plan_hint 编排(并行组同批聚拢/顺序依赖源在前),供 LLM 看到
@@ -359,7 +356,12 @@ pub mod conv_state;
pub mod command_lock;
pub(crate) mod helpers;
pub(crate) mod router_setup;
pub(crate) mod stall_detector;
pub(crate) use router_setup::resolve_model_with_override;
pub(crate) use stall_detector::{
check_retry_exhausted, count_recent_failures, last_round_has_match_evidence,
summarize_tool_results,
};
pub(crate) use helpers::try_continue_agent_loop;
pub(crate) use helpers::infer_goal_from_tool_calls;
pub(crate) use helpers::{
@@ -911,37 +913,6 @@ async fn stream_round(
}
}
// ── check_retry_exhausted: 判断 InitFailed Retryable 是否已耗尽重试预算 ──
// 返回 Some(error) 表示已耗尽(调用方应返回 InitFailedExhausted);None 表示可继续重试。
fn check_retry_exhausted(
conv_id: &str,
candidate: &AiProviderRecord,
retry_attempt: usize,
max_retries: usize,
retry_deadline: tokio::time::Instant,
error: &str,
) -> Option<String> {
let is_last = retry_attempt >= max_retries;
let now = tokio::time::Instant::now();
let budget_exhausted = now >= retry_deadline;
if !is_last && !budget_exhausted {
return None;
}
// 本 candidate 重试耗尽 / 预算耗尽 → 交外层切下一 candidate。
// 不在此 emit AiError(可能切下一 candidate 成功,emit 会留残留气泡);
// 携带 error 交外层「全 candidate 耗尽」时 emit 最后一条 AiError。
let reason = if budget_exhausted { "预算耗尽" } else { "耗尽" };
tracing::warn!(
conv_id = %conv_id,
provider = %candidate.name,
attempt = retry_attempt + 1,
budget_exhausted = budget_exhausted,
"[ai] 候选 {} 流前失败重试{}({}次),切下一 provider",
candidate.name, reason, max_retries + 1,
);
Some(error.to_string())
}
// ── emit_retry_attempt: 退避日志 + emit AiStreamRetry + sleep ──
// 前置:未耗尽重试预算(check_retry_exhausted 已返 None)。本函数完成退避后返回。
async fn emit_retry_attempt(
@@ -2223,122 +2194,6 @@ async fn heartbeat_loop(
}
}
// ── count_recent_failures: L1 断路器失败计数(扁平抽自原嵌套 5 层块) ──
// 读末尾连续 Tool 消息,失败包按结构化信号(tool_name + 错误类别)归一 key,
// 追加进滚动窗口 fail_window(超窗 CIRCUIT_BREAKER_WINDOW 淘汰最旧)。
// 返回 (max_count, sample_key):窗口内同 key 最高频次 + 该 key;max_count=0 表示无失败。
//
// 两处相对原实现的关键修正(原 key=内容前 40 字符 + 全 loop 累加不衰减):
// 1. key 归一:失败 envelope 内容(不同路径/参数)前 40 字符易把同类失败拆成多 key
// (治不住)或异类失败误聚成一类(误熔断)。改读结构化信号——
// is_failure_content 已结构化判成败,本函数按 tool_name + 错误类别(exit_code/错误关键词
// 哈希)归一 key,同一工具的同类错误稳定聚到同一 key。
// 2. 滚动窗口:全 loop 累加会让早期偶发失败与后期叠加触发误熔断。VecDeque 仅保留最近 N 条,
// 计数只反映最近失败密度——长任务中途偶发 1~2 次不触发,真正连续死循环才触发。
async fn count_recent_failures(
session_arc: &Arc<Mutex<AiSession>>,
conv_id: &str,
fail_window: &mut std::collections::VecDeque<String>,
) -> (u32, String) {
// BE2只克隆尾部 BREAKER_TAIL_N 条,避免 all_messages_clone 全量深克隆
// (长对话每轮 O(n) clone 浪费)。窗口覆盖末尾 Tool 结果与其前驱 assistant tool_calls
// (同轮工具回填通常 1~5 条,40 足富余;超窗工具映射退化为 unknown_tool,仅失败分组更粗,不误熔断)。
let messages = {
let session = session_arc.lock().await;
match session.conv_read(conv_id) {
Some(conv) => conv.messages.recent_messages(BREAKER_TAIL_N),
None => Vec::new(),
}
};
// 末尾连续 Tool 消息(本轮工具回填结果)。
let recent_tool_results: Vec<&ChatMessage> = messages
.iter().rev()
.take_while(|m| matches!(m.role, MessageRole::Tool))
.collect();
if recent_tool_results.is_empty() {
return (0u32, String::new());
}
// tool_call_id → tool_name 反查表:遍历 Assistant 消息的 tool_calls(开销与消息数线性)。
// tool_result 消息只带 tool_call_id,工具名在其前驱 Assistant 头里。
let mut id_to_name: std::collections::HashMap<&str, &str> = std::collections::HashMap::new();
for m in &messages {
if let MessageRole::Assistant = m.role {
if let Some(calls) = m.tool_calls.as_ref() {
for c in calls {
id_to_name.insert(c.id.as_str(), c.function.name.as_str());
}
}
}
}
// 本轮失败的归一 key 追加进窗口(顺序:消息正序,即工具执行顺序)。
for m in recent_tool_results.into_iter().rev() {
if is_failure_content(&m.content) {
let tool_name = m
.tool_call_id
.as_deref()
.and_then(|id| id_to_name.get(id).copied())
.unwrap_or("unknown_tool");
let key = failure_key(tool_name, &m.content);
fail_window.push_back(key);
while fail_window.len() > CIRCUIT_BREAKER_WINDOW {
fail_window.pop_front();
}
}
}
// 窗口内同 key 频次最高者 = 熔断候选。
let mut freq: std::collections::HashMap<&str, u32> = std::collections::HashMap::new();
for k in fail_window.iter() {
*freq.entry(k.as_str()).or_insert(0) += 1;
}
freq.into_iter()
.max_by_key(|(_, v)| *v)
.map(|(k, v)| (v, k.to_string()))
.unwrap_or((0u32, String::new()))
}
/// 归一失败 key:tool_name + 错误类别。保证同一工具的同类错误稳定映射到同一 key。
///
/// - run_command 等带 exit_code → `tool_name::exit=N`(非零退出码即失败类别)。
/// - status=failed/error envelope → `tool_name::error:<错误文本稳定哈希>`:
/// 用 fxhash 风格简单 a 哈希避免裸截内容前缀(前 40 字符易把同类失败拆多 key
/// 或异类误聚)。哈希值非敏感信息(仅断路器分组用),无碰撞顾虑(N=20 窗口内几无碰撞)。
/// - 内容解析失败(非 JSON)→ is_failure_content 已返回 false 不会进本函数,
/// 此处兜底返回 `tool_name::unknown` 防御。
fn failure_key(tool_name: &str, content: &str) -> String {
let Ok(v) = serde_json::from_str::<serde_json::Value>(content) else {
return format!("{}::unknown", tool_name);
};
if let Some(exit) = v.get("exit_code").and_then(|x| x.as_i64()) {
return format!("{}::exit={}", tool_name, exit);
}
// 取 error 字段文本做稳定哈希;无 error 字段则按 status 兜底。
let bucket = match v.get("status").and_then(|s| s.as_str()) {
Some("failed") => "failed",
Some("error") => "error",
_ => "other",
};
let err_text = v
.get("error")
.and_then(|e| e.as_str())
.unwrap_or("")
.trim();
if err_text.is_empty() {
return format!("{}::{}", tool_name, bucket);
}
let h = fnv1a_32(err_text.as_bytes());
format!("{}::{}:{:08x}", tool_name, bucket, h)
}
/// a 32-bit 哈希(零依赖,纯函数,断路器 key 归一专用;非密码学用途)。
fn fnv1a_32(bytes: &[u8]) -> u32 {
let mut hash: u32 = 0x811c9dc5;
for &b in bytes {
hash ^= b as u32;
hash = hash.wrapping_mul(0x0100_0193);
}
hash
}
// ============================================================
// 目标验证续轮(GOAL_VERIFY_ENABLED 门控)——治"过早宣称"的机制化兜底
// ============================================================
@@ -2351,40 +2206,6 @@ fn fnv1a_32(bytes: &[u8]) -> u32 {
// 主入口保障:三态兜底(applied 字段缺失/解析异常 → 视为无证据,不误伤也不阻断)、
// 上限熔断(goal_verify_rounds 达 MAX 强制收敛)、调工具重置计数(LLM 在行动非空转)。
/// 检查本轮(末尾连续 Tool 消息)是否有 applied.matched_target=true 的验证证据。
///
/// 读尾部 BREAKER_TAIL_N 条消息,取末尾连续 Tool 消息,解析 content JSON 的
/// `applied.matched_target` 字段。任一为 true 即返 true(本轮有达成证据)。
/// 三态兜底:字段缺失/非 JSON/解析异常 → 返 false(无证据),不阻断也不误判。
async fn last_round_has_match_evidence(
session_arc: &Arc<Mutex<AiSession>>,
conv_id: &str,
) -> bool {
let messages = {
let session = session_arc.lock().await;
match session.conv_read(conv_id) {
Some(conv) => conv.messages.recent_messages(BREAKER_TAIL_N),
None => Vec::new(),
}
};
for m in messages.iter().rev() {
if !matches!(m.role, MessageRole::Tool) {
break;
}
let Ok(v) = serde_json::from_str::<serde_json::Value>(&m.content) else {
continue;
};
if v.get("applied")
.and_then(|a| a.get("matched_target"))
.and_then(|x| x.as_bool())
.unwrap_or(false)
{
return true;
}
}
false
}
/// 注入目标验证续轮 nudge(强制续轮,非软提示)。
///
/// 文案随 goal_verify_rounds 递进:第 1 次请验证,第 2 次(将达上限)明确最后机会。
@@ -2418,99 +2239,6 @@ async fn insert_goal_verify_nudge(
}
}
// ── is_failure_content: 纯结构化判定(只读字段,绝不解析内容文本) ──
// 原理:工具成败是执行层的结构化事实(exit_code / status),断路器只读字段。内容文本(无论含
// error/失败/任何词)绝不参与判定 —— 这样读含 error 字样的代码、搜"失败"的结果等成功工具内容
// 永远不会被判失败(根除 conv 09e7abfa 的误熔断)。
// 失败 content 由 audit/mod.rs 包成 JSON envelope {"status":"failed","error":...};
// 成功 content 是 result JSON(run_command 含 exit_code,其他工具无 status 字段即成功)或 namespace:// 占位符。
// 非 JSON(namespace 占位符 / 重试 guard skip 纯文本)→ 无结构化失败信号 → 不判(false)。
fn is_failure_content(content: &str) -> bool {
let Ok(v) = serde_json::from_str::<serde_json::Value>(content) else {
return false; // namespace:// 占位符 / 重试 guard skip 等 → 无结构化失败信号,不判
};
if let Some(exit) = v.get("exit_code").and_then(|x| x.as_i64()) {
return exit != 0; // run_command 等:非零退出 = 真失败
}
matches!(v.get("status").and_then(|s| s.as_str()), Some("failed") | Some("error"))
|| v.get("error").is_some()
}
// ── summarize_tool_results: tool_result view-only 摘要压缩(扁平抽自原嵌套 map 块) ──
// 超阈值的 Tool 消息用 extract_key_info 压缩 content;非 Tool / 未超阈值 原样返回。
// 仅改入参 Vec(clone),不改 ContextManager 持久化(下次 build_for_request 从 DB 全量重建)。
fn summarize_tool_results(
messages: Vec<ChatMessage>,
history_tokens_snapshot: u32,
conv_id: &str,
iteration: usize,
) -> Vec<ChatMessage> {
let est = TokenEstimator::default();
let mut compressed_bytes: usize = 0;
let mut original_bytes: usize = 0;
let mut summarized_count: usize = 0;
// 构建 tool_call_id -> tool_name 映射:role=Tool 消息只带 tool_call_id(调用 ID),
// 真实工具名(read_file/run_command/...)在前序 role=Assistant 消息的 tool_calls[].function.name。
// 旧实现误把 tool_call_id(如 call_f892)当 tool_name 传给 extract_key_info,导致:
// 1) 摘要头注释显示 call_f892 而非 read_file(显示 bug);
// 2) extract_key_info 的 read_file 豁免永远命中不了(传入的不是真实名)。
// 用 owned String(非 &str 借用)避开与 messages.into_iter() 的所有权冲突;
// tool_call 数量极少(单轮几次),克隆开销可忽略。
let mut id_to_name: std::collections::HashMap<String, String> = std::collections::HashMap::new();
for m in &messages {
if let Some(calls) = m.tool_calls.as_ref() {
for c in calls {
id_to_name.insert(c.id.clone(), c.function.name.clone());
}
}
}
let msgs: Vec<ChatMessage> = messages
.into_iter()
.map(|mut m| {
if !matches!(m.role, df_ai::provider::MessageRole::Tool) {
return m;
}
let content_len = m.content.len();
let content_tokens = est.estimate_text(&m.content);
if !should_summarize_tool_result(content_len, history_tokens_snapshot, content_tokens) {
return m;
}
// 解析真实工具名:tool_call_id 查表;查不到(异常/老数据)退化为 "tool"。
let resolved_name = m
.tool_call_id
.as_ref()
.and_then(|id| id_to_name.get(id))
.map(|s| s.as_str())
.unwrap_or("tool");
// read_file 豁免压缩:AI 定向读代码,压缩 content 阉割代码分析能力。
// read_file handler 自带 limit 硬上限 2000 行,无爆 prompt 风险。
if resolved_name == "read_file" {
tracing::debug!(
conv_id = %conv_id, iteration,
content_len,
"[ai] read_file 工具结果豁免压缩(AI 定向读代码)"
);
return m;
}
original_bytes += content_len;
let compressed = extract_key_info(&m.content, resolved_name);
compressed_bytes += compressed.len();
summarized_count += 1;
m.content = compressed;
m
})
.collect();
if summarized_count > 0 {
tracing::info!(
conv_id = %conv_id, iteration, summarized_count, original_bytes, compressed_bytes,
"[ai] tool_result view-only 摘要压缩(不改持久化)"
);
}
msgs
}
// ── emit_circuit_breaker_tripped: L1 断路器熔断统一 emit(扁平抽自原嵌套块) ──
// 开关 CIRCUIT_BREAKER_HELP_EVENT: true=求助卡(默认), false=AiError 错误气泡(兜底回退)。
fn emit_circuit_breaker_tripped(app_handle: &AppHandle, conv_id: &str, max_count: u32, sample_key: &str) {