neo4j-aura-graph-analytics-skill

适用场景

• 在 Aura Graph Analytics GDS 会话中运行 GDS 算法
• 创建 GdsSessions 或使用 AuraGraphDataScience
• 使用 gds.graph.project.remote(...) 远程投影已连接的 Neo4j 数据
• 使用 AuraDB Cypher API 投影，并指定 { memory: ... } 或 { sessionId: ... }
• 处理来自非 Neo4j 来源的图数据（Pandas、Spark、CSV）
• 按需／管道工作负载——临时会话，按会话分钟计费
• 在分析期间与实时数据库完全隔离

不适用场景

• 启用了嵌入式 GDS 插件的 Aura Pro → neo4j-gds-skill
• 启用了嵌入式 GDS 插件的自托管 Neo4j → neo4j-gds-skill
• 编写 Cypher 查询 → neo4j-cypher-skill
• Snowflake Graph Analytics → neo4j-snowflake-graph-analytics-skill

---

部署决策表

部署方式	使用
AuraDB Free	此技能 — 最大 m_2GB，1 个并发会话，不计费
Aura Pro + 启用 Graph Analytics 插件（轻量级探索，共享资源）	neo4j-gds-skill
Aura Pro / Pro Trial + 会话（隔离计算）	此技能 — 最高 128 GB（Pro）／8 GB（Pro Trial），100／3 个并发会话
AuraDB + Python 客户端会话	此技能
AuraDB + Cypher API	此技能用于 AGA 专属投影／会话说明；查询编写使用 neo4j-cypher-skill
自托管 Neo4j + AGA 会话	此技能
自托管 Neo4j + 嵌入式插件	neo4j-gds-skill
非 Neo4j 数据（Pandas、Spark）	此技能（独立模式）

---

默认设置

• 必须使用 graphdatascience >= 2.0；建议使用 >= 2.2
• 2.0 端点：不带 v2 前缀 — gds.page_rank.*、gds.graph.node_properties.*、gds.graph.construct(...)
• 全程使用 snake_case 参数
• 创建会话后调用 gds.verify_connectivity() — 验证会话；如果已连接，还会验证源数据库
• 大型会话前估算内存
• 设置 TTL；默认空闲时长为 1h，最长 7d（硬性上限为 7 天）
• 完成后关闭会话：gds.delete() 或 sessions.delete(session_name=...)，即可停止计费
• 使用 AuraAPICredentials.from_env() 和 DbmsConnectionInfo.from_env() — 切勿将凭据硬编码

---

安装

pip install "graphdatascience>=2.2"     # 2.2 is the current stable release

2.0–2.2 的要求：Python >= 3.10、neo4j 驱动 5.26–7.0、pandas 2–3、pyarrow 21–25、numpy <3。

客户端 1.x（旧版）

2.0 对客户端进行了重命名和重新组织。固定到 1.22 版本（graphdatascience<2）？映射如下：

1.x	2.0
gds.v2.<endpoint>	gds.<endpoint> — 移除了 v2 前缀；移除了 1.x 中无类型的端点
gds.graph.project(graph_name, query)（远程）	gds.graph.project.cypher(graph_name, query)
gds.graph.project_native(...)	gds.graph.project.native(...)
GraphV2 / ModelV2	Graph / Model — from graphdatascience import Graph
Graph.drop(failIfMissing=) / Model.drop(failIfMissing=)	fail_if_missing=
gds.v2.verify_session_connectivity() / gds.v2.verify_db_connectivity()	gds.verify_connectivity() — 1.x 中也已存在；移除了 v2 命名空间
run_cypher(..., retryable=)	已移除 — 始终重试
gds.graph.project.cypher(database=...)	已移除 — 投影前调用 gds.set_database(...)
gds.graph.node_labels.mutate(write_concurrency=, job_id=)	已移除这些参数
ArrowEndpointVersion.from_arrow_info	check_version_compatibility

