Datalog 风格推理

# 安装 Semantica 包
!pip install -qU semantica

# ── 推理 ──────────────────────────────────────────────────────────────
from semantica.reasoning import (
    DatalogReasoner,   # 原生 Datalog 不动点引擎
    DatalogFact,       # 冻结数据类：谓词 + 参数元组
    DatalogRule,       # 数据类：头 + 体（list[BodyAtom]）
    ExplanationGenerator,  # 生成自然语言解释
    InferenceResult,   # 由 ExplanationGenerator 消费的结果数据类
    Rule,              # 由 ExplanationGenerator 使用的规则数据类
    RuleType,          # 枚举：IMPLICATION | EQUIVALENCE | CONSTRAINT | TRANSFORMATION
)

# ── 知识图谱 ────────────────────────────────────────────────────────
from semantica.kg import (
    GraphBuilder,    # 从实体+关系源构建知识图谱字典
    GraphAnalyzer,   # 中心性、社区、连通性、指标
)

# ── 上下文 ────────────────────────────────────────────────────────────────
from semantica.context import ContextGraph  # 内存图谱：add_node/add_edge/find_*

print("All Semantica classes imported successfully.")

# ── 步骤 1：创建引擎 ──────────────────────────────────────────────────
dr = DatalogReasoner()

# ── 步骤 2：加载 EDB（基础事实） ───────────────────────────────────────
# 语法：predicate(constant1, constant2)  — 常量必须小写
edb_facts = [
    "parent(tom, bob)",
    "parent(bob, ann)",
    "parent(ann, pat)",
]
for f in edb_facts:
    dr.add_fact(f)

print(f"EDB loaded: {len(dr._all_facts)} ground facts")

# ── 步骤 3：添加 IDB 规则（Horn 子句） ──────────────────────────────────
# 语法：head(Vars) :- body_atom1(Vars), body_atom2(Vars).
# 变量以大写开头；末尾的 '.' 是可选的
dr.add_rule("ancestor(X, Y) :- parent(X, Y).")
dr.add_rule("ancestor(X, Y) :- parent(X, Z), ancestor(Z, Y).")  # 递归

print(f"Rules loaded: {len(dr._rules)}")

# ── 步骤 4：不动点求值 ────────────────────────────────────────────
# derive_all() 运行半朴素自底向上求值，直到不再出现新事实
all_facts: list[str] = dr.derive_all()

ancestor_strs = sorted(f for f in all_facts if f.startswith("ancestor"))
print(f"Derived {len(ancestor_strs)} ancestor facts:")
for f in ancestor_strs:
    print(" ", f)

# ── 步骤 5：查询 ──────────────────────────────────────────────────────────
# 使用 '?varname' 占位符——query() 在需要时会自动调用 derive_all()
# 返回：list[dict]  例如 [{"Y": "bob"}, {"Y": "ann"}, {"Y": "pat"}]

descendants = dr.query("ancestor(tom, ?Y)")
print("All descendants of tom:", sorted(r["Y"] for r in descendants))

ancestors_of_pat = dr.query("ancestor(?X, pat)")
print("All ancestors of pat: ", sorted(r["X"] for r in ancestors_of_pat))

all_pairs = dr.query("ancestor(?X, ?Y)")
print(f"\nAll ancestor pairs ({len(all_pairs)}):")
for row in sorted(all_pairs, key=lambda r: (r["X"], r["Y"])):
    print(f"  {row['X']:6s} → {row['Y']}")

# ── 构建一个软件依赖知识图谱 ────────────────────────────────────────
entities = [
    {"id": "pythonsdk",   "name": "Python SDK",   "type": "Component"},
    {"id": "restapi",     "name": "REST API",     "type": "Component"},
    {"id": "authservice", "name": "Auth Service", "type": "Component"},
    {"id": "database",    "name": "Database",     "type": "Component"},
    {"id": "dashboard",   "name": "Dashboard",    "type": "Component"},
    {"id": "analytics",   "name": "Analytics",    "type": "Component"},
]
relationships = [
    {"source": "pythonsdk",   "target": "restapi",     "type": "depends_on"},
    {"source": "restapi",     "target": "authservice", "type": "depends_on"},
    {"source": "authservice", "target": "database",    "type": "depends_on"},
    {"source": "dashboard",   "target": "restapi",     "type": "depends_on"},
    {"source": "dashboard",   "target": "analytics",   "type": "depends_on"},
    {"source": "analytics",   "target": "database",    "type": "depends_on"},
]

