Skip to main content
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

from graphiti_core.driver.falkordb_driver import FalkorDriver

driver = FalkorDriver(
    host="localhost",
    port=6379,
    username=None,
    password=None,
    database="default_db"
)

​
Parameters

​
host
str
default:"localhost"
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"
​
port
int
default:"6379"
The port on which FalkorDB is listening.The default Redis port is 6379. FalkorDB Cloud may use different ports.Example: 6379
​
username
str | None
default:"None"
The username for authentication.Optional for on-premises deployments. Required for FalkorDB Cloud.Example: "admin"
​
password
str | None
default:"None"
The password for authentication.Optional for on-premises deployments. Required for FalkorDB Cloud.Example: "your-password"
​
falkor_db
FalkorDB | None
default:"None"
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)
​
database
str
default:"default_db"
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. 
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. 
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. 
await driver.close()
Returns: None

​
build_indices_and_constraints

Create all required indices for Graphiti. 
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. 
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. 
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. 
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. 
sanitized = driver.sanitize("[email protected], 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

driver.provider  # GraphProvider.FALKORDB
Returns the GraphProvider.FALKORDB enum value.

​
fulltext_syntax

driver.fulltext_syntax  # '@'
The prefix used for fulltext field queries in RedisSearch syntax.

​
default_group_id

driver.default_group_id  # '\\_'
The default group ID used for FalkorDB queries.

​
Operations Properties

The driver exposes specialized operation interfaces: 
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)

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

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

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

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

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

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: 
FALKORDB_HOST=localhost
FALKORDB_PORT=6379
FALKORDB_USERNAME=admin
FALKORDB_PASSWORD=your-password
Load in your application: 
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: 
pip install graphiti-core[falkordb]
Or install the FalkorDB client directly: 
pip install falkordb

​
Datetime Handling

FalkorDB does not support native datetime objects. The driver automatically converts: 
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: 
from graphiti_core.driver.falkordb import STOPWORDS

print(len(STOPWORDS))  # 604 stopwords
Stopwords are automatically removed when building fulltext queries.

Drivers Overview

Learn about the driver architecture

Graphiti

Main Graphiti class documentation

Neo4j Driver

Alternative production-ready driver

Kuzu Driver

Embedded analytics driver

Build docs developers (and LLMs) love

Get started for freeTalk to us