Graph Store 模块 · Semantica Cookbook

1概览

Graph Store 模块为属性图数据库提供了一个统一接口。它支持多种后端（Neo4j、FalkorDB），并提供用于存储、查询和分析图数据的全面功能。

关键特性

多后端支持：Neo4j（企业版）、FalkorDB（基于 Redis）
完整 CRUD 操作：创建、读取、更新、删除节点和关系
Cypher 查询语言：借助 OpenCypher 支持执行复杂的图查询
图分析：内置中心性、社区检测、路径查找等算法
批量操作：带进度跟踪的优化批量数据加载
事务支持：具备回滚能力的 ACID 事务
索引管理：创建和管理索引以优化性能
便捷函数：面向常见操作的简单函数式 API

学习目标

学完本 notebook 后，你将能够：

使用不同后端初始化和配置 GraphStore
对节点和关系执行 CRUD 操作
执行 Cypher 查询以完成复杂的图操作
使用图分析算法（最短路径、邻居、中心性）
更新和删除图数据
使用批量操作进行高效的数据加载
使用便捷函数和配置管理
为你的用例选择合适的后端

2安装

核心安装

# 安装 Semantica
pip install semantica

# 或安装所有可选依赖
pip install semantica[all]

后端特定依赖

# 用于 Neo4j（需要 Neo4j 服务器）
pip install neo4j

# 用于 FalkorDB（需要 Redis/FalkorDB 服务器）
pip install falkordb

Docker 设置（可选）

对于 FalkorDB，你可以在 Docker 中运行它：

# 启动 FalkorDB 容器（端口 6379/3000）
docker run -p 6379:6379 -p 3000:3000 -it --rm \
  -v ./data:/var/lib/falkordb/data \
  falkordb/falkordb

3后端对比

后端	最适合	部署方式	特性
Neo4j	企业应用、生产系统	服务器/云	完整 Cypher、APOC 过程、多数据库
FalkorDB	LLM 应用、实时系统、高性能	基于 Redis	超快、稀疏矩阵运算

建议：企业生产系统使用 Neo4j，高性能实时应用使用 FalkorDB。

# 安装 Semantica
!pip install semantica

4步骤 1：初始化 Graph Store

使用你偏好的后端初始化一个 GraphStore 实例。在本教程中，我们将使用 Neo4j（需要运行中的服务器）。

from semantica.graph_store import GraphStore

# Neo4j AuraDB 连接详情
# 将这些值替换为你实际的 AuraDB 凭据
store = GraphStore(
    backend="neo4j",
    uri="Your URI",  # 你的 AuraDB 实例 URI
    user="neo4j",
    password="Your Password"  # 请在此输入你的密码
)

# 连接到数据库
store.connect()
print("Connected to graph database successfully!")

5步骤 2：节点操作

创建节点

节点代表图中的实体。每个节点可以具有： - 标签：类别/类型（例如 Person、Company、Location） - 属性：键值对（例如 {"name": "Alice", "age": 30}）

# 创建带标签和属性的单个节点
apple = store.create_node(
    labels=["Company"],
    properties={"name": "Apple Inc.", "founded": 1976, "industry": "Technology"}
)
print(f"Created company node: {apple.get('properties', {}).get('name')} (ID: {apple.get('id')})")

tim_cook = store.create_node(
    labels=["Person"],
    properties={"name": "Tim Cook", "title": "CEO", "age": 63}
)
print(f"Created person node: {tim_cook.get('properties', {}).get('name')} (ID: {tim_cook.get('id')})")

cupertino = store.create_node(
    labels=["Location"],
    properties={"name": "Cupertino", "state": "California", "country": "USA"}
)
print(f"Created location node: {cupertino.get('properties', {}).get('name')} (ID: {cupertino.get('id')})")

# 批量创建多个节点（对大型数据集更高效）
other_companies = store.create_nodes([
    {"labels": ["Company"], "properties": {"name": "Microsoft", "founded": 1975}},
    {"labels": ["Company"], "properties": {"name": "Google", "founded": 1998}},
    {"labels": ["Company"], "properties": {"name": "Amazon", "founded": 1994}},
])
print(f"Created {len(other_companies)} company nodes in batch")

6步骤 3：关系操作

创建关系