# GraphBuilder 验证、去重并打包数据
builder = GraphBuilder(merge_entities=True, resolve_conflicts=False)
kg = builder.build([{"entities": entities, "relationships": relationships}])

print(f"KG built — entities: {len(kg['entities'])}, relationships: {len(kg['relationships'])}")

# ── 在推理之前分析图谱结构 ───────────────────────────
# GraphAnalyzer 提供中心性、社区、连通性和指标
analyzer = GraphAnalyzer()
metrics = analyzer.compute_metrics(graph=kg)

print("Graph structure:")
print(f"  Nodes      : {metrics['num_nodes']}")
print(f"  Edges      : {metrics['num_edges']}")
if "density" in metrics:
    print(f"  Density    : {metrics['density']:.3f}")
if "is_connected" in metrics:
    print(f"  Connected  : {metrics['is_connected']}")

# ── 将知识图谱关系加载为 EDB 事实 ────────────────────────────────────
# GraphBuilder 输出的字典使用与 DatalogReasoner.add_fact()
# 原生理解的相同的 source/target/type 结构
dr = DatalogReasoner()

for rel in kg["relationships"]:
    dr.add_fact(rel)   # 字典路径：{"source": ..., "target": ..., "type": ...}

print(f"EDB loaded: {len(dr._all_facts)} dependency facts")

# ── 传递依赖闭包 ─────────────────────────────────────────
# 'depends_on' 是 add_fact 从 'type' 推断出的谓词名
dr.add_rule("transitive_dep(X, Y) :- depends_on(X, Y).")
dr.add_rule("transitive_dep(X, Y) :- depends_on(X, Z), transitive_dep(Z, Y).")

dr.derive_all()

# 传递依赖于数据库的所有组件
db_deps = sorted(r["X"] for r in dr.query("transitive_dep(?X, database)"))
print("Components that transitively depend on Database:")
for c in db_deps:
    print(" ", c)

# pythonsdk 传递依赖于什么？
sdk_chain = sorted(r["Y"] for r in dr.query("transitive_dep(pythonsdk, ?Y)"))
print(f"\nPython SDK full dependency chain: {sdk_chain}")

# ── 构建一个内存 ContextGraph ───────────────────────────────────────
# ContextGraph.add_node / add_edge 是构建内存知识图谱的规范方式
cg = ContextGraph()

# 节点
for person in ["alice", "bob", "carol", "dave", "eve"]:
    cg.add_node(person, node_type="person", name=person.capitalize())

# 有向 "follows" 边
for src, dst in [("alice", "bob"), ("bob", "carol"), ("carol", "dave"), ("alice", "eve"), ("eve", "carol")]:
    cg.add_edge(src, dst, edge_type="follows")

# 验证图谱构建正确
nodes = cg.find_nodes(node_type="person")
edges = cg.find_edges(edge_type="follows")
print(f"ContextGraph — nodes: {len(nodes)}, edges: {len(edges)}")
print("Edges:", [(e.get("source", e.get("source_id")), e.get("target", e.get("target_id"))) for e in edges])

# ── load_from_graph() 直接摄取 ContextGraph ───────────────────
dr = DatalogReasoner()
n_loaded = dr.load_from_graph(cg)  # 内部调用 cg.find_edges() + cg.find_nodes()
print(f"Facts loaded from ContextGraph: {n_loaded}")

# ── 通过传递 'follows' 实现影响力可达 ──────────────────────────────
dr.add_rule("influence(X, Y) :- follows(X, Y).")
dr.add_rule("influence(X, Y) :- follows(X, Z), influence(Z, Y).")

dr.derive_all()

# alice 能影响到谁？
alice_reach = sorted(r["Y"] for r in dr.query("influence(alice, ?Y)"))
print(f"Alice's influence reach : {alice_reach}")

# 谁能影响到 dave？
reach_dave = sorted(r["X"] for r in dr.query("influence(?X, dave)"))
print(f"Who can influence dave  : {reach_dave}")

# 完整影响力矩阵
all_influence = dr.query("influence(?X, ?Y)")
print(f"\nTotal influence pairs: {len(all_influence)}")

