Tutorial: Build Your First SochDB Application

Time: 15 minutes Difficulty: Beginner Prerequisites: Python 3.9+ or Rust 1.85+ (edition 2024)

In this tutorial you'll build a small agent-memory store that keeps user preferences and conversation history, runs semantic search over past messages with vector embeddings, and assembles a token-efficient LLM context — all on an embedded SochDB database (no server required).

Which Python package?

pip install sochdb ships two importable packages under the same name. This tutorial uses the pure-Python ctypes SDK (0.5.9) and its Database class — the broad embedded + server SDK. The separate native PyO3 engine (2.0.3) exposes lower-level primitives such as HnswIndex/BM25Index/TableDatabase; you do not need it here.

---

What You'll Build

A personal-assistant memory store with:

• User preferences storage (path-based keys)
• Conversation history tracking (prefix scans)
• Semantic search with vector embeddings (embedded HNSW index)
• Token-efficient LLM context assembly

---

Step 1: Project Setup

Python

# Create project directory
mkdir agent-memory && cd agent-memory

# Create virtual environment
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# Install dependencies (numpy is used to generate example embeddings)
pip install sochdb numpy

Rust

# Create new project
cargo new agent-memory
cd agent-memory

# Add the SochDB crate (core engine 2.0.3)
cargo add sochdb

Your Cargo.toml should now contain:

[dependencies]
sochdb = "2.0.3"

License

The core engine (the sochdb crate, server, and MCP) is AGPL-3.0-or-later (commercial licensing available). The language SDKs (Python, Node.js, Go) are Apache-2.0.

---

Step 2: Initialize the Database

Python

Create main.py:

from sochdb import Database
import json

# Open an embedded database (creates it if it doesn't exist).
db = Database.open("./agent_memory")

print("Database initialized at ./agent_memory")

db.close()

Run it:

python main.py
# Output: Database initialized at ./agent_memory

Rust

Create src/main.rs:

use sochdb::Connection;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Connection is the durable, WAL-backed embedded database (aliased as Database).
    let conn = Connection::open("./agent_memory")?;
    println!("Database initialized at ./agent_memory");
    Ok(())
}

Run it:

cargo run
# Output: Database initialized at ./agent_memory

---

Step 3: Store User Preferences

SochDB keys and values are bytes. We store each preference under a path-style key so individual fields can be read or updated independently, and read them all back with a prefix scan.

Python

Add to main.py:

def store_preferences(db, user_id: str, prefs: dict):
    """Store user preferences using path-based keys."""
    # The transaction auto-commits on clean exit and auto-aborts on exception.
    with db.transaction() as txn:
        for key, value in prefs.items():
            path = f"users/{user_id}/preferences/{key}"
            txn.put(path.encode(), json.dumps(value).encode())
    print(f"Stored preferences for user {user_id}")


def get_preferences(db, user_id: str) -> dict:
    """Retrieve all preferences for a user via a prefix scan."""
    prefs = {}
    prefix = f"users/{user_id}/preferences/"
    for key, value in db.scan_prefix(prefix.encode()):
        pref_name = key.decode().split("/")[-1]
        prefs[pref_name] = json.loads(value.decode())
    return prefs


# Test it
db = Database.open("./agent_memory")

store_preferences(db, "alice", {
    "theme": "dark",
    "language": "en",
    "notifications": True,
    "model": "claude-sonnet",
})

prefs = get_preferences(db, "alice")
print(f"Alice's preferences: {prefs}")

db.close()

Run it:

python main.py
# Output:
# Stored preferences for user alice
# Alice's preferences: {'theme': 'dark', 'language': 'en', ...}

Prefix-scan safety

db.scan_prefix(prefix) enforces a minimum 2-byte prefix and raises ValueError for shorter prefixes — this prevents accidental full-database (cross-tenant) scans. Use db.scan(start, end) for explicit range scans when you genuinely need them.

Rust

Add to src/main.rs (the durable Connection exposes path helpers and prefix scans directly):

use sochdb::Connection;

