This document provides a comprehensive, OSS-grade reference for the engram.Engram asynchronous client.
Note
All methods (except context managers) are asynchronous and must be awaited. The code snippets assume you have instantiated an engram client.
Engram(
settings: EngramSettings | None = None,
*,
database_url: str | None = None,
openai_api_key: str | None = None,
memory_policy: str | MemoryPolicy | None = None,
)Initializes the Engram client. Providers are lazy-loaded upon connect(). Use memory_policy to configure how facts are typed and slotted (valid strings: "default", "legal", "coding_agent").
await engram.connect() -> NoneEstablishes the database connection pool and initializes configured providers.
await engram.close() -> NoneSafely tears down the database connection pool and releases provider resources.
await engram.add(
content: str,
agent_id: str,
*,
main_content: str | None = None,
user_id: str | None = None,
session_id: str | None = None,
memory_type: str = "semantic",
metadata: dict | None = None,
) -> MemoryStores a single atomic memory fact. The content is the exact fact to embed and index (mapped to .fact internally). main_content is used for storing the raw context the fact was extracted from, but it is not embedded for search.
await engram.add_batch(memories: list[dict[str, Any]]) -> list[Memory]Efficiently adds multiple memories in a single transaction with batch-vectorization. The memories list should contain dictionaries with content, agent_id, and other optional arguments corresponding to add().
await engram.add_conversation(
user_message: str,
assistant_response: str,
agent_id: str,
*,
user_id: str | None = None,
session_id: str | None = None,
conversation_history: list[dict[str, str]] | None = None,
conversation_summary: str | None = None,
metadata: dict | None = None,
search_limit: int = 10,
update_summary: bool = True,
extract_assistant_response: bool = False,
) -> ConversationResultProcesses a conversation turn to intelligently extract, compare, and store new memories using the configured LLM provider. Set extract_assistant_response=True only if you want the assistant's own statements converted into stored memories.
ConversationResult is list-compatible — iterating, len()-ing, indexing, or truth-testing it yields the written memories, exactly like the previous list[Memory] return, so existing callers need no change. Its .decisions field additionally exposes a FactDecision(fact, operation, applied, reason, memory_id, target_id) for every extracted fact, including those that resolved to NOOP or could not be applied. This makes a skipped update visible and debuggable (e.g. operation="NOOP", applied=False, reason="Duplicate of …") instead of silently absent from the returned list. If you write on this path, inspect .decisions to detect a fact that was extracted but not stored.
Use as the sole writer to a memory space.
add_conversation()searches the agent's existing memories to decide ADD/UPDATE/DELETE. If you also store raw turns in the same space withadd_batch(), the freshly-extracted fact is found already present verbatim in its own raw row, so the decision step judges it identical and NOOPs it — supersession never fires. Either letadd_conversation()own a memory space, or keep the two writers in separate namespaces. (Theexamples/chatbot.pyexample deliberately usesadd_batch()only for this reason.)
await engram.get(memory_id: str) -> MemoryFetches a single memory by ID and updates its internal access count.
await engram.update(
memory_id: str,
*,
content: str | None = None,
importance: float | None = None,
metadata: dict | None = None
) -> MemoryEdits a memory directly in-place. Use with caution, as it bypasses version lineage tracking.
await engram.revise(
memory_id: str,
*,
content: str | None = None,
importance: float | None = None,
metadata: dict | None = None,
reason: str | None = None
) -> MemoryUpdates a memory by creating a new active head in the lineage and marking the old memory as superseded. Essential for user corrections.
await engram.get_current(memory_id: str) -> MemoryFetches the active head of a memory's lineage. If passed an old (superseded) ID, it resolves forward to the current version.
await engram.get_lineage(memory_id: str) -> MemoryLineageRetrieves the full version history (all updates and supersessions) of a specific memory.
await engram.explain_memory(memory_id: str) -> MemoryExplanationReturns an LLM-generated explanation of a memory's intent and context based on its graph relations.
await engram.reinforce(memory_id: str, importance_boost: float = 0.1) -> MemoryIncreases the importance score of a memory, typically used when a recalled memory was proven highly useful.
await engram.forget(memory_id: str) -> boolSoft-deletes a memory from the active search index. Returns True if successfully deleted.
await engram.purge(agent_id: str, user_id: str | None = None) -> intHard-deletes all memories belonging to an agent (and optionally a specific user). Returns the number of deleted records.
await engram.list_recent(agent_id: str, user_id: str | None = None, limit: int = 10) -> list[Memory]Returns the most recently created or updated active memories.
await engram.get_history(
agent_id: str,
*,
user_id: str | None = None,
limit: int = 50,
include_superseded: bool = True,
memory_types: list[str] | None = None,
since: datetime | None = None,
until: datetime | None = None
) -> list[MemoryHistoryEvent]Retrieves a chronological event timeline of added, revised, and superseded facts. Useful for rendering user-facing memory audit logs.
await engram.get_memories(
agent_id: str,
*,
user_id: str | None = None,
session_id: str | None = None,
metadata_filter: dict | None = None,
memory_types: list[str] | None = None,
limit: int = 200
) -> list[Memory]A plain filtered read operation without vector ranking. Ideal for bulk exporting or aggregating specific memory segments.
await engram.search(
query: str,
agent_id: str,
*,
user_id: str | None = None,
limit: int = 10,
min_score: float = 0.0,
metadata_filter: dict | None = None,
memory_types: list[str] | None = None,
mode: str = "hybrid",
rerank: bool = False,
include_superseded: bool = False,
) -> list[SearchResult]Core hybrid search combining semantic similarity, full-text keywords, and importance/decay. Setting rerank=True activates local cross-encoder re-sorting (requires the sentence-transformers dependency).
await engram.deep_search(
query: str,
agent_id: str,
*,
user_id: str | None = None,
limit: int = 10,
min_score: float = 0.0,
metadata_filter: dict | None = None,
memory_types: list[str] | None = None,
mode: str = "hybrid",
n_queries: int = 4,
rerank: bool = False,
) -> list[SearchResult]Expands broad queries by asking the LLM to generate n_queries permutations, runs them concurrently, and fuses the results using Reciprocal Rank Fusion (RRF).
await engram.recall(
question: str,
agent_id: str,
*,
user_id: str | None = None,
question_date: datetime | None = None,
limit: int = 10,
compose_answer: bool = True,
) -> RecallAnswerThe high-level autonomous operator. Classifies user intent (current, historical, event, lineage, temporal_chain, verify, or chat), maps natural language timeframes (e.g., "yesterday"), and composes a fully grounded answer. temporal_chain runs one search per event anchor and merges evidence chronologically; verify pairs the affirmative topic search with a counter-evidence search for an explicit denial (superseded rows included). Set compose_answer=False to skip the final LLM composition.
await engram.recall_critical(
agent_id: str,
*,
user_id: str | None = None,
limit: int = 50,
memory_types: list[str] | None = None
) -> list[Memory]Reads active critical memories (like constraints, base logic, or profile info) directly via metadata, bypassing vector similarity rank.
await engram.trace_recall(
query: str,
agent_id: str,
*,
user_id: str | None = None,
limit: int = 20,
min_score: float = 0.0,
max_tokens: int = 2000,
expected_terms: list[str] | None = None,
use_deep_search: bool = True,
memory_types: list[str] | None = None,
token_counter: Callable | None = None
) -> RecallTraceA diagnostic utility. Returns a full observability trace of retrieval performance, including hit/miss metrics against expected_terms and truncation flags.
await engram.get_context_block(
query: str,
agent_id: str,
*,
user_id: str | None = None,
session_id: str | None = None,
limit: int = 10,
min_score: float = 0.0,
max_tokens: int | None = None,
header: str = "## Relevant memories",
token_counter: Callable | None = None,
memory_types: list[str] | None = None,
group_by_type: bool = False,
rerank: bool = False
) -> strFormats retrieved search results directly into a budgeted, prompt-ready markdown block. If session_id is provided and has a rolling summary, the summary is prepended.
await engram.relate(
source_id: str,
target_id: str,
relation_type: str = "related_to",
weight: float = 1.0,
metadata: dict | None = None
) -> NoneCreates a directional edge between two memories to form associative networks. Common relation types: related_to, supports, causes.
await engram.traverse(
start_memory_id: str,
max_depth: int = 3,
direction: str = "outbound",
relation_types: list[str] | None = None,
min_weight: float = 0.0,
limit: int = 50
) -> list[TraversalResult]Performs a multi-hop traversal starting from a single memory using recursive Postgres CTEs.
await engram.traverse_many(
start_memory_ids: list[str],
*,
max_depth: int = 2,
direction: str = "any",
relation_types: list[str] | None = None,
min_weight: float = 0.0,
limit_per_seed: int = 25,
total_limit: int = 100,
skip_missing: bool = True
) -> list[TraversalResult]Traverses the memory graph concurrently from multiple starting seeds and returns deduplicated, ranked results.
engram.render_graph_context(
results: list[TraversalResult],
*,
max_tokens: int | None = None,
token_counter: Callable | None = None,
include_paths: bool = False,
header: str = "## Related memory graph"
) -> strUtility to render traversal results into a markdown context block suitable for LLM injection.
async with engram.session(
agent_id: str,
user_id: str | None = None,
metadata: dict | None = None
) -> SessionContext manager that creates a conversational session boundary. Memory additions scoped to the active session can automatically maintain a rolling conversational summary.
await engram.start_task(
goal: str,
agent_id: str,
*,
user_id: str | None = None,
session_id: str | None = None,
metadata: dict | None = None
) -> TaskRunInitializes a durable event ledger for a long-running autonomous agent.
await engram.get_task(task_run_id: str, *, include_deleted: bool = False) -> TaskRun
await engram.list_tasks(*, agent_id: str | None = None, user_id: str | None = None, status: str | None = None, limit: int = 100, include_deleted: bool = False) -> list[TaskRun]Query existing tasks based on their status or ownership.
await engram.pause_task(task_run_id: str, *, outcome: str | None = None) -> TaskRun
await engram.complete_task(task_run_id: str, *, outcome: str | None = None) -> TaskRun
await engram.fail_task(task_run_id: str, *, outcome: str | None = None) -> TaskRun
await engram.cancel_task(task_run_id: str, *, outcome: str | None = None) -> TaskRun
await engram.soft_delete_task(task_run_id: str) -> TaskRunMethods to manage the lifecycle and terminal outcomes of a task.
await engram.record_event(
*,
agent_id: str,
role: str,
event_type: str,
content: str = "",
task_run_id: str | None = None,
session_id: str | None = None,
user_id: str | None = None,
payload: dict | None = None,
metadata: dict | None = None
) -> AgentEventLogs a single, immutable action, observation, or tool call to the event ledger.
await engram.record_turn(
task_run_id: str,
user_message: str,
assistant_response: str,
*,
agent_id: str | None = None,
user_id: str | None = None,
session_id: str | None = None,
tool_calls: list[dict] | None = None,
artifacts: list[dict] | None = None,
metadata: dict | None = None,
enqueue_processing: bool = True
) -> list[AgentEvent]Writes a complete conversation turn (user, assistant, and tools) in one transaction and optionally queues a background analysis job for derivation.
await engram.search_events(
query: str,
*,
agent_id: str | None = None,
task_run_id: str | None = None,
session_id: str | None = None,
user_id: str | None = None,
event_types: list[str] | None = None,
roles: list[str] | None = None,
since: datetime | None = None,
until: datetime | None = None,
limit: int = 50,
include_deleted: bool = False,
mode: str = "hybrid"
) -> list[AgentEvent]Perform hybrid search directly on the event ledger to recall prior actions or specific tool calls.
await engram.create_checkpoint(
task_run_id: str,
summary: str,
*,
completed_steps: list[str] | None = None,
pending_steps: list[str] | None = None,
decisions: list[str] | None = None,
blockers: list[str] | None = None,
artifacts: list[dict] | None = None,
source_event_ids: list[str] | None = None,
metadata: dict | None = None
) -> TaskCheckpointSaves a compact state snapshot summarizing the agent's progress, pending goals, and blockers.
await engram.build_context(
task_run_id: str,
*,
query: str = "",
max_tokens: int = 200000,
token_counter: Callable | None = None,
recent_event_limit: int = 40,
memory_limit: int = 25,
checkpoint_limit: int = 3,
include_graph: bool = True
) -> ContextBuildResultAssembles a unified prompt context encompassing the latest checkpoints, recent events, relevant semantic memories, and graph traversal logic.
await engram.record_long_input(
task_run_id: str,
text: str,
*,
title: str | None = None,
agent_id: str | None = None,
user_id: str | None = None,
session_id: str | None = None,
metadata: dict | None = None,
max_chunk_tokens: int = 700,
extract_with_llm: bool = True,
max_facts_per_chunk: int = 6
) -> LongInputIngestionReportIngests massive text contexts (e.g., source code, PR diffs, or contracts) and recursively chunks them, extracting actionable and citable facts.
await engram.build_long_input_context(
task_run_id: str,
*,
query: str,
max_tokens: int = 4000,
source_chunk_limit: int = 6,
expected_terms: list[str] | None = None,
token_counter: Callable | None = None
) -> LongInputContextResultRetrieves targeted, source-anchored chunks from a previously recorded long input for precise answering.
await engram.process_memory_jobs(*, limit: int = 10) -> list[MemoryJob]Manually processes a batch of deferred derivation jobs queued by operations like record_turn.
await engram.run_memory_worker(
*,
batch_size: int = 10,
interval_seconds: float = 1.0,
stop_event: asyncio.Event | None = None,
max_iterations: int | None = None
) -> intRuns a continuous loop to consume background memory jobs. Suitable for launching in an asyncio.create_task() daemon.
await engram.health_check() -> dict[str, Any]Runs diagnostic checks verifying the database connection, schema version, and accessibility of configured LLM/Embedding providers.