手动本体 + Snowflake 映射 · Semantica Cookbook

本 notebook 回答了一个特定的工作流：

"我想自己设计本体——而不是让 AI 从我的表中推断——然后将 Snowflake 数据显式地映射到它上面。"

本 notebook 演示了什么

步骤	发生了什么	由谁控制
1	设计本体类和属性	你（Python 字典）
2	用具体化建模 n 元事实	你（`AssociativeClassBuilder`）
3	从 Snowflake 拉取行	Semantica `SnowflakeIngestor`
4	将列映射到本体对齐的图谱	你（显式转换）
5	验证 + 导出 OWL / SHACL	Semantica `OntologyEngine`
6	加载到三元组存储并查询	Semantica `TripletStore`

本 notebook 不做什么

不做 LLM 驱动的本体生成
不做模式内省或表到类的推断
不做"根据我的数据建议本体"

标准覆盖

特性	状态
OWL 2（Turtle / RDF-XML）	支持
SHACL 1.1 shapes	支持
SPARQL 1.1	支持
具体化 / n 元事实	通过 `AssociativeClassBuilder` 支持
SPARQL 1.2（reifier 注解、`LATERAL`）	计划中
SHACL 1.2（`sh:severity` 扩展、SHACL-AF）	计划中

# 安装 Semantica 包
!pip install -qU semantica

# 导入所需模块：Snowflake 摄取、图谱构建、本体与三元组存储
import os
from typing import Any, Dict, List

from semantica.ingest import SnowflakeIngestor
from semantica.kg.methods import build_kg
from semantica.ontology import AssociativeClassBuilder, OntologyEngine
from semantica.triplet_store import TripletStore

1步骤 1：在 Python 中手工设计本体

你显式地定义每一个类和属性。在此阶段不会从 Snowflake 读取任何内容。

属于你的设计决策： - 存在哪些类以及它们的含义 - 哪些属性是数据类型属性，哪些是对象属性 - 定义域、值域和基数约束 - 哪些属性是必需的（稍后由 SHACL 强制执行）

这个字典随你的代码一起版本化。当你的数据库模式变化时，它不会改变。

BASE_URI = "https://example.com/hr/"

# 你的本体——由你设计，而非由 Semantica 推断。
ontology: Dict[str, Any] = {
    "name": "EmploymentDomainOntology",
    "uri": f"{BASE_URI}EmploymentDomainOntology",
    "namespace": {"base_uri": BASE_URI},

    # 由你决定类的分类体系
    "classes": [
        {"name": "Person",          "uri": f"{BASE_URI}Person"},
        {"name": "Organization",    "uri": f"{BASE_URI}Organization"},
        {"name": "Role",            "uri": f"{BASE_URI}Role"},
        # EmploymentEvent 是一个具体化节点。
        # 它连接 Person + Organization + Role，并携带薪资/日期上下文。
        {"name": "EmploymentEvent", "uri": f"{BASE_URI}EmploymentEvent"},
    ],

    # 每个属性都携带完整 URI，因此 TripletStore 将其存储为 hr:<name>
    # 而不是默认的 urn:property:<name>。
    # 这确保使用 PREFIX hr: 的 SPARQL 查询与实际存储的内容匹配。
    "properties": [
        # 数据类型属性
        {"name": "name",      "uri": f"{BASE_URI}name",      "type": "datatype", "domain": "Person",          "range": "string",  "required": True},
        {"name": "legalName", "uri": f"{BASE_URI}legalName", "type": "datatype", "domain": "Organization",    "range": "string",  "required": True},
        {"name": "title",     "uri": f"{BASE_URI}title",     "type": "datatype", "domain": "Role",            "range": "string",  "required": True},
        {"name": "startDate", "uri": f"{BASE_URI}startDate", "type": "datatype", "domain": "EmploymentEvent", "range": "date"},
        {"name": "endDate",   "uri": f"{BASE_URI}endDate",   "type": "datatype", "domain": "EmploymentEvent", "range": "date"},
        {"name": "salary",    "uri": f"{BASE_URI}salary",    "type": "datatype", "domain": "EmploymentEvent", "range": "decimal"},

        # 对象属性——具体化辐条（必需）
        {"name": "employee",  "uri": f"{BASE_URI}employee",  "type": "object",   "domain": "EmploymentEvent", "range": "Person",       "required": True},
        {"name": "employer",  "uri": f"{BASE_URI}employer",  "type": "object",   "domain": "EmploymentEvent", "range": "Organization", "required": True},
        {"name": "role",      "uri": f"{BASE_URI}role",      "type": "object",   "domain": "EmploymentEvent", "range": "Role",         "required": True},

        # 快捷边——直接 person→org / person→role，无需穿越事件节点
        {"name": "worksFor",  "uri": f"{BASE_URI}worksFor",  "type": "object",   "domain": "Person",          "range": "Organization"},
        {"name": "hasRole",   "uri": f"{BASE_URI}hasRole",   "type": "object",   "domain": "Person",          "range": "Role"},
    ],
}

