YantrikDB Python API Reference — record, recall, think, graph

YantrikDB

The main database class.

Constructor

db = YantrikDB(
    db_path="memory.db",
    embedding_dim=384,
    embedder=None,       # optional: sentence_transformers model
)

Parameter	Type	Default	Description
`db_path`	`str`	`"memory.db"`	Path to the database file (`:memory:` for in-memory)
`embedding_dim`	`int`	`384`	Embedding vector dimensions
`embedder`	`object`	`None`	A `SentenceTransformer` model instance for auto-embedding

Core Memory

`record(text, **kwargs) → str`

Store a memory. Returns the record ID.

rid = db.record(
    "User likes Python",
    importance=0.7,          # 0.0-1.0
    valence=0.3,             # -1.0 to 1.0 (emotional)
    memory_type="semantic",  # episodic, semantic, procedural
    domain="work",
    source="user",
    certainty=0.8,           # 0.0-1.0
    namespace="default",
    idempotency_key=None,    # v0.10: exactly-once writes (see below)
)

Idempotency keys (v0.10). Pass idempotency_key to make the write exactly-once: retrying with the same key and an identical payload returns the original rid with zero writes — no duplicate row, no importance-stats advance, no certainty change. Repetition is not corroboration. The same key with a different payload raises a typed yantrikdb.IdempotencyConflict carrying the existing rid. Derive keys from a stable external identity (a message id, "conv:{id}:turn:{n}") — never from the text itself, or every edit becomes a conflict.

Routing: with an explicit embedding, the key binds to the exact bytes you sent. With no embedding on an engine-embedder host, the write routes through a drift-safe digest that excludes the generated vector — an embedder upgrade between retries can never fake a conflict. A key with a Python-side fallback embedder raises ValueError (that deployment can’t do keys).

`record_batch(inputs) → list[str]`

Store multiple memories at once. Each input is a dict with the same fields as record(). Batches are all-or-nothing on failure (v0.10): a rejected batch writes nothing at all.

rids = db.record_batch([
    {"text": "User likes Python", "importance": 0.7},
    {"text": "User works at Acme Corp", "importance": 0.9},
])

Items may carry an idempotency_key (v0.10) — a duplicate item returns the original rid at its position; an in-batch repeat of the same key+payload writes once. Keyed batch items require an explicit embedding (pass text-only keyed writes through record() instead, which routes drift-safe).

`recall(query, top_k=10, **kwargs) → list[dict]`

Retrieve memories using relevance-conditioned scoring.

results = db.recall(
    "What programming languages?",
    top_k=5,
    memory_type="semantic",       # optional filter
    namespace="default",          # optional filter
    include_superseded=False,     # v0.10 default: current truth only
)
for r in results:
    print(f"[{r['score']:.3f}] {r['text']}")

Returns a list of dicts with text, score, rid, importance, valence, certainty, etc.

Behavioral change in v0.10: recall excludes superseded records by default. If record B supersedes record A (via a supersedes link), default recall returns B and hides A; pass include_superseded=True for history access. Current-by-default is what an agent should act on — the outcome that matters is correct, current context after restart and correction, not raw recall volume. Disputed records are different: an open dispute keeps both sides visible, each carrying the other in disputed_with — the engine never silently picks a winner.

`recall_with_response(query, top_k=10) → dict`

Like recall() but includes confidence calibration with certainty_reasons explaining why confidence is high or low.

resp = db.recall_with_response("What is the user's timezone?", top_k=3)
# resp['results'], resp['certainty'], resp['certainty_reasons']

`recall_refine(original_query, refinement, top_k=5) → list[dict]`

Refine a low-confidence recall with a follow-up query.

`get(rid) → dict | None`

Retrieve a specific memory by record ID.

`forget(rid) → bool`

Tombstone a memory by record ID.

`correct(rid, reason, new_text=None, metadata_merge=None, new_importance=None, new_valence=None) → dict`

Fix an incorrect memory in place (v0.9.4+): the rid is stable, the original is not tombstoned, and the full revision history is preserved. A text change is vector-coherent — the memory is re-embedded so it is retrieved under its new meaning, never the old one.

result = db.correct(rid, "user changed employers", new_text="User works at Google, not Meta")
assert result["corrected_rid"] == rid          # same record, revised
assert result["original_tombstoned"] is False  # in-place, history kept

Scalar/metadata-only corrections (importance, valence, metadata_merge) skip re-embedding entirely. Note: a metadata_merge cannot launder provenance — flipping kind to fact on an inference-sourced record is refused by the write gate (v0.10) regardless of whether the text changes.

