qilbeedb 0.2.1

Python SDK for QilbeeDB - Enterprise Graph Database with Agent Memory

QilbeeDB Python SDK

Official Python client library for QilbeeDB - Enterprise Graph Database with Bi-Temporal Agent Memory.

Installation

pip install qilbeedb

Quick Start

Authentication

QilbeeDB supports two authentication methods:

JWT Authentication (for humans/admins)

from qilbeedb import QilbeeDB

# Connect and login with username/password
db = QilbeeDB("http://localhost:7474")
db.login("admin", "password")

API Key Authentication (recommended for applications)

# Option 1: Initialize with API key
db = QilbeeDB({
    "uri": "http://localhost:7474",
    "api_key": "qilbee_live_your_api_key_here"
})

# Option 2: Switch to API key after JWT login
db = QilbeeDB("http://localhost:7474")
db.login("admin", "password")
db.set_api_key("qilbee_live_your_api_key_here")

Basic Graph Operations

from qilbeedb import QilbeeDB

# Connect to database
db = QilbeeDB("http://localhost:7474")
db.login("admin", "password")

# Get or create a graph
graph = db.graph("social")

# Create nodes
alice = graph.create_node(
    labels=["Person"],
    properties={"name": "Alice", "age": 30}
)

bob = graph.create_node(
    labels=["Person"],
    properties={"name": "Bob", "age": 35}
)

# Create relationship
knows = graph.create_relationship(
    alice.id,
    "KNOWS",
    bob.id,
    properties={"since": 2020}
)

# Query nodes
people = graph.find_nodes("Person")
for person in people:
    print(f"{person['name']} is {person['age']} years old")

Cypher Queries

# Execute Cypher query
result = graph.query(
    "MATCH (p:Person) WHERE p.age > $age RETURN p.name, p.age ORDER BY p.age DESC",
    parameters={"age": 25}
)

for record in result:
    print(f"{record['p.name']}: {record['p.age']}")

# Query statistics
print(f"Execution time: {result.stats.execution_time_ms}ms")

Query Builder

from qilbeedb.query import Query

# Build query fluently
result = (
    Query(graph)
    .match("(p:Person)")
    .where("p.age > $age", age=25)
    .return_clause("p.name", "p.age")
    .order_by("p.age", desc=True)
    .limit(10)
    .execute()
)

Agent Memory

QilbeeDB provides enterprise-grade agent memory with automatic persistence. All episodes are stored in RocksDB with write-ahead logging, ensuring durability across server restarts.

from qilbeedb.memory import Episode

# Create agent memory
memory = db.agent_memory(
    "agent-001",
    max_episodes=10000,
    min_relevance=0.1
)

# Store conversation (automatically persisted to disk)
episode = Episode.conversation(
    "agent-001",
    "What is the capital of France?",
    "The capital of France is Paris."
)
episode_id = memory.store_episode(episode)
print(f"Stored episode: {episode_id}")  # Episode survives server restart

# Store observation
obs = Episode.observation(
    "agent-001",
    "User seems interested in European geography"
)
memory.store_episode(obs)

# Store action with result
action = Episode.action(
    "agent-001",
    "Searched knowledge base for European capitals",
    "Found 47 capital cities"
)
memory.store_episode(action)

# Retrieve recent memories
recent = memory.get_recent_episodes(10)
for ep in recent:
    print(f"[{ep.episode_type}] {ep.content}")

# Get a specific episode by ID
retrieved = memory.get_episode(episode_id)

# Search memories
results = memory.search_episodes("France")

# Delete a specific episode
memory.delete_episode(episode_id)

# Get statistics
stats = memory.get_statistics()
print(f"Total episodes: {stats.total_episodes}")
print(f"Average relevance: {stats.avg_relevance:.2f}")

# Consolidate and forget
memory.consolidate()
memory.forget(min_relevance=0.2)

# Clear all memories for an agent
memory.clear()

Semantic Search

Find memories by meaning, not just keywords. Semantic search uses vector embeddings to find conceptually similar content.