迁移指南：Neo4j GDS Python 客户端 2.0 迁移

2.0 新增内容：GdsSessions.estimate(algorithms=[...]) 按算法估算内存；GdsSessions.get_or_create(show_progress=...)；仅限关键字参数的 GdsSessions.delete(session_name=|session_id=) 在未删除任何内容时返回 False；在 gds.graph.project / generate / construct / filter / sample 中设置 overwrite=True 会先删除同名图；gds.graph.drop(...) 可接受多个图 → list[GraphInfo]。

2.1 新增内容：gds.run_cypher(query, auto_commit=True)，用于 CALL { … } IN TRANSACTIONS（2.0 默认的可重试事务不支持此语法）；gds.db_driver() → 会话客户端的托管 neo4j.Driver（由 gds.close() 关闭）；QueryMode 接受 "READ" / "WRITE" 字符串。

---

关键模式

第 1 步 — 身份验证

from graphdatascience.session import AuraAPICredentials, GdsSessions

sessions = GdsSessions(api_credentials=AuraAPICredentials.from_env())
# Reads: AURA_CLIENT_ID, AURA_CLIENT_SECRET, AURA_PROJECT_ID (optional)
# Create API credentials in Aura Console → Account → API credentials

属于多个项目或组织的成员：设置 AURA_PROJECT_ID 或传入 project_id= — 2.1 在推导默认项目时会优先检查组织。

第 2 步 — 估算内存

from graphdatascience.session import AlgorithmCategory, SessionMemory

# Per-algorithm + config — preferred
memory = sessions.estimate(
    node_count=1_000_000,
    relationship_count=5_000_000,
    algorithms=["wcc", "louvain", "fast_rp"],
)
# or with config:
memory = sessions.estimate(
    node_count=1_000_000,
    relationship_count=5_000_000,
    algorithms={"fast_rp": {"embedding_dimension": 128}},
)
# Coarse category estimate — 1.x style, still available
memory = sessions.estimate(
    node_count=1_000_000,
    relationship_count=5_000_000,
    algorithm_categories=[
        AlgorithmCategory.CENTRALITY,
        AlgorithmCategory.NODE_EMBEDDING,
        AlgorithmCategory.COMMUNITY_DETECTION,
    ],
)
# Returns SessionMemory tier, e.g. SessionMemory.m_8GB
# Fixed tiers: m_2GB … m_512GB — see references/limitations.md

第 3 步 — 创建会话

模式 A — 已连接 AuraDB：

from graphdatascience.session import DbmsConnectionInfo, SessionMemory, CloudLocation
from datetime import timedelta

# Reads: AURA_INSTANCEID (takes precedence) or NEO4J_URI, plus NEO4J_USERNAME,
# NEO4J_PASSWORD, NEO4J_DATABASE
db_connection = DbmsConnectionInfo.from_env()
# Explicit: DbmsConnectionInfo(aura_instance_id=..., username=..., password=...)

gds = sessions.get_or_create(
    session_name="my-analysis",
    memory=memory,
    db_connection=db_connection,
    ttl=timedelta(hours=2),
)
gds.verify_connectivity()

模式 B — 自行管理的 Neo4j：

# Same from_env() — set NEO4J_URI (e.g. "bolt://my-server:7687"), no AURA_INSTANCEID
gds = sessions.get_or_create(
    session_name="my-analysis-sm",
    memory=SessionMemory.m_8GB,
    db_connection=DbmsConnectionInfo.from_env(),
    ttl=timedelta(hours=2),
    cloud_location=CloudLocation("gcp", "europe-west1"),
)
gds.verify_connectivity()

模式 C — 独立运行（无 Neo4j 数据库）：

gds = sessions.get_or_create(
    session_name="my-standalone",
    memory=SessionMemory.m_4GB,
    ttl=timedelta(hours=1),
    cloud_location=CloudLocation("gcp", "europe-west1"),
)
gds.verify_connectivity()

