From fee7b51fd5429673c3608418d892998896bea44d Mon Sep 17 00:00:00 2001 From: Bogdan Szabo Date: Mon, 19 Oct 2026 22:44:28 +0200 Subject: [PATCH] Search answers a service with each hit's score, ranks and source, and pages Signed-off-by: Bogdan Szabo --- crates/utopia-core/src/models.rs | 11 ++ crates/utopia-search/src/lib.rs | 68 +++++++- crates/utopia-server/src/api/search_routes.rs | 33 +++- .../utopia-server/src/api/token_read_tests.rs | 81 +++++++--- crates/utopia-server/src/auth.rs | 1 + crates/utopia-server/src/retrieval.rs | 149 ++++++++++++++++-- crates/utopia-server/src/retrieval_tests.rs | 130 +++++++++++++++ crates/utopia-store/src/documents.rs | 17 +- .../a_search_reads_the_base_as_it_was.rs | 4 +- ...ty-from-the-person-scope-from-the-token.md | 8 +- docs/design/access-and-audit.md | 6 +- docs/design/chat-and-mcp.md | 9 +- web/src/docs/ingest.md | 31 ++++ 13 files changed, 488 insertions(+), 60 deletions(-) diff --git a/crates/utopia-core/src/models.rs b/crates/utopia-core/src/models.rs index 0178d60a6..3320fe736 100644 --- a/crates/utopia-core/src/models.rs +++ b/crates/utopia-core/src/models.rs @@ -475,6 +475,17 @@ pub struct ChunkView { pub filename: String, } +/// 分块带上它来自哪个来源。`external_id` 只有推送来源（api / statements）才有： +/// 推送方给的那个 id，去掉库里加的 `{kind}:` 前缀，好让它拿回去对自己的记录（#1271） +#[derive(Debug, Clone, Serialize, sqlx::FromRow)] +pub struct SourcedChunk { + #[serde(flatten)] + #[sqlx(flatten)] + pub chunk: ChunkView, + pub source_id: Option, + pub external_id: Option, +} + #[derive(Debug, Clone, Serialize, sqlx::FromRow)] pub struct LlmSettings { pub workspace_id: Uuid, diff --git a/crates/utopia-search/src/lib.rs b/crates/utopia-search/src/lib.rs index 4fe45b517..f518825bd 100644 --- a/crates/utopia-search/src/lib.rs +++ b/crates/utopia-search/src/lib.rs @@ -180,28 +180,53 @@ impl SearchIndex { } } +/// 融合后的一条：RRF 总分，和它在每一路里的名次（0 起；没进那一路是 None）。 +/// 分数只在同一次查询里可比，换一句话就是另一把尺子 +#[derive(Debug, Clone, PartialEq)] +pub struct Fused { + pub id: String, + pub score: f64, + pub ranks: Vec>, +} + /// Reciprocal Rank Fusion：融合多路召回的排名（k=60 为经验常数）。 pub fn rrf_fuse(lists: &[Vec], limit: usize) -> Vec { + rrf_fuse_scored(lists, limit) + .into_iter() + .map(|fused| fused.id) + .collect() +} + +/// 同 [`rrf_fuse`]，但把分数和各路名次一并交回来 +pub fn rrf_fuse_scored(lists: &[Vec], limit: usize) -> Vec { const K: f64 = 60.0; - let mut scores: HashMap = HashMap::new(); - for list in lists { + let mut by_id: HashMap = HashMap::new(); + for (channel, list) in lists.iter().enumerate() { for (rank, id) in list.iter().enumerate() { - *scores.entry(id.clone()).or_default() += 1.0 / (K + rank as f64 + 1.0); + let fused = by_id.entry(id.clone()).or_insert_with(|| Fused { + id: id.clone(), + score: 0.0, + ranks: vec![None; lists.len()], + }); + fused.score += 1.0 / (K + rank as f64 + 1.0); + fused.ranks[channel] = Some(rank); } } - let mut ranked: Vec<(String, f64)> = scores.into_iter().collect(); + let mut ranked: Vec = by_id.into_values().collect(); // 同分按 id 定序，否则 HashMap 的随机顺序会改变上限内保留的证据和口径。 ranked.sort_by(|a, b| { - b.1.partial_cmp(&a.1) + b.score + .partial_cmp(&a.score) .unwrap_or(std::cmp::Ordering::Equal) - .then_with(|| a.0.cmp(&b.0)) + .then_with(|| a.id.cmp(&b.id)) }); - ranked.into_iter().take(limit).map(|(id, _)| id).collect() + ranked.truncate(limit); + ranked } #[cfg(test)] mod rrf_tests { - use super::rrf_fuse; + use super::{rrf_fuse, rrf_fuse_scored, Fused}; fn ids(list: &[&str]) -> Vec { list.iter().map(|s| s.to_string()).collect() @@ -222,6 +247,33 @@ mod rrf_tests { assert_eq!(rrf_fuse(&lists, 10), ids(&["a", "c", "b"])); } + /// 分数就是各路名次倒数之和，名次记在各自那一路的位置上（#1271） + #[test] + fn scored_fusion_reports_the_sum_and_each_channels_rank() { + let lists = vec![ids(&["a", "b", "c"]), ids(&["c", "a"])]; + let fused = rrf_fuse_scored(&lists, 10); + assert_eq!( + fused, + vec![ + Fused { + id: "a".into(), + score: 1.0 / 61.0 + 1.0 / 62.0, + ranks: vec![Some(0), Some(1)], + }, + Fused { + id: "c".into(), + score: 1.0 / 63.0 + 1.0 / 61.0, + ranks: vec![Some(2), Some(0)], + }, + Fused { + id: "b".into(), + score: 1.0 / 62.0, + ranks: vec![Some(1), None], + }, + ] + ); + } + /// 空的那一路不是失败的那一路：它什么分都不加，也不影响别人 #[test] fn an_empty_list_contributes_nothing() { diff --git a/crates/utopia-server/src/api/search_routes.rs b/crates/utopia-server/src/api/search_routes.rs index 0b776059e..10ae4066d 100644 --- a/crates/utopia-server/src/api/search_routes.rs +++ b/crates/utopia-server/src/api/search_routes.rs @@ -2,20 +2,23 @@ use axum::extract::{Path, State}; use axum::Json; use serde::Deserialize; use serde_json::json; -use utopia_core::models::Role; use utopia_core::AppError; use uuid::Uuid; -use crate::auth::AuthUser; +use crate::auth::SessionOrToken; use crate::error::ApiResult; use crate::retrieval; use crate::state::AppState; +/// 不认识的字段答 422：过滤条件还没做，悄悄忽略一个 `filters` 会让调用方以为它生效了 #[derive(Deserialize)] +#[serde(deny_unknown_fields)] pub struct SearchReq { pub q: String, #[serde(default)] pub top_k: Option, + #[serde(default)] + pub offset: usize, /// 记录轴（0019）：按**那一刻库里有的东西**检索。YYYY-MM-DD 或 RFC3339。 /// 命中的是当时活着的块、当时还没被删的文档 #[serde(default)] @@ -24,7 +27,7 @@ pub struct SearchReq { pub async fn search( State(state): State, - AuthUser(user): AuthUser, + reader: SessionOrToken, Path(kb_id): Path, Json(req): Json, ) -> ApiResult> { @@ -32,10 +35,28 @@ pub async fn search( if q.is_empty() { return Err(AppError::invalid("empty_query", "Search query cannot be empty").into()); } - let kb = utopia_store::access::require_kb(&state.pool, &user, kb_id, Role::Viewer).await?; + let kb = super::documents_routes::require_reader_kb(&state, &reader, kb_id).await?; let top_k = req.top_k.unwrap_or(10).min(50); + if req.offset.saturating_add(top_k) > retrieval::MAX_DEPTH { + return Err(AppError::invalid( + "page_too_deep", + format!("offset + top_k may not exceed {}", retrieval::MAX_DEPTH), + ) + .into()); + } let as_of = super::graph_routes::parse_instant("as_of", req.as_of.as_deref())?; - let chunks = retrieval::hybrid(&state, kb_id, kb.workspace_id, q, top_k, as_of, None).await?; - Ok(Json(json!({ "results": chunks }))) + let opts = retrieval::SearchOpts { + top_k, + offset: req.offset, + as_of, + }; + let page = retrieval::hybrid_scored(&state, kb_id, kb.workspace_id, q, opts, None).await?; + Ok(Json(json!({ + "results": page.results, + "offset": req.offset, + "top_k": top_k, + "has_more": page.has_more, + "vector_used": page.vector_used, + }))) } diff --git a/crates/utopia-server/src/api/token_read_tests.rs b/crates/utopia-server/src/api/token_read_tests.rs index 22887e4fe..339e7654f 100644 --- a/crates/utopia-server/src/api/token_read_tests.rs +++ b/crates/utopia-server/src/api/token_read_tests.rs @@ -1,8 +1,8 @@ //! 个人令牌读图（0014 决定 7）：搜实体、实体详情、认知历史、事实证据、图总览、邻域、本体、 -//! 某类的实例、文档的抽取结果和整库导出这些读路由收任何一枚活着的令牌。角色照旧是上限，`kb_ids` 排除了的库答 404；改实体仍只认会话。 +//! 某类的实例、文档的抽取结果、整库导出和分块检索这些读路由收任何一枚活着的令牌。角色照旧是上限，`kb_ids` 排除了的库答 404；改实体仍只认会话。 use crate::state::AppState; use axum::body::Body; -use axum::http::{Request, StatusCode}; +use axum::http::{Method, Request, StatusCode}; use serde_json::json; use std::sync::Arc; use tower::ServiceExt; @@ -87,28 +87,49 @@ impl Fixture { .1 } - fn read_paths(&self) -> [String; 10] { + fn read_paths(&self) -> [(Method, String); 11] { let (kb, person, entity, fact, doc) = (self.kb, self.person, self.entity, self.fact, self.doc); [ - format!("/kbs/{kb}/entities?q=Zhang"), - format!("/kbs/{kb}/entities/{entity}"), - format!("/kbs/{kb}/entities/{entity}/history"), - format!("/kbs/{kb}/facts/{fact}/evidence"), - format!("/kbs/{kb}/graph/overview"), - format!("/kbs/{kb}/graph/neighborhood?entity={entity}"), - format!("/kbs/{kb}/ontology"), - format!("/kbs/{kb}/ontology/entity-types/{person}/entities"), - format!("/documents/{doc}/extractions"), - format!("/kbs/{kb}/export?format=turtle"), + (Method::GET, format!("/kbs/{kb}/entities?q=Zhang")), + (Method::GET, format!("/kbs/{kb}/entities/{entity}")), + (Method::GET, format!("/kbs/{kb}/entities/{entity}/history")), + (Method::GET, format!("/kbs/{kb}/facts/{fact}/evidence")), + (Method::GET, format!("/kbs/{kb}/graph/overview")), + ( + Method::GET, + format!("/kbs/{kb}/graph/neighborhood?entity={entity}"), + ), + (Method::GET, format!("/kbs/{kb}/ontology")), + ( + Method::GET, + format!("/kbs/{kb}/ontology/entity-types/{person}/entities"), + ), + (Method::GET, format!("/documents/{doc}/extractions")), + (Method::GET, format!("/kbs/{kb}/export?format=turtle")), + (Method::POST, format!("/kbs/{kb}/search")), ] } - async fn get(&self, path: &str, token: &str) -> anyhow::Result { + /// 每个请求都带同一个检索体；GET 路由不读它 + async fn send(&self, method: Method, path: &str, token: &str) -> anyhow::Result { + self.send_json(method, path, token, r#"{"q":"Zhang"}"#) + .await + } + + async fn send_json( + &self, + method: Method, + path: &str, + token: &str, + body: &str, + ) -> anyhow::Result { let request = Request::builder() + .method(method) .uri(format!("/api/v1{path}")) .header("authorization", format!("Bearer {token}")) - .body(Body::empty())?; + .header("content-type", "application/json") + .body(Body::from(body.to_string()))?; Ok(self.app.clone().oneshot(request).await?.status()) } @@ -128,8 +149,8 @@ async fn a_read_token_of_a_base_viewer_gets_200_on_every_graph_read() -> anyhow: }; let result = async { let token = f.read_token(f.kb).await; - for path in f.read_paths() { - let status = f.get(&path, &token).await?; + for (method, path) in f.read_paths() { + let status = f.send(method, &path, &token).await?; anyhow::ensure!(status == StatusCode::OK, "{path}: {status}"); } Ok(()) @@ -146,8 +167,8 @@ async fn a_token_limited_to_another_base_gets_404_on_every_graph_read() -> anyho }; let result = async { let token = f.read_token(f.other_kb).await; - for path in f.read_paths() { - let status = f.get(&path, &token).await?; + for (method, path) in f.read_paths() { + let status = f.send(method, &path, &token).await?; anyhow::ensure!(status == StatusCode::NOT_FOUND, "{path}: {status}"); } Ok(()) @@ -168,7 +189,7 @@ async fn an_export_by_token_is_audited_under_its_owner_with_the_token_id() -> an .await? .token_id; let path = format!("/kbs/{}/export?format=turtle", f.kb); - let status = f.get(&path, &token).await?; + let status = f.send(Method::GET, &path, &token).await?; anyhow::ensure!(status == StatusCode::OK, "{path}: {status}"); let audited: (Option, serde_json::Value) = sqlx::query_as( @@ -188,3 +209,23 @@ async fn an_export_by_token_is_audited_under_its_owner_with_the_token_id() -> an f.cleanup().await?; result } + +#[tokio::test] +async fn a_search_past_depth_100_or_with_an_unknown_filters_field_gets_422() -> anyhow::Result<()> { + let Some(f) = Fixture::new().await? else { + return Ok(()); + }; + let token = f.read_token(f.kb).await; + let path = format!("/kbs/{}/search", f.kb); + let mut statuses = Vec::new(); + for body in [ + r#"{"q":"Zhang","top_k":50,"offset":51}"#, + r#"{"q":"Zhang","filters":{"source_id":"x"}}"#, + ] { + statuses.push(f.send_json(Method::POST, &path, &token, body).await); + } + f.cleanup().await?; + let statuses = statuses.into_iter().collect::>>()?; + assert_eq!(statuses, [StatusCode::UNPROCESSABLE_ENTITY; 2]); + Ok(()) +} diff --git a/crates/utopia-server/src/auth.rs b/crates/utopia-server/src/auth.rs index a40be2926..1dee4d37b 100644 --- a/crates/utopia-server/src/auth.rs +++ b/crates/utopia-server/src/auth.rs @@ -181,6 +181,7 @@ impl FromRequestParts for AuthUser { /// 整库导出（决定 7）。 /// /// 按推送时的 `external_id` 读回一篇文档（#1276）也是文档原件，开给令牌的理由写在决定 7。 +/// 分块检索（#1271）同样按决定 7 开给令牌。 /// /// 令牌只收窄：`pat` 在的时候，路由先问 [`Self::require_write`] 或 /// [`Self::reaches`]，角色照旧由 `require_kb` 与工作区角色判。 diff --git a/crates/utopia-server/src/retrieval.rs b/crates/utopia-server/src/retrieval.rs index 4f2cbccca..4a2938107 100644 --- a/crates/utopia-server/src/retrieval.rs +++ b/crates/utopia-server/src/retrieval.rs @@ -16,12 +16,42 @@ use crate::llm_util; use crate::state::AppState; -use utopia_core::models::ChunkView; +use serde::Serialize; +use std::collections::HashMap; +use utopia_core::models::{ChunkView, SourcedChunk}; use utopia_core::AppResult; use uuid::Uuid; const RECALL_PER_CHANNEL: usize = 24; +/// 翻页最深到第几条（#1271）：再深就是在拿检索当导出用 +pub const MAX_DEPTH: usize = 100; + +pub struct SearchOpts { + pub top_k: usize, + pub offset: usize, + pub as_of: Option>, +} + +/// 一条命中连同它的分数（#1271）。`score` 是 RRF 总分，只在同一次查询里能排先后， +/// 跨查询不可比；两个名次从 1 数，没进那一路是 None +#[derive(Debug, Serialize)] +pub struct ScoredChunk { + #[serde(flatten)] + pub chunk: SourcedChunk, + pub score: f64, + pub bm25_rank: Option, + pub vector_rank: Option, +} + +#[derive(Debug)] +pub struct SearchPage { + pub results: Vec, + pub has_more: bool, + /// 向量那一路跑了没有；false 就是模块头说的静默降级 + pub vector_used: bool, +} + pub async fn hybrid( state: &AppState, kb_id: Uuid, @@ -31,32 +61,128 @@ pub async fn hybrid( as_of: Option>, embeddings: Option<&mut llm_util::EmbedCache>, ) -> AppResult> { + let (hits, _) = fused_hits( + state, + kb_id, + workspace_id, + query, + RECALL_PER_CHANNEL, + as_of, + embeddings, + ) + .await?; + Ok(hits + .into_iter() + .take(top_k) + .map(|hit| hit.chunk.chunk) + .collect()) +} + +/// 一页命中（#1271）。每一页都向每路要同样的 `MAX_DEPTH` 条，所以各页切的是 +/// 同一张融合表：页与页不重不漏。到了 `MAX_DEPTH` 的那一页没有下一页（再翻就是 422）， +/// `has_more` 是 false +pub async fn hybrid_scored( + state: &AppState, + kb_id: Uuid, + workspace_id: Uuid, + query: &str, + opts: SearchOpts, + embeddings: Option<&mut llm_util::EmbedCache>, +) -> AppResult { + let (hits, vector_used) = fused_hits( + state, + kb_id, + workspace_id, + query, + MAX_DEPTH, + opts.as_of, + embeddings, + ) + .await?; + let end = opts.offset + opts.top_k; + let has_more = end < MAX_DEPTH && hits.len() > end; + let results = hits + .into_iter() + .skip(opts.offset) + .take(opts.top_k) + .collect(); + Ok(SearchPage { + results, + has_more, + vector_used, + }) +} + +/// 两路各取 `per_channel` 条、RRF 融合、按记录轴过滤后的整张表，连同向量那一路跑了没有 +async fn fused_hits( + state: &AppState, + kb_id: Uuid, + workspace_id: Uuid, + query: &str, + per_channel: usize, + as_of: Option>, + embeddings: Option<&mut llm_util::EmbedCache>, +) -> AppResult<(Vec, bool)> { let bm25 = { let search = state.search.clone(); let kb = kb_id.to_string(); let query = query.to_string(); - tokio::task::spawn_blocking(move || search.search(&kb, &query, RECALL_PER_CHANNEL)) + tokio::task::spawn_blocking(move || search.search(&kb, &query, per_channel)) }; // `join!` 不是 `try_join!`：向量那一路失败是降级不是错误（见 vector_channel）， // `try_join!` 会把模型故障变成检索报错 let (bm25, vector) = tokio::join!( bm25, - vector_channel(state, kb_id, workspace_id, query, as_of, embeddings) + vector_channel( + state, + kb_id, + workspace_id, + query, + per_channel, + as_of, + embeddings + ) ); let bm25 = bm25 .map_err(|e| utopia_core::AppError::Other(anyhow::anyhow!("BM25 检索线程退出：{e}")))? .map_err(utopia_core::AppError::Other)?; - let lists = channel_lists(bm25.into_iter().map(|h| h.chunk_id).collect(), vector?); + let vector = vector?; + let vector_used = vector.is_some(); + let lists = channel_lists(bm25.into_iter().map(|h| h.chunk_id).collect(), vector); // Keep the bounded channel candidates until their record-time and document // filters have run: newer hits must not consume a historical query's limit. let candidate_limit = lists.iter().map(Vec::len).sum(); - let fused = utopia_search::rrf_fuse(&lists, candidate_limit); - let ids: Vec = fused.iter().filter_map(|s| s.parse().ok()).collect(); - let mut chunks = - utopia_store::documents::chunks_by_ids(&state.pool, kb_id, &ids, as_of).await?; - chunks.truncate(top_k); - Ok(chunks) + let fused: Vec<(Uuid, utopia_search::Fused)> = + utopia_search::rrf_fuse_scored(&lists, candidate_limit) + .into_iter() + .filter_map(|f| Some((f.id.parse().ok()?, f))) + .collect(); + let ids: Vec = fused.iter().map(|(id, _)| *id).collect(); + let live = utopia_store::documents::chunks_by_ids(&state.pool, kb_id, &ids, as_of).await?; + let mut fused: HashMap = fused.into_iter().collect(); + let hits = live + .into_iter() + .filter_map(|chunk| { + let f = fused.remove(&chunk.chunk.id)?; + Some(ScoredChunk { + chunk, + score: f.score, + bm25_rank: position(&f.ranks, 0), + vector_rank: position(&f.ranks, 1), + }) + }) + .collect(); + Ok((hits, vector_used)) +} + +/// 第 `channel` 路里的名次，从 1 数 +fn position(ranks: &[Option], channel: usize) -> Option { + ranks + .get(channel) + .copied() + .flatten() + .map(|rank| rank as u32 + 1) } /// 向量那一路。没配嵌入模型、嵌入请求失败、回了空，都是 `Ok(None)`：检索照常， @@ -70,6 +196,7 @@ async fn vector_channel( kb_id: Uuid, workspace_id: Uuid, query: &str, + per_channel: usize, as_of: Option>, cache: Option<&mut llm_util::EmbedCache>, ) -> AppResult>> { @@ -101,7 +228,7 @@ async fn vector_channel( &state.pool, kb_id, &vector, - RECALL_PER_CHANNEL as i64, + per_channel as i64, as_of, ) .await?; diff --git a/crates/utopia-server/src/retrieval_tests.rs b/crates/utopia-server/src/retrieval_tests.rs index aa85174d7..5f7d9da64 100644 --- a/crates/utopia-server/src/retrieval_tests.rs +++ b/crates/utopia-server/src/retrieval_tests.rs @@ -182,6 +182,41 @@ async fn both_channels_fuse_with_bm25_first() -> anyhow::Result<()> { .mount(&server) .await; f.point_embeddings_at(&server).await?; + let source = Uuid::now_v7(); + sqlx::raw_sql(&format!( + "INSERT INTO sources(id,kb_id,kind,name) VALUES ('{source}','{}','api','ci'); + UPDATE documents SET source_id='{source}', external_key='api:obs-7' WHERE id='{}'", + f.kb, f.doc + )) + .execute(&f.pool) + .await?; + let opts = super::SearchOpts { + top_k: 8, + offset: 0, + as_of: None, + }; + let page = super::hybrid_scored(&f.state, f.kb, f.ws, "apple harvest", opts, None).await?; + let hits: Vec<_> = page + .results + .iter() + .map(|h| { + ( + h.score, + h.bm25_rank, + h.vector_rank, + h.chunk.source_id, + h.chunk.external_id.as_deref(), + ) + }) + .collect(); + assert!(page.vector_used); + assert_eq!( + hits, + vec![ + (2.0 / 61.0, Some(1), Some(1), Some(source), Some("obs-7")), + (1.0 / 62.0, None, Some(2), Some(source), Some("obs-7")), + ] + ); let ids = f.search("apple harvest").await?; assert_eq!( ids, @@ -264,3 +299,98 @@ async fn historical_search_keeps_live_candidates_before_limiting_results() -> an ); Ok(()) } + +/// `count` chunks replace the fixture's two (#1271): BM25 ranks them c0 first (more +/// "apple" in the same length), the vector channel ranks them the other way round. +async fn seed_ranked_chunks(f: &Fixture, count: usize) -> anyhow::Result { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/embeddings")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "data": [{ "embedding": [1.0, 0.0, 0.0, 0.0] }] + }))) + .mount(&server) + .await; + f.point_embeddings_at(&server).await?; + sqlx::query("DELETE FROM chunks WHERE document_id=$1") + .bind(f.doc) + .execute(&f.pool) + .await?; + let mut indexed = Vec::new(); + for seq in 0..count { + let id = Uuid::now_v7(); + let text = std::iter::repeat_n("apple", count - seq) + .chain(std::iter::repeat_n("pear", seq)) + .collect::>() + .join(" "); + sqlx::query( + "INSERT INTO chunks(id,kb_id,document_id,seq,text,embedding) + VALUES($1,$2,$3,$4,$5,$6::text::vector)", + ) + .bind(id) + .bind(f.kb) + .bind(f.doc) + .bind(seq as i32) + .bind(&text) + .bind(format!("[{},1,0,0]", seq + 1)) + .execute(&f.pool) + .await?; + indexed.push((id.to_string(), text)); + } + f.state + .search + .reindex_document(&f.kb.to_string(), &f.doc.to_string(), &indexed)?; + Ok(server) +} + +fn page(offset: usize, top_k: usize) -> super::SearchOpts { + super::SearchOpts { + top_k, + offset, + as_of: None, + } +} + +/// Pages slice one fused list (#1271). A page asking each channel for fewer +/// candidates than a deeper page fuses a different list, so two pages of fifteen +/// would not join into one of thirty. +#[tokio::test] +async fn two_pages_of_fifteen_join_into_the_first_thirty_hits() -> anyhow::Result<()> { + let Some(f) = fixture().await? else { + return Ok(()); + }; + let _server = seed_ranked_chunks(&f, 30).await?; + let walk = |opts| super::hybrid_scored(&f.state, f.kb, f.ws, "apple", opts, None); + let first = walk(page(0, 15)).await?; + let second = walk(page(15, 15)).await?; + let whole = walk(page(0, 30)).await?; + let ids = + |hits: &[super::ScoredChunk]| hits.iter().map(|h| h.chunk.chunk.id).collect::>(); + let joined = [ids(&first.results), ids(&second.results)].concat(); + f.cleanup().await?; + assert_eq!(whole.results.len(), 30); + assert_eq!( + joined, + ids(&whole.results), + "no hit repeats and none is skipped" + ); + assert_eq!( + (first.has_more, second.has_more, whole.has_more), + (true, false, false) + ); + Ok(()) +} + +/// The next page past depth 100 is a 422, so the page ending there has no more +#[tokio::test] +async fn a_page_ending_at_depth_100_of_101_hits_has_no_more() -> anyhow::Result<()> { + let Some(f) = fixture().await? else { + return Ok(()); + }; + let _server = seed_ranked_chunks(&f, 101).await?; + let last = super::hybrid_scored(&f.state, f.kb, f.ws, "apple", page(50, 50), None).await; + f.cleanup().await?; + let last = last?; + assert_eq!((last.results.len(), last.has_more), (50, false)); + Ok(()) +} diff --git a/crates/utopia-store/src/documents.rs b/crates/utopia-store/src/documents.rs index 45c12acf1..6f1e30fe4 100644 --- a/crates/utopia-store/src/documents.rs +++ b/crates/utopia-store/src/documents.rs @@ -2,7 +2,7 @@ use chrono::{DateTime, Utc}; use pgvector::Vector; use serde::Serialize; use sqlx::{PgPool, Postgres, Transaction}; -use utopia_core::models::{ChunkView, Document, DocumentPage}; +use utopia_core::models::{ChunkView, Document, DocumentPage, SourcedChunk}; use utopia_core::{AppError, AppResult}; use utopia_ingest::ChunkPiece; use uuid::Uuid; @@ -1928,7 +1928,7 @@ pub async fn vector_search( Ok(rows.into_iter().map(|(id,)| id).collect()) } -/// 按 id 集合取分块（带文档名），保持传入顺序。 +/// 按 id 集合取分块（带文档名与来源），保持传入顺序。 /// /// `as_of`：记录轴（0019）——那一刻还活着的块、还没被删的文档。文档的删除 /// 同样是记录轴上的事件（#268 留了墓碑），所以两个条件一起倒。 @@ -1937,10 +1937,13 @@ pub async fn chunks_by_ids( kb_id: Uuid, ids: &[Uuid], as_of: Option>, -) -> AppResult> { - let rows: Vec = sqlx::query_as(&format!( - "SELECT c.id, c.document_id, c.seq, c.text, d.filename +) -> AppResult> { + let rows: Vec = sqlx::query_as(&format!( + "SELECT c.id, c.document_id, c.seq, c.text, d.filename, d.source_id, + CASE WHEN s.kind IN ('api', 'statements') + THEN substr(d.external_key, length(s.kind) + 2) END AS external_id FROM chunks c JOIN documents d ON d.id = c.document_id + LEFT JOIN sources s ON s.id = d.source_id WHERE c.kb_id = $1 AND c.id = ANY($2) AND {live} AND {doc_live}", live = crate::record_axis::chunk_live_at("c", as_of.map(|_| 3)), @@ -1952,8 +1955,8 @@ pub async fn chunks_by_ids( .fetch_all(pool) .await?; // 恢复 RRF 排名顺序 - let mut by_id: std::collections::HashMap = - rows.into_iter().map(|c| (c.id, c)).collect(); + let mut by_id: std::collections::HashMap = + rows.into_iter().map(|c| (c.chunk.id, c)).collect(); Ok(ids.iter().filter_map(|id| by_id.remove(id)).collect()) } diff --git a/crates/utopia-store/tests/store/a_search_reads_the_base_as_it_was.rs b/crates/utopia-store/tests/store/a_search_reads_the_base_as_it_was.rs index 7e0826033..6baccb786 100644 --- a/crates/utopia-store/tests/store/a_search_reads_the_base_as_it_was.rs +++ b/crates/utopia-store/tests/store/a_search_reads_the_base_as_it_was.rs @@ -182,7 +182,7 @@ async fn a_superseded_chunk_keeps_its_vector_and_answers_as_of_then() -> anyhow: None, ) .await?; - let ids: Vec = live_now.iter().map(|c| c.id).collect(); + let ids: Vec = live_now.iter().map(|c| c.chunk.id).collect(); assert!( !ids.contains(&f.deleted_doc_chunk), "删掉的文档不出现在现在" @@ -196,7 +196,7 @@ async fn a_superseded_chunk_keeps_its_vector_and_answers_as_of_then() -> anyhow: Some(t("2026-05-15T00:00:00Z")), ) .await?; - let ids: Vec = live_then.iter().map(|c| c.id).collect(); + let ids: Vec = live_then.iter().map(|c| c.chunk.id).collect(); assert!( ids.contains(&f.deleted_doc_chunk), "五月中那篇文档还没被删，它的块该拿得回来" diff --git a/docs/decisions/0014-identity-from-the-person-scope-from-the-token.md b/docs/decisions/0014-identity-from-the-person-scope-from-the-token.md index 8c10d626a..9825b075f 100644 --- a/docs/decisions/0014-identity-from-the-person-scope-from-the-token.md +++ b/docs/decisions/0014-identity-from-the-person-scope-from-the-token.md @@ -1,6 +1,6 @@ # 0014 · Identity from the person, scope from the token -- **Status**: Implemented · 2026-10-16 · #180, #318, #1243, #1262 · open: tokens across workspaces +- **Status**: Implemented · 2026-10-16 · #180, #318, #1243, #1262, #1271 · open: tokens across workspaces - **Written**: 2026-09-01 · condensed into English 2026-09-03 - **Related**: migration `0014_data_source_grants` gave data sources a grant layer; this is the same question where a machine knocks. [0004](0004-language-and-localization.md) has the server speak English only, MCP error codes included. [0015](0015-recording-a-sentence-is-not-asserting-a-fact.md) removes the main objection to opening `remember` over MCP. @@ -120,6 +120,10 @@ push documents in. mirror to hold a write token for a read-only job. The control is the `kb.exported` audit row, which names the owner and, through `via_token`, the token. + Chunk search (`POST /kbs/{id}/search`, added 2026-10-24, #1271) takes a token on the same + terms, so the service can ask the base about what it mirrored. Each hit carries its score, its + BM25 and vector ranks and its source; `offset` pages through one fused list, at most 100 deep. + ## Dead ends - **Base-level machine tokens**, one per base and unrelated to a person. It would be a third @@ -148,7 +152,7 @@ push documents in. and document content" now also counts them. 2026-10-19: the graph overview and neighbourhood join them, and so does reading a pushed document back by its `external_id` (#1276). The ontology and a class's instances follow (#1265), then a document's extractions and the - base's export (#1266). + base's export (#1266). 2026-10-24: chunk search joins them (#1271). ## Open questions diff --git a/docs/design/access-and-audit.md b/docs/design/access-and-audit.md index 2f533da5a..6353d9a34 100644 --- a/docs/design/access-and-audit.md +++ b/docs/design/access-and-audit.md @@ -19,10 +19,10 @@ writes an audit row `mcp.tool_called`; the tokens page shows the plaintext once snippet [0014]. A `write` token also sets up a base as its owner: creating a base and a `statements` or `api` source, reading the source's push key, importing an ontology and deciding a kind word, with the owner's role unchanged as the ceiling; any token lists a workspace's bases; a token limited by `kb_ids` -creates no base. These routes, document content and eight graph reads (entity search, detail, +creates no base. These routes, document content, eight graph reads (entity search, detail, history, fact evidence, overview, neighbourhood, ontology, a class's instances; Viewer suffices, -`kb_ids` answers 404), a document's extractions and the base's export (its `kb.exported` row -names the token) take `SessionOrToken`; apart from MCP, every other route is session-only [0014 d6, d7]. `ingest_token` on a source is a different kind: plaintext, push-only [0014]. +`kb_ids` answers 404), a document's extractions, the base's export (its `kb.exported` row +names the token) and chunk search take `SessionOrToken`; apart from MCP, every other route is session-only [0014 d6, d7]. `ingest_token` on a source is a different kind: plaintext, push-only [0014]. **A client reads a pushed document back by its `external_id`.** `GET /kbs/{id}/sources/{sid}/documents/by-external-id?external_id=…` finds an `api` or `statements` diff --git a/docs/design/chat-and-mcp.md b/docs/design/chat-and-mcp.md index d29b880cf..eaa3fbe57 100644 --- a/docs/design/chat-and-mcp.md +++ b/docs/design/chat-and-mcp.md @@ -72,7 +72,14 @@ sources for the citation numbers it repeats (#943). "now" only; neither takes `at` [0035, 0019, 0022]. Confirmed mappings and a schema document reach the prompt through retrieval; a conventions document works today where a rule would be exact [0011, 0036]. A turn embeds its question once: choosing the mappings, `search_chunks` and the -graph lookups share the turn's embedding cache (#971). +graph lookups share the turn's embedding cache (#971). The search route answers with each hit's RRF score, its +1-based rank in each channel, `vector_used` (false when the vector channel was skipped) and +`offset`/`has_more` paging to a depth of 100; a pushed source's hit carries its `external_id`. An +RRF score ranks hits within one query only; it is not comparable across queries (#1271). Every +page fuses 100 candidates per channel where chat fuses 24, so the search page can rank the same +query differently from chat and from before #1271. Pages join up only while the base is unchanged +between requests, and each channel counts its ranks before the final `as_of` filter, so a +historical page can skip rank numbers. **`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 diff --git a/web/src/docs/ingest.md b/web/src/docs/ingest.md index 35bc59479..b577f0251 100644 --- a/web/src/docs/ingest.md +++ b/web/src/docs/ingest.md @@ -197,6 +197,37 @@ The credential is a personal access token (or a signed-in session) of someone wh `content` is the current version's text; older versions are at `GET /api/v1/documents/{id}/content?version=N`. An identity marked "Not in source", a document in the trash or purged, and an identity this source never pushed are all `404`. +## Searching what you pushed + +A service can search the base's chunks with the same personal access token. The push token is not accepted here either. + +``` +POST {your-utopia-base}/api/v1/kbs/{kb_id}/search +Authorization: Bearer utp_pat_… +Content-Type: application/json + +{ "q": "harvest report", "top_k": 20, "offset": 0 } +``` + +`q` is required. `top_k` defaults to 10 and is capped at 50; `offset` defaults to 0. `as_of` (an RFC 3339 instant) reads the base as it was then; ranks are counted before `as_of` filters the hits. Any other field is a `422`. + +```json +{ + "results": [ + { + "id": "0199…", "document_id": "0199…", "seq": 3, "text": "…", "filename": "obs-000412.json", + "source_id": "0199…", "external_id": "obs-000412", + "score": 0.0328, "bm25_rank": 1, "vector_rank": 2 + } + ], + "offset": 0, "top_k": 20, "has_more": true, "vector_used": true +} +``` + +`external_id` is the identity you pushed the document under, so you can match a hit to your own record; it is `null` for documents that did not come from an API or Statements push. `score` orders hits within this one query and means nothing across queries. `bm25_rank` and `vector_rank` count from 1 and are `null` when the hit was not in that channel. `vector_used` is `false` when no embedding model answered and the hits are keyword-only. + +Page by raising `offset` by `top_k` while `has_more` is `true`. Pages stop at a depth of 100: a request with `offset + top_k` above 100 is a `422` (`page_too_deep`), and the page that reaches 100 has `has_more: false`. Pages join without repeats only while the base does not change between requests. + --- ## Shared semantics