ontology

2步骤 2：具体化——建模 n 元事实

二元三元组的问题： 一个简单的三元组 (Alice, worksFor, Acme) 无法携带额外的上下文，例如薪资、开始日期或角色。标准 RDF 具体化和 OWL n 元模式通过引入一个中间节点来解决这个问题。

Semantica 的 AssociativeClassBuilder 是这一模式的 Pythonic API：

EmploymentEvent
    ├── employee  → Person          (required)
    ├── employer  → Organization    (required)
    ├── role      → Role            (required)
    ├── startDate → xsd:date
    ├── endDate   → xsd:date
    └── salary    → xsd:decimal

关于 SPARQL 1.1 与 SPARQL 1.2： - SPARQL 1.1（当前）： 显式地穿越事件节点——?event hr:employee ?person ; hr:salary ?salary - SPARQL 1.2（计划中）： 草案中的 reifier 注解语法允许直接将上下文附加到三元组上，而无需单独的中间节点。一旦该规范获批，Semantica 将采用它。

关于 SHACL 1.1 与 SHACL 1.2： - SHACL 1.1（当前）： 为所有 required 属性导出 sh:NodeShape + sh:PropertyShape 约束，并在加载时强制执行。 - SHACL 1.2（计划中）： sh:severity 配置扩展和 SHACL-AF 规则已在路线图上。

assoc_builder = AssociativeClassBuilder()

employment_assoc = assoc_builder.create_associative_class(
    name="EmploymentEvent",
    connects=["Person", "Organization", "Role"],
    temporal=True,  # 添加 startDate / endDate 处理
    properties={
        "startDate": "xsd:date",
        "endDate":   "xsd:date",
        "salary":    "xsd:decimal",
    },
)

validation_result = assoc_builder.validate_associative_class(employment_assoc)

# AssociativeClass 是一个数据类——使用属性访问，而不是 .get()
print("AssociativeClass structure:")
print(f"  name:       {employment_assoc.name}")
print(f"  connects:   {employment_assoc.connects}")
print(f"  temporal:   {employment_assoc.temporal}")
print(f"  properties: {list(employment_assoc.properties.keys())}")
print(f"\nValidation passed: {validation_result}")

3步骤 3：摄取 Snowflake 行（仅抽取）

SnowflakeIngestor 检索行——仅此而已。它不会： - 检查你的表模式 - 建议类或属性 - 从列名推断关系

设置 USE_LIVE_SNOWFLAKE=true 以及下面的环境变量以连接到真实的数据仓库。否则将使用桩数据。

# 从 Snowflake 摄取雇佣事实行；未启用实时连接时返回桩数据
def fetch_rows_from_snowflake() -> List[Dict[str, Any]]:
    # 未设置 USE_LIVE_SNOWFLAKE 时返回桩数据
    if os.getenv("USE_LIVE_SNOWFLAKE", "false").lower() != "true":
        return [
            {
                "EMPLOYEE_ID":   "E100",
                "EMPLOYEE_NAME": "Alice Johnson",
                "ORG_ID":        "O10",
                "ORG_NAME":      "Acme Corp",
                "ROLE_ID":       "R7",
                "ROLE_TITLE":    "Senior Engineer",
                "START_DATE":    "2025-01-15",
                "END_DATE":      None,
                "SALARY":        160000,
            },
            {
                "EMPLOYEE_ID":   "E101",
                "EMPLOYEE_NAME": "Bob Singh",
                "ORG_ID":        "O10",
                "ORG_NAME":      "Acme Corp",
                "ROLE_ID":       "R9",
                "ROLE_TITLE":    "Data Architect",
                "START_DATE":    "2024-09-01",
                "END_DATE":      None,
                "SALARY":        185000,
            },
        ]

    # 连接真实 Snowflake 数据仓库
    ingestor = SnowflakeIngestor(
        account=os.getenv("SNOWFLAKE_ACCOUNT"),
        user=os.getenv("SNOWFLAKE_USER"),
        password=os.getenv("SNOWFLAKE_PASSWORD"),
        warehouse=os.getenv("SNOWFLAKE_WAREHOUSE"),
        database=os.getenv("SNOWFLAKE_DATABASE"),
        schema=os.getenv("SNOWFLAKE_SCHEMA", "PUBLIC"),
    )
    query = (
        "SELECT EMPLOYEE_ID, EMPLOYEE_NAME, "
        "ORG_ID, ORG_NAME, ROLE_ID, ROLE_TITLE, "
        "START_DATE, END_DATE, SALARY "
        "FROM HR_EMPLOYMENT_FACT"
    )
    # 执行查询并返回结果行
    data = ingestor.ingest_query(query)
    ingestor.close()
    return data.data


