orbiter.memory.base - Orbiter Docs

Memory store protocol, typed memory item hierarchy, and status lifecycle.

Module Path

python

from orbiter.memory.base import (
    MemoryStore,
    MemoryItem,
    MemoryMetadata,
    MemoryStatus,
    MemoryError,
    SystemMemory,
    HumanMemory,
    AIMemory,
    ToolMemory,
)

MemoryError

Base exception for memory operations.

python

class MemoryError(Exception): ...

MemoryStatus

Memory item status lifecycle.

python

class MemoryStatus(StrEnum):
    DRAFT = "draft"
    ACCEPTED = "accepted"
    DISCARD = "discard"

Valid Transitions

From	Allowed To
`DRAFT`	`ACCEPTED`, `DISCARD`
`ACCEPTED`	`DISCARD`
`DISCARD`	(terminal — no transitions)

MemoryMetadata

Scoping metadata for memory items. All fields are optional.

Inherits: pydantic.BaseModel (frozen)

Constructor

python

MemoryMetadata(
    user_id: str | None = None,
    session_id: str | None = None,
    task_id: str | None = None,
    agent_id: str | None = None,
    extra: dict[str, Any] = {},
)

Field	Type	Default	Description
`user_id`	`str \| None`	`None`	User scope
`session_id`	`str \| None`	`None`	Session scope
`task_id`	`str \| None`	`None`	Task scope
`agent_id`	`str \| None`	`None`	Agent scope
`extra`	`dict[str, Any]`	`{}`	Extensible metadata

MemoryItem

Base class for all memory entries.

Inherits: pydantic.BaseModel

Constructor

python

MemoryItem(
    content: str,
    memory_type: str,
    id: str = <auto-generated UUID>,
    status: MemoryStatus = MemoryStatus.ACCEPTED,
    metadata: MemoryMetadata = MemoryMetadata(),
    created_at: str = <ISO-8601 now>,
    updated_at: str = <ISO-8601 now>,
)

Field	Type	Default	Description
`id`	`str`	Auto-generated UUID hex	Unique identifier
`content`	`str`	(required)	The stored content
`memory_type`	`str`	(required)	Discriminator for subtype dispatch
`status`	`MemoryStatus`	`ACCEPTED`	Current lifecycle status
`metadata`	`MemoryMetadata`	`MemoryMetadata()`	Scoping information
`created_at`	`str`	ISO-8601 now	Creation timestamp
`updated_at`	`str`	ISO-8601 now	Last-update timestamp

Methods

transition()

python

def transition(self, new_status: MemoryStatus) -> None

Transition to a new status. Updates updated_at.

Raises: MemoryError if the transition is invalid.

Memory Subtypes

SystemMemory

System/initialization memory (e.g. system prompts).

python

class SystemMemory(MemoryItem):
    memory_type: str = "system"

HumanMemory

User/human message memory.

python

class HumanMemory(MemoryItem):
    memory_type: str = "human"

AIMemory

AI/assistant response memory.

python

class AIMemory(MemoryItem):
    memory_type: str = "ai"
    tool_calls: list[dict[str, Any]] = []

Extra Field	Type	Default	Description
`tool_calls`	`list[dict[str, Any]]`	`[]`	Tool invocations in this response

ToolMemory

Tool execution result memory.

python

class ToolMemory(MemoryItem):
    memory_type: str = "tool"
    tool_call_id: str = ""
    tool_name: str = ""
    is_error: bool = False

Extra Field	Type	Default	Description
`tool_call_id`	`str`	`""`	ID of the originating tool call
`tool_name`	`str`	`""`	Name of the tool
`is_error`	`bool`	`False`	Whether this is an error result

MemoryStore (Protocol)

Protocol for pluggable memory backends. All methods are async.

python

@runtime_checkable
class MemoryStore(Protocol):
    async def add(self, item: MemoryItem) -> None: ...
    async def get(self, item_id: str) -> MemoryItem | None: ...
    async def search(
        self,
        *,
        query: str = "",
        metadata: MemoryMetadata | None = None,
        memory_type: str | None = None,
        status: MemoryStatus | None = None,
        limit: int = 10,
    ) -> list[MemoryItem]: ...
    async def clear(
        self,
        *,
        metadata: MemoryMetadata | None = None,
    ) -> int: ...

Methods

add()

python

async def add(self, item: MemoryItem) -> None

Persist a memory item.

get()

python

async def get(self, item_id: str) -> MemoryItem | None

Retrieve a memory item by ID. Returns None if not found.

search()

python

async def search(
    self,
    *,
    query: str = "",
    metadata: MemoryMetadata | None = None,
    memory_type: str | None = None,
    status: MemoryStatus | None = None,
    limit: int = 10,
) -> list[MemoryItem]

Search memory items with optional filters.

Parameter	Type	Default	Description
`query`	`str`	`""`	Keyword search query
`metadata`	`MemoryMetadata \| None`	`None`	Scope filter
`memory_type`	`str \| None`	`None`	Filter by type (e.g. `"human"`, `"ai"`)
`status`	`MemoryStatus \| None`	`None`	Filter by status
`limit`	`int`	`10`	Maximum results

clear()

python

async def clear(
    self,
    *,
    metadata: MemoryMetadata | None = None,
) -> int

Remove memory items matching the filter. Returns count removed. If metadata is None, clears everything.

Example

python

import asyncio
from orbiter.memory.base import (
    HumanMemory, AIMemory, MemoryMetadata, MemoryStatus,
)

async def main():
    meta = MemoryMetadata(user_id="user-1", session_id="sess-1")

    # Create items
    msg = HumanMemory(content="What is Python?", metadata=meta)
    reply = AIMemory(content="Python is a programming language.", metadata=meta)

    # Status lifecycle
    draft = HumanMemory(content="Draft message", status=MemoryStatus.DRAFT)
    draft.transition(MemoryStatus.ACCEPTED)
    assert draft.status == MemoryStatus.ACCEPTED

asyncio.run(main())