fn store_preference(conn: &Connection, user_id: &str, key: &str, value: &str)
    -> Result<(), Box<dyn std::error::Error>>
{
    let path = format!("users/{user_id}/preferences/{key}");
    conn.put_path(&path, value.as_bytes())?;
    Ok(())
}

fn get_preferences(conn: &Connection, user_id: &str)
    -> Result<Vec<(String, String)>, Box<dyn std::error::Error>>
{
    let prefix = format!("users/{user_id}/preferences/");
    let mut prefs = Vec::new();
    for (path, value) in conn.scan_path(&prefix)? {
        let name = path.rsplit('/').next().unwrap_or(&path).to_string();
        prefs.push((name, String::from_utf8_lossy(&value).into_owned()));
    }
    Ok(prefs)
}

---

Step 4: Track Conversation History

Messages are stored under a per-user history/ prefix keyed by an ISO timestamp, so a prefix scan returns the whole conversation and we can sort it client-side.

from datetime import datetime

def add_message(db, user_id: str, role: str, content: str) -> str:
    """Append a message to conversation history."""
    timestamp = datetime.now().isoformat()
    message_id = timestamp.replace(":", "-").replace(".", "-")

    with db.transaction() as txn:
        path = f"users/{user_id}/history/{message_id}"
        message = {"role": role, "content": content, "timestamp": timestamp}
        txn.put(path.encode(), json.dumps(message).encode())

    return message_id


def get_recent_messages(db, user_id: str, limit: int = 10) -> list:
    """Get recent messages (most recent first)."""
    messages = []
    prefix = f"users/{user_id}/history/"
    for _, value in db.scan_prefix(prefix.encode()):
        messages.append(json.loads(value.decode()))
    messages.sort(key=lambda m: m["timestamp"], reverse=True)
    return messages[:limit]


# Test it
db = Database.open("./agent_memory")

add_message(db, "alice", "user", "What is the capital of France?")
add_message(db, "alice", "assistant", "The capital of France is Paris.")
add_message(db, "alice", "user", "What's the population?")
add_message(db, "alice", "assistant", "Paris has about 2.1 million people in the city proper.")

history = get_recent_messages(db, "alice", limit=2)
print("Recent messages:")
for msg in history:
    print(f"  [{msg['role']}]: {msg['content'][:50]}...")

db.close()

---

Step 5: Add Vector Search (Semantic Memory)

SochDB ships an embedded HNSW index. The 0.5.9 SDK exposes it directly on the database with three helpers: db.create_index(...), db.insert_vectors(...), and db.search(...). Vector indexing is not done through SQL DDL — there is no CREATE INDEX in the pure-Python SQL engine; use db.create_index instead.

import numpy as np

def create_embeddings(texts: list[str]) -> np.ndarray:
    """Create mock embeddings (replace with a real model in production).

    In production use sentence-transformers, an embedding API, etc.
    Here we generate deterministic 384-dim float32 vectors.
    """
    rng = np.random.default_rng(42)
    return rng.standard_normal((len(texts), 384)).astype(np.float32)


def build_memory_index(db, user_id: str):
    """Build a vector index from conversation history."""
    ids, messages = [], []
    prefix = f"users/{user_id}/history/"
    for i, (_, value) in enumerate(db.scan_prefix(prefix.encode())):
        msg = json.loads(value.decode())
        ids.append(i)               # integer ids map back to messages
        messages.append(msg["content"])

    if not messages:
        print("No messages to index")
        return None

    embeddings = create_embeddings(messages)

    # Create an HNSW index (dimension 384). Defaults: max_connections=32,
    # ef_construction=256 — sensible for general-purpose embeddings.
    db.create_index("alice_memory", dimension=384)
    db.insert_vectors("alice_memory", ids, embeddings.tolist())

    print(f"Built index 'alice_memory' with {len(messages)} vectors")
    return messages


def search_memory(db, messages: list[str], query: str, k: int = 3):
    """Search conversation memory semantically."""
    query_embedding = create_embeddings([query])[0].tolist()
    # db.search returns a list of (id, distance) tuples.
    results = db.search("alice_memory", query_embedding, k=k)
    return [(messages[idx], dist) for idx, dist in results]


