Memory API Reference

from marsys.agents.memory import ConversationMemory

ConversationMemory(description: Optional[str] = None)

memory = ConversationMemory(description="You are helpful")

# Add messages
msg_id = memory.add(role="user", content="Hello")
memory.add(role="assistant", content="Hi there!")

# Retrieve messages
all_msgs = memory.retrieve_all()  # List[Dict]
recent = memory.retrieve_recent(5)

# Clear
memory.reset_memory()  # Keeps system prompt

from marsys.agents.memory import ManagedConversationMemory, ManagedMemoryConfig

ManagedConversationMemory(
    config: Optional[ManagedMemoryConfig] = None,
    trigger_strategy: Optional[TriggerStrategy] = None,
    process_strategy: Optional[ProcessStrategy] = None,
    retrieval_strategy: Optional[RetrievalStrategy] = None,
    description: Optional[str] = None
)

@dataclass
class ManagedMemoryConfig:
    threshold_tokens: int = 150_000
    image_token_estimate: int = 800
    trigger_events: List[str] = ["add", "get_messages"]
    cache_invalidation_events: List[str] = ["add", "remove_by_id", "delete_memory"]
    token_counter: Optional[Callable] = None
    active_context: ActiveContextPolicyConfig = ...

    @property
    def compaction_target_tokens(self) -> int:
        """Derived: threshold_tokens * (1 - min_reduction_ratio)."""

@dataclass
class ActiveContextPolicyConfig:
    enabled: bool = True
    mode: str = "compaction"  # "compaction" | "sliding_window"
    processor_order: List[str] = ["tool_truncation", "summarization", "backward_packing"]
    excluded_processors: List[str] = []
    min_reduction_ratio: float = 0.4
    tool_truncation: ToolTruncationConfig = ...
    summarization: SummarizationConfig = ...
    backward_packing: BackwardPackingConfig = ...

@dataclass
class SummarizationConfig:
    enabled: bool = True
    grace_recent_messages: int = 1
    output_max_tokens: int = 6000
    include_original_instruction: bool = True
    include_image_payload_bytes: bool = False
    max_retained_images: Optional[int] = None  # Payload recovery defaults to 5 when unset

from marsys.agents.memory import KGMemory

KGMemory(
    model: Union[BaseLocalModel, BaseAPIModel, LocalProviderAdapter],
    description: Optional[str] = None
)

from marsys.agents.memory import MemoryManager

MemoryManager(
    memory_type: str = "conversation_history",
    description: Optional[str] = None,
    model: Optional[Union[BaseLocalModel, LocalProviderAdapter]] = None,
    memory_config: Optional[ManagedMemoryConfig] = None,
    token_counter: Optional[Callable] = None
)

# Standard memory
manager = MemoryManager(
    memory_type="conversation_history",
    description="System prompt"
)

# With token management
manager = MemoryManager(
    memory_type="managed_conversation",
    memory_config=ManagedMemoryConfig(
        threshold_tokens=100000
    )
)

# Knowledge graph
manager = MemoryManager(
    memory_type="kg",
    model=your_model
)

# Use it
manager.add(role="user", content="Hello")
msgs = manager.get_messages()
manager.save_to_file("memory.json")

from marsys.agents.memory import MemoryResetEvent
from marsys.coordination.event_bus import EventBus

bus = EventBus()
manager.set_event_context(agent_name="Researcher", event_bus=bus, session_id="run_123")

from marsys.coordination.status.events import CompactionEvent

CompactionEvent(
    session_id: str,
    agent_name: str,
    status: str,              # "started" | "completed" | "failed"
    pre_tokens: int = 0,
    post_tokens: int = 0,
    pre_messages: int = 0,
    post_messages: int = 0,
    duration: Optional[float] = None,
    stages_run: Optional[List[str]] = None,
)

from marsys.agents.memory import Message

Message(
    role: str,
    content: Optional[Union[str, Dict[str, Any], List[Dict[str, Any]]]] = None,
    message_id: str = auto_generated,
    name: Optional[str] = None,
    tool_calls: Optional[List[ToolCallMsg]] = None,
    agent_calls: Optional[List[AgentCallMsg]] = None,
    structured_data: Optional[Dict[str, Any]] = None,
    images: Optional[List[str]] = None,
    tool_call_id: Optional[str] = None,
    reasoning_details: Optional[List[Dict[str, Any]]] = None
)

# Simple message
msg = Message(role="user", content="Hello")

# With tool calls
msg = Message(
    role="assistant",
    content=None,
    tool_calls=[
        ToolCallMsg(
            id="call_123",
            call_id="call_123",
            type="function",
            name="search",
            arguments='{"query": "AI"}'
        )
    ]
)

# Tool response
msg = Message(
    role="tool",
    content='{"result": "found"}',
    tool_call_id="call_123",
    name="search"
)

# With images
msg = Message(
    role="user",
    content="Describe this",
    images=["path/to/image.jpg"]
)

from marsys.agents.memory import MessageContent

MessageContent(
    thought: Optional[str] = None,
    next_action: Optional[str] = None,
    action_input: Optional[Dict[str, Any]] = None
)

from marsys.agents.memory import ToolCallMsg

ToolCallMsg(
    id: str,
    call_id: str,
    type: str,
    name: str,
    arguments: str
)

from marsys.agents.memory import AgentCallMsg

agent_call = AgentCallMsg(
    agent_name="DataProcessor",
    request="Process the sales data"
)