Python API

Complete reference for the CortexaDB Python package.

CortexaDB

The main database class.

CortexaDB.open(path, **kwargs)

Opens or creates a database at the specified path.

Parameters:

Returns: CortexaDB

Example:

from cortexadb import CortexaDB
from cortexadb.providers.openai import OpenAIEmbedder

# With embedder
db = CortexaDB.open("agent.mem", embedder=OpenAIEmbedder())

# With manual dimension
db = CortexaDB.open("agent.mem", dimension=128, sync="batch")

# With HNSW indexing
db = CortexaDB.open("agent.mem", dimension=128, index_mode={
    "type": "hnsw", "m": 16, "ef_search": 50, "metric": "cos"
})

CortexaDB.replay(log_path, db_path, **kwargs)

Rebuilds a database from a replay log file.

Parameters:

Returns: CortexaDB

Example:

db = CortexaDB.replay("session.log", "restored.mem", strict=False)
report = db.last_replay_report

Memory Operations

.add(text=None, vector=None, metadata=None, collection=None)

Stores a new memory entry. If an embedder is configured and no embedding is provided, the text is auto-embedded.

Parameters:

Returns: int - The assigned memory ID

Example:

mid = db.add("User prefers dark mode")
mid = db.add("text", metadata={"source": "onboarding"})
mid = db.add("text", vector=[0.1, 0.2, ...], collection="agent_a")

.query(text=None, vector=None)

Starts a fluent query builder to search across the database.

Methods:

Example:

hits = db.search("What does the user prefer?", limit=5, use_graph=True)

for hit in hits:
    print(f"ID: {hit.id}, Score: {hit.score:.3f}")

.get(mid)

Retrieves a full memory entry by ID.

Parameters:

Returns: Memory

Raises: CortexaDBNotFoundError if the memory doesn't exist.

Example:

mem = db.get(42)
print(mem.id)          # 42
print(mem.content)     # b"User prefers dark mode"
print(mem.collection)  # "default"
print(mem.metadata)    # {"source": "onboarding"}
print(mem.created_at)  # 1709654400
print(mem.importance)  # 0.0
print(mem.embedding)   # [0.1, 0.2, ...] or None

.delete(mid)

Permanently deletes a memory and updates all indexes.

Parameters:

Raises: CortexaDBNotFoundError if the memory doesn't exist.

Example:

db.delete(42)

Graph Operations

.connect(from_id, to_id, relation)

Creates a directed edge between two memories.

Parameters:

Example:

db.connect(1, 2, "relates_to")
db.connect(1, 3, "caused_by")

Both memories must be in the same collection. Cross-collection edges are forbidden.

.get_neighbors(mid)

Returns all outgoing edges from a memory.

Parameters:

Returns: list[Edge] where each Edge has to (int) and relation (str) fields.

Example:

neighbors = db.get_neighbors(1)
for edge in neighbors:
    print(f"→ {edge[0]} ({edge[1]})")

Document Ingestion

.ingest(text, strategy="recursive", chunk_size=512, overlap=50, metadata=None, collection=None)

Chunks text and stores each chunk as a memory.

Parameters:

Returns: list[int] - Memory IDs of stored chunks

.load(file_path, strategy="markdown", chunk_size=512, overlap=50, metadata=None, collection=None)

Loads a file, chunks it, and stores each chunk.

Parameters:

Supported formats: .txt, .md, .json, .docx (requires cortexadb[docs]), .pdf (requires cortexadb[pdf])

Example:

db.load("README.md", strategy="markdown")
db.load("paper.pdf", strategy="recursive", chunk_size=1024)

Collections

.collection(name, readonly=False)

Returns a scoped view of the database for a specific collection.

Parameters:

Returns: Collection

Example:

col = db.collection("agent_a")
mid = col.add("text")
hits = col.query("query").execute()
col.delete(mid)
col.ingest("long text")

Maintenance Operations

.compact()

Reclaims disk space by removing deleted entries from segment files.

.flush()

Forces all pending writes to be synced to disk.

.checkpoint()

Creates a binary snapshot of the current state and truncates the WAL. Also saves the HNSW index if using HNSW mode.

.stats()

Returns database statistics.

Returns: Stats

stats = db.stats()
print(stats.entries)              # Total memory count
print(stats.indexed_embeddings)   # Embeddings in vector index
print(stats.wal_length)           # WAL file size in bytes
print(stats.vector_dimension)     # Configured vector dimension
print(stats.storage_version)      # Storage format version

Replay Properties

.last_replay_report

Diagnostic report from the most recent replay() call.

Type: dict

.last_export_replay_report

Diagnostic report from the most recent export_replay() call.

Type: dict

Export

.export_replay(path)

Exports the current database state as a replay log.

Parameters:

Types

Hit

Query result from .query().execute().

Memory

Full memory entry from .get().

Stats

Database statistics from .stats().

ChunkResult

Result from chunk().

Standalone Functions

chunk(text, strategy="recursive", chunk_size=512, overlap=50)

Chunks text without storing it.

Returns: list[ChunkResult]

from cortexadb import chunk

chunks = chunk("Long text...", strategy="recursive", chunk_size=512, overlap=50)

Exceptions