get_or_create() 是幂等的；会按名称重新连接到现有会话。

Step 4 — 项目图

从已连接的 Neo4j 投影（远程投影）：

query = """
    CALL () {
        MATCH (p:Person)
        OPTIONAL MATCH (p)-[r:KNOWS]->(p2:Person)
        RETURN p AS source, r AS rel, p2 AS target,
               p {.age, .score} AS sourceNodeProperties,
               p2 {.age, .score} AS targetNodeProperties
    }
    RETURN gds.graph.project.remote(source, target, {
        sourceNodeLabels:     labels(source),
        targetNodeLabels:     labels(target),
        sourceNodeProperties: sourceNodeProperties,
        targetNodeProperties: targetNodeProperties,
        relationshipType:     type(rel)
    })
"""

G, result = gds.graph.project.cypher(
    graph_name="my-graph",
    query=query,
    undirected_relationship_types=["KNOWS"],
)
print(f"Projected {G.node_count()} nodes, {G.relationship_count()} relationships")

多模式 MATCH 必须使用 CALL () { ... }。多个标签或关系类型请在 CALL 中使用 UNION。 远程查询必须使用 gds.graph.project.remote(...)；图名称应传给 gds.graph.project.cypher(...)，而不是查询。包含 gds.graph.project 但未包含 .remote 的查询会自动重写，并发出警告。在查询中使用 undirectedRelationshipTypes / inverseIndexedRelationshipTypes 会导致 ValueError——请将它们作为方法参数传入。 只有数值型节点属性可以投影到会话中；流式读取时，请通过 db_node_properties 获取字符串属性。 独立会话无法进行远程投影——会抛出 ValueError: Remote projection is only supported for attached Sessions. 1.x 的备用方案：gds.graph.project(graph_name=..., query=...)。

原生远程投影（无需 Cypher 查询） — gds.graph.project.native(...) 会根据标签／类型筛选条件，从已连接的数据库进行投影：

G, result = gds.graph.project.native(
    "my-graph",
    ["Person"],                              # node_label_filter
    ["KNOWS"],                               # relationship_type_filter
    node_properties=["age", "score"],
    undirected_relationship_types=["KNOWS"],
)

仅适用于已连接的会话。对于按标签／类型筛选的投影，请使用 project.native；对于转换、计算属性或 UNION 异构模式，请使用 project.cypher。

AuraDB Cypher API 投影：

CYPHER runtime=parallel
MATCH (source)
OPTIONAL MATCH (source)-->(target)
RETURN gds.graph.project(
  'my-graph',
  source,
  target,
  {},
  { memory: '2GB' }
)

现有显式会话：

CYPHER runtime=parallel
MATCH (source)
OPTIONAL MATCH (source)-->(target)
RETURN gds.graph.project(
  'my-graph',
  source,
  target,
  {},
  { sessionId: '00000000-11111111' }
)

Cypher API 使用 gds.graph.project(...)，而非 gds.graph.project.remote(...)。请将 memory、ttl、sessionId、batchSize 放入第五个配置参数中。

通过 Cypher API 管理会话：

CALL gds.session.getOrCreate('test-session', '2GB', duration({minutes: 30}))
YIELD id, name, status
RETURN id, name, status

CALL gds.session.list()
YIELD id, name, status, memory
RETURN id, name, status, memory

当隐式 Cypher API 会话中的所有投影图都被删除时，该会话也会随之删除。

从 Pandas DataFrame 投影（独立模式）：

import pandas as pd

nodes_df = pd.DataFrame([
    {"nodeId": 0, "labels": "Person", "age": 30},
    {"nodeId": 1, "labels": "Person", "age": 25},
])
rels_df = pd.DataFrame([
    {"sourceNodeId": 0, "targetNodeId": 1, "relationshipType": "KNOWS"},
])

G = gds.graph.construct("my-graph", [nodes_df], [rels_df])