# 摄取行数据并预览前两行
rows = fetch_rows_from_snowflake()
rows[:2]

4步骤 4：将行显式映射到本体概念

这是语义转换层——让你的本体成为现实的部分。

Semantica 不会猜测哪一列变成哪个实体或属性。每一项赋值都是你编写并拥有的代码：

稳定的节点 ID——确定性的、防冲突的，从业务键派生
类赋值——与你在步骤 1 中声明的内容匹配
属性路由——每个列值都进入正确的本体属性
具体化接线——EmploymentEvent 被链接到它的三个参与者

当你的 Snowflake 模式变化时，只有这个函数需要更新。本体保持稳定。

def map_rows_to_kg(rows: List[Dict[str, Any]]) -> Dict[str, Any]:
    entities: Dict[str, Dict[str, Any]] = {}
    relationships: List[Dict[str, Any]] = []

    for row in rows:
        # 从业务键派生的稳定、确定性的节点 ID
        person_id = f"person:{row['EMPLOYEE_ID']}"
        org_id    = f"org:{row['ORG_ID']}"
        role_id   = f"role:{row['ROLE_ID']}"
        # 事件 ID 包含全部三个参与者 + 开始日期，这样
        # 重新雇用的员工会得到一个不同的事件节点，而不是被覆盖。
        event_id  = f"employment:{row['EMPLOYEE_ID']}:{row['ORG_ID']}:{row['START_DATE']}"

        # 实体——"type" 必须与步骤 1 中的类名匹配
        entities[person_id] = {
            "id": person_id,
            "type": "Person",
            "properties": {"name": row["EMPLOYEE_NAME"]},
        }
        entities[org_id] = {
            "id": org_id,
            "type": "Organization",
            "properties": {"legalName": row["ORG_NAME"]},
        }
        entities[role_id] = {
            "id": role_id,
            "type": "Role",
            "properties": {"title": row["ROLE_TITLE"]},
        }

        # 具体化节点——过滤掉 None 值，这样 TripletStore 不会
        # 将 None 字符串化为字面量 "None"（针对开放式任职）。
        event_props = {
            "startDate": row["START_DATE"],
            "endDate":   row["END_DATE"],
            "salary":    row["SALARY"],
        }
        entities[event_id] = {
            "id": event_id,
            "type": "EmploymentEvent",
            "properties": {k: v for k, v in event_props.items() if v is not None},
        }

        # 关系类型使用完整 URI，这样 TripletStore 存储 hr:<type>
        # 而不是默认的 urn:property:<type>，保持 SPARQL 一致。
        relationships.extend([
            # 快捷边——在不需要上下文时实现快速 SPARQL
            {"source": person_id, "target": org_id,    "type": f"{BASE_URI}worksFor"},
            {"source": person_id, "target": role_id,   "type": f"{BASE_URI}hasRole"},
            # 具体化辐条——通过事件节点获得完整上下文
            {"source": event_id,  "target": person_id, "type": f"{BASE_URI}employee"},
            {"source": event_id,  "target": org_id,    "type": f"{BASE_URI}employer"},
            {"source": event_id,  "target": role_id,   "type": f"{BASE_URI}role"},
        ])

    return build_kg([{"entities": list(entities.values()), "relationships": relationships}])


kg = map_rows_to_kg(rows)
print(f"Entities built:      {len(kg.get('entities', []))}")
print(f"Relationships built: {len(kg.get('relationships', []))}")

sample = next((e for e in kg["entities"] if e["type"] == "EmploymentEvent"), None)
print(f"\nSample EmploymentEvent node: {sample}")

5步骤 5：验证本体并导出 OWL + SHACL

OntologyEngine 验证你的本体字典并将其序列化为符合标准的文件。

输出文件： - employment_manual_ontology.ttl — OWL 2 Turtle - employment_manual_shapes.ttl — SHACL 1.1 节点和属性 shapes

标准状态：

标准	Semantica 支持
SPARQL 1.1	完整
SHACL 1.1（`sh:NodeShape`、`sh:PropertyShape`、`sh:minCount`、`sh:datatype`、`sh:class`）	完整
SPARQL 1.2（reifier 注解语法、`LATERAL`）	已跟踪 — 尚未实现
SHACL 1.2（`sh:severity` 配置、SHACL-AF 扩展）	已跟踪 — 尚未实现

# 验证本体并导出 OWL + SHACL 标准文件
engine = OntologyEngine(base_uri=BASE_URI)

# 验证本体并生成 OWL / SHACL 序列化
validation = engine.validate(ontology)
owl_ttl    = engine.to_owl(ontology,   format="turtle")
shacl_ttl  = engine.to_shacl(ontology, format="turtle")

