0.1.10

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

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.