# ── 使用 GraphBuilder 构建 RBAC 图谱 ────────────────────────────────────
rbac_entities = [
    # 用户
    {"id": "alice",  "type": "User", "name": "Alice"},
    {"id": "bob",    "type": "User", "name": "Bob"},
    {"id": "carol",  "type": "User", "name": "Carol"},
    {"id": "dave",   "type": "User", "name": "Dave"},
    # 角色
    {"id": "admin",  "type": "Role", "name": "Administrator"},
    {"id": "editor", "type": "Role", "name": "Editor"},
    {"id": "viewer", "type": "Role", "name": "Viewer"},
    # 权限
    {"id": "read",         "type": "Permission"},
    {"id": "write",        "type": "Permission"},
    {"id": "delete",       "type": "Permission"},
    {"id": "manage_users", "type": "Permission"},
]
rbac_relationships = [
    # 用户 → 角色分配
    {"source": "alice",  "target": "admin",  "type": "has_role"},
    {"source": "bob",    "target": "editor", "type": "has_role"},
    {"source": "carol",  "target": "viewer", "type": "has_role"},
    {"source": "dave",   "target": "editor", "type": "has_role"},
    # 角色层级（admin 继承自 editor，editor 继承自 viewer）
    {"source": "admin",  "target": "editor", "type": "role_inherits"},
    {"source": "editor", "target": "viewer", "type": "role_inherits"},
    # 角色 → 权限授予
    {"source": "viewer", "target": "read",         "type": "role_has_perm"},
    {"source": "editor", "target": "write",        "type": "role_has_perm"},
    {"source": "admin",  "target": "delete",       "type": "role_has_perm"},
    {"source": "admin",  "target": "manage_users", "type": "role_has_perm"},
]

builder = GraphBuilder(merge_entities=True, resolve_conflicts=False)
rbac_kg = builder.build([{"entities": rbac_entities, "relationships": rbac_relationships}])

print(f"RBAC KG — entities: {len(rbac_kg['entities'])}, relationships: {len(rbac_kg['relationships'])}")

# ── 分析 RBAC 图谱结构 ──────────────────────────────────────────
analyzer = GraphAnalyzer()
metrics = analyzer.compute_metrics(graph=rbac_kg)
centrality = analyzer.calculate_centrality(rbac_kg, centrality_type="degree")

print(f"RBAC graph — {metrics['num_nodes']} nodes, {metrics['num_edges']} edges")
if isinstance(centrality, dict) and "degree" in centrality:
    top = sorted(centrality["degree"].items(), key=lambda x: x[1], reverse=True)[:3]
    print("Top-3 nodes by degree centrality:", top)

# ── 将 RBAC 知识图谱加载到 DatalogReasoner ────────────────────────────────────
dr = DatalogReasoner()

for rel in rbac_kg["relationships"]:
    dr.add_fact(rel)   # {source, target, type} → predicate(source, target)

# ── IDB 规则：传递角色层级 ─────────────────────────────────
dr.add_rule("effective_role(R, R2) :- role_inherits(R, R2).")
dr.add_rule("effective_role(R, R2) :- role_inherits(R, Z), effective_role(Z, R2).")

# ── IDB 规则：继承的权限 ─────────────────────────────────────
dr.add_rule("role_can(R, P) :- role_has_perm(R, P).")
dr.add_rule("role_can(R, P) :- effective_role(R, R2), role_has_perm(R2, P).")

# ── IDB 规则：用户有效权限 ────────────────────────────────
dr.add_rule("can(U, P) :- has_role(U, R), role_can(R, P).")

dr.derive_all()

print("User permissions derived via role-hierarchy inference:")
for user in ["alice", "bob", "carol", "dave"]:
    perms = sorted(r["P"] for r in dr.query(f"can({user}, ?P)"))
    print(f"  {user:6s}: {perms}")

# ── ExplanationGenerator — 可供审计的自然语言解释 ──────────────────
# ExplanationGenerator 与 InferenceResult 对象一起工作。
# 我们手动构造一个来表示推导出的 Datalog 结论。

explainer = ExplanationGenerator(detail_level="detailed")

