> ## Documentation Index
> Fetch the complete documentation index at: https://mintlify.com/getzep/graphiti/llms.txt
> Use this file to discover all available pages before exploring further.
# FalkorDriver
> FalkorDB graph database driver implementation
The `FalkorDriver` class provides connectivity to FalkorDB, a Redis-based graph database with RedisSearch integration.
## Overview
FalkorDB is a multi-tenant graph database built on Redis. The FalkorDriver provides:
* Redis-based graph storage
* RedisSearch fulltext indexing
* Multi-tenant database support
* Connection to FalkorDB Cloud or on-premises
* Automatic datetime conversion
## Constructor
```python theme={null}
from graphiti_core.driver.falkordb_driver import FalkorDriver
driver = FalkorDriver(
host="localhost",
port=6379,
username=None,
password=None,
database="default_db"
)
```
### Parameters
The hostname where FalkorDB is running.
For on-premises deployments, this is typically `"localhost"`.
For FalkorDB Cloud, use the provided cloud endpoint.
Example: `"localhost"` or `"cloud.falkordb.com"`
The port on which FalkorDB is listening.
The default Redis port is 6379. FalkorDB Cloud may use different ports.
Example: `6379`
The username for authentication.
Optional for on-premises deployments. Required for FalkorDB Cloud.
Example: `"admin"`
The password for authentication.
Optional for on-premises deployments. Required for FalkorDB Cloud.
Example: `"your-password"`
An existing FalkorDB client instance to reuse.
If provided, the driver will use this client instead of creating a new one.
Useful for sharing connections or custom client configurations.
Example: `FalkorDB(host="localhost", port=6379)`
The name of the database (graph) to connect to.
FalkorDB supports multiple graphs. Each graph is isolated.
Example: `"graphiti"` or `"user_graph"`
## Methods
### execute\_query
Execute a Cypher query and return results.
```python theme={null}
result = await driver.execute_query(
"MATCH (n:Entity) RETURN n LIMIT 10"
)
```
**Parameters:**
* `cypher_query_` (str) - The Cypher query to execute
* `**kwargs` - Query parameters (automatically converted for FalkorDB)
**Returns:** `tuple[list[dict], list[str], None]` - Records, headers, and summary
FalkorDB does not support native datetime objects. The driver automatically converts datetime values to ISO format strings.
### session
Create a new database session.
```python theme={null}
session = driver.session(database="graphiti")
try:
result = await session.run("MATCH (n) RETURN count(n)")
finally:
await session.close()
```
**Parameters:**
* `database` (str | None) - Optional database name override
**Returns:** `FalkorDriverSession` - Session object
### close
Close the driver and release connections.
```python theme={null}
await driver.close()
```
**Returns:** `None`
### build\_indices\_and\_constraints
Create all required indices for Graphiti.
```python theme={null}
await driver.build_indices_and_constraints(delete_existing=False)
```
**Parameters:**
* `delete_existing` (bool) - If True, delete existing indices before creating new ones
**Returns:** `None`
This method is automatically called during driver initialization.
### health\_check
Verify connectivity to FalkorDB.
```python theme={null}
try:
await driver.health_check()
print("Connection healthy")
except Exception as e:
print(f"Connection failed: {e}")
```
**Returns:** `None` - Raises exception if connection fails
### clone
Create a copy of the driver targeting a different database.
```python theme={null}
default_driver = FalkorDriver(host="localhost", database="default_db")
graphs_driver = default_driver.clone("user_graphs")
```
**Parameters:**
* `database` (str) - The name of the database to target
**Returns:** `GraphDriver` - New driver instance
### build\_fulltext\_query
Build a RedisSearch-compatible fulltext query string.
```python theme={null}
query = driver.build_fulltext_query(
query="California Attorney General",
group_ids=["main", "politics"],
max_query_length=128
)
# Returns: '(@group_id:"main"|"politics") (California | Attorney | General)'
```
**Parameters:**
* `query` (str) - The search query text
* `group_ids` (list\[str] | None) - Optional list of group IDs to filter by
* `max_query_length` (int) - Maximum query length (default: 128)
**Returns:** `str` - RedisSearch formatted query
FalkorDB uses RedisSearch syntax with the `@` prefix for field queries. The driver automatically formats queries and escapes reserved characters.
### sanitize
Sanitize query text by replacing FalkorDB special characters with whitespace.
```python theme={null}
sanitized = driver.sanitize("user@example.com, test-query!")
# Returns: "user example com test query "
```
**Parameters:**
* `query` (str) - The text to sanitize
**Returns:** `str` - Sanitized text with special characters replaced
## Properties
### provider
```python theme={null}
driver.provider # GraphProvider.FALKORDB
```
Returns the `GraphProvider.FALKORDB` enum value.
### fulltext\_syntax
```python theme={null}
driver.fulltext_syntax # '@'
```
The prefix used for fulltext field queries in RedisSearch syntax.
### default\_group\_id
```python theme={null}
driver.default_group_id # '\\_'
```
The default group ID used for FalkorDB queries.
### Operations Properties
The driver exposes specialized operation interfaces:
```python theme={null}
driver.entity_node_ops # FalkorEntityNodeOperations
driver.episode_node_ops # FalkorEpisodeNodeOperations
driver.community_node_ops # FalkorCommunityNodeOperations
driver.saga_node_ops # FalkorSagaNodeOperations
driver.entity_edge_ops # FalkorEntityEdgeOperations
driver.episodic_edge_ops # FalkorEpisodicEdgeOperations
driver.community_edge_ops # FalkorCommunityEdgeOperations
driver.has_episode_edge_ops # FalkorHasEpisodeEdgeOperations
driver.next_episode_edge_ops # FalkorNextEpisodeEdgeOperations
driver.search_ops # FalkorSearchOperations
driver.graph_ops # FalkorGraphMaintenanceOperations
```
## Usage Examples
### Basic Connection (On-Premises)
```python theme={null}
import asyncio
from graphiti_core.driver.falkordb_driver import FalkorDriver
async def main():
# Connect to local FalkorDB
driver = FalkorDriver(
host="localhost",
port=6379
)
try:
# Verify connection
await driver.health_check()
print("Connected to FalkorDB")
# Execute a query
result, headers, _ = await driver.execute_query(
"MATCH (n:Entity) RETURN count(n) as count"
)
print(f"Entity count: {result[0]['count']}")
finally:
await driver.close()
asyncio.run(main())
```
### FalkorDB Cloud Connection
```python theme={null}
import os
from dotenv import load_dotenv
from graphiti_core.driver.falkordb_driver import FalkorDriver
load_dotenv()
driver = FalkorDriver(
host=os.environ.get("FALKORDB_HOST", "localhost"),
port=int(os.environ.get("FALKORDB_PORT", "6379")),
username=os.environ.get("FALKORDB_USERNAME"),
password=os.environ.get("FALKORDB_PASSWORD"),
database="graphiti"
)
```
### With Graphiti
```python theme={null}
from graphiti_core import Graphiti
from graphiti_core.driver.falkordb_driver import FalkorDriver
from graphiti_core.nodes import EpisodeType
from datetime import datetime, timezone
import json
# Create driver
driver = FalkorDriver(
host="localhost",
port=6379,
database="graphiti"
)
# Initialize Graphiti
graphiti = Graphiti(graph_driver=driver)
# Build indices
await graphiti.build_indices_and_constraints()
# Add an episode
await graphiti.add_episode(
name="Episode 1",
episode_body="Kamala Harris is the Attorney General of California.",
source=EpisodeType.text,
source_description="biographical information",
reference_time=datetime.now(timezone.utc)
)
# Search
results = await graphiti.search(
"Who is the California Attorney General?"
)
print(results)
```
### Multi-Tenant Usage
```python theme={null}
from graphiti_core.driver.falkordb_driver import FalkorDriver
# Create a base driver
base_driver = FalkorDriver(
host="localhost",
port=6379
)
# Create drivers for different tenants (graphs)
user1_driver = base_driver.clone("user_1_graph")
user2_driver = base_driver.clone("user_2_graph")
shared_driver = base_driver.clone("shared_graph")
# Each driver operates on its own isolated graph
await user1_driver.execute_query(
"CREATE (n:Entity {name: 'User 1 Entity'})"
)
await user2_driver.execute_query(
"CREATE (n:Entity {name: 'User 2 Entity'})"
)
# Queries are isolated by graph
result1, _, _ = await user1_driver.execute_query(
"MATCH (n:Entity) RETURN n.name"
) # Returns only User 1 entities
result2, _, _ = await user2_driver.execute_query(
"MATCH (n:Entity) RETURN n.name"
) # Returns only User 2 entities
```
### Fulltext Search with RedisSearch
```python theme={null}
from graphiti_core.driver.falkordb_driver import FalkorDriver
driver = FalkorDriver(host="localhost")
# Build a RedisSearch query
query = driver.build_fulltext_query(
query="California politics government",
group_ids=["politics", "history"]
)
print(query)
# Output: (@group_id:"politics"|"history") (California | politics | government)
# Use in a search
result, _, _ = await driver.execute_query(
f"""
CALL db.idx.fulltext.queryNodes('entities', '{query}')
YIELD node
RETURN node
LIMIT 10
"""
)
```
### Reusing an Existing FalkorDB Client
```python theme={null}
from falkordb.asyncio import FalkorDB
from graphiti_core.driver.falkordb_driver import FalkorDriver
# Create a custom FalkorDB client
falkor_client = FalkorDB(
host="localhost",
port=6379,
username="admin",
password="password"
)
# Reuse it in multiple drivers
driver1 = FalkorDriver(falkor_db=falkor_client, database="graph1")
driver2 = FalkorDriver(falkor_db=falkor_client, database="graph2")
# Both drivers share the same connection pool
```
## Environment Variables
Common environment variables for FalkorDB configuration:
```bash theme={null}
FALKORDB_HOST=localhost
FALKORDB_PORT=6379
FALKORDB_USERNAME=admin
FALKORDB_PASSWORD=your-password
```
Load in your application:
```python theme={null}
import os
from dotenv import load_dotenv
from graphiti_core.driver.falkordb_driver import FalkorDriver
load_dotenv()
driver = FalkorDriver(
host=os.environ.get("FALKORDB_HOST", "localhost"),
port=int(os.environ.get("FALKORDB_PORT", "6379")),
username=os.environ.get("FALKORDB_USERNAME"),
password=os.environ.get("FALKORDB_PASSWORD")
)
```
## Installation
FalkorDB support requires the optional `falkordb` extra:
```bash theme={null}
pip install graphiti-core[falkordb]
```
Or install the FalkorDB client directly:
```bash theme={null}
pip install falkordb
```
## Datetime Handling
FalkorDB does not support native datetime objects. The driver automatically converts:
```python theme={null}
from datetime import datetime, timezone
# Datetime objects are automatically converted to ISO strings
await driver.execute_query(
"CREATE (e:Episode {created_at: $created_at})",
created_at=datetime.now(timezone.utc)
)
# Stored as: "2024-03-15T10:30:00+00:00"
```
The conversion happens automatically in:
* Query parameters
* List values
* Nested dictionary values
## Stopwords
FalkorDB's RedisSearch automatically filters common stopwords. The driver includes a comprehensive stopword list that mirrors RedisSearch behavior:
```python theme={null}
from graphiti_core.driver.falkordb import STOPWORDS
print(len(STOPWORDS)) # 604 stopwords
```
Stopwords are automatically removed when building fulltext queries.
## Related
Learn about the driver architecture
Main Graphiti class documentation
Alternative production-ready driver
Embedded analytics driver