必需列——节点：nodeId（int）、labels（str）。关系：sourceNodeId、targetNodeId、relationshipType。在调用 construct() 之前，请删除字符串型节点属性——会话仅接受数值型属性。

Step 5 — 运行算法

# Mutate — chain results without writing to DB
gds.page_rank.mutate(G, mutate_property="pagerank", damping_factor=0.85)
gds.fast_rp.mutate(G,
    mutate_property="embedding",
    embedding_dimension=128,
    feature_properties=["pagerank"],
    random_seed=42,
)

# Stream — inspect results as DataFrame
df = gds.page_rank.stream(G)
print(df.sort_values("score", ascending=False).head(10))

# Write — persist to connected Neo4j DB (connected modes only)
gds.louvain.write(G, write_property="community")

ML 管道：gds.pipeline.node_classification / link_prediction / node_regression——这是 2.0 中唯一可用的 API。 1.x 的备用方案：gds.v2.page_rank.mutate(...)；未指定类型的 1.x 端点（如 gds.pageRank.mutate(...)）在 2.0 中已移除。 插件算法参考 → neo4j-gds-skill；AGA 的限制有所不同。

Step 6 — 异步作业轮询

长时间运行的算法——非阻塞的 compute() 会返回一个 JobHandle：

import time

job = gds.page_rank.compute(G, damping_factor=0.85)   # algorithm config only — no mutate/write args
while not job.done():
    time.sleep(5)
    print(f"Job status: {job.status().status} {job.status().progress_percent()}")
if not job.status().succeeded():                       # status string "done" / "aborted" / "running"
    raise RuntimeError(f"Algorithm job failed: {job.status().status}")
df = job.stream(wait=False)                            # raises JobNotFinishedError if not done
# or: job.mutate(mutate_property="pagerank")  /  job.write(write_properties="pagerank") → WriteJobHandle

句柄方法：.job_id()、.status() → JobStatus（.status、.progress、.succeeded()、.aborted()、.running()）、.done()、.wait(*, termination_flag=None)、.cancel()、.summary(wait=True)、.stream(wait=True)、.mutate(mutate_property=..., wait=True)、.write(write_properties=...)。

GraphSAGE 预测作为作业 [2.2]：gds.graph_sage.compute(G, model_name)（经典模型）、gds.graph_sage.supervised.compute(G, model_name, feature_properties)（添加 predicted_class、predicted_probabilities）、gds.graph_sage.unsupervised.compute(G, model_name, feature_properties)——每个都会返回 JobHandle；通过 .stream() / .mutate() / .write() 实例化结果。 异步投影会返回 ProjectionJobHandle（gds.graph.project.native_async(...)、cypher_async(...)）；写回操作会返回 WriteJobHandle。列出／恢复作业：

gds.jobs.list()                     # JobInfo per job: job_id, name
handle = gds.jobs.get(G, job_id)    # concrete handle type for the job

Step 7 — 获取结果

# Stream node properties
result_df = gds.graph.node_properties.stream(
    G,
    node_properties=["pagerank", "embedding"],
    db_node_properties=["name"],   # connected modes only — fetches string props from DB
)
result_df.head(10)

独立模式：没有 db_node_properties；请联接源 DataFrame：

result_df = gds.graph.node_properties.stream(G, ["pagerank"])
result_df.merge(nodes_df[["nodeId", "name"]], how="left")

Step 8 — 写回并清理

# Write node properties to connected Neo4j
gds.graph.node_properties.write(G, ["pagerank", "embedding"])

# Write relationship properties
gds.graph.relationships.write(G, "SIMILAR", ["score"])

# Query connected DB from session
gds.run_cypher("MATCH (n:Person) RETURN count(n)")

# Drop projected graph
gds.graph.drop(G)

# Delete session
sessions.delete(session_name="my-analysis")
# or: gds.delete()

请在删除前写回；会话关闭时，未写回的结果会丢失。

会话管理

# List active sessions
from pandas import DataFrame
DataFrame(sessions.list())