from qilbeedb import QilbeeDB
from qilbeedb.memory import Episode

db = QilbeeDB("http://localhost:7474")
db.login("admin", "password")
memory = db.agent_memory("my-agent")

# Store some episodes
memory.store_episode(Episode.conversation(
    "my-agent",
    "What is machine learning?",
    "ML is a subset of artificial intelligence..."
))
memory.store_episode(Episode.conversation(
    "my-agent",
    "How do neural networks work?",
    "Neural networks are computing systems inspired by the brain..."
))

# Semantic search - finds related content even with different words
results = memory.semantic_search("AI training techniques", limit=5)
for result in results:
    print(f"Score: {result.score:.2f}")
    print(f"Episode: {result.episode.content}")

# Hybrid search - combines keyword and semantic matching
results = memory.hybrid_search(
    "machine learning",
    limit=10,
    semantic_weight=0.5  # 0.0 = keyword only, 1.0 = semantic only
)
for result in results:
    print(f"Combined: {result.score:.2f}")
    print(f"Keyword: {result.keyword_score}, Semantic: {result.semantic_score}")

# Find episodes similar to a specific episode
similar = memory.find_similar_episodes(episode_id, limit=5)
for result in similar:
    print(f"Similar episode (score {result.score:.2f}): {result.episode.id}")

# Check semantic search configuration
status = memory.get_semantic_search_status()
if status["enabled"]:
    print(f"Model: {status['model']}")
    print(f"Indexed: {status['indexedEpisodes']} episodes")

For more information, see the Semantic Search documentation.

Context Manager

with QilbeeDB("http://localhost:7474") as db:
    graph = db.graph("mydata")
    # Your code here
# Connection automatically closed

API Reference

QilbeeDB

Main database client.

Methods:

• graph(name: str) -> Graph - Get or create graph
• list_graphs() -> List[str] - List all graphs
• delete_graph(name: str) -> bool - Delete graph
• health() -> Dict - Get health status
• agent_memory(agent_id: str, **config) -> AgentMemory - Create agent memory
• login(username: str, password: str) -> Dict - Login with JWT authentication
• logout() -> None - Logout and clear authentication
• is_authenticated() -> bool - Check if authenticated
• set_api_key(api_key: str) -> None - Switch to API key authentication
• refresh_token() -> str - Manually refresh JWT access token
• get_audit_logs(**filters) -> Dict - Query audit logs (admin only)
• get_failed_logins(limit: int) -> List - Get failed login events
• get_user_audit_events(username: str, limit: int) -> List - Get events for a user
• get_security_events(limit: int) -> List - Get security-related events
• get_locked_accounts() -> Dict - Get all locked accounts (admin only)
• get_lockout_status(username: str) -> Dict - Get lockout status for a user (admin only)
• lock_account(username: str, reason: str) -> Dict - Manually lock an account (admin only)
• unlock_account(username: str) -> Dict - Unlock an account (admin only)

Graph

Graph operations.

Methods:

• create_node(labels: List[str], properties: Dict) -> Node - Create node
• get_node(node_id: int) -> Node - Get node by ID
• update_node(node: Node) -> Node - Update node
• delete_node(node_id: int) -> bool - Delete node
• create_relationship(from_node, rel_type: str, to_node, properties: Dict) -> Relationship - Create relationship
• find_nodes(label: str, properties: Dict, limit: int) -> List[Node] - Find nodes
• get_relationships(node, direction: str) -> List[Relationship] - Get relationships
• query(cypher: str, parameters: Dict) -> QueryResult - Execute Cypher query

Node

Graph node.

Attributes:

• id: int - Node ID
• labels: List[str] - Node labels
• properties: Dict - Node properties

Methods:

• get(key, default) -> Any - Get property
• __getitem__(key) -> Any - Access property: node["name"]
• __setitem__(key, value) - Set property: node["age"] = 31

Relationship

Graph relationship.

Attributes:

• id: int - Relationship ID
• type: str - Relationship type
• start_node: int - Start node ID
• end_node: int - End node ID
• properties: Dict - Relationship properties

