Python API Reference

Brain(path: str | Path)

from agentic_memory import Brain

brain = Brain("my_agent.amem")
brain = Brain(Path("/data/agents/assistant.amem"))

Brain.add_fact(
    content: str,
    confidence: float = 1.0,
    metadata: dict[str, str] | None = None
) -> Event

Brain.add_decision(
    content: str,
    confidence: float = 1.0,
    metadata: dict[str, str] | None = None
) -> Event

Brain.add_inference(
    content: str,
    confidence: float = 1.0,
    metadata: dict[str, str] | None = None
) -> Event

Brain.add_correction(
    content: str,
    confidence: float = 1.0,
    metadata: dict[str, str] | None = None
) -> Event

Brain.add_skill(
    content: str,
    confidence: float = 1.0,
    metadata: dict[str, str] | None = None
) -> Event

Brain.add_episode(
    content: str,
    confidence: float = 1.0,
    metadata: dict[str, str] | None = None
) -> Event

Brain.link(
    source: int,
    target: int,
    edge_type: str | EdgeType,
    weight: float = 1.0
) -> Edge

Brain.facts(session: int | None = None) -> list[Event]

Brain.decisions(session: int | None = None) -> list[Event]

Brain.traverse(
    start: int,
    depth: int = 3,
    edge_types: list[str | EdgeType] | None = None
) -> TraversalResult

Brain.resolve(event_id: int) -> Event

Brain.impact(event_id: int, depth: int = 5) -> ImpactResult

Brain.info() -> BrainInfo

Brain.session_info(session: int) -> SessionInfo

Brain.search(
    query: str,
    top_k: int = 10,
    event_type: str | EventType | None = None,
    session: int | None = None,
    min_confidence: float = 0.0
) -> list[SearchResult]

Brain.search_text(
    query: str,
    top_k: int = 10,
    event_type: str | None = None
) -> list[TextMatch]

Brain.search(
    query: str,
    top_k: int = 10
) -> list[HybridMatch]

Brain.centrality(
    metric: str = "pagerank",
    event_type: str | None = None
) -> list[CentralityResult]

Brain.shortest_path(
    src: int,
    dst: int,
    weighted: bool = False
) -> PathResult

Brain.revise(node_id: int) -> RevisionReport

Brain.gaps(
    min_severity: float = 0.0,
    max_age_days: int = 30
) -> GapReport

Brain.analogy(
    node_id: int,
    top_k: int = 5
) -> list[Analogy]

Brain.consolidate(
    dry_run: bool = True
) -> ConsolidationReport

Brain.drift(
    topic: str | None = None
) -> DriftReport

@dataclass
class TextMatch:
    node_id: int
    content: str
    bm25_score: float
    event_type: str

@dataclass
class HybridMatch:
    node_id: int
    content: str
    combined_score: float
    bm25_score: float
    vector_score: float

@dataclass
class CentralityResult:
    node_id: int
    score: float
    content: str

@dataclass
class PathResult:
    path: list[int]
    total_cost: float
    found: bool

@dataclass
class RevisionReport:
    target_id: int
    affected_nodes: list[int]
    confidence_losses: dict[int, float]
    unsupported: list[int]
    total_impact: float

@dataclass
class GapReport:
    gaps: list[dict]
    health_score: float
    total_gaps: int

@dataclass
class Analogy:
    node_id: int
    structural_score: float
    content_score: float
    combined_score: float

@dataclass
class ConsolidationReport:
    duplicates: list[tuple[int, int, float]]
    contradictions: list[tuple[int, int]]
    promotions: list[int]
    orphans: list[int]
    dry_run: bool

@dataclass
class DriftReport:
    topics: list[dict]
    overall_stability: float

MemoryAgent(
    brain: Brain,
    provider: LLMProvider,
    auto_link: bool = True,
    extraction_prompt: str | None = None
)

MemoryAgent.chat(
    message: str,
    context: list[Event] | None = None
) -> str

MemoryAgent.last_extraction: list[Event]

@dataclass
class Event:
    id: int                          # Unique node ID within the brain
    event_type: EventType            # Fact, Decision, Inference, Correction, Skill, Episode
    content: str                     # The textual content
    session: int                     # Session ID this event belongs to
    confidence: float                # Confidence score (0.0 to 1.0)
    timestamp: datetime              # UTC timestamp of creation
    metadata: dict[str, str]         # Optional key-value metadata

@dataclass
class Edge:
    source: int                      # Source node ID
    target: int                      # Target node ID
    edge_type: EdgeType              # Relationship type
    weight: float                    # Edge weight (0.0 to 1.0)