关系连接节点，代表实体之间的连接。每个关系具有： - 类型：关系类型（例如 CEO_OF、LOCATED_IN、KNOWS） - 属性：键值对（例如 {"since": 2011}） - 方向：从 start_node_id 到 end_node_id

# 在节点之间创建关系
ceo_rel = store.create_relationship(
    start_node_id=tim_cook["id"],
    end_node_id=apple["id"],
    rel_type="CEO_OF",
    properties={"since": 2011}
)
print(f"Created relationship: {ceo_rel.get('type')} (ID: {ceo_rel.get('id')})")

location_rel = store.create_relationship(
    start_node_id=apple["id"],
    end_node_id=cupertino["id"],
    rel_type="HEADQUARTERED_IN",
    properties={"since": 1977}
)
print(f"Created relationship: {location_rel.get('type')} (ID: {location_rel.get('id')})")

7步骤 4：查询节点和关系

检索节点

你可以按标签、属性或节点 ID 查询节点。

# 按标签获取节点
companies = store.get_nodes(labels=["Company"], limit=10)
print(f"Found {len(companies)} companies:")
for company in companies:
    name = company.get('properties', {}).get('name', 'Unknown')
    founded = company.get('properties', {}).get('founded', 'N/A')
    print(f"  - {name} (founded: {founded})")

# 按 ID 获取特定节点
if apple.get('id'):
    node = store.get_node(node_id=apple["id"])
    print(f"\nRetrieved node by ID: {node.get('properties', {}).get('name')}")

# 获取某个节点的关系
relationships = store.get_relationships(node_id=apple["id"], direction="both")
print(f"Found {len(relationships)} relationships for Apple:")
for rel in relationships:
    rel_type = rel.get('type', 'Unknown')
    props = rel.get('properties', {})
    print(f"  - {rel_type}: {props}")

# 按类型和方向获取关系
if tim_cook.get('id'):
    outgoing = store.get_relationships(
        node_id=tim_cook["id"],
        rel_type="CEO_OF",
        direction="out"
    )
    print(f"\nOutgoing CEO_OF relationships: {len(outgoing)}")

8步骤 5：Cypher 查询执行

执行 Cypher 查询

Cypher 是一种强大的图查询语言，允许你表达复杂的图模式和操作。Graph Store 模块在所有后端上支持 OpenCypher 语法。

# 执行 Cypher 查询以查找 CEO 关系
results = store.execute_query("""
    MATCH (p:Person)-[r:CEO_OF]->(c:Company)
    RETURN p.name as person, c.name as company, r.since as since
""")

print("CEO Relationships:")
for record in results.get("records", []):
    person = record.get('person', 'Unknown')
    company = record.get('company', 'Unknown')
    since = record.get('since', 'N/A')
    print(f"  - {person} is CEO of {company} since {since}")

# 参数化查询（更安全、更高效）
results = store.execute_query(
    "MATCH (c:Company) WHERE c.founded > $year RETURN c.name, c.founded ORDER BY c.founded",
    parameters={"year": 1990}
)

print("Companies founded after 1990:")
for record in results.get("records", []):
    name = record.get('c.name', 'Unknown')
    founded = record.get('c.founded', 'N/A')
    print(f"  - {name} (founded: {founded})")

9步骤 6：图分析

内置分析算法

Graph Store 模块提供了若干图分析算法，用于分析你的图结构。

# 获取节点的邻居（遍历图）
if apple.get('id'):
    neighbors = store.get_neighbors(
        node_id=apple["id"],
        direction="both",
        depth=2
    )

    print(f"Found {len(neighbors)} neighbors (up to depth 2) for Apple:")
    for neighbor in neighbors:
        name = neighbor.get('properties', {}).get('name', 'Unknown')
        labels = neighbor.get('labels', [])
        print(f"  - {name} ({', '.join(labels)})")

# 查找两个节点之间的最短路径
if tim_cook.get('id') and cupertino.get('id'):
    path = store.shortest_path(
        start_node_id=tim_cook["id"],
        end_node_id=cupertino["id"],
        max_depth=5
    )

    if path:
        print(f"Shortest path found:")
        print(f"  - Path length: {path.get('length')}")
        print(f"  - Nodes in path: {len(path.get('nodes', []))}")
        print(f"  - Relationships: {len(path.get('relationships', []))}")
    else:
        print("No path found between the nodes")

10步骤 7：更新和删除操作

