aifastdb 0.1.5

Python bindings for AiFastDb - High-performance AI-friendly database engine

aifastdb

Python bindings for AiFastDb — a high-performance AI-friendly database engine written in Rust.

Features

• Core Database — Key-value store with HNSW vector search, batch operations, and hybrid (text + vector) recall
• SocialGraphV2 — Social network graph with path finding, network analysis, family trees, and concurrent sharding
• DocumentStore — JSONL-based document store with tag/type indexing and async flush
• Federation — Heterogeneous storage federation (Graph + Document + Vector) with cross-database queries and 2PC transactions
• Embedding — Built-in embedding generation (dummy / MiniLM / BGE / Gemma) with numpy output
• Reasoning — LLM-based reasoning via Ollama or OpenAI (intent, logic, summarize)
• pandas integration — One-line DataFrame conversion for any query result

Installation

pip install aifastdb

# With optional extras:
pip install aifastdb[pandas]    # pandas DataFrame support
pip install aifastdb[numpy]     # numpy array support
pip install aifastdb[all]       # everything

Requirements

• Python >= 3.8
• No additional dependencies for core functionality (self-contained Rust binary)

Quick Start

Core Database

from aifastdb import AiFastDb

with AiFastDb("./my_db") as db:
    # Store a memory
    db.remember("doc1", "The quick brown fox", tags=["animal", "classic"])

    # Retrieve by ID
    doc = db.get("doc1")
    print(doc["content"])  # "The quick brown fox"

    # Hybrid search (text + vector)
    results = db.recall(query="brown fox")
    for r in results:
        print(f"  {r['node']['id']}  score={r['score']:.3f}")

    # Delete
    db.forget("doc1")

Social Graph

from aifastdb import SocialGraphV2

with SocialGraphV2("./social_data") as sg:
    # Create people
    alice = sg.add_person({"name": "Alice", "age": 30})
    bob   = sg.add_person({"name": "Bob",   "age": 28})
    carol = sg.add_person({"name": "Carol", "age": 32})

    # Add friendships
    sg.add_friend(alice["id"], bob["id"])
    sg.add_friend(bob["id"], carol["id"])

    # Find shortest path
    path = sg.find_path(alice["id"], carol["id"])
    print(f"Path hops: {path['hops']}")

    # Mutual friends
    mutual = sg.get_mutual_friends(alice["id"], carol["id"])
    print(f"Mutual friends: {[m['name'] for m in mutual]}")

    # Network stats
    stats = sg.get_network_stats(bob["id"])
    print(f"Bob's degree centrality: {stats['degree_centrality']:.2f}")

Historical Graph

from aifastdb import SocialGraphV2

with SocialGraphV2("./historical_data") as sg:
    session = sg.begin_import_session(
        {
            "dataset_key": "shiji",
            "source_name": "Shiji",
            "import_batch_id": "batch-001",
        }
    )

    batch = sg.add_historical_entities_batch_detailed(
        [
            {
                "entity_id": "hist:person:hanxin",
                "entity_type": "person",
                "name": "Han Xin",
            },
            {
                "entity_type": "person",
                "name": "  ",
            },
        ],
        {"continue_on_error": True, "emit_checkpoint_every": 1},
        session["session_id"],
    )
    print(batch["created"], batch["failed"], batch["items"][1]["reason_code"])

    timeline = sg.get_entity_timeline("hist:person:hanxin")
    buckets = sg.get_entity_timeline_buckets("hist:person:hanxin")
    projection = sg.export_entity_history_projection("hist:person:hanxin")

The historical graph layer in SocialGraphV2 includes import sessions, batch diagnostics, first-class claim/fact/evidence entities, entity-centered timeline bundles, conflict clusters, projection export, and historical vector or hybrid retrieval helpers.

Document Store

from aifastdb import DocumentStore

with DocumentStore("./docs.jsonl") as store:
    store.put("note-1", "Meeting at 3pm", doc_type="note", tags=["work"])
    store.put("note-2", "Buy groceries",  doc_type="note", tags=["personal"])

    # Query by tag
    work_notes = store.find_by_tag("work")
    print(f"Work notes: {len(work_notes)}")

    # All documents
    all_docs = store.all()
    print(f"Total documents: {len(all_docs)}")

Federation (Multi-Store)

from aifastdb import Federation

with Federation() as fed:
    fed.register_graph("social", "./social_data")
    fed.register_document("docs", "./docs.jsonl")

    # Query across stores
    result = fed.query("social", {"target": "person", "limit": 10})
    print(f"Found {result['total']} persons")

    # Write to document store
    fed.write("docs", {
        "op": "CreateDocument",
        "id": "fed-doc-1",
        "content": "Created via federation",
        "doc_type": "note",
    })

    # Cross-store transaction
    tx = fed.begin_transaction(["social", "docs"])
    tx.write("social", {"op": "CreateEntity", "name": "Eve", "entity_type": "person"})
    tx.write("docs",   {"op": "CreateDocument", "id": "eve-doc", "content": "Eve's note"})
    tx.commit()

