From 0818b3b6635351499fe1deb698c58dec105f5cd7 Mon Sep 17 00:00:00 2001 From: WaylandYang Date: Wed, 9 Sep 2026 20:59:56 +0800 Subject: [PATCH] The schema document is searched, not extracted Co-Authored-By: Claude Fable 5.1 Signed-off-by: WaylandYang --- crates/utopia-core/src/models.rs | 55 +++++++++ .../src/api/datasource_routes.rs | 8 +- crates/utopia-server/src/api/graph_routes.rs | 14 +++ .../utopia-server/src/api/sources_routes.rs | 9 ++ crates/utopia-server/src/pipeline.rs | 23 ++++ crates/utopia-store/src/documents.rs | 13 ++- ...hema_document_is_searched_not_extracted.rs | 106 ++++++++++++++++++ ..._the_schema_document_leaves_extraction.sql | 24 ++++ web/src/api.ts | 2 + web/src/i18n/en.ts | 1 + web/src/i18n/zh.ts | 1 + web/src/pages/Library.tsx | 5 +- 12 files changed, 253 insertions(+), 8 deletions(-) create mode 100644 crates/utopia-store/tests/a_schema_document_is_searched_not_extracted.rs create mode 100644 migrations/0045_the_schema_document_leaves_extraction.sql diff --git a/crates/utopia-core/src/models.rs b/crates/utopia-core/src/models.rs index e7400ec4a..a75a55ae5 100644 --- a/crates/utopia-core/src/models.rs +++ b/crates/utopia-core/src/models.rs @@ -142,6 +142,61 @@ pub struct Source { pub created_at: DateTime, } +impl Source { + /// 这个来源下的文档要不要进抽取。 + /// + /// **缺省是要。** 只有 config 里 `{"extract": false}` 明说了才不抽——schema + /// 文档就是这样（0035 决定 7）：它是给问数检索表结构的语料，不是事实的来源， + /// 进抽取的结果是每个列名变成一个实体。开关记在来源上而不是文档上，因为 + /// 「只检索、不学习」是这一整个来源的性质；也没有拿来源的名字当规则，那是 + /// 命名约定冒充类型保证（0009）。 + /// + /// 值不是布尔的按没写处理：一个手滑不该让一整个来源静默停抽。 + /// `documents::queue_extraction` 里的 SQL 判的是同一件事，改一处要改两处。 + pub fn extracts(&self) -> bool { + self.config + .get("extract") + .and_then(serde_json::Value::as_bool) + .unwrap_or(true) + } +} + +#[cfg(test)] +mod source_extracts_tests { + use super::Source; + + fn with(config: serde_json::Value) -> Source { + Source { + id: uuid::Uuid::nil(), + kb_id: uuid::Uuid::nil(), + kind: "folder".into(), + name: "x".into(), + config, + icon: None, + sync_interval_minutes: None, + sync_cron: None, + last_sync_at: None, + last_sync_status: "never".into(), + last_sync_error: None, + last_sync_added: 0, + ingest_token: None, + created_at: chrono::Utc::now(), + } + } + + #[test] + fn only_an_explicit_false_turns_extraction_off() { + // 老来源的 config 是 `{}`，watch_folder 的是 `{"path": …}`：都照旧抽取 + assert!(with(serde_json::json!({})).extracts()); + assert!(with(serde_json::json!({ "path": "/x" })).extracts()); + assert!(with(serde_json::json!({ "extract": true })).extracts()); + // 不是布尔的按没写处理，而不是按 false + assert!(with(serde_json::json!({ "extract": "no" })).extracts()); + assert!(with(serde_json::json!({ "extract": 0 })).extracts()); + assert!(!with(serde_json::json!({ "extract": false })).extracts()); + } +} + /// 来源配置里**用来鉴权**的那几个键。凭据只进不出：列表与创建 / 更新的响应都剔掉， /// 更新时客户端没传或传空串就保留库里的原值，审计里也不落。 /// diff --git a/crates/utopia-server/src/api/datasource_routes.rs b/crates/utopia-server/src/api/datasource_routes.rs index 3409b4177..c77990ca2 100644 --- a/crates/utopia-server/src/api/datasource_routes.rs +++ b/crates/utopia-server/src/api/datasource_routes.rs @@ -312,7 +312,11 @@ async fn sync_schema_doc(state: &AppState, kb_id: Uuid, ds_id: Uuid) -> anyhow:: )); } - // per-KB "Data schemas" 容器来源（folder：纯容器语义） + // per-KB "Data schemas" 容器来源（folder：纯容器语义）。 + // + // **这份文档只做检索语料，不进抽取**（0035 决定 7）。它跟别的文档一样进抽取的 + // 时候，抽取器把每个列名都当成了实体——宽表语料上四十个概念实体里二十八个是 + // 列名（#553）。「不抽取」记在来源的 config 上，流水线读它 let folder = match sqlx::query_as::<_, (Uuid,)>( "SELECT id FROM sources WHERE kb_id = $1 AND kind = 'folder' AND name = 'Data schemas'", ) @@ -327,7 +331,7 @@ async fn sync_schema_doc(state: &AppState, kb_id: Uuid, ds_id: Uuid) -> anyhow:: kb_id, "folder", "Data schemas", - &serde_json::json!({}), + &serde_json::json!({ "extract": false }), Some("database"), None, None, diff --git a/crates/utopia-server/src/api/graph_routes.rs b/crates/utopia-server/src/api/graph_routes.rs index eb846059d..401e7f719 100644 --- a/crates/utopia-server/src/api/graph_routes.rs +++ b/crates/utopia-server/src/api/graph_routes.rs @@ -474,6 +474,20 @@ pub async fn extract( ) -> ApiResult> { let doc = utopia_store::documents::get(&state.pool, document_id).await?; require_kb(&state, &user, doc.kb_id, Role::Editor).await?; + // 来源说了不抽取的（schema 文档，0035 决定 7）：说清楚为什么，而不是排一个 + // 流水线到了那一步又跳过的任务 + if let Some(source_id) = doc.source_id { + if !utopia_store::sources::get(&state.pool, source_id) + .await? + .extracts() + { + return Err(utopia_core::AppError::invalid( + "source_not_extracted", + "Documents under this source are searched, not extracted", + ) + .into()); + } + } // 手动触发 = 强制全量：清增量标记、解雇在跑的任务、置 queued、建任务，一个事务办完 let job_id = utopia_store::documents::queue_extraction_one(&state.pool, document_id).await?; state.emit_document(doc.kb_id, document_id); diff --git a/crates/utopia-server/src/api/sources_routes.rs b/crates/utopia-server/src/api/sources_routes.rs index 32d9c988a..19f38c892 100644 --- a/crates/utopia-server/src/api/sources_routes.rs +++ b/crates/utopia-server/src/api/sources_routes.rs @@ -551,6 +551,15 @@ pub async fn re_extract( if source.kb_id != kb_id { return Err(utopia_core::AppError::NotFound.into()); } + // `queue_extraction` 自己也会把这种来源的文档滤掉，那样这里回的是「排了 0 篇」。 + // 点按钮的人该听到的是为什么 + if !source.extracts() { + return Err(utopia_core::AppError::invalid( + "source_not_extracted", + "Documents under this source are searched, not extracted", + ) + .into()); + } // 任务由 queue_extraction 与状态同事务建好，这里只负责推送 let ids = utopia_store::documents::queue_extraction(&state.pool, kb_id, Some(source_id)).await?; diff --git a/crates/utopia-server/src/pipeline.rs b/crates/utopia-server/src/pipeline.rs index 0ced3d11c..56ce766b5 100644 --- a/crates/utopia-server/src/pipeline.rs +++ b/crates/utopia-server/src/pipeline.rs @@ -8,6 +8,16 @@ use utopia_core::models::{LlmSettings, Proposer}; use utopia_llm::LlmClient; use uuid::Uuid; +/// 这份文档的来源要不要抽取。没有来源的文档（直接上传、记忆片段）照旧抽。 +async fn source_extracts(state: &AppState, source_id: Option) -> anyhow::Result { + let Some(id) = source_id else { + return Ok(true); + }; + Ok(utopia_store::sources::get(&state.pool, id) + .await? + .extracts()) +} + /// 每次送多少条去嵌入。 /// /// 与 ontology_index 的 64 不同，这里没量过，先不动：那边的注释说批大小要拿真实文本量， @@ -80,6 +90,19 @@ async fn run(state: &AppState, document_id: Uuid) -> anyhow::Result<()> { utopia_store::documents::set_ready(&state.pool, document_id, text_len, chunk_count).await?; + // 来源说了不抽取的，到这里为止：可搜、可问，不进图。 + // + // schema 文档就是这一类（0035 决定 7）——它是给问数检索表结构的语料，进抽取的 + // 结果是抽取器把每个列名当成一个实体（宽表语料上四十个概念实体里二十八个是 + // 列名，#553）。状态记成 `skipped` 而不是留在 `none`：`none` 在 Library 里读作 + // 「还没排到」，而它永远不会排到 + if !source_extracts(state, doc.source_id).await? { + utopia_store::documents::set_graph_status(&state.pool, document_id, "skipped").await?; + state.emit_document(doc.kb_id, document_id); + tracing::info!(%document_id, chunks = chunk_count, "文档处理完成，来源不抽取"); + return Ok(()); + } + // 两段式：索引就绪后，若配置了对话模型则排队图谱抽取（不阻塞可搜可问） if settings.as_ref().is_some_and(|s| s.chat_ready()) { utopia_store::documents::set_graph_status(&state.pool, document_id, "queued").await?; diff --git a/crates/utopia-store/src/documents.rs b/crates/utopia-store/src/documents.rs index f822f635a..622434b9c 100644 --- a/crates/utopia-store/src/documents.rs +++ b/crates/utopia-store/src/documents.rs @@ -1301,11 +1301,16 @@ pub async fn queue_extraction( source_id: Option, ) -> AppResult> { let mut tx = pool.begin().await?; + // 来源说了不抽取的文档不排（`Source::extracts` 的 SQL 版，两处判的是同一个键）。 + // 这条在库里挡而不是在各个入口挡：全库重建（`rebuild`）、按来源重抽、以后 + // 任何新入口都走这里，schema 文档一律不进图（0035 决定 7，#553） let ids: Vec<(Uuid,)> = sqlx::query_as( - "SELECT id FROM documents - WHERE kb_id = $1 AND deleted_at IS NULL AND status = 'ready' - AND ($2::uuid IS NULL OR source_id = $2) - ORDER BY created_at", + "SELECT d.id FROM documents d + LEFT JOIN sources s ON s.id = d.source_id + WHERE d.kb_id = $1 AND d.deleted_at IS NULL AND d.status = 'ready' + AND ($2::uuid IS NULL OR d.source_id = $2) + AND NOT coalesce(s.config -> 'extract' = 'false'::jsonb, false) + ORDER BY d.created_at", ) .bind(kb_id) .bind(source_id) diff --git a/crates/utopia-store/tests/a_schema_document_is_searched_not_extracted.rs b/crates/utopia-store/tests/a_schema_document_is_searched_not_extracted.rs new file mode 100644 index 000000000..89e92aab5 --- /dev/null +++ b/crates/utopia-store/tests/a_schema_document_is_searched_not_extracted.rs @@ -0,0 +1,106 @@ +//! 来源说了不抽取的文档，`queue_extraction` 不排——打在真库上（0035 决定 7，#553）。 +//! +//! 挂载数据源时 schema 被摄成一份 markdown，好让问数检索得到表结构。它从前跟 +//! 别的文档一样进抽取，抽取器把每个列名都当成实体：宽表语料上四十个概念实体里 +//! 二十八个是列名。「只检索、不学习」记在来源的 config 上，而这条过滤放在库里 +//! 而不是各个入口——全库重建、按来源重抽、以后任何新入口都走这一个查询。 + +use sqlx::PgPool; +use uuid::Uuid; + +async fn base(pool: &PgPool) -> anyhow::Result { + let (org, ws, kb) = (Uuid::now_v7(), Uuid::now_v7(), Uuid::now_v7()); + sqlx::query("INSERT INTO organizations (id, name) VALUES ($1, 'skip-test')") + .bind(org) + .execute(pool) + .await?; + sqlx::query("INSERT INTO workspaces (id, org_id, name) VALUES ($1, $2, 'skip-test')") + .bind(ws) + .bind(org) + .execute(pool) + .await?; + sqlx::query( + "INSERT INTO knowledge_bases (id, workspace_id, name) VALUES ($1, $2, 'skip-test')", + ) + .bind(kb) + .bind(ws) + .execute(pool) + .await?; + Ok(kb) +} + +async fn ready_document(pool: &PgPool, kb: Uuid, source: Uuid, name: &str) -> anyhow::Result { + let id = Uuid::now_v7(); + sqlx::query( + "INSERT INTO documents (id, kb_id, source_id, filename, sha256, status) + VALUES ($1, $2, $3, $4, $5, 'ready')", + ) + .bind(id) + .bind(kb) + .bind(source) + .bind(name) + .bind(format!("sha-{id}")) + .execute(pool) + .await?; + Ok(id) +} + +#[tokio::test] +async fn a_source_that_does_not_extract_keeps_its_documents_out_of_the_queue() -> anyhow::Result<()> +{ + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + let kb = base(&pool).await?; + + let run = async { + // 与 `sync_schema_doc` 建的那个一模一样：folder，config 里明说不抽取 + let schemas = utopia_store::sources::create( + &pool, + kb, + "folder", + "Data schemas", + &serde_json::json!({ "extract": false }), + Some("database"), + None, + None, + ) + .await?; + // 老来源：config 是 `{}`，照旧抽取 + let uploads = utopia_store::sources::create( + &pool, + kb, + "folder", + "Uploads", + &serde_json::json!({}), + None, + None, + None, + ) + .await?; + assert!(!schemas.extracts() && uploads.extracts()); + + let schema_doc = ready_document(&pool, kb, schemas.id, "wide-schema.md").await?; + let prose_doc = ready_document(&pool, kb, uploads.id, "report.md").await?; + + // 全库重建走的就是这一条（source_id = None） + let queued = utopia_store::documents::queue_extraction(&pool, kb, None).await?; + assert_eq!(queued, vec![prose_doc], "只有普通来源下的文档该被排上"); + assert!(!queued.contains(&schema_doc)); + + // 按来源重抽：来源本身说了不抽取，排出来的是空 + let queued = utopia_store::documents::queue_extraction(&pool, kb, Some(schemas.id)).await?; + assert!(queued.is_empty(), "不抽取的来源按来源重抽也该是空"); + + Ok::<_, anyhow::Error>(()) + } + .await; + + // 只删知识库，不删 org——用户是软删除的，测试也不该造一个产品里不存在的动作 + sqlx::query("DELETE FROM knowledge_bases WHERE id = $1") + .bind(kb) + .execute(&pool) + .await?; + run +} diff --git a/migrations/0045_the_schema_document_leaves_extraction.sql b/migrations/0045_the_schema_document_leaves_extraction.sql new file mode 100644 index 000000000..51db3407a --- /dev/null +++ b/migrations/0045_the_schema_document_leaves_extraction.sql @@ -0,0 +1,24 @@ +-- schema 文档只做检索语料，不进抽取（0035 决定 7，#553）。 +-- +-- 挂载数据源时把 schema 摄成一份 markdown，好让问数 `search_chunks` 找得到表。 +-- 这份文档从前跟别的文档一样进抽取，而抽取器看见的正是探索刚建的 Metric / +-- Dimension 两个类——于是每个列名都被抽成一个实体。宽表语料上量过：四十个 +-- 概念实体里二十八个是列名，`amt_pay` 是一个 Metric，`dw.dim_shop` 是一个 +-- Dimension 还挂着八条事实。 +-- +-- 「只检索、不学习」是这份文档所属来源的性质，记在 `sources.config`——那一列 +-- 本来就是 kind 专属配置的 JSONB。没有拿来源的名字当规则（「叫 Data schemas 的 +-- 文件夹不抽取」）：那是拿命名约定冒充类型保证，0009 警告过的那一类缺陷。 + +-- 流水线走到抽取那一步、发现来源不要抽取时，文档得有个状态说这件事。 +-- 留在 `none` 会让 Library 把它显示成「还没排到」，而它永远不会排到 +ALTER TABLE documents DROP CONSTRAINT IF EXISTS documents_graph_status_check; +ALTER TABLE documents + ADD CONSTRAINT documents_graph_status_check + CHECK (graph_status IN ('none', 'queued', 'extracting', 'done', 'failed', 'skipped')); + +-- 已有的库里那个文件夹是 `sync_schema_doc` 用固定名字建的，补上标记。 +-- 这里按名字找是一次性回填，不是运行时规则 +UPDATE sources + SET config = config || '{"extract": false}'::jsonb + WHERE kind = 'folder' AND name = 'Data schemas'; diff --git a/web/src/api.ts b/web/src/api.ts index 66ac15940..7e3b6ec2e 100644 --- a/web/src/api.ts +++ b/web/src/api.ts @@ -206,6 +206,8 @@ export interface SourceView { base_url?: string; /** jira_issues：项目 key，如 KAFKA */ project?: string; + /** false = 这个来源下的文档只检索、不抽取（schema 文档，0035 决定 7）；缺省抽取 */ + extract?: boolean; } | null; icon: string | null; sync_interval_minutes: number | null; diff --git a/web/src/i18n/en.ts b/web/src/i18n/en.ts index 826676e8d..01595288c 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -506,6 +506,7 @@ export const en = { extracting: "Extracting", done: "Done", failed: "Failed", + skipped: "Not extracted", }, sources: "Sources", allDocs: "All documents", diff --git a/web/src/i18n/zh.ts b/web/src/i18n/zh.ts index b2fc1d786..e1e20c705 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -457,6 +457,7 @@ export const zh: Strings = { extracting: "抽取中", done: "完成", failed: "失败", + skipped: "不抽取", }, sources: "来源", allDocs: "全部文档", diff --git a/web/src/pages/Library.tsx b/web/src/pages/Library.tsx index 296594bb5..58fc9044c 100644 --- a/web/src/pages/Library.tsx +++ b/web/src/pages/Library.tsx @@ -1044,8 +1044,9 @@ function SourceBar({ {S.library.viewToken} )} - {/* 全量重抽本来源：所有类型都给（有文档就能重抽） */} - {onReExtract && source.doc_count > 0 && ( + {/* 全量重抽本来源：所有类型都给（有文档就能重抽）——除了说了不抽取的 + 那种（schema 文档）：后端会拒，按钮就不该出现 */} + {onReExtract && source.doc_count > 0 && source.config?.extract !== false && (