//! 知识库集成 — 注入 + 提炼(嵌入生成 / 混合检索 / 上下文构建 / 对话提炼) use std::collections::HashMap; use std::sync::{Arc, LazyLock, Mutex as StdMutex}; use tokio::sync::Mutex; use df_ai::provider::{ChatMessage, CompletionRequest, LlmProvider, MessageRole}; // F-01 阶段5: 知识提炼 / 嵌入两路路由。 // - 提炼: TaskRequirements(Standard,无工具)— 在对话 provider 的 model_configs 池中选。 // - 嵌入: TaskRequirements(Capability::Embedding,无工具)— 在 embedding provider 的池中选, // 池空兜底 config.embedding_model(行为不变,现有 KnowledgeConfig 配置即生效)。 use df_ai::router::{ is_likely_image_model, select_model_id, Modality, TaskRequirements, }; use df_storage::crud::{AiConversationRepo, AiMessageRepo, AiProviderRepo, KnowledgeRepo}; use df_storage::db::Database; use df_storage::models::{AiProviderRecord, KnowledgeRecord}; use df_types::types::new_id; use crate::state::{AppState, ExtractTrigger, LlmConcurrency, log_slow_lock}; use crate::commands::ai::audit::truncate_chars; use crate::commands::err_str; use super::{AiSession}; // ============================================================ // 知识注入/提炼参数(集中收口,不散魔法数) // ============================================================ /// 注入检索 query 最多保留的关键词个数(按出现频次取前 N) /// /// 多词会放大 SQL 命中面并稀释「标题命中 > 内容命中」的排序精度,6 个足以覆盖 /// 一句提问的信息量(ASCII 专有名词 + 核心汉字 2-gram)。 const KNOWLEDGE_QUERY_MAX_KEYWORDS: usize = 6; /// ASCII 词最短长度(单字符词如 a/x/i 全表乱命中,过滤) const KNOWLEDGE_QUERY_MIN_ASCII_LEN: usize = 2; /// 自动提炼窗口:取最近 N 条 user/assistant 消息送 LLM 提炼 /// /// 原 6 条窗口过窄,长会话后半程的经验不在提炼视野内;40 条覆盖近期完整上下文, /// 同时不撑爆提炼调用的 token 预算(max_tokens 4096 的输出侧不受影响,输入侧由 /// 提炼模型上下文窗口兜底)。 const KNOWLEDGE_EXTRACT_WINDOW: usize = 40; /// 自动提炼频次:距上次提炼至少新增 N 条用户消息才再提炼 /// /// 取代原「每会话仅提炼一次」的 bool 硬去重——长会话后半程的新经验此前永不提炼。 const KNOWLEDGE_EXTRACT_MIN_NEW_USER_MSGS: usize = 10; /// 自动提炼频次水位(conv_id → 上次提炼进度),见 [`EXTRACT_WATERMARKS`]。 #[derive(Clone, Copy)] enum ExtractWatermark { /// 提炼进行中(spawn 前置位,完成/失败后改写,防并发两路同时触发) InFlight, /// 上次成功提炼完成时的用户消息累计条数(含已压缩消息,保证单调递增) Done(usize), } /// 提炼频次水位表(进程内,不落库)。 /// /// 原「每会话仅提炼一次」的 `knowledge_extracted` bool 去重使长会话后半程的新经验 /// 永不提炼(知识库走查 2026-09-29);改为水位制:距上次提炼累计新增 /// [`KNOWLEDGE_EXTRACT_MIN_NEW_USER_MSGS`] 条用户消息可再提炼。 /// 放本模块静态而非 PerConvState:后者归属会话状态层(本次改动边界外),且水位随进程 /// 生命周期存续即可——重启丢失后最多多提炼一轮,无正确性风险。 static EXTRACT_WATERMARKS: LazyLock>> = LazyLock::new(|| StdMutex::new(HashMap::new())); /// 统计会话的用户消息累计条数(提炼频次水位的计量单位) fn count_user_rounds<'a>(msgs: impl Iterator) -> usize { msgs.filter(|m| matches!(m.role, MessageRole::User)).count() } // ============================================================ // 检索 query 分词(整句 LIKE 命中率≈0 的根治,见 search_keywords) // ============================================================ /// 中文判断(CJK 统一表意文字区,覆盖日常中文) fn is_cjk_char(c: char) -> bool { matches!(c, '\u{4E00}'..='\u{9FFF}') } /// ASCII 词字符(字母数字下划线) fn is_ascii_word_char(c: char) -> bool { c.is_ascii_alphanumeric() || c == '_' } /// 高频虚词停用表(2-gram 粒度):中文疑问/指代/客套组合 + 英文虚词。 /// ASCII 词抽取时已统一小写,英文停用词按小写收录;命中即弃,不参与检索。 const QUERY_STOPWORDS: &[&str] = &[ "怎么", "如何", "什么", "为何", "帮我", "看看", "这个", "那个", "一下", "可以", "应该", "需要", "是否", "如果", "但是", "所以", "因为", "还是", "就是", "没有", "一个", "这些", "那些", "然后", "现在", "已经", "我们", "你们", "他们", "自己", "时候", "我看", "请把", "觉得", "感觉", "好像", "对于", "关于", "the", "and", "for", "with", "this", "that", "from", "have", "what", "how", "can", "you", "are", "was", "not", "but", "about", "into", "use", "please", "help", "want", "need", "should", "would", "could", "there", "here", "when", "where", "which", "your", ]; /// 从用户消息抽取检索关键词:ASCII 词(≥2 字符,统一小写)+ CJK 2-gram,按频次取前 N /// /// 背景(知识库走查 2026-09-29):原把整条用户消息当 LIKE pattern,中文长句几乎不可能 /// 作为连续子串命中知识正文,注入检索近乎失效。不上分词器:提问多为短句, /// 2-gram + 停用词已够;不依赖 df-ai 的同思路实现(df-storage 在依赖链下层,反向依赖会成环), /// 故此处自实现最小分词。停用词过滤同时压掉客套话与指代词的噪音 2-gram。 /// /// 返回按出现频次降序的前 [`KNOWLEDGE_QUERY_MAX_KEYWORDS`] 个,同频次 ASCII 词优先于 /// CJK 2-gram(2-gram 是无词典近似噪音多,等频抢占容量时精确词让位),组内按首次 /// 出现顺序稳定排序(检索 query 确定,行为可测)。 pub(crate) fn extract_query_keywords(text: &str) -> Vec { let mut tokens: Vec = Vec::new(); let mut ascii_buf = String::new(); let mut cjk_buf: Vec = Vec::new(); for c in text.chars() { if is_ascii_word_char(c) { // ASCII 词打断了汉字连续段:先结算汉字 2-gram 再收集 ASCII flush_cjk_bigrams(&mut cjk_buf, &mut tokens); ascii_buf.push(c.to_ascii_lowercase()); } else { if !ascii_buf.is_empty() { tokens.push(std::mem::take(&mut ascii_buf)); } if is_cjk_char(c) { cjk_buf.push(c); } else { // 空白/标点等分隔符:结算当前汉字段 flush_cjk_bigrams(&mut cjk_buf, &mut tokens); } } } if !ascii_buf.is_empty() { tokens.push(ascii_buf); } flush_cjk_bigrams(&mut cjk_buf, &mut tokens); // 过滤(过短/停用词)后按频次降序取前 N;同频次 ASCII 词优先于 CJK 2-gram, // 组内按首现顺序稳定排序 let ascii_word = |t: &str| t.starts_with(is_ascii_word_char); let mut freq: Vec<(String, usize, usize)> = Vec::new(); for (idx, t) in tokens.into_iter().enumerate() { if t.chars().count() < KNOWLEDGE_QUERY_MIN_ASCII_LEN || QUERY_STOPWORDS.contains(&t.as_str()) { continue; } if let Some(entry) = freq.iter_mut().find(|(tok, _, _)| tok == &t) { entry.1 += 1; } else { freq.push((t, 1, idx)); } } freq.sort_by(|a, b| { b.1.cmp(&a.1) .then_with(|| ascii_word(&b.0).cmp(&ascii_word(&a.0))) .then_with(|| a.2.cmp(&b.2)) }); freq .into_iter() .take(KNOWLEDGE_QUERY_MAX_KEYWORDS) .map(|(t, _, _)| t) .collect() } /// 结算 CJK 连续段:剥离段内停用词后两两切 2-gram。 /// /// 中文无空格分词,2-gram 是无词典近似的最小可靠粒度;孤单汉字不产出 /// (单字命中率低噪音高)。段长 <2 时 windows(2) 自然产出空,无需特判。 /// /// 停用词须在切 gram 前替换为分隔符打断连续段:切完再整词过滤滤不掉 /// 停用词与邻字的跨词组合 gram("帮我看看这个怎么修复"会产生"看这/个怎"噪音)。 /// 表内 CJK 停用词均为 2 字词,ASCII 停用词不会命中纯汉字段,统一替换无副作用; /// 剥离后单字残留不成 gram 自然丢弃,与孤单汉字语义一致。 fn flush_cjk_bigrams(buf: &mut Vec, out: &mut Vec) { let mut segment: String = buf.iter().collect(); for w in QUERY_STOPWORDS { if segment.contains(w) { segment = segment.replace(w, "\u{0}"); } } for part in segment.split('\u{0}') { for w in part.chars().collect::>().windows(2) { out.push(w.iter().collect()); } } buf.clear(); } /// 按配置构建 embedding provider + model。None = 配置缺失/provider 不存在。 /// /// F-01 阶段5: model 选择路由优先 — 在 embedding provider 的 model_configs 池中按 /// TaskRequirements(needs_tool_use=false)选最优;池空(未拉取)兜底 config.embedding_model。 /// /// 注:TaskRequirements 无 capabilities 字段(router 仅按 needs_tool_use 过滤 ToolUse), /// 嵌入模型由 embedding provider 的池构成(用户在 KnowledgeSettings 配 embedding provider, /// 该 provider 的 model_configs 通常即嵌入模型),故 needs_tool_use=false 宽松过滤即可命中。 async fn resolve_embed_provider( providers: &AiProviderRepo, config: &crate::state::KnowledgeConfig, ) -> Option<(Box, String)> { let id = config.embedding_provider_id.as_ref()?; let fallback_model = config.embedding_model.clone().unwrap_or_else(|| "embedding-3".to_string()); let rec = match providers.get_by_id(id).await { Ok(Some(r)) => r, _ => { tracing::warn!("embedding provider 不存在: {}", id); return None; } }; // build_provider_for 含空 key 早失败:Err → 返回 None 触发 LIKE 降级(与原 embed 失败降级行为一致) let provider = match super::secret::build_provider_for(&rec) { Ok(p) => p, Err(e) => { tracing::warn!("embedding provider 密钥不可用(降级 LIKE): {}", e); return None; } }; // 嵌入路由:needs_tool_use=false(排除 ToolUse 专要求,允许嵌入模型入选)。 // select_model_id None(池空/无匹配)→ 兜底 config.embedding_model(行为不变)。 let embed_req = TaskRequirements { modalities: vec![Modality::Text], needs_tool_use: false, estimated_context: 0, tier: None, }; let model = select_model_id(&embed_req, &rec.model_configs).unwrap_or(fallback_model); Some((provider, model)) } /// 生成文本嵌入(向量检索用) /// /// 用配置指定的 embedding provider(必须 openai_compat 类型),失败返回 None(降级 LIKE)。 async fn generate_embedding( state: &AppState, text: &str, config: &crate::state::KnowledgeConfig, ) -> Option> { let (provider, model) = resolve_embed_provider(&state.ai_providers, config).await?; // 截断防超 token 上限(8192 token ≈ 8000 中文字) let input: String = text.chars().take(8000).collect(); match provider.embed(&model, vec![input]).await { Ok(mut vecs) if !vecs.is_empty() => Some(vecs.remove(0)), Ok(_) => None, Err(e) => { tracing::warn!("embedding 生成失败(降级 LIKE): {}", e); None } } } /// 标记某条知识 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, provider: Arc, 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, ) { let config = state.knowledge_config.lock().await.clone(); spawn_embedding_for_record(&state.db, &config, record).await; } /// [`spawn_embedding_for_knowledge`] 的 db+config 版,嵌入语义与发布路径完全一致: /// vector_enabled 关闭跳过;provider 解析失败标 failed(修好配置后可补偿重试)。 /// /// 供只持 db 无 AppState 上下文的调用方使用(AI 工具 handler 捕获仅 Arc, /// 自动晋升在 spawn 的后台任务内同样拿不到 AppState),配置经 /// [`crate::state::load_knowledge_config`] 从 KV 现读,与 IPC 路径同一真相源。 pub(crate) async fn spawn_embedding_for_record( db: &Arc, config: &crate::state::KnowledgeConfig, record: &KnowledgeRecord, ) { if !config.vector_enabled { return; } let Some((provider, model)) = resolve_embed_provider(&AiProviderRepo::new(db), config).await else { let repo = KnowledgeRepo::new(db); mark_embedding_failed(&repo, &record.id, "provider 解析失败").await; return; }; // 单条:Box → Arc 装箱复用 spawn_embedding_task(统一嵌入逻辑,DRY)。 spawn_embedding_task( 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 跨子任务共享,避免 N 条 N 次重建。 /// - **vector_enabled 关闭时返回 0**:与发布路径一致,无 provider 无意义。 /// - **provider 整体不可用**:解析失败时逐条标 failed(语义同发布路径 resolve 失败分支)。 pub async fn retry_failed_embeddings(state: &AppState) -> anyhow::Result { 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.ai_providers, &config).await; tauri::async_runtime::spawn(async move { match resolved { Some((provider, model)) => { // Arc 共享:同一 provider 跨所有子任务复用(真并发,无重复重建)。 let provider: Arc = 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); for record in failed { mark_embedding_failed(&repo, &record.id, "retry provider 解析失败").await; } } } }); Ok(count) } /// 混合检索: LIKE 关键词 + 向量语义,合并去重加权 /// /// 双信号(两路都命中)排最前,LIKE 单信号次之,向量单信号第三。 /// vector_enabled 关闭或 embed 失败时纯 LIKE(零外部依赖降级)。 /// LIKE 路 query 先经 [`extract_query_keywords`] 分词(整句 LIKE 命中率≈0); /// 向量路 query 同步改用关键词拼接(虚词稀释 embedding 语义),分词为空时回退原句。 async fn hybrid_search( state: &AppState, query: &str, limit: usize, config: &crate::state::KnowledgeConfig, ) -> Vec { let keywords = extract_query_keywords(query); let keyword_results = match state .knowledge .search_keywords(&keywords, config.include_candidates, limit) .await { Ok(v) => v, Err(e) => { tracing::warn!(error = %e, "[ai] 知识关键词检索失败,降级空结果"); Vec::new() } }; if !config.vector_enabled { return keyword_results; } let embed_query = if keywords.is_empty() { query.to_string() } else { keywords.join("、") }; let query_vec = match generate_embedding(state, &embed_query, config).await { Some(v) => v, None => return keyword_results, // embed 失败降级 }; let vector_results = match state.knowledge.search_vector(&query_vec, limit).await { Ok(v) => v, Err(e) => { tracing::warn!(error = %e, "[ai] 知识向量检索失败,降级空结果"); Vec::new() } }; merge_hybrid_results(keyword_results, vector_results, limit) } /// 混合检索三层合并去重(纯函数,抽自 hybrid_search) /// /// 排序:双信号(LIKE + 向量均命中)> 仅 LIKE 单信号 > 仅向量单信号(且 cos≥0.3)。 /// cos<0.3 的向量单信号结果丢弃防噪音;limit 截断;按 id 去重。 pub(crate) fn merge_hybrid_results( keyword_results: Vec, vector_results: Vec<(KnowledgeRecord, f32)>, limit: usize, ) -> Vec { // 合并去重: 双信号 > LIKE 单信号 > 向量单信号(相似度<0.3 的向量结果丢弃防噪音) let keyword_ids: std::collections::HashSet = keyword_results.iter().map(|r| r.id.clone()).collect(); let mut merged = Vec::new(); let mut seen = std::collections::HashSet::new(); // 1. 双信号 for (rec, score) in &vector_results { if keyword_ids.contains(&rec.id) && seen.insert(rec.id.clone()) { tracing::debug!("混合检索双信号: {} (cos={:.2})", rec.title, score); merged.push(rec.clone()); } } // 2. LIKE 单信号 for rec in &keyword_results { if seen.insert(rec.id.clone()) { merged.push(rec.clone()); } } // 3. 向量单信号(过滤低相似度) for (rec, score) in &vector_results { if *score >= 0.3 && seen.insert(rec.id.clone()) { merged.push(rec.clone()); } } merged.truncate(limit); merged } /// 构建知识库上下文片段,拼入 Chat system prompt /// /// 流程: 开关检查 → 混合检索 top-3(克制) → 命中条目 reuse_count +1 + 记录引用事件(fire-and-forget) → markdown 格式化 /// 关闭时返回空串(零开销);无结果返回空串。 /// /// F-260619-04 P1 消息级溯源:`user_message_id` 为触发本轮检索的 user 消息 id, /// 传入 referenced 事件溯源(用户问题命中知识)。None 表示无 user 消息/老数据无 id, /// 降级 conv:{conv_id} 对话级,展示侧兼容。 pub(crate) async fn build_knowledge_context( state: &AppState, conv_id: &str, query: &str, config: &crate::state::KnowledgeConfig, user_message_id: Option<&str>, ) -> String { if !config.auto_inject { return String::new(); } let results = hybrid_search(state, query, 3, config).await; if results.is_empty() { return String::new(); } // 命中条目:复用计数 +1 + candidate 复用达标自动晋升 + 记录引用事件(fire-and-forget,单个 spawn 任务批量处理) let db = state.db.clone(); // (id, 是否 candidate):晋升检查只对 candidate 发起,published 免一次注定空跑的 UPDATE let hits: Vec<(String, bool)> = results .iter() .map(|r| (r.id.clone(), r.status == "candidate")) .collect(); let conv_id = conv_id.to_string(); let query_clone = query.to_string(); let user_message_id = user_message_id.map(|s| s.to_string()); // 晋升开关与阈值取配置快照值拷贝进闭包(决策单未拍板前默认保守,开关默认关); // 配置整体快照一并拷入,晋升成功后补嵌入(spawn_embedding_for_record)用 let embed_config = config.clone(); let auto_publish_enabled = config.auto_publish_enabled; let auto_publish_threshold = config.auto_publish_reuse_threshold; tauri::async_runtime::spawn(async move { let repo = KnowledgeRepo::new(&db); let timeline = crate::commands::knowledge_timeline::KnowledgeTimeline::new(&db); for (id, is_candidate) in &hits { if let Err(e) = repo.increment_reuse_count(id).await { tracing::warn!("reuse_count +1 失败(非阻断): {}", e); } // 复用即信任:开关开启时 candidate 达到阈值自动晋升 published(SQL 侧原子判定, // 重复调用幂等)。默认关(决策卡 C3 未拍板前保守),开启后阈值取配置。 if *is_candidate && auto_publish_enabled { match repo.auto_publish_if_reused(id, auto_publish_threshold).await { Ok(true) => { tracing::info!("[knowledge] candidate 复用达标({auto_publish_threshold} 次),自动晋升 published: {id}"); // 晋升留痕:生命线记 status_changed(source_ref=auto_publish), // fire-and-forget 与上方 record_referenced 同风格,失败仅 warn 非阻断。 timeline.record_auto_publish(id, auto_publish_threshold).await; // 晋升即进检索面,补嵌入对齐 IPC 发布路径(失败标 failed 可补偿重试, // 非阻断;vector_enabled 关闭时内部直接跳过) if let Ok(Some(rec)) = repo.get_by_id(id).await { spawn_embedding_for_record(&db, &embed_config, &rec).await; } } Ok(false) => {} Err(e) => tracing::warn!("candidate 自动晋升检查失败(非阻断): {e}"), } } timeline .record_referenced(id, &conv_id, &query_clone, user_message_id.as_deref()) .await; } }); let mut out = String::from("## 相关知识库\n"); for r in &results { let kind = r.kind.clone(); let title = r.title.clone(); // 截断 content 防膨胀(注入侧最多 500 字符;走 audit 正典,截断时尾部补 … 提示有省略) let snippet: String = truncate_chars(&r.content, 500); out.push_str(&format!("- [{}] {}: {} (复用 {} 次)\n", kind, title, snippet, r.reuse_count)); } out } /// 取末条 active user 消息文本 + id,构建知识库上下文段(DRY:注入 [`inject_knowledge_into_prompt`] /// 与 Anthropic prompt caching 易变段共用)。 /// /// 同一条消息取 text + id(②口径修复):单次反向扫描,避免 text 过滤 is_active 而 id 不过滤 /// 导致两值取自不同消息。auto_inject 关 / 无命中返回空串。 pub(crate) async fn build_knowledge_context_for_conv( state: &AppState, conv_id: &str, config: &crate::state::KnowledgeConfig, ) -> String { let (last_user_text, user_message_id) = { let __lock_t = std::time::Instant::now(); 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()); log_slow_lock(__lock_t, "build_knowledge_context_for_conv:582"); match found { Some(m) => (m.content.clone(), m.id.clone()), None => (String::new(), None), } }; build_knowledge_context(state, conv_id, &last_user_text, config, user_message_id.as_deref()).await } /// 知识注入 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 { let knowledge_context = build_knowledge_context_for_conv(state, conv_id, config).await; if knowledge_context.is_empty() { system_prompt } else { // 知识拼 system_prompt 后(固定 base 前置,提升 provider prompt cache 命中率) format!("{}\n\n---\n{}", system_prompt, knowledge_context) } } /// 判断是否应触发提炼,满足则后台 spawn 提炼 task(非阻断) /// /// 守卫: auto_extract 开 + trigger_mode == OnComplete + 消息数 ≥ min_messages /// + 频次水位(提炼中 InFlight,或距上次提炼新增用户消息 /// < [`KNOWLEDGE_EXTRACT_MIN_NEW_USER_MSGS`] 则跳过)。 /// 原「每会话仅提炼一次」的 bool 硬去重已由水位制取代——长会话后半程的新经验 /// 此前永不提炼(知识库走查 2026-09-29)。 /// /// TOCTOU 防护保持:守卫判定 + **预置位**(InFlight 水位 + knowledge_extracted 哨兵) /// 在**同一个锁临界区**内原子完成,并发窗口内的后来者读到占用即跳过,防重复提炼刷 /// candidate。spawn 后按结果修正:0 条 / Err 清位回退水位(允许下次重试), /// ≥1 条记 Done 水位(此后须再攒够阈值条用户消息才再提炼)。 pub(crate) async fn maybe_spawn_extraction( session_arc: &Arc>, db: &Arc, conv_id: &str, provider_config: &AiProviderRecord, config: &crate::state::KnowledgeConfig, llm_concurrency: LlmConcurrency, ) -> anyhow::Result<()> { if !config.auto_extract { return Ok(()); } if config.trigger_mode != ExtractTrigger::OnComplete { return Ok(()); } // 频次水位守卫 + 消息数守卫 + 预置位在同一锁临界区内原子完成(TOCTOU 防护与原 P1 修复同思路: // 审批续跑 try_continue → 重 spawn run_agentic_loop → 正常完成块会二次触发本函数, // 无占位致并发重复提炼刷 candidate)。 // 清位/回退:trigger_extraction_now(手动按钮)强制清哨兵 + 水位,允许用户立即手动重提炼。 { let __lock_t = std::time::Instant::now(); let mut session = session_arc.lock().await; // 一次 conv_read 读两个字段(消息总数守卫 + 用户消息水位计量),map 后借用即结束, // 后续 session.conv()(&mut self)不再冲突。 let (msg_count, user_rounds) = session .conv_read(conv_id) .map(|c| (c.messages.len(), count_user_rounds(c.messages.iter()))) .unwrap_or((0, 0)); // 守卫 1:频次水位(InFlight=提炼中防并发;Done=新增用户消息不足阈值)。 // 替代原 bool 硬去重;跳过走 debug 留痕(改水位制后此路径高频,warn 会刷屏)。 let blocked = { let wms = EXTRACT_WATERMARKS.lock().expect("提炼水位互斥锁中毒"); match wms.get(conv_id) { Some(ExtractWatermark::InFlight) => true, Some(ExtractWatermark::Done(last)) => { user_rounds.saturating_sub(*last) < KNOWLEDGE_EXTRACT_MIN_NEW_USER_MSGS } None => false, } }; if blocked { log_slow_lock(__lock_t, "maybe_spawn_extraction:692"); tracing::debug!( conv_id = %conv_id, "[knowledge] 跳过自动提炼:提炼中或距上次提炼新增用户消息不足 {} 条(如需立即重提炼用手动按钮)", KNOWLEDGE_EXTRACT_MIN_NEW_USER_MSGS ); return Ok(()); } // 守卫 2:消息数。conv_read 未建返 0(< min_messages 自然跳过,语义=空对话不注入知识)。 if msg_count < config.min_messages as usize { log_slow_lock(__lock_t, "maybe_spawn_extraction:705"); return Ok(()); } // 原子预置位(TOCTOU 核心):释放锁前先占提炼槽位,杜绝并发窗口内重复 spawn。 session.conv(conv_id).knowledge_extracted = true; EXTRACT_WATERMARKS .lock() .expect("提炼水位互斥锁中毒") .insert(conv_id.to_string(), ExtractWatermark::InFlight); log_slow_lock(__lock_t, "maybe_spawn_extraction:717"); } 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 { match extract_knowledge_from_conversation(&db, &conv_id, &provider_config, &llm_concurrency).await { Ok(inserted) => { if inserted > 0 { // ≥1 条:哨兵保持置位,水位记为当前用户消息进度 // (此后须再新增 ≥ 阈值条用户消息才会再提炼)。 let session = session_arc.lock().await; let rounds = session .conv_read(&conv_id) .map(|c| count_user_rounds(c.messages.iter())) .unwrap_or(0); EXTRACT_WATERMARKS .lock() .expect("提炼水位互斥锁中毒") .insert(conv_id.clone(), ExtractWatermark::Done(rounds)); tracing::info!( conv_id = %conv_id, inserted, rounds, "[knowledge] 提炼成功,水位记为当前进度,新增用户消息达阈值后才再提炼" ); } else { // 0 条:清哨兵回滚预置位;水位记 Done 保节奏(原回退水位会让下一轮对话结束 // 立即重提炼,同窗口反复失败时形成逐轮风暴;记进度后须再攒够阈值条新用户消息)。 let mut session = session_arc.lock().await; let rounds = session .conv_read(&conv_id) .map(|c| count_user_rounds(c.messages.iter())) .unwrap_or(0); session.conv(&conv_id).knowledge_extracted = false; EXTRACT_WATERMARKS .lock() .expect("提炼水位互斥锁中毒") .insert(conv_id.clone(), ExtractWatermark::Done(rounds)); tracing::info!( conv_id = %conv_id, rounds, "[knowledge] 提炼 0 条(无可提炼内容),水位记为当前进度,新增用户消息达阈值后才再试" ); } } Err(e) => { // 失败:清哨兵回滚预置位;水位同样记 Done 保节奏(原回退水位在持续故障 // (截断/网络/密钥)时逐轮重试刷爆;达阈值后才再试,手动按钮仍可立即重提炼)。 tracing::warn!("知识提取失败(非阻断): {}", e); let mut session = session_arc.lock().await; let rounds = session .conv_read(&conv_id) .map(|c| count_user_rounds(c.messages.iter())) .unwrap_or(0); session.conv(&conv_id).knowledge_extracted = false; EXTRACT_WATERMARKS .lock() .expect("提炼水位互斥锁中毒") .insert(conv_id.clone(), ExtractWatermark::Done(rounds)); tracing::info!( conv_id = %conv_id, rounds, "[knowledge] 提炼失败,水位记为当前进度,新增用户消息达阈值后才再试" ); } } }); Ok(()) } /// 手动触发提炼(ManualOnly 模式 / 前端按钮调用) /// /// fire-and-forget:立即返回,后台执行 LLM 提炼(避免 IPC 长时间阻塞)。 /// /// P1 修复(提炼重复触发):手动路径强制清 knowledge_extracted 去重标志, /// 允许用户手动重提炼(语义=手动覆盖自动去重)。 pub async fn trigger_extraction_now(state: &AppState) -> Result { let conv_id = { 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; EXTRACT_WATERMARKS .lock() .expect("提炼水位互斥锁中毒") .remove(cid); } 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 { 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; let rounds = session .conv_read(&conv_id) .map(|c| count_user_rounds(c.messages.iter())) .unwrap_or(0); EXTRACT_WATERMARKS .lock() .expect("提炼水位互斥锁中毒") .insert(conv_id.clone(), ExtractWatermark::Done(rounds)); } } Err(e) => tracing::warn!("手动提炼失败(非阻断): {}", e), } }); Ok(true) } /// 知识提炼提示词 — 强制 JSON 输出,含矛盾知识约束 const EXTRACTION_SYSTEM_PROMPT: &str = "你是知识提炼引擎,从 AI 对话中识别可复用的经验。\ 只提取真正通用、可被未来对话复用的知识,过滤一次性闲聊/项目特定的临时内容。\n\n\ 输出严格的 JSON 数组(不要 markdown 代码块包裹),每个元素 schema:\n\ {\"kind\": \"review_rule|prompt_template|pitfall|architecture_pattern|diagnosis|deployment_note|workflow_optimization\", \ \"title\": \"简短标题\", \"content\": \"完整可复用内容\", \ \"tags\": [\"标签\"], \"confidence\": \"high|medium|low\", \"reasoning\": \"为何值得沉淀\"}\n\n\ 规则:\n\ 1. 如果适用范围有限制(如仅适用特定语言/框架/场景),必须在 content 或 tags 中明确标注\n\ 2. confidence: high=对话中可直接观察的明确模式, medium=合理推断, low=推测性弱信号\n\ 3. 无可提炼内容时返回空数组 []\n\ 4. 输出纯 JSON,无任何额外文字\n\ 5. 最多提取 5 条,按复用价值降序排列(只保留高价值条目,宁缺毋滥)"; /// 从对话中提炼知识,产出 candidate 写入知识库 /// /// 流程: 读对话消息 → 过滤 user/assistant 取最近 [`KNOWLEDGE_EXTRACT_WINDOW`] 条 → /// LLM 提炼(强制 JSON) → 解析 → 批量插入 candidate /// /// 返回值: 成功插入的 candidate 条数(inserted)。调用方(maybe_spawn_extraction)据此置 /// `knowledge_extracted` 去重标志(≥1 才置位,0 条不置位—允许下次自动触发重试)。 async fn extract_knowledge_from_conversation( db: &Arc, conv_id: &str, provider_config: &AiProviderRecord, llm_concurrency: &LlmConcurrency, ) -> anyhow::Result { let conv_repo = AiConversationRepo::new(db); let conv = conv_repo .get_by_id(conv_id) .await? .ok_or_else(|| anyhow::anyhow!("对话不存在: {}", conv_id))?; // 对话标题(生命线溯源用,空标题降级为占位,避免字节切片风险) let conv_title = conv .title .clone() .filter(|t| !t.trim().is_empty()) .unwrap_or_else(|| "未命名对话".to_string()); // 消息读取:优先 ai_messages 表(消息拆分存储真相源),表空时 fallback 旧 messages JSON 列(老库兼容) // 注意:DB 查询 Err 不能吞成空 Vec(否则 records.is_empty() 误为真 → 走旧 messages JSON 回退, // 语义错:本应报 DB 故障)。这里显式 match:Ok 正常流程,Err 记 warn 后跳过本轮知识抽取。 let msg_repo = AiMessageRepo::new(db); let records = match msg_repo.list_by_conversation(conv_id).await { Ok(records) => records, Err(e) => { tracing::warn!( error = %e, conv_id, "[KNOWLEDGE-EXTRACT] list_by_conversation 失败,跳过本轮知识抽取" ); return Ok(0); // DB 故障,不当空数据回退(0 条,不置去重标志,允许下次重试) } }; let messages: Vec = if !records.is_empty() { records.iter().map(crate::commands::ai::commands::record_to_message).collect() } else { // 老库未迁移或坏数据:回退旧 messages JSON 列(与 ai_conversation_switch 同一兼容路径) let has_legacy = conv.messages != "[]" && !conv.messages.is_empty(); if has_legacy { tracing::warn!( conv_id, "[KNOWLEDGE-EXTRACT] ai_messages 表为空,回退旧 messages JSON 列(老库兼容)" ); } // 损坏 → match Err 分流:勿 unwrap_or_default 吞成空 Vec(空 Vec 会让 knowledge 抽取基于空上下文, // 误产空知识)。坏数据 warn 后跳过本轮(0 条,不置去重标志,允许下次重试),与 DB 故障同语义。 match serde_json::from_str::>(&conv.messages) { Ok(v) => v, Err(e) => { tracing::warn!( error = %e, conv_id, "[KNOWLEDGE-EXTRACT] 旧 messages JSON 列解析失败,跳过本轮知识抽取(勿基于空上下文抽取)" ); return Ok(0); } } }; // 过滤 user/assistant,取最近 KNOWLEDGE_EXTRACT_WINDOW 条 // (原 6 条窗口过窄:长会话后半程的经验不在提炼视野内;放宽后单次提炼可覆盖近期完整上下文) let recent: Vec<&ChatMessage> = messages .iter() .filter(|m| matches!(m.role, MessageRole::User | MessageRole::Assistant)) .rev() .take(KNOWLEDGE_EXTRACT_WINDOW) .collect(); if recent.len() < 4 { return Ok(0); // 太短,不值得提炼(0 条,不置去重标志) } // F-260619-04 P1 消息级溯源:取末条 assistant 消息 id(本轮 AI 产出知识的载体)。 // 反向扫描全量 messages(不限 recent 6 条,确保取到对话最新 assistant,即便它不在 // 提炼窗口内也属于本轮 AI 产出)。老数据消息无 id → None,source_ref 降级 None, // 展示侧兼容 conv: 旧格式 + 无 source_ref。 let last_assistant_msg_id: Option<&str> = messages .iter() .rev() .find(|m| matches!(m.role, MessageRole::Assistant)) .and_then(|m| m.id.as_deref()); // 构造提炼消息: system 指令 + 对话内容(user 角色) let mut conv_text = String::new(); for m in recent.iter().rev() { let role = match m.role { MessageRole::User => "用户", MessageRole::Assistant => "助手", _ => continue, }; conv_text.push_str(&format!("[{}]: {}\n\n", role, m.content)); } let extract_messages = vec![ ChatMessage::system(EXTRACTION_SYSTEM_PROMPT), ChatMessage::user(&format!("请从以下对话中提炼可复用知识:\n\n{}", conv_text)), ]; // F-01 阶段5: 知识提炼路由 — TaskRequirements(Standard,无工具)。 // select_model_id None(池空/无匹配)→ 兜底 default_model(行为不变)。 let extract_req = TaskRequirements { modalities: vec![Modality::Text], needs_tool_use: false, estimated_context: 0, tier: None, }; let extract_model = select_model_id(&extract_req, &provider_config.model_configs) .unwrap_or_else(|| provider_config.default_model.clone()); // 防御:选中/兜底模型是图像模型时跳过 LLM 提取(避免文本任务 404 刷屏,治 default_model 配错为图像模型)。 if is_likely_image_model(&extract_model, &provider_config.model_configs) { tracing::warn!( "知识提取跳过:模型 {} 是图像模型,文本任务不可用(检查 provider default_model 应为文本模型)", extract_model ); return Ok(0); } let request = CompletionRequest { model: extract_model, messages: extract_messages, temperature: Some(0.3), max_tokens: Some(4096), stream: false, tools: None, tool_choice: None, reasoning_content: None, }; let provider: Box = match super::secret::build_provider_for(provider_config) { Ok(p) => p, // 空 key 早失败:上层(maybe_spawn_extraction/trigger_extraction_now)已 warn log 降级,语义一致 Err(e) => return Err(anyhow::anyhow!("provider 密钥不可用: {}", e)), }; // LLM 并发限流（知识提炼属独立调用，纳入双层 Semaphore） // F-09 B 批5: per_conv 改 HashMap,知识提炼针对本对话,用 conv_id 共享该 conv 限流槽。 let _global_permit = llm_concurrency.acquire_global().await; let _per_conv_permit = llm_concurrency.acquire_per_conv(conv_id).await; let resp = provider.complete(request).await?; let raw = resp.text.trim(); // 容错:剥离可能的 ```json ... ``` 包裹 let json_str = strip_code_fence(raw); let items: Vec = match serde_json::from_str(json_str) { Ok(v) => v, Err(e) => match salvage_truncated_json_array(json_str) { // 截断打捞:max_tokens 截断常把数组拦腰砍断,先救回已完整闭合的条目 Some(salvaged) if !salvaged.is_empty() => { tracing::warn!("知识提炼截断打捞: 救回 {} 条,丢弃尾部残条", salvaged.len()); salvaged } // 打捞后 0 条才整批丢弃;空输出(content 空/null)与解析失败单独留痕区分 _ => { if raw.is_empty() { tracing::warn!( "知识提炼输出为空(content 空或 null),整批丢弃, completion_tokens={}", resp.usage.completion_tokens ); } else { tracing::warn!( "知识提炼 JSON 解析失败且打捞无果,整批丢弃(非阻断): {} | completion_tokens={} | 原始: {}", e, resp.usage.completion_tokens, raw ); } return Ok(0); } }, }; let knowledge_repo = KnowledgeRepo::new(db); let timeline = crate::commands::knowledge_timeline::KnowledgeTimeline::new(db); let mut inserted = 0; for item in &items { let kind = match item.get("kind").and_then(|v| v.as_str()) { Some(k) => k.to_string(), None => continue, }; let title = item.get("title").and_then(|v| v.as_str()).unwrap_or("").to_string(); let content = item.get("content").and_then(|v| v.as_str()).unwrap_or("").to_string(); if title.is_empty() || content.is_empty() { continue; } let tags = item.get("tags").map(|v| serde_json::to_string(v).unwrap_or_else(|_| "[]".into())); let confidence = item.get("confidence").and_then(|v| v.as_str()).map(|s| s.to_string()); // 回填 AI 判断依据(prompt 要求的 reasoning 字段,此前被丢弃) let reasoning = item.get("reasoning").and_then(|v| v.as_str()).map(|s| s.to_string()); let now = crate::commands::now_millis(); let record = KnowledgeRecord { id: new_id(), kind, title, content, tags, status: "candidate".to_string(), confidence, reuse_count: 0, verified: false, source_project: None, // F-260619-04 P1 消息级溯源:conv_msg:{assistant_msg_id} 精确定位本轮 AI 产出。 // last_assistant_msg_id 为 None(老数据无 id)→ 降级 conv:{conv_id} 对话级, // 展示侧双格式解析兼容。 source_ref: Some(match last_assistant_msg_id { Some(mid) => format!("conv_msg:{}", mid), None => format!("conv:{}", conv_id), }), reasoning: reasoning.clone(), embedding_status: None, created_at: now.clone(), updated_at: now, }; match knowledge_repo.insert(record.clone()).await { Ok(_) => { inserted += 1; tracing::info!( "AI 提炼知识候选: {} [confidence={}]", record.title, record.confidence.as_deref().unwrap_or("?") ); // 生命线:AI 提炼产生(fire-and-forget) timeline .record_extracted( &record.id, conv_id, &conv_title, reasoning.as_deref().unwrap_or(""), last_assistant_msg_id, ) .await; } Err(e) => tracing::warn!("知识候选插入失败(非阻断): {}", e), } } if inserted > 0 { tracing::info!("知识提炼完成: 对话 {} 产出 {} 条 candidate", conv_id, inserted); } Ok(inserted) } /// 剥离 LLM 输出可能的 ```json ... ``` 代码块包裹 fn strip_code_fence(s: &str) -> &str { let s = s.trim(); if let Some(rest) = s.strip_prefix("```json") { return rest.trim().trim_end_matches("```").trim(); } if let Some(rest) = s.strip_prefix("```") { return rest.trim().trim_end_matches("```").trim(); } s } /// 截断 JSON 打捞 — 从被截断的数组输出中救回已完整闭合的顶层条目。 /// /// 背景:提炼输出被 max_tokens 拦腰截断(`[{"a":1},{"b":` )时 serde 整批报错, /// 此前整批丢弃。扫描跳过字符串字面量内部(含 `\"` 转义)的括号,记录数组元素层 /// (depth 归 1)最后一次 `}` 的位置 — 等价于"从尾向前扫到最后一个完整顶层 `}`" /// (引号开闭与转义只有正向扫描可可靠判定,正向记录末次闭合与反向扫描结果一致)。 /// 截到该位补 `]` 重解析,serde 通过且非空才由调用方入库;空串/纯垃圾/无完整条目 /// 返回 None(调用方走原丢弃路径)。 fn salvage_truncated_json_array(raw: &str) -> Option> { let bytes = raw.as_bytes(); let mut in_string = false; let mut escaped = false; let mut depth: i64 = 0; let mut last_elem_close: Option = None; for (i, &b) in bytes.iter().enumerate() { if in_string { if escaped { escaped = false; // 转义字符本身,不参与状态判定 } else if b == b'\\' { escaped = true; } else if b == b'"' { in_string = false; } continue; } match b { b'"' => in_string = true, b'{' | b'[' => depth += 1, b'}' | b']' => { depth -= 1; if depth == 1 && b == b'}' { // 数组元素层闭合:最后一个可打捞条目的右边界 last_elem_close = Some(i); } if depth <= 0 { // 顶层 `]`(完整数组)或深度为负(垃圾前缀):数组层已到界,停止 break; } } _ => {} } } let cut = last_elem_close?; let candidate = format!("{}]", &raw[..=cut]); serde_json::from_str(&candidate).ok() } #[cfg(test)] mod tests { use super::*; use df_storage::models::KnowledgeRecord; // ---------- merge_hybrid_results ---------- fn kr(id: &str, title: &str) -> KnowledgeRecord { KnowledgeRecord { id: id.to_string(), kind: "snippet".to_string(), title: title.to_string(), content: String::new(), tags: None, status: "published".to_string(), confidence: None, reuse_count: 0, verified: false, 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(), } } #[test] fn merge_empty_inputs() { let out = merge_hybrid_results(vec![], vec![], 5); assert!(out.is_empty()); } #[test] fn merge_dual_signal_ranks_first() { // r1 同时命中双信号 → 应排在首位 let kw = vec![kr("r1", "kw1"), kr("r2", "kw2")]; let vec_results = vec![(kr("r1", "kw1-vec"), 0.8)]; let out = merge_hybrid_results(kw, vec_results, 5); assert_eq!(out.len(), 2); assert_eq!(out[0].id, "r1", "双信号 r1 必须排首"); assert_eq!(out[1].id, "r2"); } #[test] fn merge_keyword_only_after_dual() { // r2 仅 LIKE,应在双信号之后 let kw = vec![kr("only-kw", "kw-only")]; let vec_results = vec![(kr("dual", "dual-vec"), 0.7)]; // dual 不在 kw 集合 → 非双信号,走向量单信号(0.7≥0.3) let out = merge_hybrid_results(kw, vec_results, 5); // 顺序:无双信号 → LIKE 单信号(only-kw)→ 向量单信号(dual) assert_eq!(out.len(), 2); assert_eq!(out[0].id, "only-kw"); assert_eq!(out[1].id, "dual"); } #[test] fn merge_vector_threshold_filters_below_03() { // cos=0.29 < 0.3 → 向量单信号结果被滤掉 let kw = vec![]; let vec_results = vec![(kr("low", "low-vec"), 0.29)]; let out = merge_hybrid_results(kw, vec_results, 5); assert!(out.is_empty(), "cos=0.29 应被过滤"); } #[test] fn merge_vector_threshold_keeps_at_031() { // cos=0.31 ≥ 0.3 → 保留 let kw = vec![]; let vec_results = vec![(kr("ok", "ok-vec"), 0.31)]; let out = merge_hybrid_results(kw, vec_results, 5); assert_eq!(out.len(), 1); assert_eq!(out[0].id, "ok"); } #[test] fn merge_vector_threshold_boundary_exact_03() { // 边界:cos 恰好 0.3 → 保留(>= 比较) let kw = vec![]; let vec_results = vec![(kr("edge", "edge-vec"), 0.3)]; let out = merge_hybrid_results(kw, vec_results, 5); assert_eq!(out.len(), 1, "cos=0.3 边界应保留(>= 比较)"); assert_eq!(out[0].id, "edge"); } #[test] fn merge_truncates_to_limit() { // limit 截断 let kw: Vec = (0..10).map(|i| kr(&format!("k{i}"), "t")).collect(); let out = merge_hybrid_results(kw, vec![], 3); assert_eq!(out.len(), 3); } #[test] fn merge_dedups_across_signals() { // 同一 id 多路命中只出现一次(双信号路径优先) let kw = vec![kr("dup", "dup-kw")]; let vec_results = vec![(kr("dup", "dup-vec"), 0.9), (kr("v2", "v2-vec"), 0.5)]; let out = merge_hybrid_results(kw, vec_results, 5); assert_eq!(out.len(), 2, "dup 去重只出现一次"); assert_eq!(out[0].id, "dup", "dup 双信号排首"); assert_eq!(out[1].id, "v2"); } #[test] fn merge_dual_signal_not_duplicated_in_keyword_pass() { // 双信号记录已被 seen 标记,LIKE 单信号遍历时不会重复入列 let kw = vec![kr("both", "both-kw"), kr("kwonly", "kwo")]; let vec_results = vec![(kr("both", "both-vec"), 0.6)]; let out = merge_hybrid_results(kw, vec_results, 5); let both_count = out.iter().filter(|r| r.id == "both").count(); assert_eq!(both_count, 1); assert_eq!(out.len(), 2); } #[test] fn merge_all_three_signal_types_present() { // 三类信号齐全:dual(双)+ kw-only(LIKE)+ vec-only(向量) let kw = vec![kr("dual", "d-kw"), kr("kwonly", "k-kw")]; let vec_results = vec![ (kr("dual", "d-vec"), 0.85), (kr("veconly", "v-vec"), 0.45), ]; let out = merge_hybrid_results(kw, vec_results, 10); assert_eq!(out.len(), 3); // 排序:dual → kwonly → veconly assert_eq!(out[0].id, "dual"); assert_eq!(out[1].id, "kwonly"); assert_eq!(out[2].id, "veconly"); } #[test] fn merge_limit_truncates_after_sorting() { // 截断发生在排序之后:limit=1 时即便有双信号也只留首条 let kw = vec![kr("kw1", "k1")]; let vec_results = vec![(kr("dual", "dv"), 0.9)]; // dual 不在 kw,故无双信号;顺序 kw1 → dual let out = merge_hybrid_results(kw, vec_results, 1); assert_eq!(out.len(), 1); assert_eq!(out[0].id, "kw1"); } // ---------- 检索 query 分词(extract_query_keywords) ---------- #[test] fn keywords_ascii_words_lowercased() { let out = extract_query_keywords("修复 KnowledgeRepo 的 Search_Bug"); assert!(out.contains(&"knowledgerepo".to_string()), "ASCII 词应小写: {:?}", out); assert!(out.contains(&"search_bug".to_string()), "下划线应保留在词内: {:?}", out); } #[test] fn keywords_cjk_bigrams_extracted() { let out = extract_query_keywords("修复知识库检索"); // 连续汉字两两切:修复/复知/知识/识库/检索 assert!(out.contains(&"修复".to_string())); assert!(out.contains(&"知识".to_string())); assert!(out.contains(&"检索".to_string())); assert!(out.contains(&"复知".to_string()), "跨词 2-gram 允许存在(无词典近似的已知代价)"); } #[test] fn keywords_stopwords_filtered() { let out = extract_query_keywords("帮我看看这个怎么修复"); // 帮我/看看/这个/怎么全为停用词,只剩修复 assert_eq!(out, vec!["修复".to_string()], "停用词应被过滤: {:?}", out); } #[test] fn keywords_short_ascii_dropped() { let out = extract_query_keywords("use a x axis_config"); // 单字符 a/x 被过滤(use 为英文停用词),保留 axis_config assert_eq!(out, vec!["axis_config".to_string()]); } #[test] fn keywords_single_cjk_char_dropped() { // 孤单汉字不成 2-gram,不产出 let out = extract_query_keywords("好 的"); assert!(out.is_empty(), "单字与停用词应全过滤: {:?}", out); } #[test] fn keywords_frequency_ranking_and_cap() { let text = "tokio 出现多次 tokio 然后 tokio 还有 rust 与 rust 以及 postgres"; let out = extract_query_keywords(text); // 频次降序:tokio(3) > rust(2) > postgres(1);「然后」为停用词 assert_eq!(out[0], "tokio"); assert_eq!(out[1], "rust"); assert!(out.contains(&"postgres".to_string())); // 同频次按首现顺序稳定:构造 8 个互异词验证上限 6 let many = extract_query_keywords("alpha beta gamma delta epsilon zeta eta theta"); assert_eq!(many.len(), 6, "应截断到前 6 个: {:?}", many); assert_eq!(many[0], "alpha", "同频次按首现顺序"); } #[test] fn keywords_empty_or_noise_returns_empty() { assert!(extract_query_keywords("").is_empty()); assert!(extract_query_keywords(" ,。!?").is_empty(), "纯标点应产出空"); assert!(extract_query_keywords("帮我看看一下这个").is_empty(), "全停用词应产出空"); } #[test] fn keywords_mixed_ascii_cjk_segmented() { let out = extract_query_keywords("用cargo编译失败error E0499"); assert!(out.contains(&"cargo".to_string())); assert!(out.contains(&"e0499".to_string()), "字母数字混合词应整体保留: {:?}", out); assert!(out.contains(&"编译".to_string())); assert!(out.contains(&"失败".to_string())); } } #[cfg(test)] mod salvage_tests { use super::*; /// 截断点在第二个元素中段:已完整闭合的首条目应救回 #[test] fn salvages_complete_prefix_of_truncated_array() { let raw = r#"[{"kind":"pitfall","title":"t1","content":"c1"},{"kind":"review_r"#; let out = salvage_truncated_json_array(raw).expect("应打捞出 1 条"); assert_eq!(out.len(), 1); assert_eq!(out[0]["title"], "t1"); } /// 括号与引号出现在字符串字面量内(含转义):不得误判为结构边界 #[test] fn salvage_respects_string_literals_and_escapes() { let raw = r#"[{"title":"含}花括号与\"引号","content":"a\\"},{"kind":"pit"#; let out = salvage_truncated_json_array(raw).expect("应打捞出 1 条"); assert_eq!(out.len(), 1); assert_eq!(out[0]["title"], "含}花括号与\"引号"); assert_eq!(out[0]["content"], "a\\"); } /// 完整数组 + 尾部脏字符:全部条目救回(等价重解析) #[test] fn complete_array_with_trailing_garbage_salvages_all() { let raw = r#"[{"kind":"pitfall","title":"t1"},{"kind":"diagnosis","title":"t2"}] 附带文字"#; let out = salvage_truncated_json_array(raw).expect("应打捞 2 条"); assert_eq!(out.len(), 2); assert_eq!(out[1]["title"], "t2"); } /// 垃圾/空输出不误收:返回 None 走原丢弃路径 #[test] fn garbage_or_empty_yields_none() { assert!(salvage_truncated_json_array("").is_none(), "空输出无物可打捞"); assert!(salvage_truncated_json_array("模型输出了一段散文,没有 JSON").is_none()); assert!(salvage_truncated_json_array(r#"}{"a":1}"#).is_none(), "垃圾前缀深度为负"); assert!(salvage_truncated_json_array(r#"{"kind":"pitfall""#).is_none(), "无完整闭合元素"); assert!(salvage_truncated_json_array("[").is_none()); } }