AgentMemory

Bi-temporal agent memory with automatic RocksDB persistence.

All episodes are automatically persisted to disk with write-ahead logging (WAL), ensuring durability across server restarts.

Methods:

• store_episode(episode: Episode) -> str - Store episode (persisted to disk)
• get_episode(episode_id: str) -> Episode - Get episode by ID
• get_recent_episodes(limit: int) -> List[Episode] - Get recent episodes
• search_episodes(query: str, limit: int) -> List[Episode] - Search episodes
• semantic_search(query: str, limit: int, min_score: float) -> List[SemanticSearchResult] - Search by meaning
• hybrid_search(query: str, limit: int, semantic_weight: float) -> List[HybridSearchResult] - Combined search
• find_similar_episodes(episode_id: str, limit: int) -> List[SemanticSearchResult] - Find similar episodes
• get_semantic_search_status() -> Dict - Get semantic search configuration status
• delete_episode(episode_id: str) -> bool - Delete episode by ID
• get_statistics() -> MemoryStatistics - Get statistics
• consolidate() -> int - Consolidate memory
• forget(min_relevance: float) -> int - Forget low-relevance episodes
• clear() -> bool - Clear all episodes for this agent

Episode

Episodic memory.

Static Methods:

• Episode.conversation(agent_id, user_input, agent_response) - Create conversation
• Episode.observation(agent_id, observation) - Create observation
• Episode.action(agent_id, action, result) - Create action

Configuration

Connection Options

# Simple URI connection (use login() afterward)
db = QilbeeDB("http://localhost:7474")

# Configuration dict with API key (recommended for applications)
db = QilbeeDB({
    "uri": "http://localhost:7474",
    "api_key": "qilbee_live_your_api_key_here",
    "timeout": 30,
    "verify_ssl": True
})

# Configuration dict with basic auth (deprecated, use login() instead)
db = QilbeeDB({
    "uri": "http://localhost:7474",
    "auth": {"username": "admin", "password": "password"},
    "timeout": 30,
    "verify_ssl": True,
    "persist_tokens": True
})

Managing API Keys

Create and manage API keys for application authentication:

import requests

# Login as admin to manage API keys
db = QilbeeDB("http://localhost:7474")
db.login("admin", "Admin123!@#")

# Create a new API key
response = db.session.post(
    "http://localhost:7474/api/v1/api-keys",
    json={"name": "my-app-key"}
)
api_key_data = response.json()
api_key = api_key_data["key"]
key_id = api_key_data["id"]

print(f"Created API key: {api_key}")

# List all API keys
response = db.session.get("http://localhost:7474/api/v1/api-keys")
api_keys = response.json()["api_keys"]

# Delete an API key
db.session.delete(f"http://localhost:7474/api/v1/api-keys/{key_id}")

# Now use the API key in your application
app_db = QilbeeDB({
    "uri": "http://localhost:7474",
    "api_key": api_key
})

Memory Configuration

memory = db.agent_memory(
    "agent-001",
    max_episodes=10000,
    min_relevance=0.1,
    auto_consolidate=True,
    auto_forget=True,
    consolidation_threshold=5000,
    episodic_retention_days=30
)

Memory Persistence

QilbeeDB provides enterprise-grade memory persistence using RocksDB as the storage backend. All agent memories are automatically persisted to disk, ensuring durability across server restarts.

Key Features

• Automatic Durability: Episodes are automatically persisted when stored - no explicit save required
• Write-Ahead Logging (WAL): Ensures crash recovery and transaction safety
• LZ4 Compression: Reduces storage footprint with minimal overhead
• Agent Isolation: Episodes are stored in separate namespaces per agent

Verifying Persistence

from qilbeedb import QilbeeDB
from qilbeedb.memory import Episode

# Store episode
db = QilbeeDB("http://localhost:7474")
db.login("admin", "password")
memory = db.agent_memory("my-agent")

episode = Episode.conversation(
    "my-agent",
    "What is QilbeeDB?",
    "QilbeeDB is a graph database with agent memory..."
)
episode_id = memory.store_episode(episode)
print(f"Stored episode: {episode_id}")

