qilbeedb 0.1.0

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

Basic Graph Operations

from qilbeedb import QilbeeDB

# Connect to database
db = QilbeeDB("http://localhost:7474")

# 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

from qilbeedb.memory import Episode

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

# Store conversation
episode = Episode.conversation(
    "agent-001",
    "What is the capital of France?",
    "The capital of France is Paris."
)
memory.store_episode(episode)

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

# Retrieve recent memories
recent = memory.get_recent_episodes(10)
for ep in recent:
    print(ep.content)

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

# 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)

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

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.

Methods:

• store_episode(episode: Episode) -> str - Store episode
• get_episode(episode_id: str) -> Episode - Get episode
• get_recent_episodes(limit: int) -> List[Episode] - Get recent episodes
• search_episodes(query: str, limit: int) -> List[Episode] - Search episodes
• get_statistics() -> MemoryStatistics - Get statistics
• consolidate() -> int - Consolidate memory
• forget(min_relevance: float) -> int - Forget low-relevance episodes
• clear() -> bool - Clear all episodes

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

db = QilbeeDB(
    uri="http://localhost:7474",
    auth=("username", "password"),
    timeout=30,
    verify_ssl=True
)

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
)

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]