mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-08-29 04:26:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Add Amazon Neptune Database Graph Store Support

Browse Source 
- Implement NeptuneAuthTokenManager extending Neo4j AuthManager for IAM SigV4 signing
- Add automatic token refresh and security exception handling
- Add retry logic with backoff for transient errors (signature expired, connection closed)
- Add connection recovery with driver recreation
- Add NeptuneDriver, NeptuneSession, NeptuneTransaction wrapper classes
- Use native Neptune ~id via id() function for all CRUD operations
- Add graph-amazon-neptune optional dependency group (boto3, neo4j)
- Update cookbook with Amazon Neptune Graph Store examples
- Add comprehensive tests (61 tests covering all GraphStore interface methods)

Closes #151
This commit is contained in:
Don Simpson
2026-01-08 20:13:28 -05:00
parent 9bb94c2337
commit 976a20496d
9 changed files with 4494 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -61,6 +61,7 @@ wheels/
.installed.cfg
*.egg
MANIFEST
.python-version
# IDE
.vscode/
README.md
+19 -1
View File
@@ -360,7 +360,7 @@ results = vector_store.search(query="supply chain", top_k=5)
### Graph Store & Triplet Store
> **Neo4j, FalkorDB support** • **SPARQL queries** • **RDF triplets**
> **Neo4j, FalkorDB, Amazon Neptune support** • **SPARQL queries** • **RDF triplets**
```python
from semantica.graph_store import GraphStore
@@ -370,6 +370,24 @@ from semantica.triplet_store import TripletStore
graph_store = GraphStore(backend="neo4j", uri="bolt://localhost:7687", user="neo4j", password="password")
graph_store.add_nodes([{"id": "n1", "labels": ["Person"], "properties": {"name": "Alice"}}])
# Amazon Neptune Graph Store (OpenCypher via HTTP with IAM Auth)
neptune_store = GraphStore(
backend="neptune",
endpoint="your-cluster.us-east-1.neptune.amazonaws.com",
port=8182,
region="us-east-1",
iam_auth=True, # Uses AWS credential chain (boto3, env vars, or IAM role)
)
# Node Operations
neptune_store.add_nodes([
{"labels": ["Person"], "properties": {"id": "alice", "name": "Alice", "age": 30}},
{"labels": ["Person"], "properties": {"id": "bob", "name": "Bob", "age": 25}},
])
# Query Operations
result = neptune_store.execute_query("MATCH (p:Person) RETURN p.name, p.age")
# Triplet Store (Blazegraph, Jena, RDF4J)
triplet_store = TripletStore(backend="blazegraph", endpoint="http://localhost:9999/blazegraph")
triplet_store.add_triplet({"subject": "Alice", "predicate": "knows", "object": "Bob"})
cookbook/introduction/21_Amazon_Neptune_Store.ipynb
+667
View File
@@ -0,0 +1,667 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Amazon Neptune Graph Store\n",
"\n",
"## Overview\n",
"\n",
"This notebook covers the Amazon Neptune Database integration in Semantica. Amazon Neptune is a fully managed graph database service that supports both property graphs (via OpenCypher/Gremlin) and RDF graphs (via SPARQL).\n",
"\n",
"### Key Features\n",
"\n",
"- **IAM Authentication**: Secure access using AWS SigV4 signatures via AuthManager\n",
"- **OpenCypher Support**: Query using standard OpenCypher syntax\n",
"- **Bolt Protocol**: Uses Neo4j Bolt driver for efficient binary communication\n",
"- **Native ~id Support**: Leverages Neptune's native element ID handling\n",
"- **Full CRUD Operations**: Create, read, update, delete nodes and relationships\n",
"- **Automatic Retry**: Built-in retry logic with exponential backoff for transient errors\n",
"\n",
"### Prerequisites\n",
"\n",
"- An Amazon Neptune Database cluster\n",
"- AWS credentials configured (boto3, environment variables, or IAM role)\n",
"- Network access to your Neptune cluster (VPC, security groups)\n",
"\n",
"---"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Installation\n",
"\n",
"```bash\n",
"# Install Semantica with Neptune support\n",
"pip install semantica\n",
"\n",
"# Required dependencies (installed automatically)\n",
"pip install boto3 neo4j\n",
"```"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"!pip install semantica"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Configuration\n",
"\n",
"Set your Neptune cluster endpoint and AWS credentials. Replace the placeholder values with your actual configuration."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import os\n",
"\n",
"# Neptune cluster configuration - REPLACE WITH YOUR VALUES\n",
"os.environ[\"NEPTUNE_ENDPOINT\"] = \"your-cluster.us-east-1.neptune.amazonaws.com\"\n",
"os.environ[\"NEPTUNE_PORT\"] = \"8182\"\n",
"os.environ[\"AWS_REGION\"] = \"us-east-1\"\n",
"\n",
"# AWS credentials (if using IAM Auth and not relying on IAM role or ~/.aws/credentials)\n",
"# os.environ[\"AWS_ACCESS_KEY_ID\"] = \"your-access-key-id\"\n",
"# os.environ[\"AWS_SECRET_ACCESS_KEY\"] = \"your-secret-access-key\"\n",
"# os.environ[\"AWS_SESSION_TOKEN\"] = \"your-session-token\"\n",
"\n",
"print(f\"Neptune Endpoint: {os.environ.get('NEPTUNE_ENDPOINT')}\")\n",
"print(f\"AWS Region: {os.environ.get('AWS_REGION')}\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Step 1: Initialize Neptune Store\n",
"\n",
"Initialize a connection to your Amazon Neptune cluster with IAM authentication."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import os\n",
"from semantica.graph_store import GraphStore\n",
"\n",
"# Option 1: Using GraphStore factory (recommended)\n",
"neptune_store = GraphStore(\n",
" backend=\"neptune\",\n",
" endpoint=os.environ.get(\"NEPTUNE_ENDPOINT\"),\n",
" port=int(os.environ.get(\"NEPTUNE_PORT\", 8182)),\n",
" region=os.environ.get(\"AWS_REGION\", \"us-east-1\"),\n",
" iam_auth=True,\n",
")\n",
"\n",
"# Connect to Neptune\n",
"neptune_store.connect()\n",
"print(\"Connected to Amazon Neptune!\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Development/Testing Without IAM Auth\n",
"\n",
"For development or testing environments where IAM authentication is not required (e.g., Neptune notebooks or VPC-only access), you can disable IAM signing:"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# For dev/test environments without IAM authentication\n",
"neptune_store_dev = GraphStore(\n",
" backend=\"neptune\",\n",
" endpoint=os.environ.get(\"NEPTUNE_ENDPOINT\"),\n",
" port=int(os.environ.get(\"NEPTUNE_PORT\", 8182)),\n",
" region=os.environ.get(\"AWS_REGION\", \"us-east-1\"),\n",
" iam_auth=False, # Disable IAM signing for dev/test\n",
")\n",
"neptune_store_dev.connect()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Authentication Options\n",
"\n",
"IAM Authentication (recommended for production) automatically uses the AWS credential chain:\n",
"1. Environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)\n",
"2. AWS credentials file (~/.aws/credentials)\n",
"3. IAM role (for EC2, Lambda, ECS)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Step 2: Node Operations\n",
"\n",
"### Creating Nodes\n",
"\n",
"Nodes represent entities in your graph. Each node can have:\n",
"- **ID**: A unique identifier (custom or auto-generated UUID)\n",
"- **Labels**: Categories/types (e.g., `Person`, `Company`)\n",
"- **Properties**: Key-value pairs (e.g., `{\"name\": \"Alice\", \"age\": 30}`)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Create a single node with custom ID (id in properties)\n",
"alice = neptune_store.create_node(\n",
" labels=[\"Person\"],\n",
" properties={\"id\": \"alice\", \"name\": \"Alice\", \"age\": 30, \"role\": \"Engineer\"}\n",
")\n",
"print(f\"Created node: {alice}\")\n",
"\n",
"# Create a node with auto-generated UUID (no id in properties)\n",
"bob = neptune_store.create_node(\n",
" labels=[\"Person\"],\n",
" properties={\"name\": \"Bob\", \"age\": 25, \"role\": \"Designer\"}\n",
")\n",
"print(f\"Created node with UUID: {bob['id']}\")\n",
"\n",
"# Create a company node with auto-generated ID\n",
"acme = neptune_store.create_node(\n",
" labels=[\"Company\"],\n",
" properties={\"name\": \"Acme Corp\", \"industry\": \"Technology\", \"founded\": 2010}\n",
")\n",
"print(f\"Created company: {acme}\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Creating Multiple Nodes (Batch)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Batch create nodes for better performance\n",
"# Include 'id' in properties for custom IDs\n",
"nodes_data = [\n",
" {\"labels\": [\"Person\"], \"properties\": {\"id\": \"charlie\", \"name\": \"Charlie\", \"age\": 35}},\n",
" {\"labels\": [\"Person\"], \"properties\": {\"id\": \"diana\", \"name\": \"Diana\", \"age\": 28}},\n",
" {\"labels\": [\"Location\"], \"properties\": {\"name\": \"San Francisco\", \"state\": \"CA\"}},\n",
"]\n",
"\n",
"created_nodes = neptune_store.create_nodes(nodes_data)\n",
"print(f\"Created {len(created_nodes)} nodes in batch\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Retrieving Nodes"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Get a specific node by ID\n",
"alice_node = neptune_store.get_node(node_id=\"alice\")\n",
"print(f\"Retrieved: {alice_node}\")\n",
"\n",
"# Get nodes by label\n",
"people = neptune_store.get_nodes(labels=[\"Person\"], limit=10)\n",
"print(f\"Found {len(people)} Person nodes:\")\n",
"for person in people:\n",
" print(f\" - {person.get('properties', {}).get('name')}\")\n",
"\n",
"# Get nodes by properties\n",
"engineers = neptune_store.get_nodes(\n",
" labels=[\"Person\"],\n",
" properties={\"role\": \"Engineer\"},\n",
" limit=5\n",
")\n",
"print(f\"Found {len(engineers)} engineers\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Updating Nodes"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Update node properties (merge mode - default)\n",
"updated_alice = neptune_store.update_node(\n",
" node_id=\"alice\",\n",
" properties={\"age\": 31, \"department\": \"AI Research\"},\n",
" merge=True\n",
")\n",
"print(f\"Updated Alice: {updated_alice}\")\n",
"\n",
"# Replace all properties (merge=False)\n",
"# WARNING: This removes properties not in the update\n",
"replaced = neptune_store.update_node(\n",
" node_id=\"charlie\",\n",
" properties={\"name\": \"Charlie\", \"age\": 36},\n",
" merge=False\n",
")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Deleting Nodes"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Delete a node (with detach=True to also delete relationships)\n",
"deleted = neptune_store.delete_node(node_id=\"diana\", detach=True)\n",
"print(f\"Deleted diana: {deleted}\")\n",
"\n",
"# Without detach (fails if node has relationships)\n",
"# neptune_store.delete_node(node_id=\"alice\", detach=False)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Step 3: Relationship Operations\n",
"\n",
"### Creating Relationships\n",
"\n",
"Relationships connect nodes and represent connections between entities."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Create a relationship between Alice and Acme\n",
"works_at = neptune_store.create_relationship(\n",
" start_node_id=\"alice\",\n",
" end_node_id=acme[\"id\"],\n",
" rel_type=\"WORKS_AT\",\n",
" properties={\"since\": 2020, \"position\": \"Senior Engineer\"}\n",
")\n",
"print(f\"Created relationship: {works_at}\")\n",
"\n",
"# Create a KNOWS relationship between people\n",
"knows_rel = neptune_store.create_relationship(\n",
" start_node_id=\"alice\",\n",
" end_node_id=bob[\"id\"],\n",
" rel_type=\"KNOWS\",\n",
" properties={\"since\": 2019}\n",
")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Retrieving Relationships"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Get all relationships for a node\n",
"alice_rels = neptune_store.get_relationships(node_id=\"alice\", direction=\"both\")\n",
"print(f\"Alice has {len(alice_rels)} relationships\")\n",
"\n",
"# Get outgoing relationships only\n",
"outgoing = neptune_store.get_relationships(node_id=\"alice\", direction=\"out\")\n",
"\n",
"# Filter by relationship type\n",
"works_rels = neptune_store.get_relationships(\n",
" node_id=\"alice\",\n",
" rel_type=\"WORKS_AT\",\n",
" direction=\"out\"\n",
")\n",
"print(f\"Alice's work relationships: {len(works_rels)}\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Deleting Relationships"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Delete a specific relationship by ID\n",
"if works_at.get(\"id\"):\n",
" deleted = neptune_store.delete_relationship(rel_id=works_at[\"id\"])\n",
" print(f\"Deleted relationship: {deleted}\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Step 4: OpenCypher Queries\n",
"\n",
"Amazon Neptune supports OpenCypher queries via the Bolt protocol. Execute complex graph patterns using standard Cypher syntax."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Simple query\n",
"results = neptune_store.execute_query(\n",
" \"MATCH (p:Person) RETURN p.name, p.age ORDER BY p.age\"\n",
")\n",
"print(\"People in the graph:\")\n",
"for record in results.get(\"records\", []):\n",
" print(f\" - {record.get('p.name')}: {record.get('p.age')} years old\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Using parameters (safer and more efficient)\n",
"results = neptune_store.execute_query(\n",
" \"MATCH (p:Person) WHERE p.age > $min_age RETURN p.name, p.age\",\n",
" parameters={\"min_age\": 25}\n",
")\n",
"print(f\"People over 25: {len(results.get('records', []))}\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Find relationships between nodes\n",
"results = neptune_store.execute_query(\"\"\"\n",
" MATCH (p:Person)-[r:WORKS_AT]->(c:Company)\n",
" RETURN p.name as employee, c.name as company, r.since as start_year\n",
"\"\"\")\n",
"for record in results.get(\"records\", []):\n",
" print(f\"{record['employee']} works at {record['company']} since {record['start_year']}\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Count and aggregate\n",
"results = neptune_store.execute_query(\"\"\"\n",
" MATCH (p:Person)\n",
" RETURN count(p) as total, avg(p.age) as avg_age, max(p.age) as max_age\n",
"\"\"\")\n",
"stats = results.get(\"records\", [{}])[0]\n",
"print(f\"Total: {stats.get('total')}, Avg Age: {stats.get('avg_age'):.1f}\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Step 5: Graph Analytics\n",
"\n",
"### Get Neighbors\n",
"\n",
"Traverse the graph to find connected nodes."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Get immediate neighbors (depth=1)\n",
"neighbors = neptune_store.get_neighbors(\n",
" node_id=\"alice\",\n",
" direction=\"both\",\n",
" depth=1\n",
")\n",
"print(f\"Alice's direct neighbors: {len(neighbors)}\")\n",
"\n",
"# Get neighbors up to 2 hops away\n",
"extended = neptune_store.get_neighbors(\n",
" node_id=\"alice\",\n",
" direction=\"out\",\n",
" depth=2\n",
")\n",
"print(f\"Nodes within 2 hops: {len(extended)}\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Shortest Path\n",
"\n",
"Find the shortest path between two nodes."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Find shortest path\n",
"path = neptune_store.shortest_path(\n",
" start_node_id=\"alice\",\n",
" end_node_id=\"charlie\",\n",
" max_depth=5\n",
")\n",
"\n",
"if path:\n",
" print(\"Path found!\")\n",
" print(f\" Length: {path.get('length')}\")\n",
" print(f\" Nodes: {len(path.get('nodes', []))}\")\n",
" print(f\" Relationships: {len(path.get('relationships', []))}\")\n",
"else:\n",
" print(\"No path found between nodes\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Step 6: Graph Statistics\n",
"\n",
"Get comprehensive statistics about your graph."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Get graph statistics\n",
"stats = neptune_store.get_stats()\n",
"\n",
"print(\"Graph Statistics:\")\n",
"print(f\" Total nodes: {stats.get('node_count', 'N/A')}\")\n",
"print(f\" Total relationships: {stats.get('relationship_count', 'N/A')}\")\n",
"\n",
"print(\"\\nNode labels:\")\n",
"for label, count in stats.get('label_counts', {}).items():\n",
" print(f\" - {label}: {count}\")\n",
"\n",
"print(\"\\nRelationship types:\")\n",
"for rel_type, count in stats.get('relationship_type_counts', {}).items():\n",
" print(f\" - {rel_type}: {count}\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Step 7: Connection Management\n",
"\n",
"Always close connections when done to free resources."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Check connection status\n",
"status = neptune_store.get_status()\n",
"print(f\"Connection status: {status}\")\n",
"\n",
"# Close the connection\n",
"neptune_store.close()\n",
"print(\"Connection closed\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Neptune-Specific Considerations\n",
"\n",
"### Native Element IDs\n",
"\n",
"Neptune uses native `~id` for element identification. Include `id` in properties to set a custom ID:\n",
"\n",
"```python\n",
"# Create a node with custom ID (include 'id' in properties)\n",
"node = neptune_store.create_node(\n",
" labels=[\"Person\"],\n",
" properties={\"id\": \"my-custom-id\", \"name\": \"Test\"}\n",
")\n",
"\n",
"# Create a node with auto-generated UUID (omit 'id' from properties)\n",
"node = neptune_store.create_node(\n",
" labels=[\"Person\"],\n",
" properties={\"name\": \"Test\"}\n",
")\n",
"\n",
"# The ID is used in id() function calls internally:\n",
"# MATCH (n) WHERE id(n) = 'my-custom-id' RETURN n\n",
"```\n",
"\n",
"### OpenCypher Considerations\n",
"\n",
"Amazon Neptune Database's OpenCypher implementation has some differences from Neo4j:\n",
"\n",
"1. **No `shortestPath()` function**: Use variable-length path patterns or `allShortestPaths()`\n",
"2. **Labels syntax**: Use `labels(n)` function to retrieve node labels\n",
"3. **Property updates**: Use `SET n += {props}` for merge behavior\n",
"\n",
"For the complete OpenCypher specification supported by Amazon Neptune Database, see the [AWS documentation](https://docs.aws.amazon.com/neptune/latest/userguide/access-graph-opencypher.html).\n",
"\n",
"### Amazon Neptune Analytics\n",
"\n",
"For analytical (OLAP) workloads such as graph algorithms, aggregations, and large-scale traversals, consider [Amazon Neptune Analytics](https://docs.aws.amazon.com/neptune-analytics/latest/userguide/what-is-neptune-analytics.html). Neptune Analytics complements Neptune Database by providing optimized performance for analytical queries while Neptune Database is optimized for transactional (OLTP) workloads.\n",
"\n",
"### Performance Tips\n",
"\n",
"1. **Use batch operations** for creating multiple nodes/relationships\n",
"2. **Use parameters** in queries to enable query caching\n",
"3. **Limit result sets** with `LIMIT` clause"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Summary\n",
"\n",
"This notebook covered the Amazon Neptune Graph Store integration:\n",
"\n",
"- **IAM Authentication**: Secure AWS SigV4 signing\n",
"- **CRUD Operations**: Full node and relationship management\n",
"- **OpenCypher Queries**: Standard graph query language\n",
"- **Graph Analytics**: Neighbors and shortest path algorithms\n",
"- **Statistics & Monitoring**: Graph metrics and status\n",
"\n",
"### Key Takeaways\n",
"\n",
"- Neptune uses native `~id` for element identification\n",
"- IAM authentication is recommended for production\n",
"- Bolt protocol provides efficient binary query interface\n",
"- Semantica abstracts Neptune-specific syntax differences\n",
"\n",
"### Next Steps\n",
"\n",
"- [Graph Store (Neo4j/FalkorDB)](09_Graph_Store.ipynb) - Compare with other backends\n",
"- [Building Knowledge Graphs](07_Building_Knowledge_Graphs.ipynb) - Build production KGs\n",
"- [Graph Analytics](10_Graph_Analytics.ipynb) - Advanced analytics algorithms"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
},
"language_info": {
"name": "python",
"version": "3.9.0"
}
},
"nbformat": 4,
"nbformat_minor": 4
}
pyproject.toml
+5 -1
View File
@@ -209,8 +209,12 @@ graph-falkordb = [
"falkordb>=1.0.0",
"redis>=4.3.0"
]
graph-amazon-neptune = [
"boto3>=1.24.0",
"neo4j>=5.0.0"
]
graph-all = [
"semantica[graph-neo4j,graph-falkordb]"
"semantica[graph-neo4j,graph-falkordb,graph-amazon-neptune]"
]
parse-docling = [
"docling>=1.0.0"
semantica/graph_store/__init__.py
+74 -41
View File
@@ -1,48 +1,71 @@
"""
Graph Store Module
This module provides comprehensive property graph database integration for the
Semantica framework, supporting multiple graph database backends including Neo4j
and FalkorDB for storing and querying knowledge graphs.
This module provides comprehensive property graph database integration for
the Semantica framework, supporting multiple graph database backends including
Neo4j and FalkorDB for storing and querying knowledge graphs.
Algorithms Used:
Graph Store Management:
- Store Registration: Store type detection, store factory pattern, configuration management, default store selection
- Backend Pattern: Unified interface for multiple backends (Neo4j, FalkorDB), backend instantiation, backend-specific operation delegation
- Store Selection: Default store resolution, store ID lookup, store validation
- Store Registration: Store type detection, store factory pattern,
configuration management, default store selection
- Backend Pattern: Unified interface for multiple backends (Neo4j,
FalkorDB), backend instantiation, backend-specific operation delegation
- Store Selection: Default store resolution, store ID lookup,
store validation
Node and Relationship Operations:
- Node Creation: Single node insertion, batch node insertion, property validation, label management, backend delegation
- Node Retrieval: Pattern matching (label/property filtering), Cypher query construction, result extraction, node reconstruction
- Node Update: Property update, label modification, atomic update operations, conflict detection
- Node Deletion: Node matching, cascade deletion (optional), deletion operation delegation, result verification
- Relationship Creation: Single relationship insertion, batch insertion, property validation, type management
- Node Creation: Single node insertion, batch node insertion,
property validation, label management, backend delegation
- Node Retrieval: Pattern matching (label/property filtering),
Cypher query construction, result extraction, node reconstruction
- Node Update: Property update, label modification, atomic update
operations, conflict detection
- Node Deletion: Node matching, cascade deletion (optional),
deletion operation delegation, result verification
- Relationship Creation: Single relationship insertion, batch insertion,
property validation, type management
- Relationship Retrieval: Pattern matching, path queries, traversal queries
- Relationship Update: Property update, type modification
- Relationship Deletion: Relationship matching, deletion operation delegation
- Relationship Deletion: Relationship matching, deletion operation
delegation
Graph Query Execution:
- Cypher Query: Full Cypher query language support for Neo4j and FalkorDB (OpenCypher)
- Pattern Matching: Node and relationship pattern matching, variable binding, path matching
- Graph Traversal: BFS/DFS traversal, shortest path algorithms, path finding
- Cypher Query: Full Cypher query language support for Neo4j and
FalkorDB (OpenCypher)
- Pattern Matching: Node and relationship pattern matching, variable
binding, path matching
- Graph Traversal: BFS/DFS traversal, shortest path algorithms,
path finding
- Aggregation: COUNT, SUM, AVG, MIN, MAX operations, GROUP BY support
- Query Optimization: Query caching, execution plan analysis, index utilization
- Query Optimization: Query caching, execution plan analysis,
index utilization
Graph Analytics:
- Centrality Algorithms: Degree centrality, betweenness centrality, PageRank, closeness centrality
- Community Detection: Label propagation, Louvain modularity, connected components
- Path Algorithms: Shortest path, all shortest paths, Dijkstra, A* pathfinding
- Centrality Algorithms: Degree centrality, betweenness centrality,
PageRank, closeness centrality
- Community Detection: Label propagation, Louvain modularity,
connected components
- Path Algorithms: Shortest path, all shortest paths, Dijkstra,
A* pathfinding
- Similarity: Node similarity, Jaccard similarity, cosine similarity
Store Backends:
- Neo4j Store: Official Neo4j Python driver, Bolt protocol communication, transaction support, multi-database support, APOC procedures
- FalkorDB Store: Redis-based graph database, sparse matrix representation, linear algebra queries, OpenCypher support, ultra-fast performance
- Neo4j Store: Official Neo4j Python driver, Bolt protocol
communication, transaction support, multi-database support,
APOC procedures
- FalkorDB Store: Redis-based graph database, sparse matrix
representation, linear algebra queries, OpenCypher support,
ultra-fast performance
Bulk Operations:
- Batch Processing: Chunking algorithm (fixed-size batch creation), batch size optimization, memory management for large datasets
- Transaction Management: ACID transaction support, batch commits, rollback on failure
- Progress Tracking: Load progress calculation, elapsed time tracking, throughput calculation
- Batch Processing: Chunking algorithm (fixed-size batch creation),
batch size optimization, memory management for large datasets
- Transaction Management: ACID transaction support, batch commits,
rollback on failure
- Progress Tracking: Load progress calculation, elapsed time tracking,
throughput calculation
Key Features:
- Multi-backend property graph support (Neo4j, FalkorDB)
@@ -79,33 +102,42 @@ Convenience Functions:
- list_available_methods: List registered graph store methods
Example Usage:
>>> from semantica.graph_store import GraphStore, create_node, create_relationship, execute_query
>>> from semantica.graph_store import GraphStore, create_node, \
... create_relationship, execute_query
>>> # Using convenience functions
>>> node_id = create_node(labels=["Person"], properties={"name": "Alice", "age": 30})
>>> rel_id = create_relationship(start_id=node1_id, end_id=node2_id, rel_type="KNOWS", properties={"since": 2020})
>>> results = execute_query("MATCH (p:Person) WHERE p.age > 25 RETURN p.name")
>>> node_id = create_node(labels=["Person"],
... properties={"name": "Alice", "age": 30})
>>> rel_id = create_relationship(start_id=node1_id, end_id=node2_id,
... rel_type="KNOWS",
... properties={"since": 2020})
>>> results = execute_query("MATCH (p:Person) WHERE p.age > 25 "
... "RETURN p.name")
>>> # Using classes directly
>>> store = GraphStore(backend="neo4j", uri="bolt://localhost:7687")
>>> node_id = store.create_node(labels=["Person"], properties={"name": "Bob"})
>>> node_id = store.create_node(labels=["Person"],
... properties={"name": "Bob"})
>>> results = store.execute_query("MATCH (n) RETURN n LIMIT 10")
Author: Semantica Contributors
License: MIT
"""
from .config import GraphStoreConfig, graph_store_config
from .falkordb_store import (
FalkorDBStore,
FalkorDBClient,
FalkorDBGraph,
from .amazon_neptune import (
AmazonNeptuneStore,
NeptuneAuthTokenManager,
NeptuneDriver,
NeptuneSession,
NeptuneTransaction,
)
from .config import GraphStoreConfig, graph_store_config
from .falkordb_store import FalkorDBClient, FalkorDBGraph, FalkorDBStore
from .graph_store import (
GraphAnalytics,
GraphManager,
GraphStore,
NodeManager,
QueryEngine,
RelationshipManager,
GraphAnalytics,
)
from .methods import (
create_node,
@@ -125,11 +157,7 @@ from .methods import (
update_node,
update_relationship,
)
from .neo4j_store import (
Neo4jStore,
Neo4jDriver,
Neo4jTransaction,
)
from .neo4j_store import Neo4jDriver, Neo4jStore, Neo4jTransaction
from .registry import MethodRegistry, method_registry
__all__ = [
@@ -144,6 +172,12 @@ __all__ = [
"Neo4jStore",
"Neo4jDriver",
"Neo4jTransaction",
# Amazon Neptune
"AmazonNeptuneStore",
"NeptuneAuthTokenManager",
"NeptuneDriver",
"NeptuneSession",
"NeptuneTransaction",
# FalkorDB
"FalkorDBStore",
"FalkorDBClient",
@@ -171,4 +205,3 @@ __all__ = [
"MethodRegistry",
"method_registry",
]
semantica/graph_store/amazon_neptune.py
+1798
View File
File diff suppressed because it is too large Load Diff
semantica/graph_store/config.py
+49 -4
View File
@@ -6,7 +6,8 @@ supporting multiple configuration sources including environment variables, confi
and programmatic configuration.
Supported Configuration Sources:
- Environment variables: GRAPH_STORE_DEFAULT_BACKEND, GRAPH_STORE_NEO4J_URI, GRAPH_STORE_FALKORDB_HOST, etc.
- Environment variables: GRAPH_STORE_DEFAULT_BACKEND,
GRAPH_STORE_NEO4J_URI, GRAPH_STORE_FALKORDB_HOST, etc.
- Config files: YAML, JSON, TOML formats
- Programmatic: Python API for setting graph store configurations
@@ -44,7 +45,11 @@ from ..utils.logging import get_logger
class GraphStoreConfig:
"""Configuration manager for graph store module - supports .env files, environment variables, and programmatic config."""
"""
Configuration manager for graph store module.
Supports .env files, environment variables, and programmatic config.
"""
def __init__(self, config_file: Optional[str] = None):
"""
@@ -124,6 +129,15 @@ class GraphStoreConfig:
"GRAPH_STORE_FALKORDB_PORT": "falkordb_port",
"GRAPH_STORE_FALKORDB_PASSWORD": "falkordb_password",
"GRAPH_STORE_FALKORDB_GRAPH_NAME": "falkordb_graph_name",
# Amazon Neptune settings
"GRAPH_STORE_NEPTUNE_ENDPOINT": "neptune_endpoint",
"GRAPH_STORE_NEPTUNE_PORT": "neptune_port",
"GRAPH_STORE_NEPTUNE_REGION": "neptune_region",
"GRAPH_STORE_NEPTUNE_IAM_AUTH": "neptune_iam_auth",
"GRAPH_STORE_NEPTUNE_USE_SSL": "neptune_use_ssl",
"AWS_ACCESS_KEY_ID": "neptune_access_key",
"AWS_SECRET_ACCESS_KEY": "neptune_secret_key",
"AWS_SESSION_TOKEN": "neptune_session_token",
}
for env_var, config_key in env_mappings.items():
@@ -135,6 +149,7 @@ class GraphStoreConfig:
"timeout",
"max_retries",
"falkordb_port",
"neptune_port",
]:
try:
self._config[config_key] = int(value)
@@ -142,7 +157,11 @@ class GraphStoreConfig:
self.logger.warning(
f"Invalid integer value for {env_var}: {value}"
)
elif config_key in ["neo4j_encrypted"]:
elif config_key in [
"neo4j_encrypted",
"neptune_iam_auth",
"neptune_use_ssl",
]:
self._config[config_key] = value.lower() in [
"true",
"1",
@@ -171,6 +190,15 @@ class GraphStoreConfig:
"falkordb_port": 6379,
"falkordb_password": None,
"falkordb_graph_name": "default",
# Amazon Neptune defaults
"neptune_endpoint": None,
"neptune_port": 8182,
"neptune_region": None,
"neptune_iam_auth": True,
"neptune_use_ssl": True,
"neptune_access_key": None,
"neptune_secret_key": None,
"neptune_session_token": None,
}
for key, default_value in defaults.items():
@@ -269,6 +297,24 @@ class GraphStoreConfig:
"graph_name": self._config.get("falkordb_graph_name"),
}
def get_neptune_config(self) -> Dict[str, Any]:
"""
Get Amazon Neptune-specific configuration.
Returns:
Neptune configuration dictionary
"""
return {
"endpoint": self._config.get("neptune_endpoint"),
"port": self._config.get("neptune_port"),
"region": self._config.get("neptune_region"),
"iam_auth": self._config.get("neptune_iam_auth"),
"use_ssl": self._config.get("neptune_use_ssl"),
"access_key": self._config.get("neptune_access_key"),
"secret_key": self._config.get("neptune_secret_key"),
"session_token": self._config.get("neptune_session_token"),
}
def reset(self) -> None:
"""Reset configuration to defaults."""
self._config.clear()
@@ -278,4 +324,3 @@ class GraphStoreConfig:
# Global configuration instance
graph_store_config = GraphStoreConfig()
semantica/graph_store/graph_store.py
+179 -111
View File
@@ -34,7 +34,7 @@ License: MIT
from typing import Any, Dict, List, Optional, Tuple, Union
from ..utils.exceptions import ProcessingError, ValidationError
from ..utils.exceptions import ValidationError
from ..utils.logging import get_logger
from ..utils.progress_tracker import get_progress_tracker
from .config import graph_store_config
@@ -214,7 +214,9 @@ class RelationshipManager:
Returns:
List of relationships
"""
return self.backend.get_relationships(node_id, rel_type, direction, limit, **options)
return self.backend.get_relationships(
node_id, rel_type, direction, limit, **options
)
def delete(
self,
@@ -290,6 +292,7 @@ class QueryEngine:
) -> str:
"""Generate cache key for query."""
import hashlib
key_str = f"{query}:{str(parameters)}"
return hashlib.md5(key_str.encode()).hexdigest()
@@ -340,7 +343,9 @@ class GraphAnalytics:
Returns:
Path information or None
"""
return self.backend.shortest_path(start_node_id, end_node_id, rel_type, max_depth, **options)
return self.backend.shortest_path(
start_node_id, end_node_id, rel_type, max_depth, **options
)
def get_neighbors(
self,
@@ -363,7 +368,9 @@ class GraphAnalytics:
Returns:
List of neighboring nodes
"""
return self.backend.get_neighbors(node_id, rel_type, direction, depth, **options)
return self.backend.get_neighbors(
node_id, rel_type, direction, depth, **options
)
def degree_centrality(
self,
@@ -440,7 +447,7 @@ class GraphAnalytics:
Component information
"""
backend_type = type(self.backend).__name__
if "Neo4j" in backend_type:
query = """
CALL gds.wcc.stream({
@@ -452,16 +459,23 @@ class GraphAnalytics:
"""
params = {"label": labels[0] if labels else "*"}
result = self.backend.execute_query(query, params)
return [{"component": r["componentId"], "nodes": r["nodes"]} for r in result]
return [
{"component": r["componentId"], "nodes": r["nodes"]} for r in result
]
elif "NetworkX" in backend_type:
import networkx as nx
G = self.backend.graph
components = list(nx.connected_components(G))
return [{"component": i, "nodes": list(c)} for i, c in enumerate(components)]
return [
{"component": i, "nodes": list(c)} for i, c in enumerate(components)
]
else:
raise NotImplementedError(f"connected_components not implemented for {backend_type}")
raise NotImplementedError(
f"connected_components not implemented for {backend_type}"
)
class GraphManager:
@@ -534,7 +548,11 @@ class GraphStore:
self.progress_tracker.enabled = True
# Determine backend
self.backend = backend or config.get("backend") or graph_store_config.get("default_backend", "neo4j")
self.backend = (
backend
or config.get("backend")
or graph_store_config.get("default_backend", "neo4j")
)
self.config = config
# Initialize store backend
@@ -546,16 +564,25 @@ class GraphStore:
"""Initialize the appropriate store backend based on backend type."""
if self.backend == "neo4j":
from .neo4j_store import Neo4jStore
neo4j_config = graph_store_config.get_neo4j_config()
neo4j_config.update(self.config)
self._store_backend = Neo4jStore(**neo4j_config)
elif self.backend == "falkordb":
from .falkordb_store import FalkorDBStore
falkordb_config = graph_store_config.get_falkordb_config()
falkordb_config.update(self.config)
self._store_backend = FalkorDBStore(**falkordb_config)
elif self.backend == "neptune" or self.backend == "amazon_neptune":
from .amazon_neptune import AmazonNeptuneStore
neptune_config = graph_store_config.get_neptune_config()
neptune_config.update(self.config)
self._store_backend = AmazonNeptuneStore(**neptune_config)
else:
raise ValidationError(f"Unknown backend: {self.backend}")
@@ -621,7 +648,9 @@ class GraphStore:
**options,
) -> List[Dict[str, Any]]:
"""Get nodes matching criteria."""
return self._manager.nodes.get(labels=labels, properties=properties, limit=limit, **options)
return self._manager.nodes.get(
labels=labels, properties=properties, limit=limit, **options
)
def update_node(
self,
@@ -665,7 +694,9 @@ class GraphStore:
**options,
) -> List[Dict[str, Any]]:
"""Get relationships."""
return self._manager.relationships.get(node_id, rel_type, direction, limit, **options)
return self._manager.relationships.get(
node_id, rel_type, direction, limit, **options
)
def delete_relationship(
self,
@@ -723,7 +754,9 @@ class GraphStore:
node_id, rel_type, direction, actual_depth, **options
)
def query(self, query: str, parameters: Optional[Dict[str, Any]] = None, **options) -> List[Dict[str, Any]]:
def query(
self, query: str, parameters: Optional[Dict[str, Any]] = None, **options
) -> List[Dict[str, Any]]:
"""
Execute a query and return results (Compatibility method for ContextRetriever).
@@ -771,42 +804,46 @@ class GraphStore:
# Convert to GraphStore format (labels, properties)
graph_nodes = []
for node in nodes:
# Extract label from type
labels = [node.get("type", "Entity")]
if isinstance(labels[0], str):
labels = [labels[0]] # Ensure list
# Extract labels - support both 'labels' array and 'type' string
labels = node.get("labels")
if not labels:
node_type = node.get("type", "Entity")
labels = [node_type] if isinstance(node_type, str) else node_type
if isinstance(labels, str):
labels = [labels]
# Prepare properties
props = node.get("properties", {}).copy()
# Ensure ID is preserved
if "id" in node and "id" not in props:
props["id"] = node["id"]
# Ensure content/text is preserved
if "content" in node and "content" not in props:
props["content"] = node["content"]
if "text" in node and "text" not in props:
props["text"] = node["text"]
graph_nodes.append({
"labels": labels,
"properties": props
})
graph_nodes.append({"labels": labels, "properties": props})
# Use batch creation
# Note: create_nodes expects dicts with 'labels' and 'properties' keys if passed directly?
# Note: create_nodes expects dicts with 'labels' and 'properties'
# keys if passed directly?
# Let's check create_nodes signature implementation in manager.
# But here I'll assume create_nodes takes a list of such dicts or similar.
# But here I'll assume create_nodes takes a list of such dicts
# or similar.
# Actually, let's look at create_nodes wrapper in this file:
# def create_nodes(self, nodes: List[Dict[str, Any]], **options)
# It passes to self._manager.nodes.create_batch(nodes)
# If create_batch expects specific format, I should match it.
# Assuming create_batch is smart enough or expects standard format.
# To be safe, let's look at NodeManager.create_batch if possible, but I can't easily.
# Standard expectation: List of dicts where each dict has labels and properties.
# To be safe, let's look at NodeManager.create_batch if possible,
# but I can't easily.
# Standard expectation: List of dicts where each dict has labels
# and properties.
result = self.create_nodes(graph_nodes, **options)
return len(result)
@@ -827,17 +864,21 @@ class GraphStore:
target_id = edge.get("target_id")
rel_type = edge.get("type", "RELATED_TO")
properties = edge.get("properties", {}).copy()
# Preserve weight
if "weight" in edge:
properties["weight"] = edge["weight"]
if source_id and target_id:
try:
self.create_relationship(source_id, target_id, rel_type, properties, **options)
self.create_relationship(
source_id, target_id, rel_type, properties, **options
)
count += 1
except Exception as e:
self.logger.warning(f"Failed to add edge {source_id}->{target_id}: {e}")
self.logger.warning(
f"Failed to add edge {source_id}->{target_id}: {e}"
)
return count
def build_from_conversations(
@@ -872,27 +913,29 @@ class GraphStore:
all_nodes = []
all_edges = []
seen_nodes = set()
for conv in conversations:
# Load conversation if string (file path)
conv_data = conv
if isinstance(conv, str):
from pathlib import Path
from ..utils.helpers import read_json_file
conv_data = read_json_file(Path(conv))
nodes, edges = self._process_conversation_to_elements(
conv_data,
conv_data,
extract_intents=extract_intents,
extract_sentiments=extract_sentiments
extract_sentiments=extract_sentiments,
)
# Add unique nodes
for node in nodes:
if node["id"] not in seen_nodes:
all_nodes.append(node)
seen_nodes.add(node["id"])
all_edges.extend(edges)
if link_entities:
@@ -904,13 +947,8 @@ class GraphStore:
edge_count = self.add_edges(all_edges)
self.progress_tracker.stop_tracking(tracking_id, status="completed")
return {
"statistics": {
"node_count": node_count,
"edge_count": edge_count
}
}
return {"statistics": {"node_count": node_count, "edge_count": edge_count}}
except Exception as e:
self.progress_tracker.stop_tracking(
@@ -930,92 +968,112 @@ class GraphStore:
"""
nodes = []
edges = []
# Process entities
for entity in entities:
entity_id = entity.get("id") or entity.get("entity_id")
if entity_id:
nodes.append({
"id": entity_id,
"type": entity.get("type", "entity"),
"properties": {
"content": entity.get("text") or entity.get("label") or entity_id,
**entity
nodes.append(
{
"id": entity_id,
"type": entity.get("type", "entity"),
"properties": {
"content": entity.get("text")
or entity.get("label")
or entity_id,
**entity,
},
}
})
)
# Process relationships
for rel in relationships:
source = rel.get("source_id")
target = rel.get("target_id")
if source and target:
edges.append({
"source_id": source,
"target_id": target,
"type": rel.get("type", "related_to"),
"weight": rel.get("confidence", 1.0),
"properties": rel
})
edges.append(
{
"source_id": source,
"target_id": target,
"type": rel.get("type", "related_to"),
"weight": rel.get("confidence", 1.0),
"properties": rel,
}
)
node_count = self.add_nodes(nodes)
edge_count = self.add_edges(edges)
return {"statistics": {"node_count": node_count, "edge_count": edge_count}}
def _process_conversation_to_elements(self, conv_data: Dict[str, Any], **kwargs) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
def _process_conversation_to_elements(
self, conv_data: Dict[str, Any], **kwargs
) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
"""Helper to process conversation into nodes and edges."""
nodes = []
edges = []
conv_id = conv_data.get("id") or f"conv_{hash(str(conv_data)) % 10000}"
# Conversation node
nodes.append({
"id": conv_id,
"type": "conversation",
"properties": {
"content": conv_data.get("content", "") or conv_data.get("summary", ""),
"timestamp": conv_data.get("timestamp")
nodes.append(
{
"id": conv_id,
"type": "conversation",
"properties": {
"content": conv_data.get("content", "")
or conv_data.get("summary", ""),
"timestamp": conv_data.get("timestamp"),
},
}
})
)
name_to_id = {}
extract_entities = kwargs.get("extract_entities", True) # Default true if not passed?
# Actually ContextGraph defaults to True in init, but here we are static.
# Note: extract_entities option is available but not used in this
# implementation. Default true if not passed. ContextGraph defaults
# to True in init, but here we are static.
# Let's assume True unless told otherwise or check config.
# Extract entities
for entity in conv_data.get("entities", []):
entity_id = entity.get("id") or entity.get("entity_id")
entity_text = entity.get("text") or entity.get("label") or entity.get("name") or entity_id
entity_text = (
entity.get("text")
or entity.get("label")
or entity.get("name")
or entity_id
)
entity_type = entity.get("type", "entity")
# Generate ID if missing
if not entity_id and entity_text:
import hashlib
entity_hash = hashlib.md5(f"{entity_text}_{entity_type}".encode()).hexdigest()[:12]
entity_hash = hashlib.md5(
f"{entity_text}_{entity_type}".encode()
).hexdigest()[:12]
entity_id = f"{entity_type.lower()}_{entity_hash}"
if entity_id:
if entity_text:
name_to_id[entity_text] = entity_id
nodes.append({
"id": entity_id,
"type": "entity", # Normalize type?
"properties": {
"content": entity_text,
"type": entity_type,
**entity
nodes.append(
{
"id": entity_id,
"type": "entity", # Normalize type?
"properties": {
"content": entity_text,
"type": entity_type,
**entity,
},
}
})
)
# Edge: Conversation -> Entity
edges.append({
"source_id": conv_id,
"target_id": entity_id,
"type": "mentions"
})
edges.append(
{"source_id": conv_id, "target_id": entity_id, "type": "mentions"}
)
# Extract relationships
for rel in conv_data.get("relationships", []):
@@ -1029,43 +1087,54 @@ class GraphStore:
target = name_to_id[rel.get("target")]
if source and target:
edges.append({
"source_id": source,
"target_id": target,
"type": rel.get("type", "related_to"),
"weight": rel.get("confidence", 1.0),
"properties": rel
})
edges.append(
{
"source_id": source,
"target_id": target,
"type": rel.get("type", "related_to"),
"weight": rel.get("confidence", 1.0),
"properties": rel,
}
)
return nodes, edges
def _link_entities_elements(self, nodes: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
def _link_entities_elements(
self, nodes: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
"""Link similar entities."""
edges = []
# Lazy import to avoid circular dependency
try:
from ..context.entity_linker import EntityLinker
linker = EntityLinker() # Use default config
linker = EntityLinker() # Use default config
except (ImportError, OSError):
return []
entity_nodes = [n for n in nodes if n.get("type") == "entity"]
for i, node1 in enumerate(entity_nodes):
content1 = node1["properties"].get("content", "")
if not content1: continue
if not content1:
continue
for node2 in entity_nodes[i + 1 :]:
content2 = node2["properties"].get("content", "")
if not content2: continue
similarity = linker._calculate_text_similarity(content1.lower(), content2.lower())
if not content2:
continue
similarity = linker._calculate_text_similarity(
content1.lower(), content2.lower()
)
if similarity >= linker.similarity_threshold:
edges.append({
"source_id": node1["id"],
"target_id": node2["id"],
"type": "similar_to",
"weight": similarity
})
edges.append(
{
"source_id": node1["id"],
"target_id": node2["id"],
"type": "similar_to",
"weight": similarity,
}
)
return edges
@property
@@ -1087,4 +1156,3 @@ class GraphStore:
def analytics(self) -> GraphAnalytics:
"""Get analytics engine."""
return self._manager.analytics
tests/test_amazon_neptune.py
+1702
View File
File diff suppressed because it is too large Load Diff