# Test it
db = Database.open("./agent_memory")
messages = build_memory_index(db, "alice")
if messages:
    for text, distance in search_memory(db, messages, "France capital city"):
        print(f"  ({distance:.3f}) {text}")
db.close()

note

With random mock embeddings the ranking is meaningless — swap create_embeddings for a real embedding model to get useful semantic results. For larger workloads or build-once/query-many pipelines, the native 2.0.3 package's HnswIndex and build_index_from_numpy give you finer control.

---

Step 6: Store Structured Data with SQL (Optional)

The 0.5.9 SDK includes a small KV-backed SQL engine. It supports a focused subset — CREATE TABLE, DROP TABLE, INSERT, SELECT (with WHERE/ORDER BY/LIMIT), UPDATE, and DELETE — over the types INT, TEXT, FLOAT, BOOL, and BLOB. Use it for small relational lookups alongside your KV data.

db = Database.open("./agent_memory")

# DDL + DML. There is no CREATE INDEX here — vector indexing is db.create_index.
db.execute("CREATE TABLE sessions (id INT, user TEXT, turns INT)")
db.execute("INSERT INTO sessions VALUES (1, 'alice', 4)")
db.execute("INSERT INTO sessions VALUES (2, 'bob', 9)")

# SELECT returns a SQLQueryResult with .rows (list of dicts), .columns,
# and .rows_affected.
result = db.execute("SELECT user, turns FROM sessions WHERE turns > 5 ORDER BY turns")
print(result.columns)   # ['user', 'turns']
for row in result.rows:
    print(row)          # {'user': 'bob', 'turns': 9}

db.close()

---

Step 7: Assemble LLM Context

Now combine preferences and history into a compact prompt. SochDB's native TOON format is well suited to LLM context — it typically uses 40–66% fewer tokens than equivalent JSON.

def build_llm_context(db, user_id: str, current_query: str) -> str:
    """Build a token-efficient context string from stored memory."""
    parts = []

    # 1. System prompt (always included)
    parts.append("You are a helpful assistant with access to user preferences and history.")

    # 2. User preferences as a compact TOON row
    prefs = get_preferences(db, user_id)
    if prefs:
        fields = ",".join(prefs.keys())
        values = ",".join(str(v) for v in prefs.values())
        parts.append(f"prefs[1]{{{fields}}}:\n{values}")

    # 3. Recent history in chronological order
    history = get_recent_messages(db, user_id, limit=5)
    if history:
        lines = [f"{m['role']}: {m['content']}" for m in reversed(history)]
        parts.append("Recent conversation:\n" + "\n".join(lines))

    # 4. Current query (always included)
    parts.append(f"Current query: {current_query}")

    return "\n\n".join(parts)


# Test it
db = Database.open("./agent_memory")

context = build_llm_context(db, "alice", "Tell me more about Paris")
print("=== LLM Context ===")
print(context)
print(f"\nContext length: {len(context)} characters")

db.close()

Native context assembly

The hand-rolled assembly above is fine for getting started. SochDB also has first-class, server-side context assembly with exact-BPE token budgeting via the AgentMemory API (write_episode / search / compile_context). That path requires a running SochDB server (it talks to a SochDBClient), so it is out of scope for this embedded tutorial — see the Python SDK guide.

---

Step 8: Complete Application

Here's the complete embedded main.py:

#!/usr/bin/env python3
"""Agent Memory System with SochDB — a simple personal-assistant memory store."""

from sochdb import Database
from datetime import datetime
import json


