Stores

Stores are the persistence layer where extracted beliefs are saved. All stores implement the Store protocol, providing a uniform API for adding, querying, searching, and removing beliefs. You can configure the store via TrackerConfig(store_type="...") or inject a store instance directly into the BeliefTracker constructor.

Choosing a Store

Store	Persistence	Multi-Worker	Best For	Setup
SQLite	✅ Disk file	❌ Single process	Local dev, prototyping, single-server apps	Zero — included with Python
Redis	✅ Redis server	✅ Distributed	Production, multi-worker, load-balanced	Requires Redis server
In-Memory	❌ Lost on restart	❌ Single process	Unit testing, ephemeral sessions	Zero — pure Python

Store Protocol Methods

All stores implement these async methods. You can use them directly if you need to interact with the store outside of the normal tracking pipeline:

Method	Signature	Returns	Description
`add_belief`	`(session_id: str, belief: Belief)`	`None`	Add or upsert a belief into the store. If a belief with the same `session_id`, `subject`, and `predicate` already exists, it is overwritten.
`get_beliefs`	`(session_id: str, conversation_id: Optional[str])`	`list[Belief]`	Retrieve all beliefs for a session. Optionally filter by `conversation_id` for multi-thread sessions.
`search_beliefs`	`(session_id: str, embedding: list[float], threshold: float, limit: int)`	`list[Belief]`	Semantic similarity search. Returns beliefs whose embedding vectors have cosine similarity ≥ `threshold` with the query vector. Used internally by the contradiction detector.
`remove_belief`	`(session_id: str, subject: str, predicate: str)`	`None`	Remove a specific belief identified by its session, subject, and predicate keys.
`update_belief`	`(session_id: str, belief: Belief)`	`None`	Update an existing belief (e.g., update confidence, value, or `last_referenced_at` timestamp).
`clear`	`(session_id: str)`	`None`	Delete all beliefs for a session. Used by `clear_session()` for GDPR erasure.
`belief_count`	`(session_id: str)`	`int`	Count beliefs without loading them into memory. Uses `SELECT COUNT(*)` (SQLite), `HLEN` (Redis), or `len()` (Memory). Much faster than `len(await get_beliefs(...))`.
`health_check`	`()`	`bool`	Verify the store backend is reachable and operational. Returns `True` if healthy. Uses `SELECT 1` (SQLite), `PING` (Redis), or always `True` (Memory).

SQLite Store (Default)

The SQLite store uses aiosqlite for fully async, non-blocking database access. Data is persisted to a single file on disk, making it ideal for single-process applications, local development, and prototyping. No external services required.

Constructor Parameters

Parameter	Type	Default	Description
`db_path`	`str`	`"beliefstate.db"`	Path to the SQLite database file. The file is created automatically if it doesn't exist. Use `":memory:"` for a non-persistent in-memory database (useful for testing).

pythonfrom beliefstate import SQLiteStore, BeliefTracker, TrackerConfig

# Via config (automatic initialization)
config = TrackerConfig(
    store_type="sqlite",
    store_kwargs={"db_path": "user_beliefs.db"}
)
tracker = BeliefTracker(config=config, adapter=adapter)

# Or inject a store instance directly
store = SQLiteStore(db_path="user_beliefs.db")
tracker = BeliefTracker(config=config, store=store)

# In-memory SQLite for unit testing
store = SQLiteStore(db_path=":memory:")

ℹ️ Database Schema & WAL Mode: The database creates a schema optimized for session isolation and turn order:

CREATE TABLE IF NOT EXISTS beliefs (
    session_id TEXT,
    subject TEXT,
    predicate TEXT,
    value TEXT,
    confidence REAL,
    turn INTEGER,
    source TEXT,
    embedding TEXT,
    created_at TIMESTAMP,
    last_referenced_at TIMESTAMP,
    PRIMARY KEY (session_id, subject, predicate)
);

Composite primary key (session_id, subject, predicate) prevents duplicate beliefs and enables efficient upserts.
WAL mode (Write-Ahead Logging) is enabled on startup, allowing concurrent readers (your user threads) to query beliefs without being blocked by background write tasks.
Embedding storage: Embedding vectors are serialized as JSON text in the embedding column. Cosine similarity search is performed in Python (not via SQL extension).

Performance Characteristics

