Files
DevFlow/src-tauri/src/commands/knowledge.rs
T
lxy 0dd95e0e6e 新增: 知识库图召回激活扩散与治理续批
- 阶段2 图召回: list_neighbors_for_seeds(多种子 1-hop 展开,排除种子与 archived)+knowledge_search 结果附 related 段(权重降序限 5)
- 蒸馏合并: knowledge_distill.rs(archived 按 kind 分组 LLM 合订本+merges 边连组)
- V54 knowledge_edges 边表与渐进披露批配套迁移/模型字段
- 提炼链 related_ids 协议+检索补 id 字段
2026-10-26 01:36:19 +08:00

667 lines
23 KiB
Rust

//! 知识库相关命令 — 共享记忆层(沉淀 / 检索注入 / 审核收件箱 / 配置)
//!
//! 11 个 command 对齐 MCP 语义(search/list/get/create/update_status/record_reuse 可对外暴露,
//! archive/get_config/save_config/extract_now/list_candidates 为内部便利方法)。
use serde::Deserialize;
use tauri::State;
use df_types::types::new_id;
use df_storage::models::{KnowledgeEventRecord, KnowledgeRecord, RelatedKnowledge};
use crate::state::{AppState, KnowledgeConfig};
use super::knowledge_timeline::KnowledgeTimeline;
use super::{err_str, now_millis};
use serde::Serialize;
/// 创建知识入参
#[derive(Debug, Deserialize)]
pub struct CreateKnowledgeInput {
/// 7 种 KnowledgeKind snake_case 之一
pub kind: String,
pub title: String,
pub content: String,
/// 标签 JSON 数组字符串
#[serde(default)]
pub tags: Option<String>,
#[serde(default)]
pub source_project: Option<String>,
#[serde(default)]
pub source_ref: Option<String>,
/// high | medium | low
#[serde(default)]
pub confidence: Option<String>,
}
/// 检索入参
#[derive(Debug, Deserialize)]
pub struct KnowledgeSearchInput {
pub query: String,
#[serde(default)]
pub kind: Option<String>,
#[serde(default)]
pub limit: Option<usize>,
}
/// 状态转换合法矩阵校验
///
/// candidate → pending_review | published | rejected | archived
/// pending_review → published | rejected | archived
/// published → archived
/// rejected → candidate(重新送审;rejected 独立分池语义:拒绝≠归档,
/// 被否决的知识可修改后重审,也可留在拒绝池沉淀——与 archived 分开可见)
/// archived → published(归档箱恢复;2026-10-05 归档功能尾巴——原 archived 为汇点,
/// 归档即不可逆,与项目/任务/灵感/会话归档「可逆语义」不一致)
/// (其他组合非法)
fn validate_transition(from: &str, to: &str) -> Result<(), String> {
let legal = match from {
"candidate" => matches!(to, "pending_review" | "published" | "rejected" | "archived"),
"pending_review" => matches!(to, "published" | "rejected" | "archived"),
"published" => matches!(to, "archived"),
"rejected" => matches!(to, "candidate"),
"archived" => matches!(to, "published"),
_ => false,
};
if legal {
Ok(())
} else {
Err(format!("非法状态转换: {from} → {to}"))
}
}
// ============================================================
// CRUD
// ============================================================
/// 列出知识 — 可按 status 筛选,status=None 时默认仅返回 published
///
/// 默认范围说明(F-260616-02 决策 a):
/// - status=None → library tab 的数据源,语义为「已发布知识库」,仅 published。
/// 不再混杂 pending_review(pending_review 归 inbox 收件箱,见 knowledge_list_candidates)。
/// - 显式传 status 时按该 status 过滤(含 archived)。
#[tauri::command]
pub async fn knowledge_list(
state: State<'_, AppState>,
status: Option<String>,
) -> Result<Vec<KnowledgeRecord>, String> {
match status {
// 显式传 status 时按该 status 过滤(含 archived)
Some(s) => state.knowledge.list_by_status(&s).await.map_err(err_str),
// 默认:library 仅 published(F-260616-02 决策 a,职责清晰:library=published,inbox=待处理)
None => state.knowledge.list_by_status("published").await.map_err(err_str),
}
}
/// 单条查询
#[tauri::command]
pub async fn knowledge_get(
state: State<'_, AppState>,
id: String,
) -> Result<KnowledgeRecord, String> {
state
.knowledge
.get_by_id(&id)
.await
.map_err(err_str)?
.ok_or_else(|| format!("知识不存在: {id}"))
}
/// 检索知识(top-N,默认 3) — MCP search tool
#[tauri::command]
pub async fn knowledge_search(
state: State<'_, AppState>,
input: KnowledgeSearchInput,
) -> Result<Vec<KnowledgeRecord>, String> {
let limit = input.limit.unwrap_or(3).min(20);
state
.knowledge
.search(&input.query, input.kind.as_deref(), limit)
.await
.map_err(err_str)
}
/// 创建知识(始终 candidate 状态) — MCP create tool
#[tauri::command]
pub async fn knowledge_create(
state: State<'_, AppState>,
input: CreateKnowledgeInput,
) -> Result<KnowledgeRecord, String> {
let now = now_millis();
let record = KnowledgeRecord {
id: new_id(),
kind: input.kind,
title: input.title,
content: input.content,
tags: input.tags,
status: "candidate".to_string(),
confidence: input.confidence,
reuse_count: 0,
verified: false,
source_project: input.source_project,
source_ref: input.source_ref,
reasoning: None,
embedding_status: None,
created_at: now.clone(),
updated_at: now,
};
state
.knowledge
.insert(record.clone())
.await
.map_err(err_str)?;
// 生命线:手动录入产生(fire-and-forget)
KnowledgeTimeline::new(&state.db)
.record_created(&record.id, "manual")
.await;
Ok(record)
}
// ============================================================
// 状态机 + 计数
// ============================================================
/// 状态转换(含合法矩阵校验) — MCP update_status tool
#[tauri::command]
pub async fn knowledge_update_status(
state: State<'_, AppState>,
id: String,
status: String,
) -> Result<bool, String> {
let record = state
.knowledge
.get_by_id(&id)
.await
.map_err(err_str)?
.ok_or_else(|| format!("知识不存在: {id}"))?;
let old_status = record.status.clone();
validate_transition(&record.status, &status)?;
// published 时一次性标 verified=true(发布审核标)
let now = now_millis();
let updated = KnowledgeRecord {
status: status.clone(),
verified: if status == "published" { true } else { record.verified },
updated_at: now,
..record
};
let result = state
.knowledge
.update_full(&updated)
.await
.map_err(err_str)?;
// 生命线:状态变更(fire-and-forget;to=published=审核通过,to=archived=归档)
KnowledgeTimeline::new(&state.db)
.record_status_change(&id, &old_status, &status)
.await;
// 发布时后台生成嵌入(只有 published 参与检索,candidate 不浪费 embed 调用;
// vector_enabled 关闭或 embed 失败时该条走 LIKE 降级,非阻断)
if status == "published" {
crate::commands::ai::spawn_embedding_for_knowledge(&state, &updated).await;
}
Ok(result)
}
/// 复用计数 +1(检索命中时调用) — MCP record_reuse tool
#[tauri::command]
pub async fn knowledge_record_reuse(
state: State<'_, AppState>,
id: String,
) -> Result<bool, String> {
state
.knowledge
.increment_reuse_count(&id)
.await
.map_err(err_str)
}
/// 收件箱 — 列出待处理条目(candidate + pending_review),按 confidence 语义排序
///
/// 语义(F-260616-02 决策 a):inbox = 「待处理」收件箱,聚合 candidate(待评估)
/// 与 pending_review(待发布审核)两种待处理状态。library 仅 published。
///
/// 实现:list_by_status 单状态查询,这里合并 candidate 与 pending_review 两路结果。
/// 两路各自已按 `confidence DESC, created_at DESC` 排序,有序合并保持同一规则。
#[tauri::command]
pub async fn knowledge_list_candidates(
state: State<'_, AppState>,
) -> Result<Vec<KnowledgeRecord>, String> {
let (candidates, pending) = tokio::try_join!(
state.knowledge.list_by_status("candidate"),
state.knowledge.list_by_status("pending_review"),
)
.map_err(err_str)?;
Ok(merge_by_confidence(candidates, pending))
}
/// 有序合并两列(各自已按 confidence DESC, created_at DESC 排序),结果保持同序。
///
/// confidence 排序权重:high=3, medium=2, low=1, 其他=0;同权重按 created_at DESC
/// (字符串毫秒时间戳字典序 = 时间序)。等价 SQL `ORDER BY CASE confidence ... DESC, created_at DESC`。
fn merge_by_confidence(
mut a: Vec<KnowledgeRecord>,
mut b: Vec<KnowledgeRecord>,
) -> Vec<KnowledgeRecord> {
use std::cmp::Ordering;
fn rank(c: &str) -> i8 {
match c {
"high" => 3,
"medium" => 2,
"low" => 1,
_ => 0,
}
}
let cmp = |x: &KnowledgeRecord, y: &KnowledgeRecord| -> Ordering {
let rx = rank(x.confidence.as_deref().unwrap_or(""));
let ry = rank(y.confidence.as_deref().unwrap_or(""));
ry.cmp(&rx) // confidence DESC
.then_with(|| y.created_at.cmp(&x.created_at)) // created_at DESC
};
a.sort_by(cmp);
b.sort_by(cmp);
let mut out = Vec::with_capacity(a.len() + b.len());
let (mut i, mut j) = (0, 0);
while i < a.len() && j < b.len() {
if cmp(&a[i], &b[j]) != Ordering::Greater {
out.push(a[i].clone());
i += 1;
} else {
out.push(b[j].clone());
j += 1;
}
}
out.extend_from_slice(&a[i..]);
out.extend_from_slice(&b[j..]);
out
}
/// 归档(软删除) — UPDATE status='archived'
#[tauri::command]
pub async fn knowledge_archive(
state: State<'_, AppState>,
id: String,
) -> Result<bool, String> {
knowledge_update_status(state, id, "archived".to_string()).await
}
/// 重新生成向量嵌入(补偿重试)— P1 修复(嵌入失败无标记)的补偿入口
///
/// 触发条件:embedding_status='failed' 的已发布知识(provider 临时不可用导致永久无向量索引)。
/// 行为:fire-and-forget,后台逐条重跑嵌入生成(成功置 done,失败仍 failed 可再次触发)。
/// 返回值:本次触发重试的条数(0 表示无 failed 条目或 vector_enabled 关闭)。
#[tauri::command]
pub async fn knowledge_retry_embedding(
state: State<'_, AppState>,
) -> Result<usize, String> {
let count = crate::commands::ai::retry_failed_embeddings(&state)
.await
.map_err(err_str)?;
Ok(count)
}
/// 淘汰批:零复用 + low 置信 + 超龄条目批量归档(P0 淘汰机制,2026-10-23)
///
/// 背景(知识库治理审查 2026-10-23):71% 条目零复用长尾稀释检索排序,检索面是
/// status 白名单,archived 天然出池。逐条留痕(status_changed,source_ref=archive_stale)
/// 可审计可回滚。默认阈值 90 天/批 500——从生产画像看 low 置信存量极少(实测 1 条),
/// 阈值偏保守,宁可分批多跑不误伤。
///
/// 归档不可逆性:archived 条目退出检索面,可通过 knowledge_update_status 手动转回
/// published(状态机合法迁移),留痕链完整。
#[tauri::command]
pub async fn knowledge_archive_stale(
state: State<'_, AppState>,
older_than_days: Option<i64>,
limit: Option<i64>,
) -> Result<Vec<String>, String> {
let days = older_than_days.unwrap_or(90).clamp(1, 3650);
let limit = limit.unwrap_or(500).clamp(1, 5000);
let ids = state
.knowledge
.archive_stale(days, limit)
.await
.map_err(err_str)?;
// 逐条留痕 fire-and-forget(与晋升留痕同风格,吞错只 warn)。
// from 记 "stale" 而非逐条旧值(批量 UPDATE 已合并原状态,淘汰批标记便于审计回滚)
let timeline = KnowledgeTimeline::new(&state.db);
for id in &ids {
timeline.record_status_change(id, "stale", "archived").await;
}
if !ids.is_empty() {
tracing::info!("[knowledge] 淘汰批: {} 条零复用低置信超龄条目归档(>{days}天)", ids.len());
}
Ok(ids)
}
/// 关联知识列表(知识中枢阶段1:详情页「关联知识」区块数据源)
///
/// 双向读边表 + 对端条目展开,archived 对端不展示(与检索面口径一致)。
#[tauri::command]
pub async fn knowledge_list_related(
state: State<'_, AppState>,
id: String,
limit: Option<i64>,
) -> Result<Vec<RelatedKnowledge>, String> {
let limit = limit.unwrap_or(10).clamp(1, 50) as usize;
state
.knowledge
.list_related(&id, limit)
.await
.map_err(err_str)
}
/// 存量蒸馏合并(治理 P0 深化,2026-10-24):对指定 kind 的已归档分组跑一次 LLM 蒸馏,
/// 产出单条合订本(candidate)并建 merges 边连组内条目。返回合订本 id(组内不足 2 条
/// 或模型判 skip 返回 None)。
#[tauri::command]
pub async fn knowledge_distill_group(
state: State<'_, AppState>,
kind: String,
limit: Option<i64>,
) -> Result<Option<String>, String> {
let provider_config = crate::commands::ai::prompt::get_active_provider(&state)
.await
.map_err(err_str)?;
crate::commands::ai::knowledge_distill::distill_archived_group(
&state.db,
&provider_config,
&state.llm_concurrency,
&kind,
limit.map(|n| n as usize),
)
.await
.map_err(err_str)
}
// ============================================================
// 配置 + 手动提炼
// ============================================================
/// 读取知识库行为配置(提取 + 注入)
#[tauri::command]
pub async fn knowledge_get_config(
state: State<'_, AppState>,
) -> Result<KnowledgeConfig, String> {
let cfg = state.knowledge_config.lock().await;
Ok(cfg.clone())
}
/// 保存知识库行为配置
#[tauri::command]
pub async fn knowledge_save_config(
state: State<'_, AppState>,
config: KnowledgeConfig,
) -> Result<bool, String> {
// P0(设置走查-2026-06-21):落 Settings KV 持久化,reload_knowledge_config 启动恢复
// (原纯内存 Arc<Mutex> 启动 default 覆盖致 8 项配置重启全丢)。
let json = serde_json::to_string(&config).map_err(|e| e.to_string())?;
{
let mut cfg = state.knowledge_config.lock().await;
*cfg = config;
}
state
.settings
.set(crate::state::KNOWLEDGE_CONFIG_KEY, &json)
.await
.map_err(|e| e.to_string())?;
Ok(true)
}
/// 手动触发提炼(ManualOnly 模式 / 用户主动点按钮)
///
/// 实际提炼逻辑在 commands::ai 模块(需访问 active provider + conversation messages)。
/// 此 command 仅作前端入口,委托给 ai 模块的提炼函数。
#[tauri::command]
pub async fn knowledge_extract_now(
state: State<'_, AppState>,
) -> Result<bool, String> {
crate::commands::ai::trigger_extraction_now(&state).await
}
// ============================================================
// 生命线:详情 / 编辑 / 事件查询
// ============================================================
/// 编辑知识入参(部分更新,仅传需改字段)
#[derive(Debug, Deserialize)]
pub struct UpdateKnowledgeInput {
#[serde(default)]
pub title: Option<String>,
#[serde(default)]
pub content: Option<String>,
#[serde(default)]
pub tags: Option<String>,
#[serde(default)]
pub confidence: Option<String>,
#[serde(default)]
pub reasoning: Option<String>,
}
/// 知识详情聚合负载(基本信息 + 全部生命线事件 + 关联知识)
#[derive(Debug, Serialize)]
pub struct KnowledgeDetailPayload {
pub knowledge: KnowledgeRecord,
pub events: Vec<KnowledgeEventRecord>,
/// 关联知识(阶段1 连边,双向读+对端展开);老前端不消费此字段无影响
#[serde(default)]
pub related: Vec<RelatedKnowledge>,
}
/// 知识详情(基本信息 + 生命线事件 + 关联知识) — 详情页一次拉全
#[tauri::command]
pub async fn knowledge_get_detail(
state: State<'_, AppState>,
id: String,
) -> Result<KnowledgeDetailPayload, String> {
let knowledge = state
.knowledge
.get_by_id(&id)
.await
.map_err(err_str)?
.ok_or_else(|| format!("知识不存在: {id}"))?;
let events = state
.knowledge_events
.list_by_knowledge(&id)
.await
.map_err(err_str)?;
// 关联边读取失败非阻断(边表为空/极老库未到 V54 时降级空列表)
let related = state.knowledge.list_related(&id, 10).await.unwrap_or_default();
Ok(KnowledgeDetailPayload { knowledge, events, related })
}
/// 编辑知识(部分更新 title/content/tags/confidence/reasoning)
#[tauri::command]
pub async fn knowledge_update(
state: State<'_, AppState>,
id: String,
input: UpdateKnowledgeInput,
) -> Result<KnowledgeRecord, String> {
let mut record = state
.knowledge
.get_by_id(&id)
.await
.map_err(err_str)?
.ok_or_else(|| format!("知识不存在: {id}"))?;
// P1 修复(KP-5 审计断档):记录本次实际改动的字段,供 updated 生命线事件审计展示。
// 只在值确实变化时计入 changed_fields(避免无改动也产生 updated 噪音事件)。
let mut changed_fields: Vec<&'static str> = Vec::new();
if let Some(v) = input.title {
if record.title != v {
changed_fields.push("title");
}
record.title = v;
}
if let Some(v) = input.content {
if record.content != v {
changed_fields.push("content");
}
record.content = v;
}
if let Some(v) = input.tags {
if record.tags.as_deref() != Some(v.as_str()) {
changed_fields.push("tags");
}
record.tags = Some(v);
}
// 空串 = 清空(前端编辑器清空 confidence/reasoning 输入框时传 "")
if let Some(v) = input.confidence {
let new_conf = if v.is_empty() { None } else { Some(v) };
if record.confidence != new_conf {
changed_fields.push("confidence");
}
record.confidence = new_conf;
}
if let Some(v) = input.reasoning {
let new_reason = if v.is_empty() { None } else { Some(v) };
if record.reasoning != new_reason {
changed_fields.push("reasoning");
}
record.reasoning = new_reason;
}
record.updated_at = now_millis();
state
.knowledge
.update_full(&record)
.await
.map_err(err_str)?;
// 生命线:编辑产生(fire-and-forget)。
// 仅在确有字段改动时记录,无改动不产噪音事件(对齐 create/update_status 的语义粒度)。
// source_ref="ui" 标识编辑来自前端 UI(未来可细分 detail/inbox,当前统一 ui)。
if !changed_fields.is_empty() {
KnowledgeTimeline::new(&state.db)
.record_updated(&id, &changed_fields, "ui")
.await;
}
Ok(record)
}
/// 查询生命线事件(可按 event_type 过滤 + limit)
#[tauri::command]
pub async fn knowledge_events(
state: State<'_, AppState>,
knowledge_id: String,
event_type: Option<String>,
limit: Option<usize>,
) -> Result<Vec<KnowledgeEventRecord>, String> {
match event_type {
Some(et) => {
let limit = limit.unwrap_or(50);
state
.knowledge_events
.list_by_knowledge_type(&knowledge_id, &et, limit)
.await
.map_err(err_str)
}
None => state
.knowledge_events
.list_by_knowledge(&knowledge_id)
.await
.map_err(err_str),
}
}
#[cfg(test)]
mod tests {
use super::validate_transition;
#[test]
fn candidate_legal_transitions() {
assert!(validate_transition("candidate", "pending_review").is_ok());
assert!(validate_transition("candidate", "published").is_ok());
assert!(validate_transition("candidate", "archived").is_ok());
}
#[test]
fn candidate_illegal_transitions() {
// 禁止自转、回退
assert!(validate_transition("candidate", "candidate").is_err());
}
#[test]
fn pending_review_legal_transitions() {
assert!(validate_transition("pending_review", "published").is_ok());
assert!(validate_transition("pending_review", "archived").is_ok());
// 分池:审核否决走 rejected(独立池),不再混入 archived
assert!(validate_transition("pending_review", "rejected").is_ok());
}
#[test]
fn pending_review_illegal_transitions() {
// 已进入审核态不能再退回 candidate
assert!(validate_transition("pending_review", "candidate").is_err());
assert!(validate_transition("pending_review", "pending_review").is_err());
}
#[test]
fn published_legal_transitions() {
// 已发布只能归档
assert!(validate_transition("published", "archived").is_ok());
}
#[test]
fn published_illegal_transitions() {
// 发布态不可再进审核、不可回 candidate、不可自转、不可重复 published
assert!(validate_transition("published", "pending_review").is_err());
assert!(validate_transition("published", "candidate").is_err());
assert!(validate_transition("published", "published").is_err());
}
#[test]
fn archived_restore_to_published_legal() {
// 归档箱恢复(2026-10-05 归档功能尾巴):archived → published 唯一出边
assert!(validate_transition("archived", "published").is_ok());
}
#[test]
fn rejected_legal_transitions() {
// 分池语义:rejected 唯一出边=回 candidate 重新送审(修改后重审)
assert!(validate_transition("candidate", "rejected").is_ok());
assert!(validate_transition("pending_review", "rejected").is_ok());
assert!(validate_transition("rejected", "candidate").is_ok());
}
#[test]
fn rejected_illegal_transitions() {
// 拒绝池不可直接发布(必须走重审 candidate → published)、不可归档/自转
for to in ["published", "pending_review", "archived", "rejected"] {
assert!(validate_transition("rejected", to).is_err(),
"rejected → {to} 应为非法");
}
}
#[test]
fn archived_illegal_transitions() {
// 归档态除恢复发布外均非法(不回 candidate/pending_review,不自转)
for to in ["candidate", "pending_review", "archived"] {
assert!(validate_transition("archived", to).is_err(),
"archived → {to} 应为非法");
}
}
#[test]
fn unknown_from_is_illegal() {
// 未知源状态一律拒绝
assert!(validate_transition("unknown", "published").is_err());
assert!(validate_transition("", "archived").is_err());
}
#[test]
fn illegal_target_is_rejected() {
// 合法源 → 未知/空目标一律拒绝
assert!(validate_transition("candidate", "unknown").is_err());
assert!(validate_transition("candidate", "").is_err());
assert!(validate_transition("pending_review", "draft").is_err());
}
#[test]
fn error_message_contains_transition() {
let err = validate_transition("archived", "candidate").unwrap_err();
assert!(err.contains("archived"), "错误信息应包含源状态: {err}");
assert!(err.contains("candidate"), "错误信息应包含目标状态: {err}");
}
}