class AgentMemory:
    def __init__(self, path: str = "./agent_memory"):
        self.db = Database.open(path)

    def close(self):
        self.db.close()

    def store_preference(self, user_id: str, key: str, value):
        path = f"users/{user_id}/preferences/{key}"
        with self.db.transaction() as txn:
            txn.put(path.encode(), json.dumps(value).encode())

    def get_preferences(self, user_id: str) -> dict:
        prefs = {}
        prefix = f"users/{user_id}/preferences/"
        for key, value in self.db.scan_prefix(prefix.encode()):
            pref_name = key.decode().split("/")[-1]
            prefs[pref_name] = json.loads(value.decode())
        return prefs

    def add_message(self, user_id: str, role: str, content: str) -> str:
        timestamp = datetime.now().isoformat()
        message_id = timestamp.replace(":", "-").replace(".", "-")
        with self.db.transaction() as txn:
            path = f"users/{user_id}/history/{message_id}"
            txn.put(
                path.encode(),
                json.dumps({"role": role, "content": content, "ts": timestamp}).encode(),
            )
        return message_id

    def get_history(self, user_id: str, limit: int = 10) -> list:
        messages = []
        prefix = f"users/{user_id}/history/"
        for _, value in self.db.scan_prefix(prefix.encode()):
            messages.append(json.loads(value.decode()))
        messages.sort(key=lambda m: m["ts"], reverse=True)
        return messages[:limit]

    def build_context(self, user_id: str, query: str) -> str:
        parts = [
            "System: You are a helpful assistant.",
            f"Preferences: {json.dumps(self.get_preferences(user_id))}",
        ]
        history = self.get_history(user_id, limit=5)
        if history:
            parts.append("Recent history:")
            for msg in reversed(history):
                parts.append(f"  {msg['role']}: {msg['content']}")
        parts.append(f"Query: {query}")
        return "\n".join(parts)


def main():
    print("SochDB Agent Memory Demo\n")

    memory = AgentMemory()
    user = "demo_user"

    # Store preferences
    memory.store_preference(user, "name", "Demo User")
    memory.store_preference(user, "theme", "dark")
    print(f"Stored preferences: {memory.get_preferences(user)}")

    # Add conversation
    memory.add_message(user, "user", "Hello, who are you?")
    memory.add_message(user, "assistant", "I'm your AI assistant with persistent memory!")
    memory.add_message(user, "user", "What can you remember?")
    print(f"Added {len(memory.get_history(user))} messages")

    # Build context
    context = memory.build_context(user, "Summarize our conversation")
    print(f"\nLLM Context ({len(context)} chars):\n")
    print(context)

    memory.close()
    print("\nDemo complete!")


if __name__ == "__main__":
    main()

Run the complete application:

python main.py

Expected output:

SochDB Agent Memory Demo

Stored preferences: {'name': 'Demo User', 'theme': 'dark'}
Added 3 messages

LLM Context (298 chars):

System: You are a helpful assistant.
Preferences: {"name": "Demo User", "theme": "dark"}
Recent history:
  user: Hello, who are you?
  assistant: I'm your AI assistant with persistent memory!
  user: What can you remember?
Query: Summarize our conversation

Demo complete!

---

What You Learned

Concept	What You Did
Path-based access	Used users/{id}/preferences/{key} for granular, O(path) lookups
Transactions	Wrapped writes in with db.transaction() for atomicity
Prefix scans	Used db.scan_prefix(prefix) for safe range queries
Vector search	Built an embedded HNSW index with db.create_index / db.insert_vectors / db.search
SQL subset	Ran CREATE TABLE/INSERT/SELECT via db.execute
Context assembly	Built a token-efficient LLM prompt

---

Next Steps

Goal	Resource
Add real embeddings	Vector Search Tutorial
Use with Claude/MCP	MCP Integration
Production deployment	Deployment Guide
Performance tuning	Performance Guide

---

Troubleshooting

ModuleNotFoundError: No module named 'sochdb'

pip install --upgrade sochdb

ValueError from scan_prefix

scan_prefix requires a prefix of at least 2 bytes to prevent accidental full-database scans. Pass a longer prefix, or use db.scan(start, end) for an explicit range.

Permission denied on the database path

mkdir -p ./agent_memory
chmod 755 ./agent_memory

Transaction failed

Transactions auto-abort on exceptions and may raise TransactionConflictError under serializable-snapshot-isolation conflicts. Retry the transaction, and check that your data types match the schema.

---

Tutorial complete — you've built a working embedded agent-memory system with SochDB.