更新节点

你可以使用 update_node 方法更新节点属性。

# 更新节点属性（合并模式 - 添加/更新属性）
if tim_cook.get('id'):
    updated = store.update_node(
        node_id=tim_cook["id"],
        properties={"age": 64, "title": "CEO & President"},
        merge=True  # 与现有属性合并
    )
    print(f"Updated node: {updated.get('properties', {}).get('name')}")
    print(f"  New age: {updated.get('properties', {}).get('age')}")
    print(f"  New title: {updated.get('properties', {}).get('title')}")

# 示例：替换所有属性（merge=False）
# updated = store.update_node(
#     node_id=node_id,
#     properties={"name": "New Name"},
#     merge=False  # 替换所有属性
# )

11步骤 8：删除操作

删除节点和关系

你可以在需要时删除节点和关系。

# 删除一个关系
if location_rel.get('id'):
    deleted = store.delete_relationship(rel_id=location_rel["id"])
    if deleted:
        print(f"Deleted relationship (ID: {location_rel['id']})")

# 删除一个节点（使用 detach=True 同时删除其关系）
# 警告：这将删除该节点及其所有关系
# 取消注释以测试：
# if cupertino.get('id'):
#     deleted = store.delete_node(node_id=cupertino["id"], detach=True)
#     if deleted:
#         print(f"Deleted node: {cupertino.get('properties', {}).get('name')}")

print("\nTip: Use detach=True to delete a node and all its relationships")
print("     Use detach=False to only delete the node (fails if relationships exist)")

12步骤 9：图统计信息

获取关于你的图的全面统计信息。

# 获取全面的图统计信息
stats = store.get_stats()

print("Graph Statistics:")
print(f"  Total nodes: {stats.get('node_count', 'N/A')}")
print(f"  Total relationships: {stats.get('relationship_count', 'N/A')}")
print(f"\nNode labels:")
for label, count in stats.get('label_counts', {}).items():
    print(f"  - {label}: {count} nodes")
print(f"\nRelationship types:")
for rel_type, count in stats.get('relationship_type_counts', {}).items():
    print(f"  - {rel_type}: {count} relationships")

13步骤 10：便捷函数

Graph Store 模块提供了便捷函数，用于更简单的、基于函数的操作。

# 使用便捷函数（类方法的替代方案）
from semantica.graph_store import (
    create_node,
    create_relationship,
    get_nodes,
    execute_query,
    shortest_path
)

# 这些函数使用一个默认的 store 实例
# 在本示例中，我们将继续使用我们创建的 store 实例

# 示例：使用便捷函数
# node = create_node(
#     labels=["Person"],
#     properties={"name": "Alice", "age": 30}
# )

print("Convenience functions available:")
print("  - create_node, create_nodes")
print("  - create_relationship, create_relationships")
print("  - get_nodes, get_relationships")
print("  - update_node, delete_node")
print("  - execute_query, shortest_path, get_neighbors")
print("  - run_analytics")

14步骤 11：索引管理

创建索引以提升查询性能，尤其是对于大型图。

# 在节点属性上创建索引以加快查找
# 这对于频繁查询的属性尤其有用

index_created = store.create_index(
    label="Company",
    property_name="name",
    index_type="btree"  # 默认索引类型
)

if index_created:
    print("Created index on Company.name for faster queries")
else:
    print("Index may already exist or not be supported by this backend")

# 注意：索引创建的支持因后端而异
# Neo4j：对各种索引类型提供完整支持
# FalkorDB：索引支持有限

15步骤 12：清理

完成后始终关闭连接以释放资源。

# 关闭连接
store.close()
print("Connection closed successfully")

16小结

本 notebook 介绍了 Graph Store 模块，这是一个支持 Neo4j 和 FalkorDB 的属性图数据库统一接口。

你学到了什么

CRUD 操作：创建、读取、更新和删除节点和关系
Cypher 查询：使用 OpenCypher 语法执行复杂的图查询
图分析：最短路径、邻居遍历和中心性算法
批量操作：面向大型数据集的高效批量数据加载
索引管理：通过索引进行性能优化

关键要点

后端选择：生产环境使用 Neo4j，高性能应用使用 FalkorDB
最佳实践：使用批量操作、参数化查询和正确的连接管理
下一步：探索高级分析、图质量和可视化模块