Read latency: ~1ms for get_beliefs() with a few hundred beliefs per session.
Write latency: ~2-5ms for add_belief() (single INSERT OR REPLACE).
Similarity search: O(N) — loads all beliefs for the session and computes cosine similarity in Python. Efficient for up to ~10,000 beliefs per session.
Limitation: Single-writer. Multiple processes writing to the same file may cause lock contention. Use Redis for multi-worker deployments.

Redis Store

The Redis store is designed for distributed, multi-worker production deployments. When your application runs multiple processes behind a load balancer (e.g., Gunicorn with multiple workers, Kubernetes pods), all workers share the same Redis-backed belief state. Redis also provides native TTL support for automatic session expiry.

Constructor Parameters

Parameter	Type	Default	Description
`redis_url`	`str`	`"redis://localhost:6379/0"`	Redis connection URI. Supports standard Redis URIs: `redis://host:port/db`, `rediss://` (TLS), `redis+sentinel://`. Password auth: `redis://:password@host:port/db`.

pythonfrom beliefstate import RedisStore

store = RedisStore(redis_url="redis://localhost:6379/0")

# Set TTL on all beliefs in a session (auto-expire after 24 hours)
await store.set_session_ttl("user_123", ttl_seconds=86400)

ℹ️ Redis Hash Layout & Session Expiry: Data is organized cleanly to prevent key namespace bloat:

Each session is stored as a single Redis HASH at key beliefstate:session:{session_id}.
Each belief within the hash uses subject::predicate as the hash field and the serialized belief JSON as the value.
set_session_ttl() calls Redis EXPIRE on the parent HASH key, enabling automatic expiration of the entire session state at once.
belief_count() uses HLEN — O(1) operation, no deserialization needed.
health_check() uses PING — verifies Redis connectivity.

Performance Characteristics

Read/Write latency: ~0.5ms per operation (network round-trip dependent).
Multi-worker safe: Redis handles concurrent access natively. No file locking issues.
Similarity search: O(N) — all beliefs for a session are loaded and compared in Python (same as SQLite). For very large sessions, consider using Redis vector search extensions.
Requires: A running Redis server (v5.0+ recommended).

In-Memory Store

The in-memory store keeps all beliefs in Python process memory using an ordered dictionary. It provides the fastest possible access but data is lost on process restart. Ideal for unit testing, short-lived sessions, or applications where belief persistence isn't needed.

Constructor Parameters

Parameter	Type	Default	Description
`max_bytes`	`int`	`104857600` (100 MB)	Maximum memory budget in bytes. When total belief data exceeds this limit, the store evicts least-recently-used (LRU) beliefs from the oldest sessions until the footprint fits. Set to `0` for unlimited (not recommended in production).

pythonfrom beliefstate import InMemoryBeliefStore

store = InMemoryBeliefStore(max_bytes=100 * 1024 * 1024)  # 100 MB limit

# Get usage stats
stats = store.get_stats()
print(stats)  # {"total_sessions": 5, "total_beliefs": 142, "utilization_percent": 12.3}

ℹ️ LRU Eviction & Footprint Estimation: The in-memory store uses a Python collections.OrderedDict:

Usage tracking: Reading or writing a belief updates its usage rank, moving the key to the end of the dictionary (MRU position).
Memory estimation: Each belief's memory footprint is estimated by checking the byte length of its JSON string representation (len(belief.model_dump_json())).
Eviction: When the size limit is reached, the store evicts least-recently-used beliefs from the oldest sessions until the memory footprint fits inside the max_bytes budget.

⚠️ Not Durable: In-memory store is NOT persistent. All beliefs are lost on process restart or crash. Use SQLite or Redis for any application that needs belief persistence.

Implementing a Custom Store

You can implement your own store by following the Store protocol. Your class must implement all the methods listed in the Store Protocol Methods table above. Inject your custom store into the tracker:

pythonclass MyCustomStore:
    async def add_belief(self, session_id: str, belief: Belief) -> None: ...
    async def get_beliefs(self, session_id: str, conversation_id=None) -> list[Belief]: ...
    async def search_beliefs(self, session_id, embedding, threshold, limit) -> list[Belief]: ...
    async def remove_belief(self, session_id, subject, predicate) -> None: ...
    async def update_belief(self, session_id, belief) -> None: ...
    async def clear(self, session_id) -> None: ...
    async def belief_count(self, session_id) -> int: ...
    async def health_check(self) -> bool: ...

tracker = BeliefTracker(config=config, store=MyCustomStore())