From 635418516e159283418db5d87f26d9cbaed36219 Mon Sep 17 00:00:00 2001 From: WaylandYang <145302500+WaylandYang@users.noreply.github.com> Date: Wed, 23 Sep 2026 17:16:21 +0800 Subject: [PATCH 1/3] A shape of statement implies a fact of another property, read once per phrase and cached Co-Authored-By: Claude Fable 5.1 Signed-off-by: WaylandYang <145302500+WaylandYang@users.noreply.github.com> --- crates/utopia-cli/src/main.rs | 2 +- crates/utopia-extract/src/implication.rs | 416 ++++++++++++++++++ crates/utopia-extract/src/lib.rs | 1 + crates/utopia-extract/src/phrase_align.rs | 2 +- crates/utopia-server/src/api/mod.rs | 5 + crates/utopia-server/src/api/review_routes.rs | 46 ++ .../src/api/review_routes_phrase_tests.rs | 76 ++++ crates/utopia-server/src/implication.rs | 304 +++++++++++++ crates/utopia-server/src/implication_tests.rs | 207 +++++++++ crates/utopia-server/src/main.rs | 11 + crates/utopia-server/src/phrase_alignment.rs | 88 ++++ .../src/phrase_alignment_tests.rs | 15 +- crates/utopia-server/src/rdf.rs | 5 + crates/utopia-store/src/alignment_queue.rs | 68 +++ crates/utopia-store/src/export.rs | 4 +- crates/utopia-store/src/implication_rules.rs | 303 +++++++++++++ .../src/implication_rules_tests.rs | 313 +++++++++++++ crates/utopia-store/src/lib.rs | 1 + crates/utopia-store/src/materialize.rs | 180 +++++++- .../a_bound_statement_becomes_a_typed_fact.rs | 6 +- ...ology-is-a-view-over-what-documents-say.md | 4 +- docs/design/ontology.md | 19 +- .../0073_a_statement_shape_implies_a_fact.sql | 67 +++ web/src/api.ts | 23 + web/src/i18n/en.ts | 7 + web/src/i18n/zh.ts | 7 + web/src/pages/Review.tsx | 56 +++ 27 files changed, 2219 insertions(+), 17 deletions(-) create mode 100644 crates/utopia-extract/src/implication.rs create mode 100644 crates/utopia-server/src/implication.rs create mode 100644 crates/utopia-server/src/implication_tests.rs create mode 100644 crates/utopia-store/src/implication_rules.rs create mode 100644 crates/utopia-store/src/implication_rules_tests.rs create mode 100644 migrations/0073_a_statement_shape_implies_a_fact.sql diff --git a/crates/utopia-cli/src/main.rs b/crates/utopia-cli/src/main.rs index 4753c7019..9ea25e369 100644 --- a/crates/utopia-cli/src/main.rs +++ b/crates/utopia-cli/src/main.rs @@ -82,7 +82,7 @@ struct ManifestDataDir { /// not a side effect of a code change. // 是迁移文件的**个数**，不是最大的编号（守卫 `schema_version_policy_compares_against_current` // 按个数比）：编号有空缺时两者不同——0071 由一个开放 PR 占着，0072 先落，个数是 71 -const CURRENT_SCHEMA_VERSION: u32 = 71; +const CURRENT_SCHEMA_VERSION: u32 = 72; fn main() -> anyhow::Result<()> { dotenvy::dotenv().ok(); diff --git a/crates/utopia-extract/src/implication.rs b/crates/utopia-extract/src/implication.rs new file mode 100644 index 000000000..d0c487a9f --- /dev/null +++ b/crates/utopia-extract/src/implication.rs @@ -0,0 +1,416 @@ +//! 蕴含规则的两次模型调用（0044 决定 3 第五片）。 +//! +//! 一是**提规则**：给对齐器刚判过的签名（或类别词），问「这种形状的陈述除了它绑到的属性， +//! 还蕴含哪条属性的事实，宾语怎么来」。答案是 (属性, 读数或 null)；null 属性 = 什么也不蕴含。 +//! 二是**读数**：给 distinct 的字，问「按这种读法它指什么」，答一个名字或一个值或 null。 +//! 读数按字缓存，一个字一辈子只问一次；提示词里只有字和读法，没有文档。 +//! +//! 回复都是紧凑 JSON，解析同 phrase_align：坏的一条计数，不毁掉整批；没答到的 id 不出现。 + +use crate::phrase_align::{candidate_line, PropertyCandidate}; +use utopia_llm::ChatMessage; + +/// 一条待提规则的形状：签名或类别词，带例句与候选属性 +#[derive(Debug, Clone)] +pub struct RuleItem<'a> { + pub id: i64, + /// phrase | kind_word + pub trigger: &'a str, + pub phrase: &'a str, + pub subject_class: Option<&'a str>, + pub object_class: Option<&'a str>, + pub object_is_value: bool, + /// 签名已绑到的属性键（类别词没有）；提的规则不能又是它 + pub bound_to: Option<&'a str>, + pub examples: &'a [String], + pub candidates: Vec>, +} + +/// 模型对一条形状的回答：Some((属性键, 读数)) 或 None（什么也不蕴含） +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RuleChoice { + pub id: i64, + pub implies: Option<(String, Option)>, +} + +const RULE_SYSTEM: &str = "\ +You find facts a careful reader draws from a statement without the text stating them. \ +Each numbered item is one shape: a relation phrase between a subject class and an object class \ +(or a value), or a kind word the documents use for a thing, with example statements. The item \ +may already be bound to a property; that binding is not the question. The question is whether \ +this shape implies a fact of ANOTHER candidate property about the subject, and how the object \ +of that fact is obtained:\n\ +- null: the object is the statement's own object (the shape implies a second property about the same pair);\n\ +- a reading: the object is read from the words of the statement's object (for a kind word, from the kind word itself). \ +Readings available, by name:\n\ +{READINGS}\n\ +Examples of what is implied: a kind word \"British film\" implies country of origin = United Kingdom \ +(reading country_of_nationality); \"located in the Piedmont region of Virginia\" implies country = \ +United States (reading country_of_place); \"released in the summer of 1952\" implies publication year \ +(reading year_of_phrase). Answer null when nothing beyond the binding is implied, when the implication \ +would only sometimes hold, or when no candidate property fits.\n\ +Answer with one JSON object and nothing else: {\"i\": [[id, \"property_key\" | null, \"reading\" | null]]}. \ +Every item id appears exactly once."; + +fn readings_text(readings: &[(&str, &str)]) -> String { + readings + .iter() + .map(|(k, d)| format!("- {k}: {d}")) + .collect::>() + .join("\n") +} + +pub fn build_rule_messages(items: &[RuleItem<'_>], readings: &[(&str, &str)]) -> Vec { + let mut user = String::new(); + for item in items { + let shape = if item.trigger == "kind_word" { + format!("kind word \"{}\"", item.phrase) + } else { + let object = if item.object_is_value { + "value".to_string() + } else { + item.object_class.unwrap_or("?").to_string() + }; + format!( + "phrase \"{}\" · subject class: {} · object: {}", + item.phrase, + item.subject_class.unwrap_or("?"), + object + ) + }; + let mut examples = String::new(); + for ex in item.examples { + examples.push_str(&format!("\n · {ex}")); + } + if examples.is_empty() { + examples.push_str(" (none)"); + } + let bound = item + .bound_to + .map(|b| format!(" · already bound to: {b}")) + .unwrap_or_default(); + let candidates = if item.candidates.is_empty() { + " (none)".to_string() + } else { + let lines: Vec = item.candidates.iter().map(candidate_line).collect(); + format!("\n{}", lines.join("\n")) + }; + user.push_str(&format!( + "Item {}: {shape}{bound}\nExamples:{examples}\nCandidates:{candidates}\n\n", + item.id + )); + } + vec![ + ChatMessage { + role: "system".into(), + content: RULE_SYSTEM.replace("{READINGS}", &readings_text(readings)), + }, + ChatMessage { + role: "user".into(), + content: user.trim_end().to_string(), + }, + ] +} + +/// 解析提规则的回复：`(裁决, 坏项数)`。同一个 id 只收第一次；属性键不在候选里、读数不在 +/// 清单里、属性就是已绑到的那条，都算坏 +pub fn parse_rule_response( + raw: &str, + items: &[RuleItem<'_>], + readings: &[(&str, &str)], +) -> Result<(Vec, usize), String> { + let v: serde_json::Value = + serde_json::from_str(extract_json(raw)).map_err(|e| e.to_string())?; + let rows = v + .get("i") + .and_then(|x| x.as_array()) + .ok_or_else(|| "no \"i\" array".to_string())?; + let mut out = Vec::new(); + let mut malformed = 0usize; + // 坏行不占 id：只有收下的答案才算答过，后面同 id 的完好答案还能收；收过再来的才是重复 + let mut seen = std::collections::HashSet::new(); + for row in rows { + let parsed = (|| { + let arr = row.as_array()?; + let id = arr.first()?.as_i64()?; + let item = items.iter().find(|i| i.id == id)?; + if seen.contains(&id) { + return None; + } + let key = arr.get(1).and_then(|x| x.as_str()); + let reading = arr.get(2).and_then(|x| x.as_str()); + let implies = match key { + None => None, + Some(k) => { + if !item.candidates.iter().any(|c| c.key == k) || item.bound_to == Some(k) { + return None; + } + if let Some(r) = reading { + if !readings.iter().any(|(name, _)| *name == r) { + return None; + } + } + // 类别词自己没有宾语：没有读数就没有宾语，这条答案没意义 + if item.trigger == "kind_word" && reading.is_none() { + return None; + } + Some((k.to_string(), reading.map(str::to_string))) + } + }; + Some(RuleChoice { id, implies }) + })(); + match parsed { + Some(choice) => { + seen.insert(choice.id); + out.push(choice); + } + None => malformed += 1, + } + } + Ok((out, malformed)) +} + +/// 一条待读的字 +#[derive(Debug, Clone)] +pub struct ReadingItem<'a> { + pub id: i64, + pub reading: &'a str, + pub phrase: &'a str, +} + +/// 读数的答案：一个名字（库里的一样东西）、一个值，或读不出来 +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ReadingAnswer { + pub id: i64, + pub name: Option, + pub value: Option, +} + +const READING_SYSTEM: &str = "\ +You read a short phrase in a stated way and answer what it names. Each numbered item gives the \ +reading and the phrase. Readings:\n\ +{READINGS}\n\ +Answer the canonical English name of the thing (a country's common name, e.g. \"United Kingdom\", \ +\"United States\"), or for year_of_phrase the four-digit year as a string, or null when the phrase \ +does not determine an answer (an ambiguous demonym, a place you cannot place, no year in the words). \ +Do not guess. Answer with one JSON object and nothing else: {\"r\": [[id, \"answer\" | null]]}. \ +Every item id appears exactly once."; + +pub fn build_reading_messages( + items: &[ReadingItem<'_>], + readings: &[(&str, &str)], +) -> Vec { + let user = items + .iter() + .map(|i| format!("Item {}: {} · \"{}\"", i.id, i.reading, i.phrase)) + .collect::>() + .join("\n"); + vec![ + ChatMessage { + role: "system".into(), + content: READING_SYSTEM.replace("{READINGS}", &readings_text(readings)), + }, + ChatMessage { + role: "user".into(), + content: user, + }, + ] +} + +/// 解析读数的回复。年份读数的答案落成值，其余落成名字；空串、非四位数的年份算坏 +pub fn parse_reading_response( + raw: &str, + items: &[ReadingItem<'_>], +) -> Result<(Vec, usize), String> { + let v: serde_json::Value = + serde_json::from_str(extract_json(raw)).map_err(|e| e.to_string())?; + let rows = v + .get("r") + .and_then(|x| x.as_array()) + .ok_or_else(|| "no \"r\" array".to_string())?; + let mut out = Vec::new(); + let mut malformed = 0usize; + let mut seen = std::collections::HashSet::new(); + for row in rows { + let parsed = (|| { + let arr = row.as_array()?; + let id = arr.first()?.as_i64()?; + let item = items.iter().find(|i| i.id == id)?; + if seen.contains(&id) { + return None; + } + let answer = arr.get(1).and_then(|x| match x { + serde_json::Value::String(s) => Some(s.trim().to_string()), + serde_json::Value::Number(n) => Some(n.to_string()), + _ => None, + }); + match answer { + None => Some(ReadingAnswer { + id, + name: None, + value: None, + }), + Some(a) if a.is_empty() => None, + Some(a) if item.reading == "year_of_phrase" => { + if a.len() == 4 && a.chars().all(|c| c.is_ascii_digit()) { + Some(ReadingAnswer { + id, + name: None, + value: Some(a), + }) + } else { + None + } + } + Some(a) => Some(ReadingAnswer { + id, + name: Some(a), + value: None, + }), + } + })(); + match parsed { + Some(answer) => { + seen.insert(answer.id); + out.push(answer); + } + None => malformed += 1, + } + } + Ok((out, malformed)) +} + +/// 回复里可能裹着 ```json 围栏或前后的话：取第一个 { 到最后一个 } +fn extract_json(raw: &str) -> &str { + match (raw.find('{'), raw.rfind('}')) { + (Some(a), Some(b)) if b > a => &raw[a..=b], + _ => raw, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const READINGS: &[(&str, &str)] = &[("country_of_nationality", "…"), ("year_of_phrase", "…")]; + + fn cand(key: &'static str) -> PropertyCandidate<'static> { + PropertyCandidate { + key, + label: key, + description: "", + kind: "relation", + domains: vec![], + ranges: vec![], + via: vec![], + } + } + fn item(id: i64, trigger: &'static str, bound: Option<&'static str>) -> RuleItem<'static> { + RuleItem { + id, + trigger, + phrase: "british film", + subject_class: Some("film"), + object_class: None, + object_is_value: false, + bound_to: bound, + examples: &[], + candidates: vec![cand("country_of_origin"), cand("genre")], + } + } + + #[test] + fn a_rule_answer_names_a_candidate_and_a_known_reading() { + let items = vec![item(0, "kind_word", None), item(1, "phrase", Some("genre"))]; + let (choices, bad) = parse_rule_response( + "```json\n{\"i\":[[0,\"country_of_origin\",\"country_of_nationality\"],[1,null,null]]}\n```", + &items, + READINGS, + ) + .unwrap(); + assert_eq!(bad, 0); + assert_eq!( + choices[0].implies, + Some(( + "country_of_origin".into(), + Some("country_of_nationality".into()) + )) + ); + assert_eq!(choices[1].implies, None); + } + + #[test] + fn bad_rule_answers_are_counted_not_believed() { + let items = vec![item(0, "kind_word", None), item(1, "phrase", Some("genre"))]; + let (choices, bad) = parse_rule_response( + // 未知属性；类别词没有读数；已绑到的属性；未知读数；重复 id + "{\"i\":[[0,\"director\",null],[0,\"country_of_origin\",null],[1,\"genre\",null],[1,\"country_of_origin\",\"made_up\"],[1,null,null],[1,null,null]]}", + &items, + READINGS, + ) + .unwrap(); + assert_eq!(choices.len(), 1); + assert_eq!(choices[0].id, 1); + assert_eq!(bad, 5); + } + + #[test] + fn readings_land_as_names_or_four_digit_years() { + let items = vec![ + ReadingItem { + id: 0, + reading: "country_of_nationality", + phrase: "british", + }, + ReadingItem { + id: 1, + reading: "year_of_phrase", + phrase: "the summer of 1952", + }, + ReadingItem { + id: 2, + reading: "year_of_phrase", + phrase: "last year", + }, + ReadingItem { + id: 3, + reading: "country_of_nationality", + phrase: "iberian", + }, + ]; + let (answers, bad) = parse_reading_response( + "{\"r\":[[0,\"United Kingdom\"],[1,1952],[2,\"recently\"],[3,null]]}", + &items, + ) + .unwrap(); + assert_eq!(bad, 1, "a non-year for year_of_phrase is malformed"); + assert_eq!(answers[0].name.as_deref(), Some("United Kingdom")); + assert_eq!(answers[1].value.as_deref(), Some("1952")); + assert_eq!( + (answers[2].name.as_deref(), answers[2].value.as_deref()), + (None, None) + ); + } + + #[test] + fn the_prompts_carry_the_readings_and_the_shape() { + let items = vec![item(0, "phrase", Some("genre"))]; + let m = build_rule_messages(&items, READINGS); + assert!(m[0].content.contains("- country_of_nationality")); + assert!(m[1] + .content + .contains("phrase \"british film\" · subject class: film · object: ?")); + assert!(m[1].content.contains("already bound to: genre")); + let r = build_reading_messages( + &[ReadingItem { + id: 7, + reading: "year_of_phrase", + phrase: "in 1952", + }], + READINGS, + ); + assert!(r[1] + .content + .contains("Item 7: year_of_phrase · \"in 1952\"")); + } +} diff --git a/crates/utopia-extract/src/lib.rs b/crates/utopia-extract/src/lib.rs index 41a4c9637..b4ce86f93 100644 --- a/crates/utopia-extract/src/lib.rs +++ b/crates/utopia-extract/src/lib.rs @@ -11,6 +11,7 @@ use utopia_llm::ChatMessage; pub mod align; pub mod governor; +pub mod implication; pub mod open; pub mod phrase_align; pub mod time; diff --git a/crates/utopia-extract/src/phrase_align.rs b/crates/utopia-extract/src/phrase_align.rs index 57cab6e3a..fef87fdde 100644 --- a/crates/utopia-extract/src/phrase_align.rs +++ b/crates/utopia-extract/src/phrase_align.rs @@ -113,7 +113,7 @@ forward; for \"X —owns→ Y\", if subsidiary_of is the only fitting candidate, 5. Never invent a key, never answer with a label, never choose for an item a key that is not \ among its candidates. One triple per item, every item answered."; -fn candidate_line(c: &PropertyCandidate<'_>) -> String { +pub fn candidate_line(c: &PropertyCandidate<'_>) -> String { let mut line = format!("- {} · {} · {}", c.key, c.label, c.kind); if !c.domains.is_empty() { line.push_str(&format!(" · domain: {}", c.domains.join(", "))); diff --git a/crates/utopia-server/src/api/mod.rs b/crates/utopia-server/src/api/mod.rs index c7536ae97..0693d1d2a 100644 --- a/crates/utopia-server/src/api/mod.rs +++ b/crates/utopia-server/src/api/mod.rs @@ -510,6 +510,11 @@ pub fn router(state: AppState, cfg: &AppConfig) -> Router { "/kbs/{id}/review/alignment/kind-words/{kind_word}", post(review_routes::decide_alignment_kind_word), ) + // 人批或驳一条蕴含规则（0044 决定 3 第五片） + .route( + "/kbs/{id}/review/alignment/rules/{rule_id}", + post(review_routes::decide_alignment_rule), + ) // 语义层映射的表态（0011）。跟消解审核并排——都是「引擎提议、人裁决」 .route( "/kbs/{id}/review/mappings/{mapping_id}", diff --git a/crates/utopia-server/src/api/review_routes.rs b/crates/utopia-server/src/api/review_routes.rs index d3fe8ba8b..9c29f3523 100644 --- a/crates/utopia-server/src/api/review_routes.rs +++ b/crates/utopia-server/src/api/review_routes.rs @@ -1227,6 +1227,52 @@ pub struct DecideAlignmentPhraseReq { /// 人定一条短语签名绑到哪个属性（#725 对齐队列）。写成人的判定，代理此后不再改它； /// 类型化图谱立刻按新绑定重算。 +#[derive(Deserialize)] +pub struct DecideAlignmentRuleReq { + pub approve: bool, +} + +/// 人批或驳一条蕴含规则（0044 决定 3 第五片）。与短语判定同一套：决定和它的后续工作 +/// 一次提交，答 202 和 job id。批准且要读数的先排 `read_phrases`（填缓存后自己排物化）， +/// 否则直接排物化——驳回也要重算，隐含行得退掉 +pub async fn decide_alignment_rule( + State(state): State, + AuthUser(user): AuthUser, + Path((kb_id, rule_id)): Path<(Uuid, Uuid)>, + Json(req): Json, +) -> ApiResult<(axum::http::StatusCode, Json)> { + require_kb(&state, &user, kb_id, Role::Editor).await?; + let rule = utopia_store::implication_rules::get(&state.pool, kb_id, rule_id) + .await? + .ok_or(utopia_core::AppError::NotFound)?; + let votes = json!({ "person": if req.approve { "approve" } else { "reject" } }); + let job_id = utopia_store::implication_rules::decide_with_delivery( + &state.pool, + kb_id, + rule_id, + req.approve, + &votes, + ) + .await? + .ok_or(utopia_core::AppError::NotFound)?; + let _ = utopia_store::audit::record( + &state.pool, + Some(kb_id), + user.id, + "alignment.rule_decided", + "implication_rule", + Some(rule_id), + json!({ "trigger": rule.trigger, "phrase": rule.phrase, "reading": rule.reading, + "approve": req.approve, "job_id": job_id }), + ) + .await; + state.emit_review(kb_id); + Ok(( + axum::http::StatusCode::ACCEPTED, + Json(json!({ "ok": true, "job_id": job_id, "status": "accepted" })), + )) +} + pub async fn decide_alignment_phrase( State(state): State, AuthUser(user): AuthUser, diff --git a/crates/utopia-server/src/api/review_routes_phrase_tests.rs b/crates/utopia-server/src/api/review_routes_phrase_tests.rs index 4f81073cb..329218280 100644 --- a/crates/utopia-server/src/api/review_routes_phrase_tests.rs +++ b/crates/utopia-server/src/api/review_routes_phrase_tests.rs @@ -245,3 +245,79 @@ async fn a_phrase_decision_is_accepted_with_its_job() -> anyhow::Result<()> { f.cleanup().await?; run } + +#[tokio::test] +async fn a_rule_decision_is_accepted_with_its_job() -> anyhow::Result<()> { + let Some(f) = Fx::new().await? else { + return Ok(()); + }; + let run = async { + let property: Uuid = + sqlx::query_scalar("SELECT id FROM relation_types WHERE kb_id=$1 AND key='based_in'") + .bind(f.kb) + .fetch_one(&f.pool) + .await?; + let votes = json!({}); + let rule = utopia_store::implication_rules::propose( + &f.pool, + f.kb, + &utopia_store::implication_rules::Proposal { + trigger: "phrase", + phrase: "based in", + subject_type_id: None, + object_type_id: None, + object_is_value: false, + conclude_property_id: property, + reading: Some("country_of_place"), + status: "proposed", + votes: &votes, + basis: "b", + statement_count: 1, + examples: &[], + }, + ) + .await? + .expect("proposed"); + let path = format!("/api/v1/kbs/{}/review/alignment/rules/{}", f.kb, rule); + // Viewer 不能批 + let (status, _) = f + .call(&f.viewer, "POST", &path, Some(json!({ "approve": true }))) + .await?; + assert_eq!(status, StatusCode::FORBIDDEN); + let (status, body) = f + .call(&f.editor, "POST", &path, Some(json!({ "approve": true }))) + .await?; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + let job_id = body["job_id"].as_i64().expect("job id"); + let kind: String = sqlx::query_scalar("SELECT kind FROM jobs WHERE id=$1") + .bind(job_id) + .fetch_one(&f.pool) + .await?; + assert_eq!( + kind, + utopia_store::implication_rules::READ_KIND, + "a reading is needed first" + ); + let r = utopia_store::implication_rules::get(&f.pool, f.kb, rule) + .await? + .unwrap(); + assert_eq!( + (r.status.as_str(), r.decided_by.as_str()), + ("approved", "person") + ); + // 换个库的 id 答 404 + let (status, _) = f + .call( + &f.editor, + "POST", + &format!("/api/v1/kbs/{}/review/alignment/rules/{}", f.other_kb, rule), + Some(json!({ "approve": false })), + ) + .await?; + assert_eq!(status, StatusCode::NOT_FOUND); + anyhow::Ok(()) + } + .await; + f.cleanup().await?; + run +} diff --git a/crates/utopia-server/src/implication.rs b/crates/utopia-server/src/implication.rs new file mode 100644 index 000000000..7b6766106 --- /dev/null +++ b/crates/utopia-server/src/implication.rs @@ -0,0 +1,304 @@ +//! 蕴含规则在服务端的两段活（0044 决定 3 第五片）：对齐结束时向模型提规则； +//! `read_phrases` 任务把已批准规则要的读数算进缓存，然后排物化。 +//! +//! 物化本身不调模型（`utopia_store::materialize`），缓存没填上的读数那一轮就不算。 + +use std::collections::HashMap; +use utopia_core::models::RelationTypeView; +use utopia_extract::implication::{ + build_reading_messages, build_rule_messages, parse_reading_response, parse_rule_response, + ReadingItem, RuleItem, +}; +use utopia_extract::phrase_align::PropertyCandidate; +use utopia_store::implication_rules::{self, Proposal, READINGS}; +use utopia_store::phrase_bindings::{self, PhraseSignature}; +use utopia_store::type_bindings::KindWordSignature; +use uuid::Uuid; + +use crate::extraction::chat_retrying_rate_limits_at; +use crate::llm_util; +use crate::state::AppState; + +const BATCH: usize = 12; + +/// 一次提规则的输入：签名（带它绑到的属性与候选）和类别词（带候选）。 +pub struct RuleAsk<'a> { + pub phrase: Option<&'a PhraseSignature>, + pub kind_word: Option<&'a KindWordSignature>, + pub bound_to: Option<&'a str>, + pub candidates: Vec<&'a RelationTypeView>, + pub basis: &'a str, +} + +/// 向模型提规则，把答案落成提案（要人批）或代理的驳回（什么也不蕴含，记下免得再问）。 +/// 返回 (提案数, 失败批次)。候选为空的形状不问 +pub async fn propose_rules( + state: &AppState, + kb_id: Uuid, + settings: &utopia_core::models::LlmSettings, + client: &utopia_llm::LlmClient, + asks: &[RuleAsk<'_>], + class_key: &HashMap, + by_key: &HashMap<&str, &RelationTypeView>, +) -> anyhow::Result<(usize, usize)> { + let pool = &state.pool; + let keys_of = |ids: &[Uuid]| -> Vec<&str> { + ids.iter() + .filter_map(|id| class_key.get(id).copied()) + .collect() + }; + let (mut proposed, mut failed) = (0usize, 0usize); + let asks: Vec<&RuleAsk<'_>> = asks.iter().filter(|a| !a.candidates.is_empty()).collect(); + for batch in asks.chunks(BATCH) { + let items: Vec> = batch + .iter() + .enumerate() + .map(|(i, a)| { + let candidates = a + .candidates + .iter() + .map(|p| PropertyCandidate { + key: &p.key, + label: &p.label, + description: &p.description, + kind: &p.kind, + domains: keys_of(&p.domains), + ranges: keys_of(&p.ranges), + via: Vec::new(), + }) + .collect(); + match (a.phrase, a.kind_word) { + (Some(s), _) => RuleItem { + id: i as i64, + trigger: "phrase", + phrase: &s.phrase, + subject_class: s.subject_type_key.as_deref(), + object_class: s.object_type_key.as_deref(), + object_is_value: s.object_is_value, + bound_to: a.bound_to, + examples: &s.examples, + candidates, + }, + (None, Some(k)) => RuleItem { + id: i as i64, + trigger: "kind_word", + phrase: &k.kind_word, + subject_class: None, + object_class: None, + object_is_value: false, + bound_to: None, + examples: &k.examples, + candidates, + }, + (None, None) => unreachable!("an ask is a phrase or a kind word"), + } + }) + .collect(); + let messages = build_rule_messages(&items, READINGS); + let reply = + match chat_retrying_rate_limits_at(state, settings, client, &messages, Some(0.0)).await + { + Ok(r) => r, + Err(e) => { + tracing::warn!(%kb_id, error = %e, "提规则调用失败，这一批留到下次"); + failed += 1; + continue; + } + }; + let (choices, malformed) = match parse_rule_response(&reply.text, &items, READINGS) { + Ok(x) => x, + Err(e) => { + tracing::warn!(%kb_id, error = %e, "提规则回复解析失败，这一批留到下次"); + failed += 1; + continue; + } + }; + if malformed > 0 { + tracing::info!(%kb_id, malformed, "提规则的回复里有坏项"); + } + for c in choices { + let Ok(i) = usize::try_from(c.id) else { + continue; + }; + let Some(a) = batch.get(i) else { continue }; + let (trigger, phrase, subject, object, value, count, examples) = + match (a.phrase, a.kind_word) { + (Some(s), _) => ( + "phrase", + s.phrase.as_str(), + s.subject_type_id, + s.object_type_id, + s.object_is_value, + s.count, + s.examples.as_slice(), + ), + (None, Some(k)) => ( + "kind_word", + k.kind_word.as_str(), + None, + None, + false, + k.count, + k.examples.as_slice(), + ), + (None, None) => continue, + }; + match c.implies { + Some((key, reading)) => { + let Some(p) = by_key.get(key.as_str()) else { + continue; + }; + let votes = + serde_json::json!({ "agent": { "property": key, "reading": reading } }); + if implication_rules::propose( + pool, + kb_id, + &Proposal { + trigger, + phrase, + subject_type_id: subject, + object_type_id: object, + object_is_value: value, + conclude_property_id: p.id, + reading: reading.as_deref(), + status: "proposed", + votes: &votes, + basis: a.basis, + statement_count: count, + examples, + }, + ) + .await? + .is_some() + { + proposed += 1; + } + } + None => { + // 「什么也不蕴含」也要落下来，不然每轮都问。落成代理驳回的一行： + // 属性列非空不可，这里记的是形状本身，用签名绑到的属性或第一个候选占位 + let placeholder = a + .bound_to + .and_then(|k| by_key.get(k)) + .or_else(|| a.candidates.first()) + .map(|p| p.id); + let Some(property) = placeholder else { + continue; + }; + let votes = serde_json::json!({ "agent": null, "reason": "nothing_implied" }); + implication_rules::propose( + pool, + kb_id, + &Proposal { + trigger, + phrase, + subject_type_id: subject, + object_type_id: object, + object_is_value: value, + conclude_property_id: property, + reading: None, + status: "rejected", + votes: &votes, + basis: a.basis, + statement_count: count, + examples, + }, + ) + .await?; + } + } + } + } + Ok((proposed, failed)) +} + +/// `read_phrases` 任务：已批准规则要的、缓存里还没有的读数，问一遍模型，落进缓存； +/// 名字解析成库里的实体（没有就建一个有名字的）；读不出来的也记，别再问。 +/// 填完排一次物化——隐含行在那里算 +pub async fn read_phrases(state: &AppState, kb_id: Uuid) -> anyhow::Result<()> { + let pool = &state.pool; + let pending = implication_rules::pending_readings(pool, kb_id).await?; + if pending.is_empty() { + tracing::info!(%kb_id, "没有待读的字"); + } else { + let kb = utopia_store::kbs::get(pool, kb_id).await?; + let settings = utopia_store::settings::get(pool, kb.workspace_id) + .await? + .ok_or_else(|| anyhow::anyhow!("Chat model not configured; cannot read phrases"))?; + let client = llm_util::chat_client(&settings) + .ok_or_else(|| anyhow::anyhow!("Chat model not configured; cannot read phrases"))?; + let (mut answered, mut failed) = (0usize, 0usize); + for batch in pending.chunks(BATCH * 2) { + let items: Vec> = batch + .iter() + .enumerate() + .map(|(i, p)| ReadingItem { + id: i as i64, + reading: &p.reading, + phrase: &p.phrase, + }) + .collect(); + let messages = build_reading_messages(&items, READINGS); + let reply = + match chat_retrying_rate_limits_at(state, &settings, &client, &messages, Some(0.0)) + .await + { + Ok(r) => r, + Err(e) => { + tracing::warn!(%kb_id, error = %e, "读数调用失败，这一批留到下次"); + failed += 1; + continue; + } + }; + let (answers, malformed) = match parse_reading_response(&reply.text, &items) { + Ok(x) => x, + Err(e) => { + tracing::warn!(%kb_id, error = %e, "读数回复解析失败，这一批留到下次"); + failed += 1; + continue; + } + }; + if malformed > 0 { + tracing::info!(%kb_id, malformed, "读数的回复里有坏项"); + } + for a in answers { + let Ok(i) = usize::try_from(a.id) else { + continue; + }; + let Some(p) = batch.get(i) else { continue }; + let entity = match &a.name { + Some(name) => { + Some(implication_rules::resolve_or_create_named(pool, kb_id, name).await?) + } + None => None, + }; + let value = a.value.as_ref().map(|v| serde_json::json!({ "value": v })); + implication_rules::record_reading( + pool, + kb_id, + &p.reading, + &p.phrase, + entity, + value.as_ref(), + ) + .await?; + answered += 1; + } + } + tracing::info!(%kb_id, pending = pending.len(), answered, failed, "读数填缓存完成"); + if failed > 0 { + anyhow::bail!("{failed} reading batches failed; the job retries"); + } + } + utopia_store::jobs::enqueue_unless_queued( + pool, + phrase_bindings::MATERIALIZE_KIND, + serde_json::json!({ "kb_id": kb_id }), + ) + .await?; + Ok(()) +} + +#[cfg(test)] +#[path = "implication_tests.rs"] +mod tests; diff --git a/crates/utopia-server/src/implication_tests.rs b/crates/utopia-server/src/implication_tests.rs new file mode 100644 index 000000000..338fb428e --- /dev/null +++ b/crates/utopia-server/src/implication_tests.rs @@ -0,0 +1,207 @@ +//! 提规则与读数走脚本化的模型端点：提案落到队列，读数落进缓存并解析成库里的实体， +//! 然后排物化。没有 `UTOPIA_DATABASE_URL` 时跳过。 +use super::*; +use axum::{extract::State, response::IntoResponse, routing::post, Json, Router}; +use serde_json::{json, Value}; +use std::sync::{Arc, Mutex}; +use utopia_store::implication_rules; + +#[derive(Clone)] +struct Model { + replies: Arc>>, + requests: Arc>>, +} +async fn reply(State(m): State, Json(body): Json) -> impl IntoResponse { + m.requests.lock().unwrap().push(body); + let text = m.replies.lock().unwrap().remove(0).to_string(); + let frame = json!({"choices":[{"delta":{"content":text}}]}); + ( + [("content-type", "text/event-stream")], + format!("data: {frame}\n\ndata: [DONE]\n\n"), + ) +} + +struct Fx { + pool: sqlx::PgPool, + state: AppState, + org: Uuid, + kb: Uuid, + film: Uuid, + country_of_origin: Uuid, + model: Model, + server: tokio::task::JoinHandle<()>, + _dir: tempfile::TempDir, +} +impl Fx { + async fn new(replies: Vec) -> anyhow::Result> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(None); + }; + let pool = sqlx::PgPool::connect(&url).await?; + utopia_store::db::migrate(&pool).await?; + let (org, ws, kb, film, coo, loud) = ( + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + ); + sqlx::raw_sql(&format!( + "INSERT INTO organizations(id,name) VALUES ('{org}','implication-server'); + INSERT INTO workspaces(id,org_id,name) VALUES ('{ws}','{org}','implication-server'); + INSERT INTO knowledge_bases(id,workspace_id,name) VALUES ('{kb}','{ws}','implication-server'); + INSERT INTO entity_types(id,kb_id,key,label,color,shape) VALUES ('{film}','{kb}','film','Film','#000','circle'); + INSERT INTO relation_types(id,kb_id,key,label,kind,temporal,description) VALUES + ('{coo}','{kb}','country_of_origin','country of origin','relation','state','the country a work comes from'); + INSERT INTO entities(id,kb_id,canonical_name,type_id,specific_type) VALUES ('{loud}','{kb}','Loud Tour','{film}','British film');" + )) + .execute(&pool) + .await?; + let model = Model { + replies: Arc::new(Mutex::new(replies)), + requests: Arc::new(Mutex::new(Vec::new())), + }; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let endpoint = format!("http://{}", listener.local_addr()?); + let router = Router::new() + .route("/chat/completions", post(reply)) + .with_state(model.clone()); + let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + utopia_store::settings::upsert( + &pool, + ws, + Some(&endpoint), + None, + Some("scripted"), + None, + None, + None, + None, + ) + .await?; + let dir = tempfile::tempdir()?; + let cfg = utopia_core::config::AppConfig { + data_dir: dir.path().to_string_lossy().into_owned(), + ..Default::default() + }; + let search = Arc::new(utopia_search::SearchIndex::open( + &dir.path().join("search"), + )?); + let state = AppState::new(pool.clone(), &cfg, search, "test-only".into()); + Ok(Some(Self { + pool, + state, + org, + kb, + film, + country_of_origin: coo, + model, + server, + _dir: dir, + })) + } + async fn cleanup(self) -> anyhow::Result<()> { + self.server.abort(); + sqlx::query("DELETE FROM jobs WHERE payload->>'kb_id'=$1") + .bind(self.kb.to_string()) + .execute(&self.pool) + .await?; + sqlx::query("DELETE FROM organizations WHERE id=$1") + .bind(self.org) + .execute(&self.pool) + .await?; + Ok(()) + } +} + +#[tokio::test] +async fn a_kind_word_is_offered_and_the_model_proposes_a_rule() -> anyhow::Result<()> { + let Some(f) = Fx::new(vec![ + json!({"i":[[0,"country_of_origin","country_of_nationality"]]}), + ]) + .await? + else { + return Ok(()); + }; + let run = async { + let kb = utopia_store::kbs::get(&f.pool, f.kb).await?; + let settings = utopia_store::settings::get(&f.pool, kb.workspace_id) + .await? + .unwrap(); + let client = llm_util::chat_client(&settings).unwrap(); + let props = utopia_store::ontology::relation_type_views(&f.pool, f.kb).await?; + let words = utopia_store::type_bindings::signatures(&f.pool, f.kb).await?; + assert_eq!(words.len(), 1); + let class_key: HashMap = HashMap::from([(f.film, "film")]); + let by_key: HashMap<&str, &RelationTypeView> = + props.iter().map(|p| (p.key.as_str(), p)).collect(); + let asks = vec![RuleAsk { + phrase: None, + kind_word: Some(&words[0]), + bound_to: None, + candidates: props.iter().collect(), + basis: "k1", + }]; + let (proposed, failed) = propose_rules( + &f.state, f.kb, &settings, &client, &asks, &class_key, &by_key, + ) + .await?; + assert_eq!((proposed, failed), (1, 0)); + let prompt = f.model.requests.lock().unwrap()[0]["messages"][1]["content"] + .as_str() + .unwrap() + .to_string(); + assert!(prompt.contains("kind word \"british film\""), "{prompt}"); + let rules = implication_rules::list(&f.pool, f.kb, Some("proposed")).await?; + assert_eq!(rules.len(), 1); + assert_eq!( + (rules[0].trigger.as_str(), rules[0].reading.as_deref()), + ("kind_word", Some("country_of_nationality")) + ); + assert_eq!(rules[0].conclude_property_id, f.country_of_origin); + // 队列里看得见 + let items = utopia_store::alignment_queue::list(&f.pool, f.kb, 10, 0).await?; + assert!(items + .iter() + .any(|i| matches!(i, utopia_store::alignment_queue::AlignmentItem::Rule { .. }))); + anyhow::Ok(()) + } + .await; + f.cleanup().await?; + run +} + +#[tokio::test] +async fn read_phrases_fills_the_cache_names_a_thing_and_queues_the_materialization( +) -> anyhow::Result<()> { + let Some(f) = Fx::new(vec![json!({"r":[[0,"United Kingdom"]]})]).await? else { + return Ok(()); + }; + let run = async { + let votes = json!({}); + let id = implication_rules::propose(&f.pool, f.kb, &implication_rules::Proposal { + trigger: "kind_word", phrase: "british film", subject_type_id: None, object_type_id: None, object_is_value: false, + conclude_property_id: f.country_of_origin, reading: Some("country_of_nationality"), status: "proposed", + votes: &votes, basis: "k1", statement_count: 1, examples: &[], + }).await?.unwrap(); + implication_rules::decide_with_delivery(&f.pool, f.kb, id, true, &votes).await?; + sqlx::query("DELETE FROM jobs WHERE payload->>'kb_id'=$1").bind(f.kb.to_string()).execute(&f.pool).await?; + read_phrases(&f.state, f.kb).await?; + let (entity, value): (Option, Option) = sqlx::query_as( + "SELECT entity_id, value FROM phrase_readings WHERE kb_id=$1 AND reading='country_of_nationality' AND phrase='british film'", + ).bind(f.kb).fetch_one(&f.pool).await?; + assert!(value.is_none()); + let name: String = sqlx::query_scalar("SELECT canonical_name FROM entities WHERE id=$1").bind(entity.unwrap()).fetch_one(&f.pool).await?; + assert_eq!(name, "United Kingdom"); + let kinds: Vec<(String,)> = sqlx::query_as("SELECT kind FROM jobs WHERE payload->>'kb_id'=$1 AND status='queued'").bind(f.kb.to_string()).fetch_all(&f.pool).await?; + assert_eq!(kinds, vec![(utopia_store::phrase_bindings::MATERIALIZE_KIND.to_string(),)]); + // 物化：一条隐含行 + let o = utopia_store::materialize::materialize(&f.pool, f.kb).await?; + assert_eq!(o.implied, 1); + anyhow::Ok(()) + } + .await; + f.cleanup().await?; + run +} diff --git a/crates/utopia-server/src/main.rs b/crates/utopia-server/src/main.rs index f02c69ce2..7e5a1cace 100644 --- a/crates/utopia-server/src/main.rs +++ b/crates/utopia-server/src/main.rs @@ -12,6 +12,7 @@ mod extraction_open; mod github_issues; mod governance; mod http_fetch; +mod implication; mod ingest_sources; mod jira_issues; mod live; @@ -537,6 +538,16 @@ async fn dispatch(st: &state::AppState, job: &utopia_store::jobs::Job) -> anyhow )), } } + // 已批准的蕴含规则要的读数（0044 决定 3 第五片）：问模型、填缓存、排物化 + utopia_store::implication_rules::READ_KIND => { + let kb_id: Uuid = job + .payload + .get("kb_id") + .and_then(|v| v.as_str()) + .and_then(|s| s.parse().ok()) + .ok_or_else(|| anyhow::anyhow!("payload 缺少 kb_id"))?; + implication::read_phrases(st, kb_id).await + } "align_types" => { let kb_id: Uuid = job .payload diff --git a/crates/utopia-server/src/phrase_alignment.rs b/crates/utopia-server/src/phrase_alignment.rs index e5d30186e..a8bdb39bf 100644 --- a/crates/utopia-server/src/phrase_alignment.rs +++ b/crates/utopia-server/src/phrase_alignment.rs @@ -470,6 +470,94 @@ async fn align_phrases_locked( } } tracing::info!(%kb_id, bound, none, undecided, skipped, failed, "短语对齐完成"); + // 提规则（0044 决定 3 第五片）：本轮刚判过的签名，和带类别词的东西，问模型「这种形状 + // 还蕴含什么」。只问本轮判过的：指纹没变的形状上一轮已经问过，答案（提案或代理驳回） + // 还在 implication_rules 里；指纹变了它就在 todo 里，自然再问 + { + let decided_now: HashMap<_, _> = phrase_bindings::bindings(pool, kb_id) + .await? + .into_iter() + .map(|b| (b.key(), b)) + .collect(); + let kind_words = utopia_store::type_bindings::signatures(pool, kb_id).await?; + let existing_rules = utopia_store::implication_rules::list(pool, kb_id, None).await?; + let asked_kind: HashSet<&str> = existing_rules + .iter() + .filter(|r| r.trigger == "kind_word") + .map(|r| r.phrase.as_str()) + .collect(); + let mut asks: Vec> = Vec::new(); + for s in &todo { + let Some(b) = decided_now.get(&s.key()) else { + continue; + }; + if b.status == "undecided" { + continue; + } + let (fitting, basis) = &considered[&s.key()]; + let bound_to = b + .relation_type_id + .and_then(|id| props.iter().find(|p| p.id == id)) + .map(|p| p.key.as_str()); + asks.push(crate::implication::RuleAsk { + phrase: Some(s), + kind_word: None, + bound_to, + candidates: fitting + .iter() + .copied() + .filter(|p| Some(p.key.as_str()) != bound_to) + .collect(), + basis, + }); + } + // 类别词：每个词问一次；候选是主语能落在它绑到的类（或没声明）的关系属性 + let kind_basis: Vec = kind_words + .iter() + .map(|k| { + phrase_bindings::basis_of( + &[], + &[], + false, + &versions + .iter() + .map(|(id, at)| (*id, *at)) + .collect::>(), + ) + ":" + + &k.kind_word + }) + .collect(); + for (k, basis) in kind_words.iter().zip(kind_basis.iter()) { + if asked_kind.contains(k.kind_word.as_str()) { + continue; + } + asks.push(crate::implication::RuleAsk { + phrase: None, + kind_word: Some(k), + bound_to: None, + candidates: props + .iter() + .filter(|p| p.kind == "relation" || p.kind == "attribute") + .collect(), + basis, + }); + } + if !asks.is_empty() { + match crate::implication::propose_rules( + state, kb_id, settings, client, &asks, &class_key, &by_key, + ) + .await + { + Ok((proposed, rule_failed)) => { + tracing::info!(%kb_id, asked = asks.len(), proposed, failed = rule_failed, "提规则完成"); + if proposed > 0 { + state.emit_review(kb_id); + } + } + Err(e) => tracing::warn!(%kb_id, error = %e, "提规则失败，下一轮再提"), + } + } + } // 绑定定了，视图跟着算：绑上的签名下的陈述成类型化行，绑定变了的行作废（0067） let typed = utopia_store::materialize::materialize(pool, kb_id).await?; tracing::info!(%kb_id, added = typed.added, merged = typed.merged, retired = typed.retired, "类型化事实按绑定算完"); diff --git a/crates/utopia-server/src/phrase_alignment_tests.rs b/crates/utopia-server/src/phrase_alignment_tests.rs index eee0b3bd5..1edd42b97 100644 --- a/crates/utopia-server/src/phrase_alignment_tests.rs +++ b/crates/utopia-server/src/phrase_alignment_tests.rs @@ -208,14 +208,19 @@ impl Fx { fn vote(key: Option<&str>, dir: Option<&str>) -> Value { json!({"b":[[0, key, dir]]}) } +/// 两票之后对齐还会问一次「这种形状还蕴含什么」（0044 决定 3 第五片）：脚本里答「没有」 +fn nothing_implied() -> Value { + json!({"i":[[0,null,null]]}) +} fn bound() -> Vec { vec![ vote(Some("based_in"), Some("forward")), vote(Some("based_in"), Some("forward")), + nothing_implied(), ] } fn none() -> Vec { - vec![vote(None, None), vote(None, None)] + vec![vote(None, None), vote(None, None), nothing_implied()] } #[tokio::test] @@ -227,7 +232,11 @@ async fn a_property_declared_on_an_ancestor_is_offered_with_its_basis_and_bound( let run = async { f.script(bound()); f.run().await?; - assert_eq!(f.requests().len(), 2, "two votes"); + assert_eq!( + f.requests().len(), + 2, + "two votes; bound to the only property, so no rule question" + ); let prompt = f.prompt_of(0); assert!(prompt.contains("based_in"), "{prompt}"); assert!( @@ -427,7 +436,7 @@ async fn an_edit_during_the_model_request_leaves_the_decision_stale() -> anyhow: f.clear_jobs().await?; f.script(bound()); f.run().await?; - assert_eq!(f.requests().len(), 4, "asked again because the basis differs, not the clock"); + assert_eq!(f.requests().len(), 5, "two votes, one rule question (nothing implied), then two votes again: the basis differs, not the clock"); assert_eq!(f.binding().await?.status, "bound"); anyhow::Ok(()) } diff --git a/crates/utopia-server/src/rdf.rs b/crates/utopia-server/src/rdf.rs index bcb2f9bc4..8daaf3535 100644 --- a/crates/utopia-server/src/rdf.rs +++ b/crates/utopia-server/src/rdf.rs @@ -499,6 +499,10 @@ pub fn emit_fact( sink.l(&stmt, &prov("invalidatedAtTime"), &dt(t))?; } sink.l(&stmt, &utopia("confidence"), &confidence(f.confidence))?; + // 规则算出来的（0044 决定 3 第五片）：不是文档直接陈述的，审计的人要看得见这一层 + if f.implied { + sink.l(&stmt, &utopia("implied"), &flag(true))?; + } if let Some(old) = f.supersedes { let old = names.fact(old); sink.r(&stmt, &utopia("supersedes"), &old)?; @@ -744,6 +748,7 @@ mod tests { recorded_at: at("2026-01-01T00:00:00Z"), invalidated_at: None, confidence: 0.9, + implied: false, supersedes: None, documents: vec![], quotes: vec![], diff --git a/crates/utopia-store/src/alignment_queue.rs b/crates/utopia-store/src/alignment_queue.rs index 2bbbad1ae..2241623c6 100644 --- a/crates/utopia-store/src/alignment_queue.rs +++ b/crates/utopia-store/src/alignment_queue.rs @@ -39,6 +39,39 @@ pub enum AlignmentItem { votes: Option, decided_at: DateTime, }, + /// 对齐器提的一条蕴含规则（0044 决定 3 第五片）：这种形状蕴含哪条属性、宾语怎么读 + Rule { + id: Uuid, + trigger: String, + phrase: String, + subject_class: Option, + object_class: Option, + object_is_value: bool, + property: String, + property_label: String, + reading: Option, + statement_count: i32, + examples: Vec, + votes: Option, + decided_at: DateTime, + }, +} + +#[derive(sqlx::FromRow)] +struct RuleRow { + id: Uuid, + trigger: String, + phrase: String, + subject_class: Option, + object_class: Option, + object_is_value: bool, + property: String, + property_label: String, + reading: Option, + statement_count: i32, + examples: Vec, + votes: Option, + decided_at: DateTime, } #[derive(sqlx::FromRow)] @@ -80,6 +113,19 @@ pub async fn list( .bind(kb_id) .fetch_all(pool) .await?; + let rules: Vec = sqlx::query_as( + "SELECT r.id, r.trigger, r.phrase, st.key AS subject_class, ot.key AS object_class, + r.object_is_value, p.key AS property, p.label AS property_label, r.reading, + r.statement_count, r.examples, r.votes, r.decided_at + FROM implication_rules r + JOIN relation_types p ON p.id = r.conclude_property_id + LEFT JOIN entity_types st ON st.id = r.subject_type_id + LEFT JOIN entity_types ot ON ot.id = r.object_type_id + WHERE r.kb_id = $1 AND r.status = 'proposed'", + ) + .bind(kb_id) + .fetch_all(pool) + .await?; let words: Vec = sqlx::query_as( "SELECT kind_word, words, votes, decided_at FROM type_bindings WHERE kb_id = $1 AND status = 'undecided'", @@ -127,6 +173,26 @@ pub async fn list( }, )); } + for r in rules { + items.push(( + r.decided_at, + AlignmentItem::Rule { + id: r.id, + trigger: r.trigger, + phrase: r.phrase, + subject_class: r.subject_class, + object_class: r.object_class, + object_is_value: r.object_is_value, + property: r.property, + property_label: r.property_label, + reading: r.reading, + statement_count: r.statement_count, + examples: r.examples, + votes: r.votes, + decided_at: r.decided_at, + }, + )); + } items.sort_by_key(|(at, _)| *at); Ok(items .into_iter() @@ -143,6 +209,8 @@ pub async fn waiting(pool: &PgPool, kb_id: Uuid) -> AppResult<(i64, Option, pub invalidated_at: Option>, pub confidence: f32, + /// 规则算出来的隐含行（0044 决定 3 第五片）：读的人要能分辨它不是陈述直接说的 + pub implied: bool, pub supersedes: Option, pub documents: Vec, pub quotes: Vec, @@ -564,7 +566,7 @@ pub async fn facts_page( f.object_id, f.object_value, f.valid_from, f.valid_from_precision, f.valid_to, f.valid_to_precision, {holds_from} AS holds_from, {holds_to} AS holds_to, - f.recorded_at, f.invalidated_at, f.confidence, f.supersedes, + f.recorded_at, f.invalidated_at, f.confidence, f.implied, f.supersedes, COALESCE(ARRAY(SELECT DISTINCT e.document_id FROM fact_evidence e WHERE e.fact_id = f.id AND e.document_id IS NOT NULL), '{{}}') AS documents, diff --git a/crates/utopia-store/src/implication_rules.rs b/crates/utopia-store/src/implication_rules.rs new file mode 100644 index 000000000..ece013fbf --- /dev/null +++ b/crates/utopia-store/src/implication_rules.rs @@ -0,0 +1,303 @@ +//! 蕴含规则（0044 决定 3 的第五片）：一种形状的陈述、或带某个类别词的东西，蕴含另一条 +//! 属性的事实。宾语要么就是陈述的宾语，要么由一个读数从宾语的字里读出来。 +//! +//! 与绑定同一套生命周期：对齐器提（`propose`，代理），工作台批（`decide_with_delivery`，人）， +//! 人的判定不被代理盖；提案带指纹（0053），输入变了对齐器会再提。执行在 `materialize` +//! 里，**没有模型调用**：读数由 `read_phrases` 任务先算进 `phrase_readings`，物化只查缓存， +//! 缓存里没有的这一轮就不算，等缓存填上再来。 + +use chrono::{DateTime, Utc}; +use serde::Serialize; +use sqlx::PgPool; +use utopia_core::{AppError, AppResult}; +use uuid::Uuid; + +/// 读数的种类。字符串进库、进提示词，所以是常量而不是枚举；执行它们的是模型， +/// 这里只定义「问什么」。加一种就是加一行——和描述一起给模型看 +pub const READINGS: &[(&str, &str)] = &[ + ( + "country_of_nationality", + "the country a nationality or demonym names (British → United Kingdom, 法国 → France)", + ), + ( + "country_of_place", + "the country a place belongs to (Piedmont region of Virginia → United States, Lyon → France)", + ), + ( + "year_of_phrase", + "the year a phrase gives, as a four-digit number (\"the summer of 1952\" → 1952)", + ), +]; + +pub fn reading_is_known(reading: &str) -> bool { + READINGS.iter().any(|(k, _)| *k == reading) +} + +/// 把一个读数的输入归一：与短语、类别词同一条规矩（空白折一个、小写、去两端） +pub fn normalize(s: &str) -> String { + crate::phrase_bindings::normalize(s) +} + +#[derive(Debug, Clone, Serialize, sqlx::FromRow)] +pub struct Rule { + pub id: Uuid, + pub trigger: String, + pub phrase: String, + pub subject_type_id: Option, + pub object_type_id: Option, + pub object_is_value: bool, + pub conclude_property_id: Uuid, + pub reading: Option, + pub status: String, + pub votes: Option, + pub decided_by: String, + pub basis: Option, + pub statement_count: i32, + pub examples: Vec, + pub decided_at: DateTime, +} + +/// 对齐器提的一条规则。 +pub struct Proposal<'a> { + /// phrase | kind_word + pub trigger: &'a str, + pub phrase: &'a str, + pub subject_type_id: Option, + pub object_type_id: Option, + pub object_is_value: bool, + pub conclude_property_id: Uuid, + pub reading: Option<&'a str>, + /// proposed（要人批）| rejected（模型说这个形状不蕴含什么——记下来免得每轮再问） + pub status: &'a str, + pub votes: &'a serde_json::Value, + pub basis: &'a str, + pub statement_count: i64, + pub examples: &'a [String], +} + +/// 记下对齐器的提案；同一条规则已经有人判过的原样留着（返回 None）。代理自己的旧行 +/// 被新提案覆盖——指纹变了对齐器才会再提，覆盖的是过期的看法 +pub async fn propose(pool: &PgPool, kb_id: Uuid, p: &Proposal<'_>) -> AppResult> { + if !matches!(p.trigger, "phrase" | "kind_word") { + return Err(AppError::Validation(format!( + "unknown trigger {:?}", + p.trigger + ))); + } + if !matches!(p.status, "proposed" | "rejected") { + return Err(AppError::Validation(format!( + "a proposal is proposed or rejected, not {:?}", + p.status + ))); + } + if let Some(r) = p.reading { + if !reading_is_known(r) { + return Err(AppError::Validation(format!("unknown reading {r:?}"))); + } + } + let phrase = normalize(p.phrase); + if phrase.is_empty() { + return Err(AppError::Validation( + "an empty phrase implies nothing".into(), + )); + } + let id = Uuid::now_v7(); + let row: Option<(Uuid,)> = sqlx::query_as( + "INSERT INTO implication_rules + (id, kb_id, trigger, phrase, subject_type_id, object_type_id, object_is_value, + conclude_property_id, reading, status, votes, decided_by, basis, statement_count, examples) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, 'agent', $12, $13, $14) + ON CONFLICT (kb_id, trigger, phrase, subject_type_id, object_type_id, object_is_value, + conclude_property_id, reading) DO UPDATE + SET status = EXCLUDED.status, votes = EXCLUDED.votes, basis = EXCLUDED.basis, + statement_count = EXCLUDED.statement_count, examples = EXCLUDED.examples, + decided_at = now() + WHERE implication_rules.decided_by = 'agent' + RETURNING id", + ) + .bind(id) + .bind(kb_id) + .bind(p.trigger) + .bind(&phrase) + .bind(p.subject_type_id) + .bind(if p.object_is_value { None } else { p.object_type_id }) + .bind(p.object_is_value) + .bind(p.conclude_property_id) + .bind(p.reading) + .bind(p.status) + .bind(p.votes) + .bind(p.basis) + .bind(i32::try_from(p.statement_count).unwrap_or(i32::MAX)) + .bind(p.examples) + .fetch_optional(pool) + .await?; + Ok(row.map(|(id,)| id)) +} + +pub async fn list(pool: &PgPool, kb_id: Uuid, status: Option<&str>) -> AppResult> { + Ok(sqlx::query_as( + "SELECT id, trigger, phrase, subject_type_id, object_type_id, object_is_value, + conclude_property_id, reading, status, votes, decided_by, basis, + statement_count, examples, decided_at + FROM implication_rules + WHERE kb_id = $1 AND ($2::text IS NULL OR status = $2) + ORDER BY decided_at, phrase", + ) + .bind(kb_id) + .bind(status) + .fetch_all(pool) + .await?) +} + +pub async fn get(pool: &PgPool, kb_id: Uuid, id: Uuid) -> AppResult> { + Ok(sqlx::query_as( + "SELECT id, trigger, phrase, subject_type_id, object_type_id, object_is_value, + conclude_property_id, reading, status, votes, decided_by, basis, + statement_count, examples, decided_at + FROM implication_rules WHERE kb_id = $1 AND id = $2", + ) + .bind(kb_id) + .bind(id) + .fetch_optional(pool) + .await?) +} + +/// 人批或驳一条规则，与它的后续工作**同一事务**提交（0051 的同一条规矩）。 +/// 批准且要读数的：先排 `read_phrases` 把缓存填上（它跑完自己会排物化）； +/// 不要读数的、或驳回的：直接排物化——驳回也要重算，隐含行得退掉 +pub async fn decide_with_delivery( + pool: &PgPool, + kb_id: Uuid, + id: Uuid, + approve: bool, + votes: &serde_json::Value, +) -> AppResult> { + let mut tx = pool.begin().await?; + let row: Option<(Option,)> = sqlx::query_as( + "UPDATE implication_rules + SET status = $3, votes = $4, decided_by = 'person', decided_at = now() + WHERE kb_id = $1 AND id = $2 + RETURNING reading", + ) + .bind(kb_id) + .bind(id) + .bind(if approve { "approved" } else { "rejected" }) + .bind(votes) + .fetch_optional(&mut *tx) + .await?; + let Some((reading,)) = row else { + tx.rollback().await?; + return Ok(None); + }; + let kind = if approve && reading.is_some() { + READ_KIND + } else { + crate::phrase_bindings::MATERIALIZE_KIND + }; + let job = crate::jobs::enqueue_with_max_attempts_tx( + &mut tx, + kind, + serde_json::json!({ "kb_id": kb_id }), + 3, + ) + .await?; + tx.commit().await?; + Ok(Some(job)) +} + +/// 填读数缓存的任务的种类。 +pub const READ_KIND: &str = "read_phrases"; + +/// 一条待读的字：哪种读数、读什么。 +#[derive(Debug, Clone, PartialEq, Eq, sqlx::FromRow)] +pub struct PendingReading { + pub reading: String, + pub phrase: String, +} + +/// 已批准的规则里，还没有缓存的 (读数, 字)。短语规则读的是陈述的宾语（实体名或字面值）， +/// 类别词规则读的是类别词自己。答过「读不出来」的也算缓存过，不再列 +pub async fn pending_readings(pool: &PgPool, kb_id: Uuid) -> AppResult> { + let sql = format!( + "SELECT DISTINCT r.reading, {phrase} AS phrase + FROM implication_rules r + JOIN facts s ON s.kb_id = r.kb_id AND s.layer = 'open' AND s.invalidated_at IS NULL + JOIN entities se ON se.id = s.subject_id + LEFT JOIN entities oe ON oe.id = s.object_id + WHERE r.kb_id = $1 AND r.status = 'approved' AND r.trigger = 'phrase' AND r.reading IS NOT NULL + AND {rule_match} + AND NOT EXISTS (SELECT 1 FROM phrase_readings pr + WHERE pr.kb_id = r.kb_id AND pr.reading = r.reading AND pr.phrase = {phrase}) + UNION + SELECT DISTINCT r.reading, r.phrase + FROM implication_rules r + WHERE r.kb_id = $1 AND r.status = 'approved' AND r.trigger = 'kind_word' AND r.reading IS NOT NULL + AND NOT EXISTS (SELECT 1 FROM phrase_readings pr + WHERE pr.kb_id = r.kb_id AND pr.reading = r.reading AND pr.phrase = r.phrase)", + phrase = object_text_sql(), + rule_match = RULE_MATCH, + ); + Ok(sqlx::query_as(&sql).bind(kb_id).fetch_all(pool).await?) +} + +/// 陈述宾语的字，归一：实体名，或字面值的 value。给读数用 +pub(crate) fn object_text_sql() -> &'static str { + "lower(btrim(regexp_replace(coalesce(oe.canonical_name, s.object_value ->> 'value', s.object_value #>> '{}', ''), '\\s+', ' ', 'g')))" +} + +/// 短语规则与陈述的签名匹配（同 materialize 里绑定的 MATCH，把 b 换成 r） +pub(crate) const RULE_MATCH: &str = + "r.phrase = lower(btrim(regexp_replace(s.phrase, '\\s+', ' ', 'g'))) + AND r.subject_type_id IS NOT DISTINCT FROM se.type_id + AND r.object_is_value = (s.object_id IS NULL) + AND (s.object_id IS NULL OR r.object_type_id IS NOT DISTINCT FROM oe.type_id)"; + +/// 记一条读数的答案：库里的一样东西、一个字面值，或两者都空（读不出来，也记，别再问）。 +pub async fn record_reading( + pool: &PgPool, + kb_id: Uuid, + reading: &str, + phrase: &str, + entity_id: Option, + value: Option<&serde_json::Value>, +) -> AppResult<()> { + sqlx::query( + "INSERT INTO phrase_readings (kb_id, reading, phrase, entity_id, value) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (kb_id, reading, phrase) DO UPDATE + SET entity_id = EXCLUDED.entity_id, value = EXCLUDED.value, answered_at = now()", + ) + .bind(kb_id) + .bind(reading) + .bind(normalize(phrase)) + .bind(entity_id) + .bind(value) + .execute(pool) + .await?; + Ok(()) +} + +/// 读数读出了一个名字：库里有这样东西就是它，没有就建一个有名字的。 +/// 名字事实照 0041 记（`names::record`）——这是召回的桥，读数读出的国家下次就能被认出来 +pub async fn resolve_or_create_named(pool: &PgPool, kb_id: Uuid, name: &str) -> AppResult { + let name = name.trim(); + if name.is_empty() { + return Err(AppError::Validation("an empty name names nothing".into())); + } + if let Some(id) = crate::resolution::existing_by_name(pool, kb_id, name).await? { + return Ok(id); + } + let id = Uuid::now_v7(); + sqlx::query("INSERT INTO entities (id, kb_id, canonical_name) VALUES ($1, $2, $3)") + .bind(id) + .bind(kb_id) + .bind(name) + .execute(pool) + .await?; + crate::names::record(pool, kb_id, id, name, None, None).await?; + Ok(id) +} + +#[cfg(test)] +#[path = "implication_rules_tests.rs"] +mod tests; diff --git a/crates/utopia-store/src/implication_rules_tests.rs b/crates/utopia-store/src/implication_rules_tests.rs new file mode 100644 index 000000000..3fcd2d089 --- /dev/null +++ b/crates/utopia-store/src/implication_rules_tests.rs @@ -0,0 +1,313 @@ +//! 蕴含规则（0044 决定 3 第五片）：提案落库、人批与后续任务同事务、待读的字、缓存、 +//! 物化算隐含行、驳回退掉。没有 `UTOPIA_DATABASE_URL` 时跳过。 +use super::*; +use crate::{materialize, phrase_bindings}; +use serde_json::json; +use sqlx::PgPool; + +struct Fx { + org: Uuid, + kb: Uuid, + film: Uuid, + place: Uuid, + country_of_origin: Uuid, + located_in: Uuid, + country: Uuid, + loud_tour: Uuid, + piedmont: Uuid, + statement: Uuid, +} + +/// 一个库：类 film / place，属性 country_of_origin / located_in / country； +/// 「Loud Tour」是 british film；一条陈述 Loud Tour —located in→ Piedmont（place） +async fn seed(pool: &PgPool) -> anyhow::Result { + let ids: Vec = (0..11).map(|_| Uuid::now_v7()).collect(); + let (org, ws, kb, film, place, coo, li, country, loud, piedmont, stmt) = ( + ids[0], ids[1], ids[2], ids[3], ids[4], ids[5], ids[6], ids[7], ids[8], ids[9], ids[10], + ); + sqlx::raw_sql(&format!( + "INSERT INTO organizations(id,name) VALUES ('{org}','implication-test'); + INSERT INTO workspaces(id,org_id,name) VALUES ('{ws}','{org}','implication-test'); + INSERT INTO knowledge_bases(id,workspace_id,name) VALUES ('{kb}','{ws}','implication-test'); + INSERT INTO entity_types(id,kb_id,key,label,color,shape) VALUES + ('{film}','{kb}','film','Film','#000','circle'), ('{place}','{kb}','place','Place','#000','circle'); + INSERT INTO relation_types(id,kb_id,key,label,kind,temporal) VALUES + ('{coo}','{kb}','country_of_origin','country of origin','relation','state'), + ('{li}','{kb}','located_in','located in','relation','state'), + ('{country}','{kb}','country','country','relation','state'); + INSERT INTO entities(id,kb_id,canonical_name,type_id,specific_type) VALUES + ('{loud}','{kb}','Loud Tour','{film}','British film'), + ('{piedmont}','{kb}','Piedmont','{place}','region'); + INSERT INTO facts(id,kb_id,subject_id,object_id,layer,phrase) VALUES + ('{stmt}','{kb}','{loud}','{piedmont}','open','located in');" + )) + .execute(pool) + .await?; + Ok(Fx { + org, + kb, + film, + place, + country_of_origin: coo, + located_in: li, + country, + loud_tour: loud, + piedmont, + statement: stmt, + }) +} + +async fn cleanup(pool: &PgPool, f: &Fx) -> anyhow::Result<()> { + sqlx::query("DELETE FROM jobs WHERE payload->>'kb_id'=$1") + .bind(f.kb.to_string()) + .execute(pool) + .await?; + sqlx::query("DELETE FROM organizations WHERE id=$1") + .bind(f.org) + .execute(pool) + .await?; + Ok(()) +} + +fn phrase_rule<'a>(f: &Fx, reading: Option<&'a str>, property: Uuid) -> Proposal<'a> { + Proposal { + trigger: "phrase", + phrase: "located in", + subject_type_id: Some(f.film), + object_type_id: Some(f.place), + object_is_value: false, + conclude_property_id: property, + reading, + status: "proposed", + votes: &serde_json::Value::Null, + basis: "b1", + statement_count: 1, + examples: &[], + } +} + +async fn implied_rows(pool: &PgPool, kb: Uuid) -> anyhow::Result)>> { + Ok(sqlx::query_as( + "SELECT subject_id, predicate_id, object_id FROM facts + WHERE kb_id=$1 AND layer='typed' AND implied AND invalidated_at IS NULL ORDER BY recorded_at", + ) + .bind(kb) + .fetch_all(pool) + .await?) +} + +#[tokio::test] +async fn a_proposal_lands_once_and_a_person_decides_it_with_its_job() -> anyhow::Result<()> { + let Some(url) = crate::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + crate::db::migrate(&pool).await?; + let f = seed(&pool).await?; + let run = async { + let p = phrase_rule(&f, Some("country_of_place"), f.country); + let id = propose(&pool, f.kb, &p).await?.expect("first proposal"); + // 同一条再提：代理的行被覆盖（返回 id），不是第二行 + assert_eq!(propose(&pool, f.kb, &p).await?, Some(id)); + assert_eq!(list(&pool, f.kb, Some("proposed")).await?.len(), 1); + // 人批：状态 approved，同事务排了 read_phrases（要读数） + let job = decide_with_delivery(&pool, f.kb, id, true, &json!({})) + .await? + .expect("decided"); + let (kind, status): (String, String) = + sqlx::query_as("SELECT kind, status FROM jobs WHERE id=$1") + .bind(job) + .fetch_one(&pool) + .await?; + assert_eq!((kind.as_str(), status.as_str()), (READ_KIND, "queued")); + let r = get(&pool, f.kb, id).await?.unwrap(); + assert_eq!( + (r.status.as_str(), r.decided_by.as_str()), + ("approved", "person") + ); + // 人判过的，代理再提也盖不掉 + assert_eq!(propose(&pool, f.kb, &p).await?, None); + // 不要读数的规则，批了直接排物化 + let plain = propose(&pool, f.kb, &phrase_rule(&f, None, f.country_of_origin)) + .await? + .unwrap(); + let job2 = decide_with_delivery(&pool, f.kb, plain, true, &json!({})) + .await? + .unwrap(); + let kind2: String = sqlx::query_scalar("SELECT kind FROM jobs WHERE id=$1") + .bind(job2) + .fetch_one(&pool) + .await?; + assert_eq!(kind2, phrase_bindings::MATERIALIZE_KIND); + anyhow::Ok(()) + } + .await; + cleanup(&pool, &f).await?; + run +} + +#[tokio::test] +async fn an_approved_rule_waits_for_its_reading_then_implies_a_fact_with_evidence( +) -> anyhow::Result<()> { + let Some(url) = crate::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + crate::db::migrate(&pool).await?; + let f = seed(&pool).await?; + let run = async { + let id = propose( + &pool, + f.kb, + &phrase_rule(&f, Some("country_of_place"), f.country), + ) + .await? + .unwrap(); + decide_with_delivery(&pool, f.kb, id, true, &json!({})).await?; + // 待读：陈述的宾语「piedmont」按 country_of_place + let pending = pending_readings(&pool, f.kb).await?; + assert_eq!( + pending, + vec![PendingReading { + reading: "country_of_place".into(), + phrase: "piedmont".into() + }] + ); + // 缓存没填：物化不算隐含行 + let o = materialize::materialize(&pool, f.kb).await?; + assert_eq!(o.implied, 0); + assert!(implied_rows(&pool, f.kb).await?.is_empty()); + // 读数读出「United States」：库里没有就建，名字事实照记 + let us = resolve_or_create_named(&pool, f.kb, "United States").await?; + assert_eq!( + resolve_or_create_named(&pool, f.kb, "united states").await?, + us, + "found by name the second time" + ); + record_reading(&pool, f.kb, "country_of_place", "Piedmont", Some(us), None).await?; + assert!( + pending_readings(&pool, f.kb).await?.is_empty(), + "cached now" + ); + let o = materialize::materialize(&pool, f.kb).await?; + assert_eq!(o.implied, 1); + let rows = implied_rows(&pool, f.kb).await?; + assert_eq!(rows, vec![(f.loud_tour, f.country, Some(us))]); + // 证据从触发它的陈述抄来；来源记着规则与陈述 + let src: (Uuid, Option) = + sqlx::query_as("SELECT rule_id, statement_id FROM implied_fact_sources") + .fetch_one(&pool) + .await?; + assert_eq!(src, (id, Some(f.statement))); + // 再跑一遍什么都不动 + let o = materialize::materialize(&pool, f.kb).await?; + assert_eq!((o.implied, o.retired), (0, 0)); + anyhow::Ok(()) + } + .await; + cleanup(&pool, &f).await?; + run +} + +#[tokio::test] +async fn rejecting_the_rule_or_losing_the_statement_retires_the_implied_row() -> anyhow::Result<()> +{ + let Some(url) = crate::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + crate::db::migrate(&pool).await?; + let f = seed(&pool).await?; + let run = async { + // 不要读数的规则：located in 的陈述还蕴含 country_of_origin = 陈述的宾语（就当测试） + let id = propose(&pool, f.kb, &phrase_rule(&f, None, f.country_of_origin)) + .await? + .unwrap(); + decide_with_delivery(&pool, f.kb, id, true, &json!({})).await?; + assert_eq!(materialize::materialize(&pool, f.kb).await?.implied, 1); + assert_eq!(implied_rows(&pool, f.kb).await?.len(), 1); + // 驳回：来源删、行作废 + decide_with_delivery(&pool, f.kb, id, false, &json!({})).await?; + let o = materialize::materialize(&pool, f.kb).await?; + assert_eq!(o.retired, 1); + assert!(implied_rows(&pool, f.kb).await?.is_empty()); + // 再批回来，行回来；陈述作废，行再退 + decide_with_delivery(&pool, f.kb, id, true, &json!({})).await?; + assert_eq!(materialize::materialize(&pool, f.kb).await?.implied, 1); + sqlx::query("UPDATE facts SET invalidated_at=now() WHERE id=$1") + .bind(f.statement) + .execute(&pool) + .await?; + assert_eq!(materialize::materialize(&pool, f.kb).await?.retired, 1); + assert!(implied_rows(&pool, f.kb).await?.is_empty()); + anyhow::Ok(()) + } + .await; + cleanup(&pool, &f).await?; + run +} + +#[tokio::test] +async fn a_kind_word_rule_implies_a_fact_for_every_thing_so_called() -> anyhow::Result<()> { + let Some(url) = crate::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + crate::db::migrate(&pool).await?; + let f = seed(&pool).await?; + let run = async { + let votes = json!({}); + let id = propose( + &pool, + f.kb, + &Proposal { + trigger: "kind_word", + phrase: "British Film", + subject_type_id: None, + object_type_id: None, + object_is_value: false, + conclude_property_id: f.country_of_origin, + reading: Some("country_of_nationality"), + status: "proposed", + votes: &votes, + basis: "k1", + statement_count: 1, + examples: &[], + }, + ) + .await? + .unwrap(); + decide_with_delivery(&pool, f.kb, id, true, &votes).await?; + assert_eq!( + pending_readings(&pool, f.kb).await?, + vec![PendingReading { + reading: "country_of_nationality".into(), + phrase: "british film".into() + }] + ); + let uk = resolve_or_create_named(&pool, f.kb, "United Kingdom").await?; + record_reading( + &pool, + f.kb, + "country_of_nationality", + "british film", + Some(uk), + None, + ) + .await?; + assert_eq!(materialize::materialize(&pool, f.kb).await?.implied, 1); + assert_eq!( + implied_rows(&pool, f.kb).await?, + vec![(f.loud_tour, f.country_of_origin, Some(uk))] + ); + // 读不出来也缓存：不再待读，也不算行 + record_reading(&pool, f.kb, "country_of_nationality", "iberian", None, None).await?; + assert!(pending_readings(&pool, f.kb).await?.is_empty()); + let _ = f.piedmont; + let _ = f.located_in; + anyhow::Ok(()) + } + .await; + cleanup(&pool, &f).await?; + run +} diff --git a/crates/utopia-store/src/lib.rs b/crates/utopia-store/src/lib.rs index 25d23a138..3c520c9fd 100644 --- a/crates/utopia-store/src/lib.rs +++ b/crates/utopia-store/src/lib.rs @@ -17,6 +17,7 @@ pub mod export; pub mod extraction_drops; pub mod governance; pub mod graph; +pub mod implication_rules; pub mod jobs; pub mod kbs; pub mod mappings; diff --git a/crates/utopia-store/src/materialize.rs b/crates/utopia-store/src/materialize.rs index 08337774e..b05c4e797 100644 --- a/crates/utopia-store/src/materialize.rs +++ b/crates/utopia-store/src/materialize.rs @@ -36,6 +36,8 @@ pub struct Outcome { pub added: u64, /// 并进已有行的陈述数 pub merged: u64, + /// 规则算出来的隐含行（0044 决定 3 第五片），新建的 + pub implied: u64, } /// 一条该物化的（陈述, 绑定）对，连陈述上要抄的东西。 @@ -128,13 +130,38 @@ async fn materialize_in_tx( .execute(&mut **tx) .await?; - // 2. 作废来源全空的类型化行：只动算出来的行（带 from_statement_id 的），人写的不碰 + // 1b. 隐含行的来源：规则不再批准、触发它的陈述死了或换了签名、实体没了或换了类别词， + // 来源就删；与 1 同一条规矩，只是来源表是另一张（implied_fact_sources） + sqlx::query(&format!( + "DELETE FROM implied_fact_sources i + USING implication_rules r + WHERE i.rule_id = r.id AND r.kb_id = $1 + AND (r.status <> 'approved' + OR (i.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 + WHERE s.id = i.statement_id AND s.layer = 'open' AND s.invalidated_at IS NULL + AND r.trigger = 'phrase' AND {RULE_MATCH})) + OR (i.entity_id IS NOT NULL AND NOT EXISTS ( + SELECT 1 FROM entities e + WHERE e.id = i.entity_id AND e.merged_into IS NULL + AND r.trigger = 'kind_word' AND {KIND} = r.phrase)))", + RULE_MATCH = crate::implication_rules::RULE_MATCH, + KIND = kind_word_sql("e.specific_type"), + )) + .bind(kb_id) + .execute(&mut **tx) + .await?; + + // 2. 作废来源全空的类型化行：只动算出来的行（带 from_statement_id 的、或规则算的），人写的不碰 let retired = sqlx::query( "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 typed_fact_sources src WHERE src.fact_id = t.id)", + AND (t.from_statement_id IS NOT NULL OR t.implied) + AND NOT EXISTS (SELECT 1 FROM typed_fact_sources src WHERE src.fact_id = t.id) + AND NOT EXISTS (SELECT 1 FROM implied_fact_sources i WHERE i.fact_id = t.id)", ) .bind(kb_id) .execute(&mut **tx) @@ -278,13 +305,160 @@ async fn materialize_in_tx( .execute(&mut **tx) .await?; } + // 3b. 已批准的规则算隐含行（0044 决定 3 第五片）。读数只查缓存：缓存里没有的这一轮 + // 不算，`read_phrases` 填上之后再来。短语规则按陈述触发，类别词规则按实体触发 + let implied = imply_in_tx(tx, kb_id).await?; Ok(Outcome { retired, added, merged, + implied, }) } +#[derive(sqlx::FromRow)] +struct Implied { + rule: Uuid, + statement: Option, + entity: Option, + subject_id: Uuid, + property: Uuid, + object_id: Option, + object_value: Option, + valid_from: Option>, + valid_from_precision: Option, + valid_from_grade: Option, + valid_to: Option>, + valid_to_precision: Option, + attested_from: Option>, + confidence: f32, +} + +async fn imply_in_tx( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + kb_id: Uuid, +) -> AppResult { + // 短语规则：签名下活着的、没 mood 的陈述；宾语是读数的答案（缓存里的实体或值）， + // 没有读数时就是陈述的宾语。已经有活着的隐含行以这条陈述为来源的不再算 + let by_statement: Vec = sqlx::query_as(&format!( + "SELECT r.id AS rule, s.id AS statement, NULL::uuid AS entity, + s.subject_id, r.conclude_property_id AS property, + CASE WHEN r.reading IS NULL THEN s.object_id ELSE pr.entity_id END AS object_id, + CASE WHEN r.reading IS NULL THEN s.object_value ELSE pr.value END AS object_value, + s.valid_from, s.valid_from_precision, s.valid_from_grade, s.valid_to, s.valid_to_precision, + s.attested_from, s.confidence + FROM implication_rules r + JOIN facts s ON s.kb_id = r.kb_id AND s.layer = 'open' AND s.invalidated_at IS NULL + JOIN entities se ON se.id = s.subject_id + LEFT JOIN entities oe ON oe.id = s.object_id + LEFT JOIN phrase_readings pr ON r.reading IS NOT NULL AND pr.kb_id = r.kb_id + AND pr.reading = r.reading AND pr.phrase = {TEXT} + WHERE r.kb_id = $1 AND r.status = 'approved' AND r.trigger = 'phrase' + AND {RULE_MATCH} + AND NOT EXISTS (SELECT 1 FROM statement_qualifiers q WHERE q.fact_id = s.id AND q.role = 'mood') + AND (r.reading IS NULL OR pr.entity_id IS NOT NULL OR pr.value IS NOT NULL) + AND NOT EXISTS (SELECT 1 FROM implied_fact_sources i JOIN facts t ON t.id = i.fact_id + WHERE i.rule_id = r.id AND i.statement_id = s.id AND t.invalidated_at IS NULL) + ORDER BY s.id", + TEXT = crate::implication_rules::object_text_sql(), + RULE_MATCH = crate::implication_rules::RULE_MATCH, + )) + .bind(kb_id) + .fetch_all(&mut **tx) + .await?; + // 类别词规则：带这个类别词的活着的实体；宾语必须来自读数（类别词自己没有宾语） + let by_entity: Vec = sqlx::query_as(&format!( + "SELECT r.id AS rule, NULL::uuid AS statement, e.id AS entity, + e.id AS subject_id, r.conclude_property_id AS property, + pr.entity_id AS object_id, pr.value AS object_value, + NULL::timestamptz AS valid_from, NULL::text AS valid_from_precision, NULL::text AS valid_from_grade, + NULL::timestamptz AS valid_to, NULL::text AS valid_to_precision, + NULL::timestamptz AS attested_from, 0.9::real AS confidence + FROM implication_rules r + JOIN entities e ON e.kb_id = r.kb_id AND e.merged_into IS NULL AND {KIND} = r.phrase + JOIN phrase_readings pr ON pr.kb_id = r.kb_id AND pr.reading = r.reading AND pr.phrase = r.phrase + WHERE r.kb_id = $1 AND r.status = 'approved' AND r.trigger = 'kind_word' AND r.reading IS NOT NULL + AND (pr.entity_id IS NOT NULL OR pr.value IS NOT NULL) + AND NOT EXISTS (SELECT 1 FROM implied_fact_sources i JOIN facts t ON t.id = i.fact_id + WHERE i.rule_id = r.id AND i.entity_id = e.id AND t.invalidated_at IS NULL) + ORDER BY e.id", + KIND = kind_word_sql("e.specific_type"), + )) + .bind(kb_id) + .fetch_all(&mut **tx) + .await?; + + let mut implied = 0u64; + for d in by_statement.iter().chain(by_entity.iter()) { + let validity = Validity { + from: d.valid_from, + from_precision: d.valid_from_precision.as_deref(), + from_grade: d.valid_from_grade.as_deref(), + to: d.valid_to, + to_precision: d.valid_to_precision.as_deref(), + attested_at: d.attested_from, + }; + let object = match (d.object_id, &d.object_value) { + (Some(o), _) => FactObject::Entity(o), + (None, Some(v)) => FactObject::Value(v), + _ => continue, + }; + // 一个东西不蕴含自己：读数把「Virginia」读成「United States」是对的，把「France」 + // 读成「France」就是一条自环 + if matches!(object, FactObject::Entity(o) if o == d.subject_id) { + continue; + } + let (fact, new) = insert_fact_on( + tx, + kb_id, + d.subject_id, + Some(d.property), + object, + validity, + d.confidence, + ) + .await?; + if new { + implied += 1; + sqlx::query("UPDATE facts SET implied = TRUE WHERE id = $1") + .bind(fact) + .execute(&mut **tx) + .await?; + } + sqlx::query( + "INSERT INTO implied_fact_sources (fact_id, rule_id, statement_id, entity_id) + VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING", + ) + .bind(fact) + .bind(d.rule) + .bind(d.statement) + .bind(d.entity) + .execute(&mut **tx) + .await?; + // 证据：短语规则抄触发它的那条陈述的引文——读的人从这句得出的结论，证据就是这句 + if let Some(statement) = d.statement { + sqlx::query( + "INSERT INTO fact_evidence (fact_id, chunk_id, quote, document_id, doc_version, + proposed_predicate, quote_start, quote_end) + SELECT $1, chunk_id, quote, document_id, doc_version, proposed_predicate, + quote_start, quote_end + FROM fact_evidence WHERE fact_id = $2 + ON CONFLICT DO NOTHING", + ) + .bind(fact) + .bind(statement) + .execute(&mut **tx) + .await?; + } + } + Ok(implied) +} + +/// 类别词的归一（同 type_bindings）：这里要在 SQL 里对上实体的 specific_type +fn kind_word_sql(col: &str) -> String { + format!("lower(btrim(regexp_replace({col}, '\\s+', ' ', 'g')))") +} + /// 库里活着的、从陈述算出来的类型化行数。 pub async fn count(pool: &PgPool, kb_id: Uuid) -> AppResult { Ok(sqlx::query_scalar( 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 index 36149b9ed..71c5cd08d 100644 --- 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 @@ -179,7 +179,7 @@ async fn a_bound_statement_becomes_a_typed_fact() -> anyhow::Result<()> { // 来源搬过来），第三条并进去（merged 1）——最后一行，三条来源，两条证据；带 mood 的 // 那条不算 let first = materialize(&pool, kb).await?; - assert_eq!(first, Outcome { retired: 0, added: 2, merged: 1 }); + assert_eq!(first, Outcome { retired: 0, added: 2, merged: 1, implied: 0 }); let live = |pool: PgPool| async move { sqlx::query_as::<_, (Uuid, Uuid, Uuid, Uuid, Uuid, Option>, Option)>( "SELECT id, subject_id, object_id, predicate_id, from_statement_id, valid_from, @@ -238,7 +238,7 @@ async fn a_bound_statement_becomes_a_typed_fact() -> anyhow::Result<()> { .execute(&pool) .await?; // 旧行的来源全不成立了：作废 1；反向重算时裸的那条先成行、带时间的再取代它：新建 2 - assert_eq!(materialize(&pool, kb).await?, Outcome { retired: 1, added: 2, merged: 0 }); + assert_eq!(materialize(&pool, kb).await?, Outcome { retired: 1, added: 2, merged: 0, implied: 0 }); let rows = live(pool.clone()).await?; assert_eq!(rows.len(), 1); assert_eq!((rows[0].1, rows[0].2), (port, bakery), "方向反了主宾对调"); @@ -258,7 +258,7 @@ async fn a_bound_statement_becomes_a_typed_fact() -> anyhow::Result<()> { .bind(kb) .execute(&pool) .await?; - assert_eq!(materialize(&pool, kb).await?, Outcome { retired: 1, added: 0, merged: 0 }); + assert_eq!(materialize(&pool, kb).await?, Outcome { retired: 1, added: 0, merged: 0, implied: 0 }); assert_eq!(utopia_store::materialize::count(&pool, kb).await?, 0); anyhow::Ok(()) } diff --git a/docs/decisions/0044-the-ontology-is-a-view-over-what-documents-say.md b/docs/decisions/0044-the-ontology-is-a-view-over-what-documents-say.md index 5432cdbc9..8fef6af08 100644 --- a/docs/decisions/0044-the-ontology-is-a-view-over-what-documents-say.md +++ b/docs/decisions/0044-the-ontology-is-a-view-over-what-documents-say.md @@ -1,6 +1,6 @@ # 0044 · The ontology is a view over what documents say -- **Status**: Accepted 2026-09-17 · cut 1 built: extraction writes open statements (#731), memory documents take the same path (#735), the typed path is deleted (#736), kind words bind to classes with two votes (#741), the contract's rules for things, phrases, tables and moods (#743, #744, #745) · cut 2 (alignment producing typed facts, identity profiles, the errata agent) not built · current state in [design/extraction](../design/extraction.md) and [design/ontology](../design/ontology.md) · replaces the staged-reading draft of this record (skim card, graded mentions, statements bound at write time), which the prototype below did not bear out · prototype scripts and measurements from 2026-09-15 are summarised in [What the prototype measured](#what-the-prototype-measured) +- **Status**: Accepted 2026-09-17 · cut 1 built: extraction writes open statements (#731), memory documents take the same path (#735), the typed path is deleted (#736), kind words bind to classes with two votes (#741), the contract's rules for things, phrases, tables and moods (#743, #744, #745) · cut 2 in progress: bindings (#751), materialisation (0067, 0068), the human decision with its job (0051, #876), the decision basis (0053, #878), **implication rules with cached readings (migration 0073; PR number added at merge)** — the parity run against the bound pass waits for the typed-graph bench (#880) to be run with a model; identity profiles (cut 4) in progress in a separate track; the errata agent (cut 6) not built · current state in [design/extraction](../design/extraction.md) and [design/ontology](../design/ontology.md) · replaces the staged-reading draft of this record (skim card, graded mentions, statements bound at write time), which the prototype below did not bear out · prototype scripts and measurements from 2026-09-15 are summarised in [What the prototype measured](#what-the-prototype-measured) - **Written**: 2026-09-16 (conventions in the [README](README.md)) - **Related**: [0022](0022-an-unknown-date-is-not-an-open-one.md) put a document's date in `attested_at` beside the world and record axes; [0025](0025-governance-reads-the-ledger-before-it-decides.md) and [0027](0027-an-automatic-merge-is-gated-by-what-it-can-undo.md) put agent decisions through a gate that weighs what they can undo; [0041](0041-a-name-is-a-claim-about-an-entity.md) made names facts and identity a matter of evidence; [0043](0043-every-review-queue-is-governed.md) sent every review queue through the governor; #714 found upload time used as the document date in extraction. @@ -68,6 +68,8 @@ This is how the facts a reader draws without the text stating them (a place's co When the ontology changes, only facts under changed signatures and rules are recomputed. A signature with no property stays in the open graph, loses nothing, and counts toward the workbench's suggestions. +**Revision 2026-09-23 (implication rules built):** a rule is keyed like a binding (a phrase signature) or by a kind word, concludes one property, and takes its object either from the statement's own object or from a *reading* of the object's words (`country_of_nationality`, `country_of_place`, `year_of_phrase`). The aligner proposes rules after it has decided the signatures of a run and for every kind word once; a person approves or rejects on the alignment queue; approval commits with its job (0051). Readings are model calls made once per distinct phrase by a `read_phrases` job and cached in `phrase_readings`, so `materialize` stays free of model calls: it reads the cache and produces facts marked `implied`, with the triggering statement's evidence, retired by source like any other typed row. What is not read cannot be implied: a phrase the model cannot place caches as "no answer" and is never asked again until it is asked about differently. + **Revision 2026-09-23:** [0053](0053-a-phrase-decision-records-the-inputs-it-considered.md) replaces "a binding goes stale by timestamp" with a recorded basis: candidates are admitted through the class hierarchy, a decision stores a fingerprint of the closures and candidates it saw, and it is stale when the current fingerprint differs. No-candidate and overflow become recorded outcomes; the requeue condition reads live signatures only (#807, #795). **Revision proposed 2026-09-21:** [0051](0051-a-human-phrase-decision-carries-its-materialization-work.md) addresses delivery after a human binding commits beyond an older materializer’s final read. It proposes a decision and its own durable job in one transaction, while retaining the current projection semantics. The asynchronous HTTP/job/UI contract remains unimplemented. diff --git a/docs/design/ontology.md b/docs/design/ontology.md index 93b868f36..6f3431c01 100644 --- a/docs/design/ontology.md +++ b/docs/design/ontology.md @@ -74,7 +74,19 @@ when the fingerprint of the current inputs differs, which is what timestamps cou parent edge added or removed, an edit committed while the model was answering [0053, #807, #795]. A signature with no admissible property is recorded as `none` (its projection retires); one with more candidates than the limit is `undecided` for the queue, not silently skipped. Kind-word -bindings still use `updated_at`, so cosmetic edits can also trigger their reevaluation. Even one edit can reopen all older automatic negative bindings on that side of the +bindings still use `updated_at`, so cosmetic edits can also trigger their reevaluation. + +**A shape of statement can imply a fact of another property** [0044 decision 3, migration 0073]. +An implication rule is keyed like a binding (a signature) or by a kind word, names the property it +concludes, and takes its object from the statement's own object or from a *reading* of the object's +words: the country a demonym names, the country a place lies in, the year a phrase gives. The +aligner proposes rules once per decided signature and once per kind word (a "nothing implied" +answer is recorded so it is not asked again until the basis changes); a person approves or rejects +them on the alignment queue, and the decision commits with its job. Readings are asked of the model +once per distinct phrase by the `read_phrases` job and cached in `phrase_readings`, including "no +answer"; materialisation never calls a model — it reads the cache, writes the implied facts with the +triggering statement's evidence, marks them `implied` (the export carries the flag), and retires +them by source like any other typed row. Even one edit can reopen all older automatic negative bindings on that side of the base, requiring two votes per eligible item through batched model requests; debouncing reduces the number of runs, not the items reconsidered. A burst of ontology edits debounces into one run rather than one run each [#757]; @@ -160,9 +172,8 @@ the prompt, a description is read by people and by the aligner. ## Proposed and not built -- **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. The prototype aligner reached 14.7% and +- **Alignment** (0044 cut 2), the rest: a signature that tells a figure from words on the value + side; the parity run against the withdrawn bound pass on the typed-graph bench (#880). 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/migrations/0073_a_statement_shape_implies_a_fact.sql b/migrations/0073_a_statement_shape_implies_a_fact.sql new file mode 100644 index 000000000..e52266bd5 --- /dev/null +++ b/migrations/0073_a_statement_shape_implies_a_fact.sql @@ -0,0 +1,67 @@ +-- 一种形状的陈述蕴含另一条属性的事实（0044 决定 3 的第五片：蕴含规则与缓存的读数）。 +-- +-- 原型量过的召回差距在这里（0044 §2）：「a 1952 British film」蕴含 country of origin， +-- 「located in the Piedmont region of Virginia」蕴含 country——读的人不用原文说就能得出， +-- 开放陈述却不会把它写成陈述。补法不是再抽一遍，是**规则**：某个签名（短语 × 两端的类） +-- 或某个类别词下的东西，蕴含某条属性的事实，宾语要么就是陈述的宾语，要么由一个「读数」 +-- 从宾语的字里读出来（民族形容词指的国家、地名所属的国家、短语给出的年份）。 +-- 对齐器提规则，工作台批，代码执行；读数按 distinct 的字算一次、缓存，物化只查缓存。 +-- +-- facts.implied 规则算出来的类型化行。不是陈述直接说的，导出与界面要能分辨 +-- implication_rules 规则本体：触发（phrase 签名 / kind_word 类别词）、结论属性、读数、 +-- 与绑定同一套 status / decided_by / basis（0053）：人的不被代理盖 +-- phrase_readings 读数缓存：(读数种类, 字) → 库里的一样东西或一个字面值；两者都空 +-- = 读不出来，也缓存住，别每轮再问 +-- implied_fact_sources 一行隐含事实的来源：哪条规则、由哪条陈述或哪个实体触发。来源全空 +-- 行就作废，与 typed_fact_sources 同一条规矩（0068） +ALTER TABLE facts ADD COLUMN implied BOOLEAN NOT NULL DEFAULT FALSE; + +CREATE TABLE implication_rules ( + id UUID PRIMARY KEY, + kb_id UUID NOT NULL REFERENCES knowledge_bases(id) ON DELETE CASCADE, + -- phrase：签名下的每条陈述触发；kind_word：带这个类别词的每个实体触发 + trigger TEXT NOT NULL CHECK (trigger IN ('phrase', 'kind_word')), + -- 归一过的短语或类别词（同 phrase_bindings.phrase / type_bindings.kind_word） + phrase TEXT NOT NULL, + subject_type_id UUID REFERENCES entity_types(id) ON DELETE CASCADE, + object_type_id UUID REFERENCES entity_types(id) ON DELETE CASCADE, + object_is_value BOOLEAN NOT NULL DEFAULT false, + conclude_property_id UUID NOT NULL REFERENCES relation_types(id) ON DELETE CASCADE, + -- 空 = 宾语就是陈述的宾语；否则是读数的种类（见 utopia_extract::implication::READINGS） + reading TEXT, + status TEXT NOT NULL CHECK (status IN ('proposed', 'approved', 'rejected')), + votes JSONB, + decided_by TEXT NOT NULL DEFAULT 'agent' CHECK (decided_by IN ('agent', 'person')), + basis TEXT, + statement_count INTEGER NOT NULL DEFAULT 0, + examples TEXT[] NOT NULL DEFAULT '{}', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + decided_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE NULLS NOT DISTINCT (kb_id, trigger, phrase, subject_type_id, object_type_id, object_is_value, + conclude_property_id, reading), + CONSTRAINT implication_rules_kind_word_shape + CHECK (trigger <> 'kind_word' OR (subject_type_id IS NULL AND object_type_id IS NULL AND NOT object_is_value)) +); +CREATE INDEX implication_rules_kb_status_idx ON implication_rules (kb_id, status); + +CREATE TABLE phrase_readings ( + kb_id UUID NOT NULL REFERENCES knowledge_bases(id) ON DELETE CASCADE, + reading TEXT NOT NULL, + phrase TEXT NOT NULL, + entity_id UUID REFERENCES entities(id) ON DELETE CASCADE, + value JSONB, + answered_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (kb_id, reading, phrase), + CHECK (entity_id IS NULL OR value IS NULL) +); + +CREATE TABLE implied_fact_sources ( + fact_id UUID NOT NULL REFERENCES facts(id) ON DELETE CASCADE, + rule_id UUID NOT NULL REFERENCES implication_rules(id) ON DELETE CASCADE, + statement_id UUID REFERENCES facts(id) ON DELETE CASCADE, + entity_id UUID REFERENCES entities(id) ON DELETE CASCADE, + PRIMARY KEY (fact_id, rule_id), + CHECK ((statement_id IS NOT NULL) <> (entity_id IS NOT NULL)) +); +CREATE INDEX implied_fact_sources_rule_idx ON implied_fact_sources (rule_id); +CREATE INDEX implied_fact_sources_statement_idx ON implied_fact_sources (statement_id) WHERE statement_id IS NOT NULL; diff --git a/web/src/api.ts b/web/src/api.ts index 2df340926..2dfe65ffa 100644 --- a/web/src/api.ts +++ b/web/src/api.ts @@ -737,6 +737,23 @@ export type AlignmentItem = entity_count: number; votes: { first?: string | null; second?: string | null } | null; decided_at: string; + } + | { + /** 对齐器提的一条蕴含规则（0044 决定 3 第五片） */ + kind: "rule"; + id: string; + trigger: "phrase" | "kind_word"; + phrase: string; + subject_class: string | null; + object_class: string | null; + object_is_value: boolean; + property: string; + property_label: string; + reading: string | null; + statement_count: number; + examples: string[]; + votes: { agent?: { property: string; reading: string | null } | null } | null; + decided_at: string; }; export interface AlignmentVote { property: string; @@ -2327,6 +2344,12 @@ export const api = { `/api/v1/kbs/${kbId}/review/alignment/phrases/${bindingId}`, { method: "POST", body: JSON.stringify({ property, direction }) }, ), + /** 人批或驳一条蕴含规则：答 202 和 job id，隐含事实在后台算（0044 决定 3 第五片） */ + decideAlignmentRule: (kbId: string, ruleId: string, approve: boolean) => + request<{ ok: boolean; job_id: number; status: "accepted" }>( + `/api/v1/kbs/${kbId}/review/alignment/rules/${ruleId}`, + { method: "POST", body: JSON.stringify({ approve }) }, + ), /** 人定一个类别词：类，或没有。它名下的实体换类，短语签名跟着重判 */ decideAlignmentKindWord: (kbId: string, kindWord: string, cls: string | null) => request<{ ok: boolean }>( diff --git a/web/src/i18n/en.ts b/web/src/i18n/en.ts index 054c12218..75461758f 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -1953,6 +1953,13 @@ export const en = { alignmentStatements: (n: number) => (n === 1 ? "1 statement" : `${n} statements`), alignmentEntities: (n: number) => (n === 1 ? "1 thing" : `${n} things`), alignmentVotes: (first: string, second: string) => `Votes: ${first} · ${second}`, + alignmentRuleImplies: (property: string) => `also implies ${property}`, + alignmentRuleObjectIsStatement: "object: the statement's own object", + alignmentRuleReading: (reading: string) => `object: read from the words as ${reading.replace(/_/g, " ")}`, + alignmentRuleKindWord: (word: string) => `things called "${word}"`, + alignmentApprove: "Approve rule", + alignmentReject: "Reject", + alignmentRuleAccepted: "Saved. Implied facts are being computed in the background.", alignmentTooMany: (n: number) => `${n} properties could apply; too many to ask the model. Pick one or leave it open.`, alignmentConflict: "This decision conflicts with the current state. Refresh and review it before trying again.", alignmentKindWordBusy: "This kind word is being updated by another operation. Please try again shortly.", diff --git a/web/src/i18n/zh.ts b/web/src/i18n/zh.ts index 838e45b94..a9e52196a 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -1712,6 +1712,13 @@ export const zh: Strings = { alignmentStatements: (n: number) => `${n} 条陈述`, alignmentEntities: (n: number) => `${n} 样东西`, alignmentVotes: (first: string, second: string) => `两票：${first} · ${second}`, + alignmentRuleImplies: (property: string) => `同时蕴含 ${property}`, + alignmentRuleObjectIsStatement: "宾语：陈述自己的宾语", + alignmentRuleReading: (reading: string) => `宾语：按「${reading.replace(/_/g, " ")}」从字里读出`, + alignmentRuleKindWord: (word: string) => `叫作「${word}」的东西`, + alignmentApprove: "批准规则", + alignmentReject: "驳回", + alignmentRuleAccepted: "已保存，隐含事实正在后台计算。", alignmentTooMany: (n: number) => `有 ${n} 条属性都可能对得上，多到没法问模型。请选一条或留在开放图谱。`, alignmentConflict: "此决定与当前状态冲突。请刷新并核对后再试。", alignmentKindWordBusy: "这个类别词正在被其他操作更新，请稍后重试。", diff --git a/web/src/pages/Review.tsx b/web/src/pages/Review.tsx index 51440376d..8ecc2ea35 100644 --- a/web/src/pages/Review.tsx +++ b/web/src/pages/Review.tsx @@ -706,6 +706,46 @@ function voteText(v: { property: string; direction: string } | string | null | u return `${v.property} · ${v.direction === "reverse" ? S.review.alignmentReverse : S.review.alignmentForward}`; } +/** 对齐器提的一条蕴含规则（0044 决定 3 第五片）：这种形状还蕴含哪条属性、宾语怎么来；人批或驳 */ +function AlignmentRuleRow({ + item, + busy, + onDecide, +}: { + item: Extract; + busy: boolean; + onDecide: (approve: boolean) => void; +}) { + const shape = + item.trigger === "kind_word" + ? S.review.alignmentRuleKindWord(item.phrase) + : `${item.subject_class ?? "?"} —${item.phrase}→ ${item.object_is_value ? "value" : (item.object_class ?? "?")}`; + return ( +
+
{shape}
+
+ {S.review.alignmentRuleImplies(item.property_label || item.property)} ·{" "} + {item.reading ? S.review.alignmentRuleReading(item.reading) : S.review.alignmentRuleObjectIsStatement} +
+ {item.examples.length > 0 && ( +
+ {item.examples.map((e, i) => ( +
{e}
+ ))} +
+ )} +
+ + +
+
+ ); +} + /** 一条短语签名：短语、两端的类、例句、两票；人选属性与方向，或「没有」 */ function AlignmentPhraseRow({ item, @@ -1365,6 +1405,12 @@ export function Review() { onSuccess: () => toast.success(S.review.alignmentAccepted), onSettled: invalidate, }); + const alignmentRuleAction = useMutation({ + mutationFn: ({ id, approve }: { id: string; approve: boolean }) => + api.decideAlignmentRule(kb!.id, id, approve), + onSuccess: () => toast.success(S.review.alignmentRuleAccepted), + onSettled: invalidate, + }); const alignmentKindWordAction = useMutation({ mutationFn: ({ kindWord, cls }: { kindWord: string; cls: string | null }) => api.decideAlignmentKindWord(kb!.id, kindWord, cls), @@ -1924,6 +1970,16 @@ export function Review() { alignmentPhraseAction.mutate({ id: item.id, property, direction }) } /> + ) : item.kind === "rule" ? ( + alignmentRuleAction.mutate({ id: item.id, approve })} + /> ) : ( Date: Wed, 23 Sep 2026 17:16:36 +0800 Subject: [PATCH 2/3] Name the PR that builds implication rules in 0044's status line Co-Authored-By: Claude Fable 5.1 Signed-off-by: WaylandYang <145302500+WaylandYang@users.noreply.github.com> --- .../0044-the-ontology-is-a-view-over-what-documents-say.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/decisions/0044-the-ontology-is-a-view-over-what-documents-say.md b/docs/decisions/0044-the-ontology-is-a-view-over-what-documents-say.md index 8fef6af08..a26d90a3d 100644 --- a/docs/decisions/0044-the-ontology-is-a-view-over-what-documents-say.md +++ b/docs/decisions/0044-the-ontology-is-a-view-over-what-documents-say.md @@ -1,6 +1,6 @@ # 0044 · The ontology is a view over what documents say -- **Status**: Accepted 2026-09-17 · cut 1 built: extraction writes open statements (#731), memory documents take the same path (#735), the typed path is deleted (#736), kind words bind to classes with two votes (#741), the contract's rules for things, phrases, tables and moods (#743, #744, #745) · cut 2 in progress: bindings (#751), materialisation (0067, 0068), the human decision with its job (0051, #876), the decision basis (0053, #878), **implication rules with cached readings (migration 0073; PR number added at merge)** — the parity run against the bound pass waits for the typed-graph bench (#880) to be run with a model; identity profiles (cut 4) in progress in a separate track; the errata agent (cut 6) not built · current state in [design/extraction](../design/extraction.md) and [design/ontology](../design/ontology.md) · replaces the staged-reading draft of this record (skim card, graded mentions, statements bound at write time), which the prototype below did not bear out · prototype scripts and measurements from 2026-09-15 are summarised in [What the prototype measured](#what-the-prototype-measured) +- **Status**: Accepted 2026-09-17 · cut 1 built: extraction writes open statements (#731), memory documents take the same path (#735), the typed path is deleted (#736), kind words bind to classes with two votes (#741), the contract's rules for things, phrases, tables and moods (#743, #744, #745) · cut 2 in progress: bindings (#751), materialisation (0067, 0068), the human decision with its job (0051, #876), the decision basis (0053, #878), **implication rules with cached readings (migration 0073, PR #882)** — the parity run against the bound pass waits for the typed-graph bench (#880) to be run with a model; identity profiles (cut 4) in progress in a separate track; the errata agent (cut 6) not built · current state in [design/extraction](../design/extraction.md) and [design/ontology](../design/ontology.md) · replaces the staged-reading draft of this record (skim card, graded mentions, statements bound at write time), which the prototype below did not bear out · prototype scripts and measurements from 2026-09-15 are summarised in [What the prototype measured](#what-the-prototype-measured) - **Written**: 2026-09-16 (conventions in the [README](README.md)) - **Related**: [0022](0022-an-unknown-date-is-not-an-open-one.md) put a document's date in `attested_at` beside the world and record axes; [0025](0025-governance-reads-the-ledger-before-it-decides.md) and [0027](0027-an-automatic-merge-is-gated-by-what-it-can-undo.md) put agent decisions through a gate that weighs what they can undo; [0041](0041-a-name-is-a-claim-about-an-entity.md) made names facts and identity a matter of evidence; [0043](0043-every-review-queue-is-governed.md) sent every review queue through the governor; #714 found upload time used as the document date in extraction. From 81c70c155455134e95bd09002d20781f3fc50a38 Mon Sep 17 00:00:00 2001 From: WaylandYang <145302500+WaylandYang@users.noreply.github.com> Date: Wed, 23 Sep 2026 17:27:18 +0800 Subject: [PATCH 3/3] Scope the implied-source assertion to its own knowledge base so parallel tests cannot leak into it Co-Authored-By: Claude Fable 5.1 Signed-off-by: WaylandYang <145302500+WaylandYang@users.noreply.github.com> --- crates/utopia-store/src/implication_rules_tests.rs | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/crates/utopia-store/src/implication_rules_tests.rs b/crates/utopia-store/src/implication_rules_tests.rs index 3fcd2d089..076f939eb 100644 --- a/crates/utopia-store/src/implication_rules_tests.rs +++ b/crates/utopia-store/src/implication_rules_tests.rs @@ -194,10 +194,14 @@ async fn an_approved_rule_waits_for_its_reading_then_implies_a_fact_with_evidenc let rows = implied_rows(&pool, f.kb).await?; assert_eq!(rows, vec![(f.loud_tour, f.country, Some(us))]); // 证据从触发它的陈述抄来；来源记着规则与陈述 - let src: (Uuid, Option) = - sqlx::query_as("SELECT rule_id, statement_id FROM implied_fact_sources") - .fetch_one(&pool) - .await?; + // 按库过滤：CI 上各测试并行共用一个库，别的库的来源行会被 fetch_one 先拿到 + let src: (Uuid, Option) = sqlx::query_as( + "SELECT i.rule_id, i.statement_id FROM implied_fact_sources i + JOIN facts t ON t.id = i.fact_id WHERE t.kb_id = $1", + ) + .bind(f.kb) + .fetch_one(&pool) + .await?; assert_eq!(src, (id, Some(f.statement))); // 再跑一遍什么都不动 let o = materialize::materialize(&pool, f.kb).await?;