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

新增: 批次工作落地(推进链/评估闭环/事件总线/并发/加固) + 技术债清理 + 文档整理

Browse Source 
后端:
- 工作流推进链(D-03):advance_task/状态机/闸门走 df-nodes Node trait,conditions 条件引擎扩展
- 想法评估闭环:启发式评分+对抗评估,df-ideas/scoring + df-storage/idea_eval_repo + idea 前端打通
- 全局事件数据总线:df-ai/context+context_helpers+augmentation 跨模块解耦
- AI planner/plan_hint/intent:aichat B 路线并行多轮基础
- patch_file 加固(TD-03/04):读改写整体锁防 lost update,expected_hash 合约闭环
- 压缩超时兜底(F-15 卡死根治)
- F-09 多会话并发:LlmConcurrency per-conv + streamingGuard 前端守护 + verify 脚本
- 知识注入 DRY/skills/audit 扩展

清理:
- aichat 技术债(误报 allow/死导入/过时注释 30 项)
- URGENT.md 删除(11 项加急全解决/迁 todo)
- 文档整理(todo/待决策/待审查/ARCHITECTURE/INDEX + 总线/技术债审查新文档)
This commit is contained in:
lxy
2026-06-21 20:51:26 +08:00
parent 330bb7f505
commit bd6a41fe6e
111 changed files with 11931 additions and 1033 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/commands/ai/knowledge_inject.rs
+266 -32
View File
@@ -85,9 +85,64 @@ async fn generate_embedding(
}
}
/// 知识条目发布时后台生成嵌入(fire-and-forget,失败仅 log)
/// 标记某条知识 embedding_status='failed'(G-2 收口:三处重复调用合一)。
///
/// 统一 warn 文案前缀(分支上下文 ctx 区分),失败本身非阻断(仅 log)。幂等(同值覆写)。
async fn mark_embedding_failed(repo: &KnowledgeRepo, id: &str, ctx: &str) {
if let Err(e) = repo.mark_embedding_failed(id).await {
tracing::warn!("标记 embedding_status=failed 失败({ctx},非阻断): {e}");
}
}
/// 后台 spawn 单条知识的嵌入生成任务(provider 已解析复用)。
///
/// 三个失败分支(写库失败 / 空向量 / provider Err)统一走 [`mark_embedding_failed`] 标 failed,
/// 可由 retry 入口补偿重试。成功(set_embedding 内已置 'done')仅 info log。
///
/// provider 以 Arc 共享:retry 路径解析一次 provider 复用传入多个并发子任务;发布路径单条同样可用。
fn spawn_embedding_task(
db: Arc<Database>,
provider: Arc<dyn LlmProvider>,
model: String,
id: String,
title: String,
content: String,
) {
tauri::async_runtime::spawn(async move {
let text = format!("{} {}", title, content);
let input: String = text.chars().take(8000).collect();
let repo = KnowledgeRepo::new(&db);
match provider.embed(&model, vec![input]).await {
Ok(vecs) if !vecs.is_empty() => {
// set_embedding 内已把 embedding_status 置 'done'。
if let Err(e) = repo.set_embedding(&id, &vecs[0]).await {
// 写库失败(非 provider 问题)同样标 failed,可重试。
tracing::warn!("嵌入写入失败(非阻断,标 failed 待补偿): {}", e);
mark_embedding_failed(&repo, &id, "写库失败").await;
} else {
tracing::info!("知识嵌入完成: {}", id);
}
}
Ok(_) => {
// provider 返回空向量(异常但非 Err):标 failed 可重试,优于静默丢弃。
mark_embedding_failed(&repo, &id, "空向量分支").await;
}
Err(e) => {
// provider 临时不可用(网络/限流/模型故障):标 failed 可补偿重试。
tracing::warn!("知识嵌入生成失败(非阻断,走 LIKE 降级,标 failed 待补偿): {}", e);
mark_embedding_failed(&repo, &id, "provider 错误").await;
}
}
});
}
/// 知识条目发布时后台生成嵌入(fire-and-forget,失败标记可补偿重试)
///
/// 由 knowledge_update_status(发布路径)调用。vector_enabled 关闭时直接跳过。
///
/// P1 修复(嵌入失败无标记):此前失败仅 warn,provider 临时不可用 → 该条永久无向量索引
/// 无人感知。现成功置 embedding_status='done'(set_embedding 内已含),失败置 'failed'
/// 并 warn,failed 条目可由 knowledge_retry_embedding IPC 触发补偿重试。
pub async fn spawn_embedding_for_knowledge(
state: &AppState,
record: &df_storage::models::KnowledgeRecord,
@@ -96,25 +151,80 @@ pub async fn spawn_embedding_for_knowledge(
if !config.vector_enabled {
return;
}
let Some((provider, model)) = resolve_embed_provider(state, &config).await else { return };
let text = format!("{} {}", record.title, record.content);
let id = record.id.clone();
let Some((provider, model)) = resolve_embed_provider(state, &config).await else {
// provider 解析失败(配置缺/密钥不可用)也标记 failed,允许用户修好配置后补偿重试。
// 老行为是静默 return(无人感知),现在落 failed 让状态可见可补。
let repo = KnowledgeRepo::new(&state.db);
mark_embedding_failed(&repo, &record.id, "provider 解析失败").await;
return;
};
// 发布路径单条:Box → Arc 装箱复用 spawn_embedding_task(统一嵌入逻辑,DRY)。
spawn_embedding_task(
state.db.clone(),
Arc::from(provider),
model,
record.id.clone(),
record.title.clone(),
record.content.clone(),
);
}
/// 补偿重试:对所有 embedding_status='failed' 的已发布知识重新生成嵌入(fire-and-forget)。
///
/// P1 修复(嵌入失败无标记)的补偿入口。由 knowledge_retry_embedding IPC 触发
/// (前端「重新生成向量」按钮)。立即返回待重试条数,后台逐条 spawn 子任务重跑
/// (成功置 done,失败仍 failed,用户可再次触发)。
///
/// 设计要点:
/// - **真并发 + IPC 立即返回**:IPC 同步拉 failed 列表算 count + 解析一次 provider/config 即返回;
/// 后台 spawn 一个总任务,**逐条 spawn 子任务**真并发跑嵌入。
/// 老实现 `for { spawn_embedding_for_knowledge().await }` 顺序串行,且每条重新
/// `knowledge_config.lock().clone()` + `build_provider_for` 重建 provider,
/// 与文档「立即返回、fire-and-forget」声明矛盾。
/// - **provider 解析一次复用**:IPC 路径解析一次 → Arc<dyn LlmProvider> 跨子任务共享,避免 N 条 N 次重建。
/// - **vector_enabled 关闭时返回 0**:与发布路径一致,无 provider 无意义。
/// - **provider 整体不可用**:解析失败时逐条标 failed(语义同发布路径 resolve 失败分支)。
pub async fn retry_failed_embeddings(state: &AppState) -> anyhow::Result<usize> {
let repo = KnowledgeRepo::new(&state.db);
let failed = repo.list_failed_embeddings().await?;
let count = failed.len();
if count == 0 {
return Ok(0);
}
tracing::info!("[knowledge] 启动 {} 条 failed 嵌入补偿重试", count);
// 解析一次 provider/config(IPC 同步,轻量:lock+clone 配置 + 一次 build_provider)。
let config = state.knowledge_config.lock().await.clone();
let db = state.db.clone();
if !config.vector_enabled {
return Ok(count); // 关闭:不动 failed 状态,仅返回条数(前端可据此提示)
}
let resolved = resolve_embed_provider(state, &config).await;
tauri::async_runtime::spawn(async move {
let input: String = text.chars().take(8000).collect();
match provider.embed(&model, vec![input]).await {
Ok(vecs) if !vecs.is_empty() => {
match resolved {
Some((provider, model)) => {
// Arc 共享:同一 provider 跨所有子任务复用(真并发,无重复重建)。
let provider: Arc<dyn LlmProvider> = Arc::from(provider);
for record in failed {
spawn_embedding_task(
db.clone(),
provider.clone(),
model.clone(),
record.id,
record.title,
record.content,
);
}
}
None => {
// provider 解析失败:逐条标 failed(允许用户修配置后再次触发补偿)。
let repo = KnowledgeRepo::new(&db);
if let Err(e) = repo.set_embedding(&id, &vecs[0]).await {
tracing::warn!("嵌入写入失败(非阻断): {}", e);
} else {
tracing::info!("知识嵌入完成: {}", id);
for record in failed {
mark_embedding_failed(&repo, &record.id, "retry provider 解析失败").await;
}
}
Ok(_) => {}
Err(e) => tracing::warn!("知识嵌入生成失败(非阻断,走 LIKE 降级): {}", e),
}
});
Ok(count)
}
/// 混合检索: LIKE 关键词 + 向量语义,合并去重加权
@@ -240,9 +350,69 @@ pub(crate) async fn build_knowledge_context(
out
}
/// 知识注入 system prompt 的单一入口(DRY:F-09 agentic + chat 五处合一)。
///
/// 把原本散落在 `try_continue_agent_loop`(agentic/mod.rs)+ `ai_chat_send` /
/// `ai_regenerate` / `ai_chat_edit` / `ai_chat_force_send`(chat.rs)五处逐行重复的
/// 「取末条 active user 文本 + id → build_knowledge_context → format 拼到 system_prompt 前」
/// 收敛至此。
///
/// **口径修复(②)**:原五处 `last_user_text` 走 `find(role==User && is_active())` 过滤
/// active,但 `user_message_id` 走 `last_user_message_id()` → `last_message_id_by_role`
/// 只判 role discriminant **不过滤 is_active**。末条 user 被压缩(`is_active=false`)后,
/// text 取到次末条 active user、id 取到末条(已压缩)user,**两值取自不同消息**,溯源错位。
/// 本 helper 在**同一次反向扫描同一条消息**取两值(text+id),根除口径漂移。
///
/// 流程:
/// 1. 单次 lock session → 读 per_conv.messages 全量克隆 → 反向扫末条 `role==User && is_active()`,
/// 同一消息取 `content`(检索 query)+ `id`(消息级溯源)。无 active user → text="" / id=None。
/// 2. `build_knowledge_context`(命中文本→混合检索→格式化)。auto_inject 关 / 无命中 → 返 ""。
/// 3. 非空 → `format!("{}\n\n---\n{}", knowledge, system_prompt)` 拼前;否则原样返回 system_prompt。
///
/// 注:`config` 由调用方从 `state.knowledge_config` lock().clone() 后传入(各调用方已在
/// 其他位置 clone 过,避免本 helper 重复加锁;亦兼容 agentic 续跑路径已 clone 的快照)。
pub(crate) async fn inject_knowledge_into_prompt(
state: &AppState,
conv_id: &str,
system_prompt: String,
config: &crate::state::KnowledgeConfig,
) -> String {
// 同一条消息取 text + id(②口径修复):单次反向扫描,避免 text 过滤 is_active 而 id 不过滤
// 导致两值取自不同消息。
let (last_user_text, user_message_id) = {
let session = state.ai_session.lock().await;
let msgs = session
.conv_read(conv_id)
.map(|c| c.messages.all_messages_clone())
.unwrap_or_default();
let found = msgs
.iter()
.rev()
.find(|m| matches!(m.role, MessageRole::User) && m.is_active());
match found {
Some(m) => (m.content.clone(), m.id.clone()),
None => (String::new(), None),
}
};
let knowledge_context =
build_knowledge_context(state, conv_id, &last_user_text, config, user_message_id.as_deref()).await;
if knowledge_context.is_empty() {
system_prompt
} else {
format!("{}\n\n---\n{}", knowledge_context, system_prompt)
}
}
/// 判断是否应触发提炼,满足则后台 spawn 提炼 task(非阻断)
///
/// 守卫: auto_extract 开 + trigger_mode == OnComplete + 消息数 ≥ min_messages
/// + 去重标志(P1 修复:本会话已成功提炼过则跳过,防重复触发刷 candidate)
///
/// TOCTOU 修复(🟡D):老实现 read(knowledge_extracted) → release lock → spawn,
/// 并发窗口内另一路 maybe_spawn_extraction 也能读到 false 各自 spawn,致重复提炼刷 candidate。
/// 现在判重 + 消息数 + **预置位**三步在**同一个锁临界区**内原子完成:置位即占用提炼槽位,
/// 并发的后来者读到 true 直接跳过。spawn 后按结果修正:0 条 / Err 清位(允许下次重试),
/// ≥1 条保持 true(已提炼,后续跳过)。
pub(crate) async fn maybe_spawn_extraction(
session_arc: &Arc<Mutex<AiSession>>,
db: &Arc<Database>,
@@ -257,24 +427,66 @@ pub(crate) async fn maybe_spawn_extraction(
if config.trigger_mode != ExtractTrigger::OnComplete {
return Ok(());
}
// 消息数守卫(总消息数,含 system/assistant/tool)
// F-260616-09 B 批4:per_conv.messages 唯一真相源(conv_id 来源:本函数入参)。
// conv_read 未建返 0(< min_messages 自然跳过,语义=空对话不注入知识)。
let msg_count = {
let session = session_arc.lock().await;
session.conv_read(conv_id).map(|c| c.messages.len()).unwrap_or(0)
};
if (msg_count as u32) < config.min_messages {
return Ok(());
// P1 修复(提炼重复触发)+ 🟡D TOCTOU 修复:判重 + 消息数守卫 + 预置位同一锁内原子完成。
// 背景:审批续跑 try_continue → 重 spawn run_agentic_loop → 正常完成块 →
// maybe_spawn_extraction 二次触发,无去重致同知识点重复 candidate 刷屏。
// 预置位语义:knowledge_extracted 在 spawn 前先置 true 占提炼槽位,并发后来者读到
// true 即跳过(等价「提炼中」哨兵);spawn 后按 inserted 结果修正(见下)。
// 清位:trigger_extraction_now(手动按钮)强制清位,允许用户手动重提炼。
{
let mut session = session_arc.lock().await;
// 一次 conv_read 读两个字段(knowledge_extracted + messages.len()),map 后借用即结束,
// 后续 session.conv()(&mut self)不再冲突。
let (already, count) = session
.conv_read(conv_id)
.map(|c| (c.knowledge_extracted, c.messages.len()))
.unwrap_or((false, 0));
// 守卫 1:已提炼过(或提炼中) → 跳过 + warn。
if already {
tracing::warn!(
conv_id = %conv_id,
"[knowledge] 跳过自动提炼:本会话已提炼过/提炼中(去重标志置位,防重复刷 candidate);如需重提炼用手动按钮"
);
return Ok(());
}
// 守卫 2:消息数。conv_read 未建返 0(< min_messages 自然跳过,语义=空对话不注入知识)。
if (count as u32) < config.min_messages {
return Ok(());
}
// 原子预置位(TOCTOU 核心):释放锁前先占提炼槽位,杜绝并发窗口内重复 spawn。
session.conv(conv_id).knowledge_extracted = true;
}
let session_arc = session_arc.clone();
let db = db.clone();
let conv_id = conv_id.to_string();
let provider_config = provider_config.clone();
let llm_concurrency = llm_concurrency.clone();
tauri::async_runtime::spawn(async move {
if let Err(e) = extract_knowledge_from_conversation(&db, &conv_id, &provider_config, &llm_concurrency).await {
tracing::warn!("知识提取失败(非阻断): {}", e);
match extract_knowledge_from_conversation(&db, &conv_id, &provider_config, &llm_concurrency).await {
Ok(inserted) => {
if inserted > 0 {
// ≥1 条:保持预置的 true(已占用槽位即最终状态),仅 log。
tracing::info!(
conv_id = %conv_id, inserted,
"[knowledge] 提炼成功,去重标志保持置位,后续自动触发将被跳过"
);
} else {
// 0 条:清位回 false,允许下次自动触发重试(对话可能后续补充了有价值内容)。
let mut session = session_arc.lock().await;
session.conv(&conv_id).knowledge_extracted = false;
tracing::info!(
conv_id = %conv_id,
"[knowledge] 提炼 0 条(无可提炼内容),清去重标志允许下次自动重试"
);
}
}
Err(e) => {
// 失败:清位回 false,允许下次触发重试(预置位已在,须回滚否则会话被锁死)。
tracing::warn!("知识提取失败(非阻断): {}", e);
let mut session = session_arc.lock().await;
session.conv(&conv_id).knowledge_extracted = false;
}
}
});
Ok(())
@@ -283,18 +495,35 @@ pub(crate) async fn maybe_spawn_extraction(
/// 手动触发提炼(ManualOnly 模式 / 前端按钮调用)
///
/// fire-and-forget:立即返回,后台执行 LLM 提炼(避免 IPC 长时间阻塞)。
///
/// P1 修复(提炼重复触发):手动路径强制清 knowledge_extracted 去重标志,
/// 允许用户手动重提炼(语义=手动覆盖自动去重)。
pub async fn trigger_extraction_now(state: &AppState) -> Result<bool, String> {
let conv_id = {
let session = state.ai_session.lock().await;
session.active_conversation_id.clone()
let mut session = state.ai_session.lock().await;
let conv_id = session.active_conversation_id.clone();
// 强制清去重标志:手动提炼是用户显式动作,允许对已提炼过的会话重提炼。
if let Some(cid) = conv_id.as_deref() {
session.conv(cid).knowledge_extracted = false;
}
conv_id
};
let conv_id = conv_id.ok_or_else(|| "当前无活跃对话".to_string())?;
let provider_config = super::prompt::get_active_provider(state).await.map_err(err_str)?;
let db = state.db.clone();
let session_arc = state.ai_session.clone();
let llm_concurrency = state.llm_concurrency.clone();
tauri::async_runtime::spawn(async move {
if let Err(e) = extract_knowledge_from_conversation(&db, &conv_id, &provider_config, &llm_concurrency).await {
tracing::warn!("手动提炼失败(非阻断): {}", e);
match extract_knowledge_from_conversation(&db, &conv_id, &provider_config, &llm_concurrency).await {
Ok(inserted) => {
// 手动提炼成功(≥1 条)后也置去重标志,保持与自动路径一致:
// 避免手动提炼后再触发自动提炼重复。用户再次手动按按钮会再次清位,行为自洽。
if inserted > 0 {
let mut session = session_arc.lock().await;
session.conv(&conv_id).knowledge_extracted = true;
}
}
Err(e) => tracing::warn!("手动提炼失败(非阻断): {}", e),
}
});
Ok(true)
@@ -316,12 +545,15 @@ const EXTRACTION_SYSTEM_PROMPT: &str = "你是知识提炼引擎,从 AI 对话
/// 从对话中提炼知识,产出 candidate 写入知识库
///
/// 流程: 读对话消息 → 过滤 user/assistant 取最后 6 条 → LLM 提炼(强制 JSON) → 解析 → 批量插入 candidate
///
/// 返回值: 成功插入的 candidate 条数(inserted)。调用方(maybe_spawn_extraction)据此置
/// `knowledge_extracted` 去重标志(≥1 才置位,0 条不置位—允许下次自动触发重试)。
async fn extract_knowledge_from_conversation(
db: &Arc<Database>,
conv_id: &str,
provider_config: &AiProviderRecord,
llm_concurrency: &LlmConcurrency,
) -> anyhow::Result<()> {
) -> anyhow::Result<usize> {
let conv_repo = AiConversationRepo::new(db);
let conv = conv_repo
.get_by_id(conv_id)
@@ -349,7 +581,7 @@ async fn extract_knowledge_from_conversation(
.take(6)
.collect();
if recent.len() < 4 {
return Ok(()); // 太短,不值得提炼
return Ok(0); // 太短,不值得提炼(0 条,不置去重标志)
}
// F-260619-04 P1 消息级溯源:取末条 assistant 消息 id(本轮 AI 产出知识的载体)。
@@ -415,7 +647,7 @@ async fn extract_knowledge_from_conversation(
Ok(v) => v,
Err(e) => {
tracing::warn!("知识提炼 JSON 解析失败,整批丢弃(非阻断): {} | 原始: {}", e, raw);
return Ok(());
return Ok(0);
}
};
@@ -457,6 +689,7 @@ async fn extract_knowledge_from_conversation(
None => format!("conv:{}", conv_id),
}),
reasoning: reasoning.clone(),
embedding_status: None,
created_at: now.clone(),
updated_at: now,
};
@@ -485,7 +718,7 @@ async fn extract_knowledge_from_conversation(
if inserted > 0 {
tracing::info!("知识提炼完成: 对话 {} 产出 {} 条 candidate", conv_id, inserted);
}
Ok(())
Ok(inserted)
}
/// 剥离 LLM 输出可能的 ```json ... ``` 代码块包裹
@@ -521,6 +754,7 @@ mod tests {
source_project: None,
source_ref: None,
reasoning: None,
embedding_status: None,
created_at: "2026-01-01".to_string(),
updated_at: "2026-01-01".to_string(),
}