Search answers a service with each hit's score, ranks and source, and pages

crates/utopia-core/src/models.rs

@@ -479,6 +479,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<Uuid>,
+    pub external_id: Option<String>,
+}
+
 #[derive(Debug, Clone, Serialize, sqlx::FromRow)]
 pub struct LlmSettings {
     pub workspace_id: Uuid,

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<Option<usize>>,
+}
+
 /// Reciprocal Rank Fusion：融合多路召回的排名（k=60 为经验常数）。
 pub fn rrf_fuse(lists: &[Vec<String>], limit: usize) -> Vec<String> {
+    rrf_fuse_scored(lists, limit)
+        .into_iter()
+        .map(|fused| fused.id)
+        .collect()
+}
+
+/// 同 [`rrf_fuse`]，但把分数和各路名次一并交回来
+pub fn rrf_fuse_scored(lists: &[Vec<String>], limit: usize) -> Vec<Fused> {
     const K: f64 = 60.0;
-    let mut scores: HashMap<String, f64> = HashMap::new();
-    for list in lists {
+    let mut by_id: HashMap<String, Fused> = 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<Fused> = 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<String> {
         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() {

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<usize>,
+    #[serde(default)]
+    pub offset: usize,
     /// 记录轴（0019）：按**那一刻库里有的东西**检索。YYYY-MM-DD 或 RFC3339。
     /// 命中的是当时活着的块、当时还没被删的文档
     #[serde(default)]
@@ -24,18 +27,36 @@ pub struct SearchReq {
 
 pub async fn search(
     State(state): State<AppState>,
-    AuthUser(user): AuthUser,
+    reader: SessionOrToken,
     Path(kb_id): Path<Uuid>,
     Json(req): Json<SearchReq>,
 ) -> ApiResult<Json<serde_json::Value>> {
     let q = req.q.trim();
     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,
+    })))
 }

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<StatusCode> {
+    /// 每个请求都带同一个检索体；GET 路由不读它
+    async fn send(&self, method: Method, path: &str, token: &str) -> anyhow::Result<StatusCode> {
+        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<StatusCode> {
         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<Uuid>, 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::<anyhow::Result<Vec<_>>>()?;
+    assert_eq!(statuses, [StatusCode::UNPROCESSABLE_ENTITY; 2]);
+    Ok(())
+}

crates/utopia-server/src/auth.rs

@@ -181,6 +181,7 @@ impl FromRequestParts<AppState> for AuthUser {
 /// 整库导出（决定 7）。
 ///
 /// 按推送时的 `external_id` 读回一篇文档（#1276）也是文档原件，开给令牌的理由写在决定 7。
+/// 分块检索（#1271）同样按决定 7 开给令牌。
 ///
 /// 令牌只收窄：`pat` 在的时候，路由先问 [`Self::require_write`] 或
 /// [`Self::reaches`]，角色照旧由 `require_kb` 与工作区角色判。