diff --git a/crates/utopia-server/src/api/search_routes.rs b/crates/utopia-server/src/api/search_routes.rs index 57212fdd5..29c89f0eb 100644 --- a/crates/utopia-server/src/api/search_routes.rs +++ b/crates/utopia-server/src/api/search_routes.rs @@ -34,6 +34,13 @@ pub struct SearchReq { /// 所以没有 `source_id` 不收 #[serde(default)] pub external_id_prefixes: Option>, + /// 这次检索算不算「用过」（RC-003）：默认算。探测、评测、健康检查传 false，不留痕 + #[serde(default = "record_by_default")] + pub record: bool, +} + +fn record_by_default() -> bool { + true } fn check_prefixes(req: &SearchReq) -> Result<(), AppError> { @@ -110,6 +117,10 @@ pub async fn search( documents, }; let page = retrieval::hybrid_scored(&state, kb_id, kb.workspace_id, q, opts, None).await?; + if req.record { + let documents = page.results.iter().map(|hit| hit.chunk.chunk.document_id); + retrieval::record_retrievals(&state, documents.collect()); + } Ok(Json(json!({ "results": page.results, "offset": req.offset, @@ -118,3 +129,18 @@ pub async fn search( "vector_used": page.vector_used, }))) } + +#[cfg(test)] +mod tests { + use super::SearchReq; + use serde_json::json; + + #[test] + fn a_search_counts_as_a_use_unless_it_says_otherwise() { + let plain: SearchReq = serde_json::from_value(json!({ "q": "x" })).unwrap(); + let probe: SearchReq = + serde_json::from_value(json!({ "q": "x", "record": false })).unwrap(); + assert!(plain.record); + assert!(!probe.record); + } +} diff --git a/crates/utopia-server/src/api/tools.rs b/crates/utopia-server/src/api/tools.rs index 08b216cf9..70a3d404f 100644 --- a/crates/utopia-server/src/api/tools.rs +++ b/crates/utopia-server/src/api/tools.rs @@ -334,6 +334,7 @@ pub async fn search_chunks( .error(); } }; + retrieval::record_retrievals(ctx.state, chunks.iter().map(|c| c.document_id).collect()); let mut lines = Vec::new(); for c in &chunks { let n = cite(sink, c.id.to_string(), |n| source_json(n, c)); diff --git a/crates/utopia-server/src/retrieval.rs b/crates/utopia-server/src/retrieval.rs index 3baf72f56..eb037404c 100644 --- a/crates/utopia-server/src/retrieval.rs +++ b/crates/utopia-server/src/retrieval.rs @@ -151,6 +151,19 @@ pub async fn hybrid_scored( }) } +/// 一次检索交出去的文档记一笔（RC-003）。放到后台写：失败只记日志，不拖慢也不打断这次读。 +/// 只有打开了衰减的源的文档会真的落行。 +pub fn record_retrievals(state: &AppState, mut document_ids: Vec) { + document_ids.sort_unstable(); + document_ids.dedup(); + let pool = state.pool.clone(); + tokio::spawn(async move { + if let Err(e) = utopia_store::decay::record_retrievals(&pool, &document_ids).await { + tracing::warn!(error = %e, "记录文档被检索失败"); + } + }); +} + /// 第 `channel` 路里的名次，从 1 数 fn position(ranks: &[Option], channel: usize) -> Option { ranks diff --git a/crates/utopia-store/src/decay.rs b/crates/utopia-store/src/decay.rs index e126ce01e..794d5bb66 100644 --- a/crates/utopia-store/src/decay.rs +++ b/crates/utopia-store/src/decay.rs @@ -5,7 +5,9 @@ use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; -use utopia_core::AppError; +use sqlx::PgPool; +use utopia_core::{AppError, AppResult}; +use uuid::Uuid; pub const DEFAULT_HALF_LIFE_DAYS: f64 = 60.0; pub const MAX_HALF_LIFE_DAYS: f64 = 365.0; @@ -52,6 +54,33 @@ impl DecaySettings { } } +/// 这些文档刚被一次检索交了出去：属于打开了衰减的源的，计数加一、最近检索时间记成现在、 +/// 半衰期加 [`RETRIEVAL_HALF_LIFE_GAIN_DAYS`] 天（封顶 [`MAX_HALF_LIFE_DAYS`]）。别的源的文档 +/// 一行也不写。一条语句写完所有文档；第一次被检索的文档从源里的半衰期起算。 +pub async fn record_retrievals(pool: &PgPool, document_ids: &[Uuid]) -> AppResult { + if document_ids.is_empty() { + return Ok(0); + } + let written = sqlx::query( + "INSERT INTO document_salience (document_id, retrieval_count, last_retrieved_at, half_life_days) + SELECT d.id, 1, now(), + LEAST($3::real, COALESCE((s.decay->>'half_life_days')::real, $4::real) + $2::real) + FROM documents d JOIN sources s ON s.id = d.source_id + WHERE d.id = ANY($1) AND s.decay IS NOT NULL + ON CONFLICT (document_id) DO UPDATE SET + retrieval_count = document_salience.retrieval_count + 1, + last_retrieved_at = now(), + half_life_days = LEAST($3::real, document_salience.half_life_days + $2::real)", + ) + .bind(document_ids) + .bind(RETRIEVAL_HALF_LIFE_GAIN_DAYS as f32) + .bind(MAX_HALF_LIFE_DAYS as f32) + .bind(DEFAULT_HALF_LIFE_DAYS as f32) + .execute(pool) + .await?; + Ok(written.rows_affected()) +} + /// 打分用到的一个文档：`key` 是文档名，`value` 是它的文字。 pub struct Memory<'a> { pub key: &'a str, diff --git a/crates/utopia-store/tests/store/a_retrieval_is_recorded_only_for_a_decaying_source.rs b/crates/utopia-store/tests/store/a_retrieval_is_recorded_only_for_a_decaying_source.rs new file mode 100644 index 000000000..9ed332b8a --- /dev/null +++ b/crates/utopia-store/tests/store/a_retrieval_is_recorded_only_for_a_decaying_source.rs @@ -0,0 +1,114 @@ +//! 检索把文档记一笔，但只记打开了衰减的源的文档（RC-003），打在真库上。 +//! +//! 要钉住的行为： +//! +//! - **关着的源一行不写** +//! - **第一次检索**：计数 1，半衰期从源里的值加 2 天 +//! - **再次检索**：计数加一、半衰期再加 2 天，最近检索时间更新 +//! - **半衰期封顶 365** +//! - **同一次里多个文档一起记**，不存在的 id 被忽略 +//! +//! 没有 `UTOPIA_DATABASE_URL` 时跳过而不是失败。自建自拆，绝不碰已有的库。 + +use sqlx::PgPool; +use utopia_store::decay::record_retrievals; +use uuid::Uuid; + +async fn salience(pool: &PgPool, document: Uuid) -> anyhow::Result> { + Ok(sqlx::query_as( + "SELECT retrieval_count, half_life_days, last_retrieved_at IS NOT NULL + FROM document_salience WHERE document_id = $1", + ) + .bind(document) + .fetch_optional(pool) + .await?) +} + +#[tokio::test] +async fn a_retrieval_is_recorded_only_for_a_decaying_source() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + utopia_store::db::migrate(&pool).await?; + + let (org, ws, kb) = (Uuid::now_v7(), Uuid::now_v7(), Uuid::now_v7()); + let (decaying, plain) = (Uuid::now_v7(), Uuid::now_v7()); + let (decaying_doc, plain_doc, old_doc) = (Uuid::now_v7(), Uuid::now_v7(), Uuid::now_v7()); + sqlx::query("INSERT INTO organizations (id, name) VALUES ($1, 'record-retrieval-test')") + .bind(org) + .execute(&pool) + .await?; + sqlx::query( + "INSERT INTO workspaces (id, org_id, name) VALUES ($1, $2, 'record-retrieval-test')", + ) + .bind(ws) + .bind(org) + .execute(&pool) + .await?; + sqlx::query( + "INSERT INTO knowledge_bases (id, workspace_id, name) VALUES ($1, $2, 'record-retrieval-test')", + ) + .bind(kb) + .bind(ws) + .execute(&pool) + .await?; + for (id, decay) in [ + (decaying, Some(r#"{"half_life_days": 100}"#)), + (plain, None), + ] { + sqlx::query("INSERT INTO sources (id, kb_id, kind, name, decay) VALUES ($1, $2, 'folder', 'f', $3::jsonb)") + .bind(id) + .bind(kb) + .bind(decay) + .execute(&pool) + .await?; + } + for (id, source) in [ + (decaying_doc, decaying), + (plain_doc, plain), + (old_doc, decaying), + ] { + sqlx::query( + "INSERT INTO documents (id, kb_id, source_id, filename, sha256, status) + VALUES ($1, $2, $3, $4, $4, 'ready')", + ) + .bind(id) + .bind(kb) + .bind(source) + .bind(id.to_string()) + .execute(&pool) + .await?; + } + + let run = async { + let missing = Uuid::now_v7(); + let written = record_retrievals(&pool, &[decaying_doc, plain_doc, missing]).await?; + assert_eq!(written, 1, "only the decaying source's document is written"); + assert_eq!(salience(&pool, plain_doc).await?, None); + assert_eq!(salience(&pool, decaying_doc).await?, Some((1, 102.0, true))); + + record_retrievals(&pool, &[decaying_doc]).await?; + assert_eq!(salience(&pool, decaying_doc).await?, Some((2, 104.0, true))); + + sqlx::query("INSERT INTO document_salience (document_id, half_life_days) VALUES ($1, 364)") + .bind(old_doc) + .execute(&pool) + .await?; + record_retrievals(&pool, &[old_doc]).await?; + assert_eq!( + salience(&pool, old_doc).await?, + Some((1, 365.0, true)), + "capped" + ); + assert_eq!(record_retrievals(&pool, &[]).await?, 0); + anyhow::Ok(()) + } + .await; + + sqlx::query("DELETE FROM organizations WHERE id = $1") + .bind(org) + .execute(&pool) + .await?; + run +} diff --git a/crates/utopia-store/tests/store/main.rs b/crates/utopia-store/tests/store/main.rs index 71ad210e5..8ad18d29e 100644 --- a/crates/utopia-store/tests/store/main.rs +++ b/crates/utopia-store/tests/store/main.rs @@ -82,6 +82,7 @@ mod a_relation_points_only_inside_its_own_kb; mod a_remembered_episode_strips_nul; mod a_retired_account; mod a_retraction_leaves_the_graph; +mod a_retrieval_is_recorded_only_for_a_decaying_source; mod a_review_has_a_summary; mod a_review_page_is_assembled_in_batches; mod a_rule_computes_what_it_concludes; diff --git a/docs/design/chat-and-mcp.md b/docs/design/chat-and-mcp.md index 3c7f504ae..3be12dadd 100644 --- a/docs/design/chat-and-mcp.md +++ b/docs/design/chat-and-mcp.md @@ -82,6 +82,10 @@ then recall only from that source's documents, live at `as_of`, whose `external_ of the prefixes, so the base's other documents cannot take the recall a filter after fusion would lose (2026-10-19). +A search that returns a document of a source that decays counts as a use of it, in the background and +never failing the read ([RC-003](../decisions/re-cinq/RC-003-a-source-may-decay.md)): `search_chunks` +always does, and the REST body may carry `record: false` for a probe, an eval or a health check. + **`remember`** writes a memory document at once; its statements go through open extraction and wait in `pending_facts`; the assistant says the sentence is recorded and its statements will be shown for confirmation; the card grows into the conversation on the SSE `pending` event, showing the