`list_memories(namespace=None, limit=100, offset=0) → list[dict]`

List all active memories with pagination.

`decay(threshold=0.01) → list[dict]`

Run decay pass — returns memories that fell below the threshold and were archived.

`stats(namespace=None) → dict`

Get database statistics: active_memories, edges, entities, open_conflicts, etc.

Errors (v0.10)

Actionable engine errors arrive as typed exception classes, all subclassing RuntimeError (existing except RuntimeError: handlers keep working). Branch on type, never on message text:

import yantrikdb

try:
    rid = db.record(text, idempotency_key=key)
except yantrikdb.IdempotencyConflict as e:
    # Same key, different payload — the first write stands.
    # The message carries the existing rid: fetch it and decide.
    ...
except yantrikdb.Backpressure:
    ...  # engine saturated, nothing written — back off and retry
except yantrikdb.ProvenanceInconsistent:
    ...  # the anti-laundering gate refused the declaration — fix it

Exception	Meaning	Retry?
`Backpressure`	Engine saturated; nothing written	Yes, with backoff
`RecallContended`	Recall lost a bounded read-contention race	Yes — nothing wrong with the query
`CorrectionDeferredDuringReembed`	Re-embedding cutover in flight	Yes, after cutover
`BatchDeferredDuringReembed`	Whole batch deferred; nothing written	Yes, identical batch
`IdempotencyConflict`	Same key, different payload; message carries existing rid	No — change key or payload
`InvalidIdempotencyKey`	Key empty/whitespace/>512 bytes	No — fix the key
`ProvenanceInconsistent`	Write refused by the anti-laundering gate	No — fix the declaration

Knowledge Graph

`relate(src, dst, rel_type, weight=1.0)`

Create a relationship in the cognitive graph.

db.relate("Alice", "Acme Corp", "works_at", weight=0.9)
db.relate("Alice", "Rust", "prefers", weight=0.8)

`get_edges(entity) → list[dict]`

Get all relationships for an entity.

`search_entities(query, limit=20) → list[dict]`

Find entities by name pattern.

`entity_profile(entity, days=90.0, namespace=None) → dict`

Get a comprehensive profile for an entity: relationship count, memory mentions, domain spread, activity timeline.

`relationship_depth(entity, namespace=None) → dict`

Composite depth score (0.0–1.0) combining sessions together, memories mentioning, domains spanning, connection count, and relationship type diversity.

depth = db.relationship_depth("Alice")
# depth['depth_score'], depth['sessions_together'], depth['domains_spanning'], ...

Cognition

`think(**kwargs) → dict`

Run autonomous cognition: consolidation, conflict detection, pattern mining, substitution category scanning, gossip triggers.

result = db.think(
    importance_threshold=0.5,
    run_consolidation=True,
    run_conflict_scan=True,
    run_pattern_mining=True,
)
# result['triggers'], result['consolidation_count'],
# result['conflicts_found'], result['patterns_new']

`get_conflicts(status=None, conflict_type=None, entity=None, priority=None, limit=50) → list[dict]`

List detected contradictions with optional filters.

conflicts = db.get_conflicts(status="open")
for c in conflicts:
    print(f"{c['conflict_type']}: {c['detection_reason']}")

`resolve_conflict(conflict_id, resolution, note=None)`

Resolve a contradiction: keep_a, keep_b, merge, keep_both.

`dismiss_conflict(conflict_id, note=None)`

Dismiss a conflict without resolving it.

`scan_conflicts() → list[dict]`

Manually trigger conflict scanning (also runs inside think()).

`get_patterns(limit=20) → list[dict]`

List all discovered behavioral patterns.

`get_personality() → dict`

Get the AI’s derived personality traits based on memory patterns.

`derive_personality() → dict`

Recalculate personality traits from current memory state.

Substitution Categories (V14)

YantrikDB maintains vocabularies of interchangeable terms (e.g., PostgreSQL/MySQL are both “databases”). When two memories differ only by a substitution, it’s flagged as a conflict rather than redundancy.

`substitution_categories() → list[dict]`

List all substitution categories with member counts.

cats = db.substitution_categories()
# [{"name": "databases", "conflict_mode": "exclusive", "member_count": 15}, ...]

`substitution_members(category_name) → list[dict]`

List members of a specific category.

members = db.substitution_members("databases")
# [{"token_normalized": "postgresql", "confidence": 0.95, "source": "seed"}, ...]