# 导出为符合标准的文件
engine.export_owl(ontology,   "employment_manual_ontology.ttl", format="turtle")
engine.export_shacl(ontology, "employment_manual_shapes.ttl",   format="turtle")

print(f"Ontology valid:      {validation.valid}")
print(f"Ontology consistent: {validation.consistent}")
print(f"OWL output:          {len(owl_ttl):,} chars → employment_manual_ontology.ttl")
print(f"SHACL output:        {len(shacl_ttl):,} chars → employment_manual_shapes.ttl")

# 预览 SHACL shapes 的前 20 行
print("\n--- SHACL shapes (first 20 lines) ---")
print("\n".join(shacl_ttl.splitlines()[:20]))

6最佳实践架构

┌──────────────────────────────────────┐
│  本体即代码（Python 字典）          │  ← 与你的应用一起版本化
│  + 用于 n 元的 AssociativeClass      │
└───────────────┬──────────────────────┘
                │ 验证 + 导出
                ▼
┌───────────────────────────────────────┐
│  OWL 2 Turtle  │  SHACL 1.1          │  ← 符合标准的产物
└───────────────┬───────────────────────┘
                │
                ▼
┌──────────────────────────────────────┐
│  Snowflake — 原始数据访问            │  ← 无模式内省
└───────────────┬──────────────────────┘
                │ 显式映射层
                ▼
┌──────────────────────────────────────┐
│  本体对齐的知识图谱                  │  ← 类型、ID、边与步骤 1 匹配
└───────────────┬──────────────────────┘
                │ 可选
                ▼
┌──────────────────────────────────────┐
│  三元组存储 + SPARQL 1.1            │
└──────────────────────────────────────┘

为什么这种拆分很重要： 如果 Semantica 从你的 Snowflake 模式推断本体，那么每次模式迁移都可能冒着悄悄改变你的语义模型的风险。采用这种模式，模式变化只会触及步骤 4 中的映射函数——本体保持稳定并处于你的控制之下。

7SPARQL 查询模式

因为我们同时编写了快捷边和具体化辐条，所以有两种查询风格可用。

简单查找——快捷边（无需上下文）

# SPARQL 查询：查找人员及其所属组织
PREFIX hr: <https://example.com/hr/>

SELECT ?personName ?orgName
WHERE {
    ?person  a hr:Person ;
             hr:name     ?personName ;
             hr:worksFor ?org .
    ?org     hr:legalName ?orgName .
}

上下文查找——通过具体化节点（薪资、日期、角色）

# SPARQL 查询：查找雇佣事件详情
PREFIX hr: <https://example.com/hr/>

SELECT ?personName ?roleTitle ?salary ?startDate
WHERE {
    ?event   a            hr:EmploymentEvent ;
             hr:employee  ?person ;
             hr:role      ?role ;
             hr:salary    ?salary ;
             hr:startDate ?startDate .
    ?person  hr:name   ?personName .
    ?role    hr:title  ?roleTitle .
}
ORDER BY DESC(?salary)

未来：SPARQL 1.2 reifier 语法

SPARQL 1.2 草案引入了注解语法，允许你将上下文直接附加到三元组上，而无需单独的中间节点。一旦该规范获批，Semantica 将采用它，上面的上下文查询或许可以更简洁地表达。

8步骤 6（可选）：加载到三元组存储并运行 SPARQL

设置 STORE_TO_TRIPLET=true 以将知识图谱加载到实时三元组存储中，并运行上下文具体化查询。

if os.getenv("STORE_TO_TRIPLET", "false").lower() == "true":
    store = TripletStore(
        backend=os.getenv("TRIPLET_BACKEND", "blazegraph"),
        endpoint=os.getenv("TRIPLET_ENDPOINT", "http://localhost:9999/blazegraph"),
        namespace=os.getenv("TRIPLET_NAMESPACE", "kb"),
    )
    store_result = store.store(knowledge_graph=kg, ontology=ontology)
    print("Store result:", store_result)

    # 上下文具体化查询——通过 EmploymentEvent 获取 person + role + salary
    query = """
        PREFIX hr: <https://example.com/hr/>

        SELECT ?personName ?roleTitle ?salary ?startDate
        WHERE {
            ?event   a            hr:EmploymentEvent ;
                     hr:employee  ?person ;
                     hr:role      ?role ;
                     hr:salary    ?salary ;
                     hr:startDate ?startDate .
            ?person  hr:name   ?personName .
            ?role    hr:title  ?roleTitle .
        }
        ORDER BY DESC(?salary)
        LIMIT 10
    """
    result = store.execute_query(query)
    print(result)
else:
    print("Skipping triplet-store load/query (set STORE_TO_TRIPLET=true to enable)")