# 构建表示权限推导链的 Rule 对象
perm_rule = Rule(
    rule_id="rbac_perm_chain",
    name="RBAC permission via role hierarchy",
    conditions=["has_role(alice, admin)", "effective_role(admin, viewer)", "role_has_perm(viewer, read)"],
    conclusion="can(alice, read)",
    rule_type=RuleType.IMPLICATION,
    confidence=1.0,
)

# 构建表示 Datalog 结论的 InferenceResult
result = InferenceResult(
    conclusion="can(alice, read)",
    rule_used=perm_rule,
    premises=[
        "has_role(alice, admin)",
        "role_inherits(admin, editor)",
        "role_inherits(editor, viewer)",
        "role_has_perm(viewer, read)",
    ],
    confidence=1.0,
)

# 生成自然语言解释
explanation = explainer.generate_explanation(result)
print("Explanation type :", explanation.explanation_type)
print("Conclusion       :", explanation.conclusion)
print("Natural language :", explanation.natural_language)
print("Reasoning steps  :", len(explanation.reasoning_path.steps))

# ── 反向查询 ───────────────────────────────────────────────────────
deleters = sorted(r["U"] for r in dr.query("can(?U, delete)"))
print("Who can delete:", deleters)

writers = sorted(r["U"] for r in dr.query("can(?U, write)"))
print("Who can write: ", writers)

# 所有 (user, permission) 对——完整策略矩阵
all_caps = dr.query("can(?U, ?P)")
print(f"\nTotal (user, permission) pairs: {len(all_caps)}")

# ── ContextGraph：组织架构图 ────────────────────────────────────────────────
org = ContextGraph()

# 将员工添加为带元数据的节点
staff = [
    ("eng1",      "engineer",  "backend"),
    ("eng2",      "engineer",  "backend"),
    ("eng3",      "engineer",  "frontend"),
    ("techlead",  "lead",      "engineering"),
    ("design1",   "designer",  "ux"),
    ("design2",   "designer",  "ux"),
    ("designlead","lead",      "design"),
    ("vpeng",     "vp",        "engineering"),
    ("cto",       "executive", "leadership"),
]
for emp_id, role, team in staff:
    org.add_node(emp_id, node_type="employee", role=role, team=team)

# 汇报关系
reports_to = [
    ("eng1", "techlead"), ("eng2", "techlead"), ("eng3", "techlead"),
    ("techlead", "vpeng"),
    ("design1", "designlead"), ("design2", "designlead"),
    ("designlead", "vpeng"),
    ("vpeng", "cto"),
]
for employee, manager in reports_to:
    org.add_edge(employee, manager, edge_type="reports_to")

# 团队归属边
teams = [
    ("eng1", "backend"),  ("eng2", "backend"),  ("eng3", "frontend"),
    ("design1", "ux"),    ("design2", "ux"),
]
for emp, team in teams:
    org.add_edge(emp, team, edge_type="in_team")
    if not org.find_nodes(node_type="team"):
        org.add_node(team, node_type="team")

print(f"ContextGraph — nodes: {len(org.find_nodes())}, edges: {len(org.find_edges())}")

# ── 将组织架构图加载到 DatalogReasoner ───────────────────────────────────
dr = DatalogReasoner()
n = dr.load_from_graph(org)   # 使用 org.find_edges() + org.find_nodes()
print(f"Facts loaded via load_from_graph(): {n}")

# ── IDB 规则 ─────────────────────────────────────────────────────────────
# 传递管理链
dr.add_rule("manages(M, E) :- reports_to(E, M).")
dr.add_rule("manages(M, E) :- reports_to(E, Z), manages(M, Z).")

# 跳级：恰好两跳汇报
dr.add_rule("skip_level(M, E) :- reports_to(E, Z), reports_to(Z, M).")

# 同一团队
dr.add_rule("same_team(X, Y) :- in_team(X, T), in_team(Y, T).")

dr.derive_all()

# ── 查询组织层级 ────────────────────────────────────────────────────
# CTO 之下的所有人
under_cto = sorted(r["E"] for r in dr.query("manages(cto, ?E)"))
print(f"CTO manages ({len(under_cto)} people): {under_cto}")

# VP Eng 的直接 + 间接下属
under_vp = sorted(r["E"] for r in dr.query("manages(vpeng, ?E)"))
print(f"VP Eng manages           : {under_vp}")

