绝尘
cf017f81e2
重构：删 5 零引用 crate（df-evolve/plugin/stages/task/traceability）+ 清死模块、ai.rs 拆 11 子 module、ai.ts 拆 6 composable、i18n 拆目录 功能：知识库全栈（df-project/scan + CRUD + 时间线 + 前端）、Settings 拆分、appSettings KV 迁移、模型池、LLM 并发 Semaphore 修复：审批持久化根治、ConditionEngine 默认拒绝、NodeRegistry unimplemented 清除、promote 补偿删除、工具结果截断 50KB、路径校验防 symlink 逃逸 文档：B-03 人工审批设计、决策记录三分档、规格契约自检、经验记录、todo 看板、PROGRESS 更新 详见 PROGRESS.md。src-tauri/儿童每日打卡应用/ 与本项目无关，已排除。
272 lines
13 KiB
Rust
272 lines
13 KiB
Rust
//! Agentic 循环 — 流式接收 → 工具执行 → 结果回传 LLM → 循环
|
||
|
||
use std::sync::Arc;
|
||
use std::sync::atomic::Ordering;
|
||
|
||
use tauri::{AppHandle, Emitter};
|
||
use tokio::sync::Mutex;
|
||
|
||
use df_ai::ai_tools::AiToolRegistry;
|
||
use df_ai::context::TokenEstimator;
|
||
use df_ai::provider::{ChatMessage, CompletionRequest, LlmProvider};
|
||
|
||
use df_storage::db::Database;
|
||
use df_storage::models::AiProviderRecord;
|
||
|
||
use crate::state::{AppState, LlmConcurrency};
|
||
|
||
use super::conversation::{save_conversation, TokenAccumulator};
|
||
use super::knowledge_inject::maybe_spawn_extraction;
|
||
use super::prompt::{build_system_prompt, get_active_provider};
|
||
use super::stream_recv::stream_llm;
|
||
use super::title::{ensure_conversation_title, spawn_ensure_title};
|
||
use super::audit::process_tool_calls;
|
||
|
||
use super::{AiChatEvent, AiSession};
|
||
|
||
/// Agentic 循环最大迭代次数
|
||
pub(crate) const MAX_AGENT_ITERATIONS: usize = 10;
|
||
|
||
/// Agentic 循环：流式接收 → 工具执行 → 结果回传 LLM → 循环
|
||
///
|
||
/// 退出条件：
|
||
/// - LLM 只返回文本（无 tool_calls）→ 正常结束
|
||
/// - 有工具需要审批 → 暂停循环（generating 保持 true），等 ai_approve 恢复
|
||
/// - 达到最大迭代次数 → 正常结束
|
||
pub(crate) async fn run_agentic_loop(
|
||
session_arc: Arc<Mutex<AiSession>>,
|
||
tools_arc: Arc<AiToolRegistry>,
|
||
db: Arc<Database>,
|
||
app_handle: AppHandle,
|
||
provider_config: AiProviderRecord,
|
||
system_prompt: String,
|
||
conv_id: String,
|
||
knowledge_config: crate::state::KnowledgeConfig,
|
||
llm_concurrency: LlmConcurrency,
|
||
) {
|
||
let provider: Box<dyn LlmProvider> = df_ai::build_provider(
|
||
&provider_config.provider_type,
|
||
&provider_config.base_url,
|
||
&provider_config.api_key,
|
||
&provider_config.default_model,
|
||
);
|
||
let tool_defs = tools_arc.tool_definitions();
|
||
// 停止信号副本：stream_llm 与每轮迭代共享读取，避免重复加锁
|
||
let stop_flag = session_arc.lock().await.stop_flag.clone();
|
||
|
||
// token 累加器:loop 生命周期内各轮叠加,退出时传 save_conversation(累加模式落库)
|
||
let mut tokens = TokenAccumulator::default();
|
||
|
||
for iteration in 0..MAX_AGENT_ITERATIONS {
|
||
// 用户请求停止 → 收尾退出（已生成文本已在上一轮入库）
|
||
if stop_flag.load(Ordering::SeqCst) {
|
||
let usage = df_ai::provider::TokenUsage {
|
||
prompt_tokens: tokens.prompt(),
|
||
completion_tokens: tokens.completion(),
|
||
total_tokens: tokens.total(),
|
||
};
|
||
// 入口 stop:本轮可能尚未 stream(首轮即停),不记 model——避免把未实际生成的 model 写入 models 数组
|
||
save_conversation(&session_arc, &db, &conv_id, Some(&usage), None).await;
|
||
// 标题生成后台化：不阻塞 Completed emit（失败有 extract_title 兜底）
|
||
spawn_ensure_title(&provider_config, &db, &conv_id, &app_handle, &session_arc, &llm_concurrency);
|
||
let mut session = session_arc.lock().await;
|
||
session.generating = false;
|
||
// generating 复位后再 emit Completed：保证前端收事件时后端已可接下一条(发送队列续发不被"正在生成中"拒绝)
|
||
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiCompleted { total_tokens: usage.total_tokens, prompt_tokens: tokens.prompt(), completion_tokens: tokens.completion(), conversation_id: Some(conv_id.clone()) });
|
||
return;
|
||
}
|
||
|
||
// 新一轮通知前端（第二轮起），前端需新建 assistant 消息
|
||
if iteration > 0 {
|
||
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiAgentRound {
|
||
round: (iteration + 1) as u32,
|
||
conversation_id: Some(conv_id.clone()),
|
||
});
|
||
}
|
||
|
||
// 构建请求消息（超预算时自动裁剪旧消息，保护工具调用三元组 + 最近 6 条）
|
||
let messages = {
|
||
let session = session_arc.lock().await;
|
||
let sys_tokens = TokenEstimator::default().estimate_text(&system_prompt);
|
||
let (history_msgs, _trimmed) = session.messages.build_for_request(sys_tokens);
|
||
let mut msgs = vec![ChatMessage::system(&system_prompt)];
|
||
msgs.extend(history_msgs);
|
||
msgs
|
||
};
|
||
|
||
// 预估输入 token(兜底:部分 provider 如 GLM 流式 usage 不报 prompt_tokens,后段用它补)
|
||
let estimated_prompt: u32 = {
|
||
let est = TokenEstimator::default();
|
||
messages.iter().map(|m| est.estimate_message(m)).sum()
|
||
};
|
||
|
||
let request = CompletionRequest {
|
||
model: provider_config.default_model.clone(),
|
||
messages,
|
||
temperature: Some(0.7),
|
||
max_tokens: Some(8192),
|
||
stream: true,
|
||
tools: if tool_defs.is_empty() { None } else { Some(tool_defs.clone()) },
|
||
tool_choice: None,
|
||
};
|
||
|
||
// LLM 并发限流（全局 + 单对话双层），仅覆盖 stream_llm 调用本身;
|
||
// 工具执行(process_tool_calls)是本地操作无 RPM 成本,permit 在 stream 后立即释放避免占槽
|
||
let _global_permit = llm_concurrency.acquire_global().await;
|
||
let _per_conv_permit = llm_concurrency.acquire_per_conv().await;
|
||
// 流式接收（内部处理 idle timeout / 断连检测 / 停止信号）
|
||
let (full_text, tool_calls_acc, round_usage) = match stream_llm(&*provider, request, &app_handle, &stop_flag, &conv_id).await {
|
||
Some(result) => result,
|
||
None => {
|
||
// 错误已在 stream_llm 中 emit，直接结束
|
||
let mut session = session_arc.lock().await;
|
||
session.generating = false;
|
||
return;
|
||
}
|
||
};
|
||
// stream 结束立即释放 permit,后续工具执行不受限流(本地操作无 RPM 成本)
|
||
drop(_global_permit);
|
||
drop(_per_conv_permit);
|
||
|
||
// 累加本轮 token:provider 流式 usage 的 prompt_tokens 为 0 时(GLM 等),用预估输入兜底
|
||
let round_prompt = if round_usage.prompt_tokens == 0 { estimated_prompt } else { round_usage.prompt_tokens };
|
||
tokens.add(round_prompt, round_usage.completion_tokens);
|
||
|
||
// 追加 assistant 消息到历史
|
||
let has_tool_calls = !tool_calls_acc.is_empty();
|
||
{
|
||
let mut session = session_arc.lock().await;
|
||
if has_tool_calls {
|
||
let mut order: Vec<u32> = tool_calls_acc.keys().copied().collect();
|
||
order.sort_unstable();
|
||
let ai_tool_calls: Vec<df_ai::provider::ToolCall> = order.iter()
|
||
.map(|i| {
|
||
let draft = &tool_calls_acc[i];
|
||
df_ai::provider::ToolCall::new(&draft.id, &draft.name, &draft.args)
|
||
})
|
||
.collect();
|
||
let mut msg = ChatMessage::assistant_with_tools(&full_text, ai_tool_calls);
|
||
msg.model = Some(provider_config.default_model.clone());
|
||
session.messages.push(msg);
|
||
} else if !full_text.is_empty() {
|
||
let mut msg = ChatMessage::assistant(&full_text);
|
||
msg.model = Some(provider_config.default_model.clone());
|
||
session.messages.push(msg);
|
||
}
|
||
}
|
||
|
||
// 停止信号：已生成文本入库后退出，不再执行后续工具调用
|
||
if stop_flag.load(Ordering::SeqCst) {
|
||
let usage = df_ai::provider::TokenUsage {
|
||
prompt_tokens: tokens.prompt(),
|
||
completion_tokens: tokens.completion(),
|
||
total_tokens: tokens.total(),
|
||
};
|
||
save_conversation(&session_arc, &db, &conv_id, Some(&usage), Some(&provider_config.default_model)).await;
|
||
// 标题生成后台化：不阻塞 Completed emit（失败有 extract_title 兜底）
|
||
spawn_ensure_title(&provider_config, &db, &conv_id, &app_handle, &session_arc, &llm_concurrency);
|
||
let mut session = session_arc.lock().await;
|
||
session.generating = false;
|
||
// generating 复位后再 emit Completed：保证前端收事件时后端已可接下一条(发送队列续发不被"正在生成中"拒绝)
|
||
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiCompleted { total_tokens: usage.total_tokens, prompt_tokens: tokens.prompt(), completion_tokens: tokens.completion(), conversation_id: Some(conv_id.clone()) });
|
||
return;
|
||
}
|
||
|
||
// 无工具调用 → 最终文本响应，循环结束
|
||
if !has_tool_calls { break; }
|
||
|
||
// 处理工具调用（Low 自动执行 / Medium+High 待审批）
|
||
let pending_count = {
|
||
let mut session = session_arc.lock().await;
|
||
process_tool_calls(&mut session, tool_calls_acc, &tools_arc, &db, &app_handle, &conv_id).await
|
||
};
|
||
|
||
// 有待审批 → 暂停循环，等待用户审批后通过 ai_approve → try_continue_agent_loop 恢复
|
||
if pending_count > 0 {
|
||
let usage = df_ai::provider::TokenUsage {
|
||
prompt_tokens: tokens.prompt(),
|
||
completion_tokens: tokens.completion(),
|
||
total_tokens: tokens.total(),
|
||
};
|
||
save_conversation(&session_arc, &db, &conv_id, Some(&usage), Some(&provider_config.default_model)).await;
|
||
return; // generating 保持 true
|
||
}
|
||
|
||
// 全部自动执行完成 → 继续下一轮
|
||
}
|
||
|
||
// 正常完成
|
||
let usage = df_ai::provider::TokenUsage {
|
||
prompt_tokens: tokens.prompt(),
|
||
completion_tokens: tokens.completion(),
|
||
total_tokens: tokens.total(),
|
||
};
|
||
// 落库 + 标题 + 知识提炼打包后台化：不阻塞 generating 复位与 Completed 事件
|
||
// save 先行(extract/title 都读已落库消息);extract 内部 fire-and-forget,与 title 可能并发
|
||
// (均受 per_conv 信号量约束,读写不同字段互不干扰)
|
||
// 并发取舍：与新对话新 loop 的 save 存在低概率并发 upsert，最多丢少量 token 累加（非功能错误，可接受）
|
||
let usage_total = usage.total_tokens;
|
||
{
|
||
let session_arc = session_arc.clone();
|
||
let db = db.clone();
|
||
let conv_id = conv_id.clone();
|
||
let provider_config = provider_config.clone();
|
||
let knowledge_config = knowledge_config.clone();
|
||
let app_handle = app_handle.clone();
|
||
let llm_concurrency = llm_concurrency.clone();
|
||
tauri::async_runtime::spawn(async move {
|
||
save_conversation(&session_arc, &db, &conv_id, Some(&usage), Some(&provider_config.default_model)).await;
|
||
// 知识提炼:需读已落库的对话消息,故在 save 之后
|
||
if let Err(e) = maybe_spawn_extraction(&session_arc, &db, &conv_id, &provider_config, &knowledge_config, llm_concurrency.clone()).await {
|
||
tracing::warn!("知识提炼触发失败(非阻断): {}", e);
|
||
}
|
||
ensure_conversation_title(&provider_config, &db, &conv_id, &app_handle, &session_arc, llm_concurrency).await;
|
||
});
|
||
}
|
||
|
||
let mut session = session_arc.lock().await;
|
||
session.generating = false;
|
||
// generating 复位后再 emit Completed：落库/标题/提炼已在后台，前端立即感知完成
|
||
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiCompleted { total_tokens: usage_total, prompt_tokens: tokens.prompt(), completion_tokens: tokens.completion(), conversation_id: Some(conv_id.clone()) });
|
||
}
|
||
|
||
/// 检查是否所有待审批已处理，如果是则恢复 agentic 循环
|
||
pub(crate) async fn try_continue_agent_loop(app: &AppHandle, state: &AppState) {
|
||
let should_continue = {
|
||
let session = state.ai_session.lock().await;
|
||
session.generating && session.pending_approvals.is_empty()
|
||
};
|
||
|
||
if !should_continue { return; }
|
||
|
||
let provider_config = match get_active_provider(state).await {
|
||
Ok(p) => p,
|
||
Err(_) => return,
|
||
};
|
||
let (lang, conv_id) = {
|
||
let session = state.ai_session.lock().await;
|
||
let lang = session.agent_language.clone().unwrap_or_else(|| "zh-CN".to_string());
|
||
let conv_id = session.active_conversation_id.clone().unwrap_or_default();
|
||
(lang, conv_id)
|
||
};
|
||
let system_prompt = build_system_prompt(state, &lang).await;
|
||
|
||
let session_arc = state.ai_session.clone();
|
||
let tools_arc = state.ai_tools.clone();
|
||
let db = state.db.clone();
|
||
let app_handle = app.clone();
|
||
let knowledge_config = state.knowledge_config.lock().await.clone();
|
||
let llm_concurrency = state.llm_concurrency.clone();
|
||
|
||
// 恢复循环前通知前端新建 assistant 消息：审批(通过/拒绝)后新一轮文本
|
||
// 不应追加到发起工具调用的旧消息，用 AiAgentRound 隔开
|
||
let _ = app.emit("ai-chat-event", AiChatEvent::AiAgentRound {
|
||
round: 0,
|
||
conversation_id: Some(conv_id.clone()),
|
||
});
|
||
|
||
tauri::async_runtime::spawn(async move {
|
||
run_agentic_loop(session_arc, tools_arc, db, app_handle, provider_config, system_prompt, conv_id, knowledge_config, llm_concurrency).await;
|
||
});
|
||
}
|