@dataclass
class BrainInfo:
    node_count: int                  # Total number of events
    edge_count: int                  # Total number of edges
    session_count: int               # Number of sessions
    file_size: int                   # File size in bytes
    sessions: list[int]              # List of session IDs
    version: int                     # File format version

@dataclass
class SessionInfo:
    id: int                          # Session ID
    node_count: int                  # Number of events in this session
    edge_count: int                  # Number of edges between session events
    start_time: datetime             # Timestamp of the first event
    end_time: datetime               # Timestamp of the last event
    event_types: dict[str, int]      # Count of each event type

@dataclass
class TraversalResult:
    nodes: list[Event]               # All nodes reached during traversal
    edges: list[Edge]                # All edges traversed
    depth_reached: int               # Maximum depth actually reached

@dataclass
class ImpactResult:
    affected: list[Event]            # Events downstream of the analyzed event
    edges: list[Edge]                # Edges in the impact graph
    total_affected: int              # Total count of affected events

@dataclass
class SearchResult:
    event: Event                     # The matching event
    score: float                     # Similarity score (0.0 to 1.0)

class EventType(str, Enum):
    FACT = "fact"
    DECISION = "decision"
    INFERENCE = "inference"
    CORRECTION = "correction"
    SKILL = "skill"
    EPISODE = "episode"

class EdgeType(str, Enum):
    CAUSED_BY = "caused_by"
    SUPPORTS = "supports"
    CONTRADICTS = "contradicts"
    SUPERSEDES = "supersedes"
    RELATED_TO = "related_to"
    PART_OF = "part_of"
    TEMPORAL_NEXT = "temporal_next"

class BrainError(Exception):
    """Raised for brain file operations: corruption, invalid IDs, I/O failures."""
    pass

class CLIError(Exception):
    """Raised when the underlying Rust CLI returns an error."""
    pass

class ProviderError(Exception):
    """Raised for LLM provider failures: API errors, rate limits, invalid responses."""
    pass

from abc import ABC, abstractmethod
from agentic_memory import Event

class LLMProvider(ABC):

    @abstractmethod
    def complete(self, prompt: str, system: str | None = None) -> str:
        """Send a prompt to the LLM and return the completion text.

        Args:
            prompt: The user/input prompt.
            system: Optional system prompt.

        Returns:
            The LLM's response text.

        Raises:
            ProviderError: If the API call fails.
        """
        ...

    @abstractmethod
    def extract_events(self, text: str) -> list[dict]:
        """Extract cognitive events from text.

        The LLM should identify facts, decisions, inferences, etc.
        in the input text and return them as structured dictionaries.

        Args:
            text: The text to extract events from.

        Returns:
            List of dicts with keys: "type", "content", "confidence".

        Raises:
            ProviderError: If extraction fails.
        """
        ...

    def embed(self, text: str) -> list[float] | None:
        """Generate an embedding vector for the given text.

        Optional. If not implemented, the default internal embedding
        model is used. Return a list of 128 floats.

        Args:
            text: The text to embed.

        Returns:
            A 128-dimensional float vector, or None to use the default.
        """
        return None

from agentic_memory.integrations import LLMProvider, ProviderError

class MyCustomProvider(LLMProvider):

    def __init__(self, api_url: str, api_key: str):
        self.api_url = api_url
        self.api_key = api_key

    def complete(self, prompt: str, system: str | None = None) -> str:
        # Call your LLM API here
        response = requests.post(
            f"{self.api_url}/completions",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={"prompt": prompt, "system": system}
        )
        if response.status_code != 200:
            raise ProviderError(f"API error: {response.status_code}")
        return response.json()["text"]

    def extract_events(self, text: str) -> list[dict]:
        extraction_prompt = f"Extract cognitive events from: {text}"
        raw = self.complete(extraction_prompt, system="Extract facts, decisions, inferences...")
        # Parse the LLM output into structured events
        return parse_extraction(raw)

# Usage
provider = MyCustomProvider("https://my-llm.example.com", "my-api-key")
agent = MemoryAgent(brain, provider)

from agentic_memory.integrations import AnthropicProvider

provider = AnthropicProvider(
    api_key: str = None,             # Defaults to ANTHROPIC_API_KEY env var
    model: str = "claude-sonnet-4-20250514",
)

from agentic_memory.integrations import OpenAIProvider

provider = OpenAIProvider(
    api_key: str = None,             # Defaults to OPENAI_API_KEY env var
    model: str = "gpt-4o",
)

from agentic_memory.integrations import OllamaProvider

provider = OllamaProvider(
    model: str = "llama3.1",
    host: str = "http://localhost:11434",
)