# After server restart, episode is still available
stats = memory.get_statistics()
print(f"Total episodes: {stats.total_episodes}")

# Retrieve the persisted episode
retrieved = memory.get_episode(episode_id)
print(f"Retrieved: {retrieved.content}")

For more information, see the Memory Persistence documentation.

Audit Logging (Admin Only)

Query and monitor security events for compliance and debugging:

from qilbeedb import QilbeeDB

# Login as admin
db = QilbeeDB("http://localhost:7474")
db.login("admin", "Admin123!@#")

# Query all audit logs
result = db.get_audit_logs(limit=100)
print(f"Total events: {result['count']}")
for event in result['events']:
    print(f"{event['event_time']}: {event['event_type']} - {event['result']}")

# Filter by event type
login_events = db.get_audit_logs(event_type="login", limit=50)

# Filter by username
user_events = db.get_audit_logs(username="admin", limit=50)

# Filter by result
failed_events = db.get_audit_logs(result="unauthorized", limit=50)

# Filter by time range
recent_events = db.get_audit_logs(
    start_time="2025-01-01T00:00:00Z",
    end_time="2025-12-31T23:59:59Z",
    limit=100
)

# Convenience methods
failed_logins = db.get_failed_logins(limit=20)
user_activity = db.get_user_audit_events("admin", limit=50)
security_events = db.get_security_events(limit=50)

Audit Event Types

Category	Event Types
Authentication	login, logout, login_failed, token_refresh, token_refresh_failed
User Management	user_created, user_updated, user_deleted, password_changed
Role Management	role_assigned, role_removed
API Keys	api_key_created, api_key_revoked, api_key_used, api_key_validation_failed
Authorization	permission_denied, access_granted
Rate Limiting	rate_limit_exceeded
Account Lockout	account_lockout_triggered, account_locked, account_unlocked

Account Lockout Management (Admin Only)

Monitor and manage locked accounts due to failed login attempts:

from qilbeedb import QilbeeDB

# Login as admin
db = QilbeeDB("http://localhost:7474")
db.login("admin", "Admin123!@#")

# Get all locked accounts
locked = db.get_locked_accounts()
print(f"Total locked accounts: {locked['count']}")
for user, status in locked['locked_users']:
    print(f"  {user}: {status['lockout_remaining_seconds']}s remaining")

# Get lockout status for a specific user
status = db.get_lockout_status("suspicious_user")
print(f"User: {status['username']}")
print(f"  Locked: {status['status']['locked']}")
print(f"  Failed attempts: {status['status']['failed_attempts']}")
print(f"  Remaining attempts: {status['status']['remaining_attempts']}")
print(f"  Lockout count: {status['status']['lockout_count']}")

# Manually lock an account (e.g., suspicious activity)
result = db.lock_account("suspicious_user", reason="Suspicious activity detected")
print(f"Lock result: {result['success']}")

# Unlock an account
result = db.unlock_account("suspicious_user")
print(f"Unlock result: {result['success']}")

Lockout Behavior

• Default max attempts: 5 failed logins
• Initial lockout duration: 15 minutes
• Progressive lockout: Duration doubles with each subsequent lockout (up to 24 hours)
• Automatic unlock: Accounts unlock automatically after lockout period expires

Error Handling

from qilbeedb.exceptions import (
    QilbeeDBError,
    ConnectionError,
    QueryError,
    AuthenticationError
)

try:
    result = graph.query("INVALID QUERY")
except QueryError as e:
    print(f"Query failed: {e}")
except ConnectionError as e:
    print(f"Connection failed: {e}")
except QilbeeDBError as e:
    print(f"Database error: {e}")

Development

Running Tests

pip install -e .[dev]
pytest tests/

Code Formatting

black qilbeedb/
flake8 qilbeedb/
mypy qilbeedb/

License

Apache License 2.0

Support

• Documentation: https://docs.qilbeedb.com
• Issues: https://github.com/your-org/qilbeedb/issues
• Email: [e-mail]