The primary interface for CausalOS is the CausalMemory class.
from causal_memory import CausalMemorymemory = CausalMemory(
db_path="causal.db",
agent_id=None,
pool_id=None,
shared=False,
fail_safe=True
)db_path: Path to the SQLite database file.agent_id: Unique identifier for this agent. Defaults tohostname:script_name.pool_id: Optional identifier for a shared causal pool.shared: IfTrue, the instance will recall records from all agents in the samepool_id.fail_safe: IfTrue(default), errors in recording/recalling are logged but won't crash your agent.
Save a new causal event.
record(
action_type: ActionType,
action_detail: str,
outcome: str,
intent: Optional[str] = None,
severity: Severity = Severity.NONE,
context: Optional[Dict[str, Any]] = None,
tags: Optional[List[str]] = None,
metadata: Optional[Dict[str, Any]] = None
) -> CausalRecordSearch for similar past events using vector similarity.
recall(
action_detail: str,
top_k: int = 5,
threshold: float = 0.6
) -> List[CausalRecord]action_detail: The query string to find similar actions.top_k: Maximum number of records to return.threshold: Cosine similarity threshold (0.0 to 1.0).
Add an observed effect to an existing record.
append_downstream(
record_id: str,
description: str,
severity: Severity
)Fetch a sequence of records, optionally filtered by session.
graph(
session_id: Optional[str] = None,
limit: int = 100
) -> List[CausalRecord]The data model representing a single memory.
id: Unique identifier (UUID).timestamp: Time of creation.agent_id,pool_id,session_id.action_type,action_detail,intent,outcome,severity.downstream: List ofDownstreamEffectobjects.
Values: NONE, LOW, MEDIUM, HIGH, CRITICAL.
Values: DB_WRITE, DB_DELETE, FILE_OP, API_CALL, SHELL, NETWORK, OTHER.