# 向 CTO 跳级汇报（CTO 之下两跳的人）
skip = sorted(r["E"] for r in dr.query("skip_level(cto, ?E)"))
print(f"CTO skip-level reports   : {skip}")

# eng1 的队友
mates = [r["Y"] for r in dr.query("same_team(eng1, ?Y)") if r["Y"] != "eng1"]
print(f"eng1's teammates         : {sorted(mates)}")

# ── 构建 InferenceResult 并解释一个组织查询 ────────────────────
explainer = ExplanationGenerator(detail_level="verbose")

mgmt_rule = Rule(
    rule_id="transitive_manages",
    name="Transitive management chain",
    conditions=["reports_to(eng1, techlead)", "manages(vpeng, techlead)"],
    conclusion="manages(vpeng, eng1)",
    rule_type=RuleType.IMPLICATION,
    confidence=1.0,
)
result = InferenceResult(
    conclusion="manages(vpeng, eng1)",
    rule_used=mgmt_rule,
    premises=["reports_to(eng1, techlead)", "reports_to(techlead, vpeng)"],
    confidence=1.0,
)

exp = explainer.generate_explanation(result)
print(exp.natural_language)

# ── 检查 DatalogRule 对象 ────────────────────────────────────────────
# dr._rules  →  List[DatalogRule]
# DatalogRule.head_predicate、.head_args、.body  (body = List[BodyAtom])
print("Rules in engine:")
for rule in dr._rules:
    body_str = ", ".join(
        f"{atom.predicate}({', '.join(atom.args)})"
        for atom in rule.body
    )
    head_str = f"{rule.head_predicate}({', '.join(rule.head_args)})"
    print(f"  {head_str}  :-  {body_str}")

# ── 检查 DatalogFact 对象 ────────────────────────────────────────────
# dr._all_facts  →  Set[DatalogFact]  (derive_all 之后 EDB + IDB 合并)
# dr._fact_index →  Dict[predicate, Set[DatalogFact]]

from collections import Counter

# 统计每个谓词的事实数量
predicate_counts = Counter(f.predicate for f in dr._all_facts)
print("Facts per predicate (EDB + derived IDB):")
for pred, count in sorted(predicate_counts.items()):
    print(f"  {pred:20s}: {count}")
print(f"\n  TOTAL: {len(dr._all_facts)}")

# ── 将 EDB 与 IDB 分开 ─────────────────────────────────────────────────
# EDB 谓词是我们通过 add_fact 添加的（不是由规则推导的）
idb_predicates = {rule.head_predicate for rule in dr._rules}
edb_predicates = {f.predicate for f in dr._all_facts} - idb_predicates

print(f"EDB predicates (base facts)  : {sorted(edb_predicates)}")
print(f"IDB predicates (derived)     : {sorted(idb_predicates)}")

# ── 示例 DatalogFact 结构 ──────────────────────────────────────────
# DatalogFact 是一个冻结数据类：predicate: str, args: Tuple[str, ...]
manages_facts = sorted(dr._fact_index.get("manages", []), key=lambda f: f.args)
print(f"First 5 'manages' DatalogFact objects ({len(manages_facts)} total):")
for fact in manages_facts[:5]:
    # 直接从数据类访问 predicate 和 args
    print(f"  DatalogFact(predicate={fact.predicate!r}, args={fact.args})")

# ── clear() 完全重置引擎 ─────────────────────────────────
print(f"Facts before clear(): {len(dr._all_facts)}")
dr.clear()
print(f"Facts after  clear(): {len(dr._all_facts)}")
print(f"Rules after  clear(): {len(dr._rules)}")

GraphBuilder.build()            → kg 字典 {entities, relationships}
                   ↓ kg["relationships"] → dr.add_fact(rel)

ContextGraph.add_node/add_edge  → 内存图谱
                   ↓ dr.load_from_graph(cg)

DatalogReasoner.add_rule()      → Horn 子句规则
DatalogReasoner.derive_all()    → 半朴素不动点
DatalogReasoner.query()         → 结果行
                   ↓ 构建 InferenceResult

ExplanationGenerator            → 自然语言解释
GraphAnalyzer                   → 推理前/后的图谱结构指标
DatalogFact / DatalogRule       → 内省引擎状态