# Reconnect to existing session
gds = sessions.get_or_create(session_name="my-analysis", memory=..., db_connection=...)

---

常见错误

错误	原因	解决方法
AuthenticationError / 401	CLIENT_ID/CLIENT_SECRET 错误	在 Aura Console → Account → API credentials 中重新生成
获取已过期会话时出现 RuntimeError	超出 TTL	使用 sessions.list() 检查；重新创建会话
SessionNotFoundError	会话已过期（超出 TTL）或名称拼写错误	使用 sessions.list() 检查；重新创建会话
GraphNotFoundError	投影已删除，或会话重新连接后未重新投影	重新运行 gds.graph.project.cypher() 或 gds.graph.construct()
ValueError: Remote projection is only supported for attached Sessions.	独立会话无法执行远程投影	改用 DataFrame 执行 gds.graph.construct(...)
NotAvailableInStandaloneSessions	此功能需要连接的数据库（例如 gds.topological_link_prediction、远程投影）	连接数据库或选择其他算法
算法作业状态为 FAILED	超出内存限制或算法不受支持	增大 SessionMemory；对于仅支持连接模式的功能，请检查 NotAvailableOutsideAura
MemoryEstimationExceeded	图比估算的大	使用实际计数重新估算；选择更高的规格
会话重新连接后结果为空	会话关闭前未写入结果	始终先写入或流式传输，再调用 gds.delete()
String node properties not supported	节点 DataFrame 中包含字符串列	在执行 gds.graph.construct() 前删除字符串列；稍后通过 db_node_properties 获取字符串
AGA not enabled for project	未启用 AGA 功能	在 Aura Console → project settings 中启用

---

参考资料

按需加载：

• references/workflows.md — AuraDB 和独立模式的完整工作流示例、Spark 集成
• references/limitations.md — AGA 与嵌入式 GDS 功能对照表、SessionMemory 规格、云区域

WebFetch

需求	URL
AGA Python 客户端文档	https://neo4j.com/docs/graph-data-science-client/current/aura-graph-analytics/
AGA Cypher API 文档	https://neo4j.com/docs/graph-data-science/current/aura-graph-analytics/cypher/
客户端迁移指南 1.x → 2.0	https://neo4j.com/docs/graph-data-science-client/current/migration-from-1x/
AuraDB 教程笔记本	https://github.com/neo4j/graph-data-science-client/blob/main/examples/graph-analytics-serverless.ipynb
GDS 算法参考	https://neo4j.com/docs/graph-data-science/current/algorithms/

---

检查清单

• 已创建 Aura API 凭据并在环境中设置（AURA_CLIENT_ID、AURA_CLIENT_SECRET）
• 连接会话：已为 DbmsConnectionInfo.from_env() 设置 AURA_INSTANCEID 或 NEO4J_URI，以及 NEO4J_USERNAME、NEO4J_PASSWORD
• 已为 Aura 项目启用 AGA 功能（Aura Console → project settings）
• 创建会话前已估算内存（sessions.estimate(..., algorithms=[...])）
• 已选择靠近数据源的云区域
• 创建会话后已调用 gds.verify_connectivity()
• 远程投影使用 gds.graph.project.cypher(graph_name, query)，并在查询中调用 gds.graph.project.remote(...)
• 将远程投影图名称传递给端点，而不是远程函数
• 将 undirected_relationship_types 作为方法参数传递，切勿放入查询中
• AuraDB Cypher API 投影使用第五个配置映射来设置 memory 或 sessionId
• 显式 Cypher API 会话使用 gds.session.getOrCreate(...)；隐式会话会随投影图一同删除
• 已设置 TTL，避免空闲会话产生意外费用
• 异步算法作业已轮询至 RUNNING_DONE 后再读取结果
• 删除前已写回结果（连接模式），或已流式传输并持久化结果（独立模式）
• 完成后已删除会话（sessions.delete(session_name=...) 或 gds.delete()）