diff --git a/crates/utopia-server/src/phrase_alignment.rs b/crates/utopia-server/src/phrase_alignment.rs index adcea627a..514786a72 100644 --- a/crates/utopia-server/src/phrase_alignment.rs +++ b/crates/utopia-server/src/phrase_alignment.rs @@ -305,6 +305,12 @@ async fn align_phrases_locked( } } tracing::info!(%kb_id, bound, none, undecided, skipped, failed, "短语对齐完成"); + // 绑定定了,视图跟着算:绑上的签名下的陈述成类型化行,绑定变了的行作废(0067) + let typed = utopia_store::materialize::materialize(pool, kb_id).await?; + tracing::info!(%kb_id, added = typed.added, retired = typed.retired, "类型化事实按绑定算完"); + if typed.added > 0 || typed.retired > 0 { + state.emit_graph(kb_id); + } // 这一轮跑着的时候世界没停:新文档带来新签名,改了的属性让刚判的绑定过期,本轮没排上 // 的触发也都落在这里。有失败的批次、有没试过的新签名、有本轮判完又过期的绑定,就再排 // 一次 diff --git a/crates/utopia-store/src/lib.rs b/crates/utopia-store/src/lib.rs index 93af005ee..c29f5b19a 100644 --- a/crates/utopia-store/src/lib.rs +++ b/crates/utopia-store/src/lib.rs @@ -19,6 +19,7 @@ pub mod graph; pub mod jobs; pub mod kbs; pub mod mappings; +pub mod materialize; pub mod members; pub mod memory; pub mod model_limits; diff --git a/crates/utopia-store/src/materialize.rs b/crates/utopia-store/src/materialize.rs new file mode 100644 index 000000000..cde67cd5c --- /dev/null +++ b/crates/utopia-store/src/materialize.rs @@ -0,0 +1,155 @@ +//! 绑上的签名下的开放陈述算成类型化事实(0044 决定 3 的第三片,见 0067)。 +//! +//! 类型化图谱是视图:一条类型化行 = 一条开放陈述 × 它签名的绑定。谓词是绑定给的属性, +//! 主宾按绑定的方向(reverse 就是陈述的宾语当主语),字面值、世界轴时间、来源时间、 +//! 置信度照抄,证据与限定各复制一份,`from_statement_id` 指回那条陈述。带 mood 限定的 +//! 陈述不算。 +//! +//! 重算是集合运算,跑多少遍结果一样:先作废「源头不成立」的类型化行(陈述作废了、签名 +//! 不再绑着、绑到了别的属性或反了方向),再给「该有而没有」的陈述补一行。绑定不变的行 +//! 不动——它们的 id、证据和记录时间都留着。没有模型调用。 + +use sqlx::PgPool; +use utopia_core::AppResult; +use uuid::Uuid; + +/// 陈述与绑定对得上的条件:短语归一后相等,两端的类相同(空也相同),宾语是不是字面值相同。 +/// 两处 SQL 共用;`s` 是开放陈述(facts),`se`/`oe` 是它两端的实体,`b` 是 phrase_bindings +const MATCH: &str = "b.kb_id = s.kb_id + AND b.phrase = lower(btrim(regexp_replace(s.phrase, '\\s+', ' ', 'g'))) + AND b.subject_type_id IS NOT DISTINCT FROM se.type_id + AND b.object_is_value = (s.object_id IS NULL) + AND (s.object_id IS NULL OR b.object_type_id IS NOT DISTINCT FROM oe.type_id)"; + +/// 一轮重算写了什么。 +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct Outcome { + /// 作废的类型化行 + pub retired: u64, + /// 新算出来的类型化行 + pub added: u64, +} + +/// 对一个库重算一遍。 +pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult { + let mut tx = pool.begin().await?; + + // 1. 作废源头不成立的:陈述死了、签名没绑着、属性或方向变了、陈述带了 mood + let retired = sqlx::query(&format!( + "UPDATE facts t + SET invalidated_at = now() + WHERE t.kb_id = $1 AND t.layer = 'typed' AND t.invalidated_at IS NULL + AND t.from_statement_id IS NOT NULL + AND NOT EXISTS ( + SELECT 1 + FROM facts s + JOIN entities se ON se.id = s.subject_id + LEFT JOIN entities oe ON oe.id = s.object_id + JOIN phrase_bindings b ON {MATCH} + WHERE s.id = t.from_statement_id + AND s.invalidated_at IS NULL + AND b.status = 'bound' + AND b.relation_type_id = t.predicate_id + AND ((b.direction = 'forward' AND t.subject_id = s.subject_id) + OR (b.direction = 'reverse' AND t.subject_id = s.object_id)) + AND NOT EXISTS (SELECT 1 FROM statement_qualifiers q + WHERE q.fact_id = s.id AND q.role = 'mood'))" + )) + .bind(kb_id) + .execute(&mut *tx) + .await? + .rows_affected(); + + // 2. 补该有而没有的:先选出(陈述, 绑定)对,再逐条落行——id 是 v7,库里没有生成 + // 它的函数。reverse 只对两样东西之间的关系有意义(字面值当不了主语) + let due: Vec<(Uuid, Uuid, String)> = sqlx::query_as(&format!( + "SELECT s.id, b.relation_type_id, b.direction + FROM facts s + JOIN entities se ON se.id = s.subject_id + LEFT JOIN entities oe ON oe.id = s.object_id + JOIN phrase_bindings b ON {MATCH} + WHERE s.kb_id = $1 AND s.layer = 'open' AND s.invalidated_at IS NULL + AND b.status = 'bound' + AND (b.direction = 'forward' OR s.object_id IS NOT NULL) + AND NOT EXISTS (SELECT 1 FROM statement_qualifiers q + WHERE q.fact_id = s.id AND q.role = 'mood') + AND NOT EXISTS (SELECT 1 FROM facts t + WHERE t.from_statement_id = s.id AND t.invalidated_at IS NULL + AND t.predicate_id = b.relation_type_id) + ORDER BY s.id" + )) + .bind(kb_id) + .fetch_all(&mut *tx) + .await?; + let mut added: Vec<(Uuid, Uuid)> = Vec::with_capacity(due.len()); + for (statement, property, direction) in &due { + let id = Uuid::now_v7(); + let reverse = direction == "reverse"; + sqlx::query( + "INSERT INTO facts (id, kb_id, subject_id, predicate_id, object_id, object_value, + valid_from, valid_from_precision, valid_to, valid_to_precision, + attested_from, attested_to, confidence, layer, from_statement_id) + SELECT $1, s.kb_id, + CASE WHEN $4 THEN s.object_id ELSE s.subject_id END, + $3, + CASE WHEN $4 THEN s.subject_id ELSE s.object_id END, + CASE WHEN $4 THEN NULL ELSE s.object_value END, + s.valid_from, s.valid_from_precision, s.valid_to, s.valid_to_precision, + s.attested_from, s.attested_to, s.confidence, 'typed', s.id + FROM facts s WHERE s.id = $2", + ) + .bind(id) + .bind(statement) + .bind(property) + .bind(reverse) + .execute(&mut *tx) + .await?; + added.push((id, *statement)); + } + + // 3. 新行抄证据与限定:证据是同一段原文的同一处引文;限定照角色词原样带过去 + if !added.is_empty() { + let ids: Vec = added.iter().map(|(id, _)| *id).collect(); + sqlx::query( + "INSERT INTO fact_evidence (fact_id, chunk_id, quote, document_id, doc_version, + proposed_predicate, quote_start, quote_end) + SELECT t.id, e.chunk_id, e.quote, e.document_id, e.doc_version, + e.proposed_predicate, e.quote_start, e.quote_end + FROM facts t + JOIN fact_evidence e ON e.fact_id = t.from_statement_id + WHERE t.id = ANY($1) + ON CONFLICT DO NOTHING", + ) + .bind(&ids) + .execute(&mut *tx) + .await?; + sqlx::query( + "INSERT INTO statement_qualifiers (fact_id, role, value, entity_id) + SELECT t.id, q.role, q.value, q.entity_id + FROM facts t + JOIN statement_qualifiers q ON q.fact_id = t.from_statement_id + WHERE t.id = ANY($1) + ON CONFLICT DO NOTHING", + ) + .bind(&ids) + .execute(&mut *tx) + .await?; + } + tx.commit().await?; + Ok(Outcome { + retired, + added: added.len() as u64, + }) +} + +/// 库里活着的、从陈述算出来的类型化行数。 +pub async fn count(pool: &PgPool, kb_id: Uuid) -> AppResult { + Ok(sqlx::query_scalar( + "SELECT count(*) FROM facts + WHERE kb_id = $1 AND layer = 'typed' AND from_statement_id IS NOT NULL + AND invalidated_at IS NULL", + ) + .bind(kb_id) + .fetch_one(pool) + .await?) +} diff --git a/crates/utopia-store/tests/a_bound_statement_becomes_a_typed_fact.rs b/crates/utopia-store/tests/a_bound_statement_becomes_a_typed_fact.rs new file mode 100644 index 000000000..b3be57940 --- /dev/null +++ b/crates/utopia-store/tests/a_bound_statement_becomes_a_typed_fact.rs @@ -0,0 +1,232 @@ +//! 类型化图谱是视图(0044 决定 3,0067):绑上的签名下的开放陈述算成类型化行,绑定变了 +//! 行跟着变,跑多少遍结果一样。 +//! +//! 一条陈述「Harbor Bakery —is based in→ Port Ellen」,签名 (is based in, organization, +//! place) 绑到 headquartered_in:算出一条类型化行,谓词是它,证据与限定抄过来, +//! `from_statement_id` 指回陈述;再跑一遍什么都不动;带 mood 的陈述不算;绑定翻成 +//! reverse,旧行作废、新行主宾对调;绑定改成 none,行作废、不再补。 +//! +//! 没有 `UTOPIA_DATABASE_URL` 时跳过而不是失败。自建自拆,绝不碰已有的库。 + +use sqlx::PgPool; +use utopia_store::materialize::{materialize, Outcome}; +use uuid::Uuid; + +#[tokio::test] +async fn a_bound_statement_becomes_a_typed_fact() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + sqlx::query("DELETE FROM organizations WHERE name = 'materialize-test'") + .execute(&pool) + .await?; + + let (org, ws, kb) = (Uuid::now_v7(), Uuid::now_v7(), Uuid::now_v7()); + sqlx::query("INSERT INTO organizations (id, name) VALUES ($1, 'materialize-test')") + .bind(org) + .execute(&pool) + .await?; + sqlx::query("INSERT INTO workspaces (id, org_id, name) VALUES ($1, $2, 'materialize-test')") + .bind(ws) + .bind(org) + .execute(&pool) + .await?; + sqlx::query( + "INSERT INTO knowledge_bases (id, workspace_id, name) VALUES ($1, $2, 'materialize-test')", + ) + .bind(kb) + .bind(ws) + .execute(&pool) + .await?; + + let run = async { + // 两个类、一条属性 + let (organization, place, hq) = (Uuid::now_v7(), Uuid::now_v7(), Uuid::now_v7()); + for (id, key) in [(organization, "organization"), (place, "place")] { + sqlx::query( + "INSERT INTO entity_types (id, kb_id, key, label, color, shape) + VALUES ($1, $2, $3, $3, '#7fd0ff', 'circle')", + ) + .bind(id) + .bind(kb) + .bind(key) + .execute(&pool) + .await?; + } + sqlx::query( + "INSERT INTO relation_types (id, kb_id, key, label, temporal) + VALUES ($1, $2, 'headquartered_in', 'headquartered in', 'state')", + ) + .bind(hq) + .bind(kb) + .execute(&pool) + .await?; + + // 两样东西,两条陈述:一条平常的,一条带 mood 的 + let (bakery, port) = (Uuid::now_v7(), Uuid::now_v7()); + for (id, type_id, name) in [(bakery, organization, "Harbor Bakery"), (port, place, "Port Ellen")] { + sqlx::query( + "INSERT INTO entities (id, kb_id, type_id, canonical_name) VALUES ($1, $2, $3, $4)", + ) + .bind(id) + .bind(kb) + .bind(type_id) + .bind(name) + .execute(&pool) + .await?; + } + let (stated, planned) = (Uuid::now_v7(), Uuid::now_v7()); + for (id, phrase) in [(stated, "is based in"), (planned, "will move to")] { + sqlx::query( + "INSERT INTO facts (id, kb_id, subject_id, object_id, layer, phrase, confidence, + valid_from, valid_from_precision) + VALUES ($1, $2, $3, $4, 'open', $5, 0.9, '2019-01-01', 'year')", + ) + .bind(id) + .bind(kb) + .bind(bakery) + .bind(port) + .bind(phrase) + .execute(&pool) + .await?; + } + let (doc, chunk) = (Uuid::now_v7(), Uuid::now_v7()); + sqlx::query( + "INSERT INTO documents (id, kb_id, filename, sha256) + VALUES ($1, $2, 'bakery.txt', 'materialize-test')", + ) + .bind(doc) + .bind(kb) + .execute(&pool) + .await?; + sqlx::query( + "INSERT INTO chunks (id, kb_id, document_id, seq, text) + VALUES ($1, $2, $3, 0, 'Harbor Bakery is based in Port Ellen.')", + ) + .bind(chunk) + .bind(kb) + .bind(doc) + .execute(&pool) + .await?; + sqlx::query( + "INSERT INTO fact_evidence (fact_id, chunk_id, quote, document_id, quote_start, quote_end) + VALUES ($1, $2, 'Harbor Bakery is based in Port Ellen.', $3, 0, 37)", + ) + .bind(stated) + .bind(chunk) + .bind(doc) + .execute(&pool) + .await?; + sqlx::query( + "INSERT INTO statement_qualifiers (fact_id, role, value) VALUES ($1, 'since', '\"2019\"'::jsonb)", + ) + .bind(stated) + .execute(&pool) + .await?; + sqlx::query( + "INSERT INTO statement_qualifiers (fact_id, role, value) VALUES ($1, 'mood', '\"will\"'::jsonb)", + ) + .bind(planned) + .execute(&pool) + .await?; + + // 两条签名都绑到 headquartered_in + let binding = |phrase: &'static str| { + let pool = pool.clone(); + async move { + sqlx::query( + "INSERT INTO phrase_bindings + (id, kb_id, phrase, subject_type_id, object_type_id, object_is_value, + relation_type_id, direction, status) + VALUES ($1, $2, $3, $4, $5, false, $6, 'forward', 'bound')", + ) + .bind(Uuid::now_v7()) + .bind(kb) + .bind(phrase) + .bind(organization) + .bind(place) + .bind(hq) + .execute(&pool) + .await + } + }; + binding("is based in").await?; + binding("will move to").await?; + + let first = materialize(&pool, kb).await?; + assert_eq!(first, Outcome { retired: 0, added: 1 }, "带 mood 的那条不算"); + let (subject, object, predicate, from, vf): (Uuid, Uuid, Uuid, Uuid, Option>) = + sqlx::query_as( + "SELECT subject_id, object_id, predicate_id, from_statement_id, valid_from + FROM facts WHERE kb_id = $1 AND layer = 'typed' AND invalidated_at IS NULL", + ) + .bind(kb) + .fetch_one(&pool) + .await?; + assert_eq!((subject, object, predicate, from), (bakery, port, hq, stated)); + assert!(vf.is_some(), "世界轴时间抄过来"); + let typed_id: Uuid = sqlx::query_scalar( + "SELECT id FROM facts WHERE from_statement_id = $1 AND invalidated_at IS NULL", + ) + .bind(stated) + .fetch_one(&pool) + .await?; + let evidence: i64 = + sqlx::query_scalar("SELECT count(*) FROM fact_evidence WHERE fact_id = $1") + .bind(typed_id) + .fetch_one(&pool) + .await?; + let qualifiers: Vec = sqlx::query_scalar( + "SELECT role FROM statement_qualifiers WHERE fact_id = $1 ORDER BY role", + ) + .bind(typed_id) + .fetch_all(&pool) + .await?; + assert_eq!((evidence, qualifiers), (1, vec!["since".to_string()]), "证据与限定各一份"); + + // 再跑一遍:什么都不动 + assert_eq!(materialize(&pool, kb).await?, Outcome::default()); + + // 绑定翻成 reverse:旧行作废,新行主宾对调 + sqlx::query("UPDATE phrase_bindings SET direction = 'reverse' WHERE kb_id = $1 AND phrase = 'is based in'") + .bind(kb) + .execute(&pool) + .await?; + assert_eq!(materialize(&pool, kb).await?, Outcome { retired: 1, added: 1 }); + let (subject, object): (Uuid, Uuid) = sqlx::query_as( + "SELECT subject_id, object_id FROM facts + WHERE from_statement_id = $1 AND invalidated_at IS NULL", + ) + .bind(stated) + .fetch_one(&pool) + .await?; + assert_eq!((subject, object), (port, bakery), "方向反了主宾对调"); + let retired: i64 = sqlx::query_scalar( + "SELECT count(*) FROM facts WHERE from_statement_id = $1 AND invalidated_at IS NOT NULL", + ) + .bind(stated) + .fetch_one(&pool) + .await?; + assert_eq!(retired, 1, "旧行留着,只是作废了"); + + // 绑定改成 none:行作废,不再补 + sqlx::query( + "UPDATE phrase_bindings SET status = 'none', relation_type_id = NULL, direction = NULL + WHERE kb_id = $1 AND phrase = 'is based in'", + ) + .bind(kb) + .execute(&pool) + .await?; + assert_eq!(materialize(&pool, kb).await?, Outcome { retired: 1, added: 0 }); + assert_eq!(utopia_store::materialize::count(&pool, kb).await?, 0); + anyhow::Ok(()) + } + .await; + + sqlx::query("DELETE FROM organizations WHERE id = $1") + .bind(org) + .execute(&pool) + .await?; + run +} diff --git a/docs/design/ontology.md b/docs/design/ontology.md index 6ea78dea9..d4a9f1bb2 100644 --- a/docs/design/ontology.md +++ b/docs/design/ontology.md @@ -71,8 +71,20 @@ signatures and dropped the judge to 87%, and on the NVDA releases, where 15 of 2 bind, to 59%: the generic "value" attribute swallowed every cash-flow row. Coverage therefore follows the kind-word bindings and the ontology's size, not the binder. What still goes wrong is a phrase that carries part of the value ("下降 1.4%" bound to a change property loses its sign) -and a table section read as a change ("changes in operating assets › accounts payable"). Nothing is materialised yet: the typed rows with `from_statement_id` are -the next slice. +and a table section read as a change ("changes in operating assets › accounts payable"). + +**A bound statement is a typed fact** [0044 cut 2, #752]. The typed graph is computed, never +written by hand: after each phrase-alignment run, every live open statement under a bound signature +becomes one `layer = 'typed'` row whose predicate is the bound property, whose ends follow the +binding's direction, and whose value, world-axis interval, attestation and confidence are copied +from the statement; its evidence rows and role-word qualifiers are copied too, and +`from_statement_id` points back at the statement. A statement with a `mood` qualifier is never +materialised. The computation is a set operation and idempotent: rows whose source no longer holds +(the statement invalidated, the signature no longer bound, the property or direction changed) are +invalidated, rows that are due and missing are added, rows whose binding is unchanged keep their +id, evidence and recorded time. Typed rows that several statements produce for the same triple are +separate rows, one per statement. On the 25-document batch the 36 bound signatures give 184 +typed rows. **Argument order is enforced, participation is guided.** A declared domain or range shapes candidates and never discards a fact; argument order is the key's encoding convention, so a fact @@ -127,10 +139,10 @@ the prompt, a description is read by people and by the aligner. ## Proposed and not built -- **Alignment** (0044 cut 2), the rest: typed rows carrying `from_statement_id`, materialised - from bound signatures and recomputed per changed signature; implication rules proposed by the - aligner, approved on the workbench, executed by code with cached readings (the sign of "下降 - 1.4%" is such a reading); a signature that tells a figure from words on the value side. The prototype aligner reached 14.7% and +- **Alignment** (0044 cut 2), the rest: implication rules proposed by the aligner, approved on + the workbench, executed by code with cached readings (the sign of "下降 1.4%" is such a + reading); a signature that tells a figure from words on the value side; merging the typed rows + that several statements produce for one triple, with the timeline's refinement rules. The prototype aligner reached 14.7% and 12.1% of gold recall in two runs against 15.5% for the withdrawn bound pass, so the bar for cut 2 is parity over two clean runs [0044, #729]. - **The workbench** (0044 cut 5): the ontology page fed by suggestions from the open graph (frequent diff --git a/docs/design/prior-work.md b/docs/design/prior-work.md index 3593d2224..3594a9c8f 100644 --- a/docs/design/prior-work.md +++ b/docs/design/prior-work.md @@ -238,7 +238,9 @@ built. [Madhavan 2007]; error propagation from extraction into fusion is the survey's standing warning [Bian 2025]; the surveys ask for provenance and re-runnable stages [Weikum 2021, Hogan 2021]. Every binding carries its two votes and a `decided_at` compared with the class's `updated_at`; - only stale bindings are re-decided; every statement keeps its quote and offsets. + only stale bindings are re-decided; every statement keeps its quote and offsets; the typed + rows are recomputed per changed signature and each carries the statement it was computed from + (#752). ### Time diff --git a/migrations/0067_a_bound_statement_becomes_a_typed_fact.sql b/migrations/0067_a_bound_statement_becomes_a_typed_fact.sql new file mode 100644 index 000000000..1b7dc4f49 --- /dev/null +++ b/migrations/0067_a_bound_statement_becomes_a_typed_fact.sql @@ -0,0 +1,18 @@ +-- 绑上的签名下的开放陈述算成类型化事实(0044 决定 3 的第三片:类型化图谱是按签名绑定 +-- 从开放图谱算出来的视图)。 +-- +-- 一条类型化行从哪条开放陈述算出来,记在 `from_statement_id` 上:谓词是绑定给的属性, +-- 主宾按绑定的方向,字面值、世界轴时间(0045 解算在开放陈述上的 valid_*)、来源时间、 +-- 置信度、证据与限定都从那条陈述抄过来。绑定变了(重判、删了、方向反了)或陈述作废了, +-- 它名下的类型化行作废;绑定不变的行不动。带 mood 限定的陈述(要、预计、若)不算—— +-- 文档没有断言它成立。 +-- +-- 老的类型化路径写的行(#736 之前)没有 from_statement_id,不受这一套管。 + +ALTER TABLE facts ADD COLUMN from_statement_id UUID REFERENCES facts(id) ON DELETE CASCADE; + +-- 只有类型化行才从陈述算出来;开放陈述自己不指向别的陈述 +ALTER TABLE facts ADD CONSTRAINT facts_from_statement_is_typed + CHECK (from_statement_id IS NULL OR layer = 'typed'); + +CREATE INDEX facts_from_statement_idx ON facts (from_statement_id) WHERE from_statement_id IS NOT NULL;