Embedding Engine

from aifastdb import EmbeddingEngine
import numpy as np

engine = EmbeddingEngine()  # dummy 384d engine

# Single embedding
vec = engine.embed("Hello, world!")
print(f"Dimension: {len(vec)}")

# Numpy output
arr = engine.embed_numpy("Hello, world!")
print(f"Shape: {arr.shape}, dtype: {arr.dtype}")

# Batch
vecs = engine.embed_batch(["Hello", "World", "Foo"])
print(f"Batch size: {len(vecs)}")

# Cosine similarity
sim = EmbeddingEngine.cosine_similarity(
    engine.embed("cat"),
    engine.embed("dog"),
)
print(f"Similarity: {sim:.4f}")

Perception Pipeline (Embed + Remember)

from aifastdb import PerceptionPipeline, DocumentStore

pipe = PerceptionPipeline()

# Auto-embed content
result = pipe.remember_with_embedding(
    "The mitochondria is the powerhouse of the cell",
    "bio-001",
    tags=["biology"],
)
print(f"Embedding dim: {result['dimension']}")

# Store the result in a DocumentStore
with DocumentStore("./knowledge.jsonl") as store:
    store.put(
        result["id"],
        result["content"],
        tags=result["tags"],
        doc_type=result["doc_type"],
    )

pandas Integration

from aifastdb import SocialGraphV2, DocumentStore
from aifastdb.pandas_ext import to_dataframe, graph_to_dataframes

# SocialGraph → DataFrame
with SocialGraphV2("./social_data") as sg:
    persons_df = to_dataframe(sg.list_persons())
    print(persons_df[["id", "name", "age"]])

    entities_df, relations_df = graph_to_dataframes(sg.export_graph())
    print(f"Entities: {len(entities_df)}, Relations: {len(relations_df)}")

# DocumentStore → DataFrame
with DocumentStore("./docs.jsonl") as store:
    docs_df = to_dataframe(store.all())
    print(docs_df.head())

LLM Gateway

from aifastdb import LlmGateway

gw = LlmGateway("./gateway_data")
gw.register_provider(
    "ollama_local",
    "Ollama Local",
    "ollama",
    "http://127.0.0.1:11434/v1",
    protocol="openai_compat",
)
gw.register_model(
    "qwen3.5:27b",
    "ollama_local",
    "qwen3.5:27b",
    "Qwen 3.5 27B",
)

conv = gw.create_conversation(
    "user-1",
    "ollama_local",
    "qwen3.5:27b",
    "qwen3.5:27b",
    title="demo",
)
reply = gw.chat(conv["id"], "user-1", "请把这句话整理成书面语")
print(reply["content"])

LLM Streaming Helper

from aifastdb import LlmGateway
from aifastdb.llm_ext import chat_stream

gw = LlmGateway("./gateway_data")

for event in chat_stream(gw, "conv-1", "user-1", "Hello!"):
    print(event["event"], event["data"])

API Reference

Core Classes

Class	Description
AiFastDb	Core database with collections, CRUD, and hybrid search
Collection	Named collection within AiFastDb
SocialGraphV2	Social graph with path finding and network analysis
DocumentStore	JSONL document store with tag/type indexing
Federation	Multi-store federation with unified query/write
FederatedTransaction	Cross-store 2PC transaction
EmbeddingEngine	Vector embedding generation
ReasoningEngine	LLM-based reasoning (Ollama/OpenAI)
PerceptionPipeline	Combined embedding + reasoning pipeline
LlmGateway	Provider/model routing, conversations, chat, memory and skills

Exception Hierarchy

RuntimeError
└── AiFastDbError
    ├── StoreNotFoundError
    ├── RecordNotFoundError
    ├── QueryError
    ├── WriteError
    ├── TransactionError
    ├── StorageError
    ├── CircuitBreakerError
    └── RateLimitError

Development

# Install Rust toolchain
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh

# Clone and build
git clone https://github.com/aifastdb/aifastdb.git
cd aifastdb/packages/python

# Create venv and install in dev mode
python -m venv .venv
source .venv/bin/activate  # or .venv\Scripts\activate on Windows
pip install maturin numpy pandas pytest
maturin develop --release

# Run tests
python -m pytest tests/ -v

If you have multiple Python interpreters or virtual environments, make sure you activate the same interpreter/environment you will use for imports/tests before running maturin develop. For an explicit target interpreter flow, build a wheel and install it with that interpreter:

maturin build --release
/path/to/python -m pip install --force-reinstall target/wheels/aifastdb-*.whl

packages/python/pyproject.toml currently enables the llm-gateway feature for editable builds, so ReasoningEngine, LlmGateway, and aifastdb.llm_ext are available in the local development environment after maturin develop.

License

See LICENSE for details. Free for personal, educational, and non-commercial use. Commercial use requires a separate license.