`learn_category_members(category_name, members, source="llm_suggested") → int`

Add new members to a category. Returns count of new members added.

count = db.learn_category_members(
    "databases",
    [("tidb", 0.35), ("surrealdb", 0.35)],
    "llm_suggested"
)

Source	Confidence	Drives conflicts?
`seed`	0.95	Yes (≥ 0.6 threshold)
`user_confirmed`	1.0	Yes
`llm_suggested`	0.35	No (below threshold)

`reclassify_conflict(conflict_id, new_type) → dict`

Reclassify a conflict and teach the system. Extracts differing tokens and learns new category members from the correction.

result = db.reclassify_conflict(conflict_id, "preference")
# result['learned_members'] — tokens added to categories from this feedback

`reset_category_to_seed(category_name) → int`

Remove all learned members from a category, keeping only seed vocabulary. Returns count of members removed.

removed = db.reset_category_to_seed("editors_tools")

Sessions

`session_start(namespace="default", client_id="default", metadata=None) → str`

Start a new conversation session. Returns session ID.

`session_end(session_id, summary=None) → dict`

End a session with optional summary.

`active_session(namespace="default", client_id="default") → dict | None`

Get the currently active session.

`session_history(namespace="default", client_id="default", limit=10) → list[dict]`

Get recent session history.

`session_abandon_stale(max_age_hours=24.0) → int`

Clean up sessions that were never properly ended.

Temporal

`stale(days=30.0, limit=50, namespace=None) → list[dict]`

Find memories that haven’t been accessed in a while and may need verification.

`upcoming(days=7.0, limit=50, namespace=None) → list[dict]`

Find memories with upcoming temporal relevance (deadlines, events).

Procedural Memory

Self-improving memory for strategies and behaviors. Tracks what works and adapts over time.

`record_procedural(text, domain="general", task_context="", effectiveness=0.5, namespace="default") → str`

Record a procedural memory (a strategy or approach that worked/didn’t work).

rid = db.record_procedural(
    "When user asks about code, show examples before explanation",
    domain="communication",
    effectiveness=0.8,
)

`surface_procedural(query_embedding, query_text=None, domain=None, top_k=5, namespace=None) → list[dict]`

Retrieve relevant procedural memories for a task context.

`reinforce_procedural(rid, outcome) → bool`

Update a procedural memory’s effectiveness score based on outcome (EMA-based adaptation).

db.reinforce_procedural(rid, outcome=0.9)  # worked well
db.reinforce_procedural(rid, outcome=0.2)  # didn't work this time

`procedural_stats(namespace=None) → list[dict]`

Get procedural memory statistics by domain: count and average effectiveness.

Triggers & Lifecycle

`get_pending_triggers(limit=10) → list[dict]`

Get undelivered proactive triggers.

`deliver_trigger(trigger_id) → bool`

Mark a trigger as delivered to the agent.

`acknowledge_trigger(trigger_id) → bool`

Mark a trigger as acknowledged by the agent.

`act_on_trigger(trigger_id) → bool`

Mark a trigger as acted upon.

`dismiss_trigger(trigger_id) → bool`

Dismiss a trigger without acting on it.

`get_trigger_history(limit=20, trigger_type=None) → list[dict]`

Get trigger history with optional type filter.

Recall Feedback

`recall_feedback(query, chosen_rid, rejected_rids=None) → bool`

Teach the scoring engine which results were useful. Improves retrieval quality over time.

db.recall_feedback(
    query="What's the user's timezone?",
    chosen_rid="019d...",
    rejected_rids=["019e...", "019f..."],
)

`learned_weights() → dict`

Inspect the current learned scoring weights.

Embedding

`set_embedder(fn_or_model)`

Set the embedding function or model.

from sentence_transformers import SentenceTransformer
model = SentenceTransformer("all-MiniLM-L6-v2")

# Pass model directly (recommended)
db = YantrikDB("memory.db", embedding_dim=384, embedder=model)

# Or set after construction
db.set_embedder(lambda text: model.encode(text).tolist())

`embed(text) → list[float]`

Generate an embedding for text using the configured embedder.

Sync & Replication

`extract_ops_since(hlc=None, limit=1000) → list[dict]`

Extract oplog entries since a given HLC for CRDT sync.

`apply_ops(ops) → dict`

Apply remote operations (merge with conflict resolution).

`archive(rid) → bool`

Move a memory to cold storage.

`hydrate(rid) → bool`

Restore a memory from cold storage.